impuls.model¶

class impuls.model.Entity(*args, **kwargs)¶

Bases: Protocol

Entity is a protocol for marshalling data between model entities and SQL and GTFS. Every entity defined in the model implements this protocol.

static sql_columns() → LiteralString¶: sql_columns returns a “(col1, col2, col3)” string used in SQL queries for this type. The number of question marks must match the number of elements returned by sql_marshall().

static sql_create_table() → LiteralString¶: sql_create_table returns the SQL CREATE TABLE statement necessary to hold entities of this type.

sql_marshall() → tuple[None | int | float | str, ...]¶: sql_marshall converts an entity into its SQL representation.

static sql_placeholder() → LiteralString¶: sql_placeholder returns a “(?, ?, ?, ?, …)” string used in SQL queries for this type. The number of question marks must match the number of elements returned by sql_marshall().

sql_primary_key() → tuple[None | int | float | str, ...]¶: sql_primary_key converts the primary key of an entity into its SQL representation. The returned tuple should have the same number of elements as sql_where_clause() has parameters.

static sql_set_clause() → LiteralString¶: sql_set_clause returns a “COLUMN_NAME = ?, OTHER_COLUMN = ?, …” string used in UPDATE statements

static sql_table_name() → LiteralString¶: sql_table_name returns the SQL table name which holds entities of this type

classmethod sql_unmarshall(row: Sequence[None | int | float | str]) → Self¶: sql_unmarshall creates an entity from its SQL representation.

static sql_where_clause() → LiteralString¶: sql_where_clause returns a “COLUMN_NAME = ? AND …” string used in SQL queries to uniquely identify entities of this type.

final class impuls.model.Agency(id: str, name: str, url: str, timezone: str, lang: str = '', phone: str = '', fare_url: str = '', extra_fields_json: str | None = None)¶

Bases: Entity, ExtraFieldsMixin

Agency represents the entity/public body/company responsible for high-level management (especially fares) of a public transportation network.

The exact meaning is up to the user, but an Agency should be the body riders associate as responsible for the transit system. For example, in Poland, for publicly run city networks this should be the city-run public transport authority (the organizer, e.g. Zarząd Transportu Miejskiego), but for train networks this should be the train company itself (e.g. Koleje Mazowieckie, even though technically the organizer is usually the voivodeship marshal).

Equivalent to GTFS’s agency.txt entries.

extra_fields_json: str | None = None¶

fare_url: str = ''¶

id: str¶

lang: str = ''¶

name: str¶

phone: str = ''¶

timezone: str¶

url: str¶

final class impuls.model.Attribution(id: str, organization_name: str, is_producer: bool = False, is_operator: bool = False, is_authority: bool = False, is_data_source: bool = False, url: str = '', email: str = '', phone: str = '', extra_fields_json: str | None = None)¶

Bases: Entity, ExtraFieldsMixin

Attribution represents a copyright or any other attribution which must be attached to the dataset.

Equivalent to GTFS’s attributions.txt entries.

email: str = ''¶

extra_fields_json: str | None = None¶

id: str¶

is_authority: bool = False¶

is_data_source: bool = False¶

is_operator: bool = False¶

is_producer: bool = False¶

organization_name: str¶

phone: str = ''¶

url: str = ''¶

final class impuls.model.Calendar(id: str, monday: bool = False, tuesday: bool = False, wednesday: bool = False, thursday: bool = False, friday: bool = False, saturday: bool = False, sunday: bool = False, start_date: Date = Date(1111, 11, 11), end_date: Date = Date(1111, 11, 11), desc: str = '', extra_fields_json: str | None = None)¶

Bases: Entity, ExtraFieldsMixin

Calendar defines a set of dates on which Trip instances operate.

Equivalent to GTFS’s calendar.txt entries.

Contrary to GTFS, Calendar entries are mandatory, even if all operating dates are defined using CalendarException instances. If this is the case, all weekdays should be set to False and start_date and end_date should be set to Date.SIGNALS_EXCEPTIONS.

compute_active_dates() → set[Date]¶

Computes the set of active dates of this Calendar, not taking exceptions into account.

Use CalendarException.reflect_in_active_dates() to take CalendarException instances into account.

property compressed_weekdays: int¶

desc: str = ''¶

end_date: Date = Date(1111, 11, 11)¶

extra_fields_json: str | None = None¶

friday: bool = False¶

id: str¶

monday: bool = False¶

saturday: bool = False¶

start_date: Date = Date(1111, 11, 11)¶

sunday: bool = False¶

thursday: bool = False¶

tuesday: bool = False¶

wednesday: bool = False¶

final class impuls.model.CalendarException(calendar_id: str, date: Date, exception_type: Type)¶

Bases: Entity

CalendarExceptions are used to override operating dates defined by a Calendar.

Equivalent to GTFS’s calendar_dates.txt entries.

Contrary to GTFS, Calendar entries are mandatory (even if empty), as calendar_id is always a foreign key referencing Calendar.id.

class Type(*values)¶

Bases: IntEnum

ADDED = 1¶

REMOVED = 2¶

static reflect_in_active_dates(active_dates: set[Date], exceptions: Iterable[CalendarException]) → set[Date]¶

Reflects a set of CalendarExceptions in a set of active dates. Warning! The provided set is both modified in-place and later returned.

The set of active dates can come from Calendar.compute_active_dates.

calendar_id: str¶

date: Date¶

exception_type: Type¶

class impuls.model.Date¶

Bases: date

Date is an extension of datetime.date to represent dates of the Impuls model.

The extension provides __str__ and from_ymd_str() methods to convert between dates and YYYY-MM-DD strings, and a few other helper functions.

add_days(delta: int) → Self¶

Returns a new day delta days off from self. Delta may be negative.

>>> Date(2012, 6, 1).add_days(4)
Date(2012, 6, 5)
>>> Date(2012, 12, 27).add_days(7)
Date(2013, 1, 3)
>>> Date(2012, 2, 28).add_days(1)
Date(2012, 2, 29)
>>> Date(2013, 2, 28).add_days(1)
Date(2013, 3, 1)

classmethod from_ymd_str(x: str) → Self¶

Parses a YYYY-MM-DD string into a Date. The separator may be omitted, or may be any non-word characters.

>>> Date.from_ymd_str("2012-06-01")
Date(2012, 6, 1)
>>> Date.from_ymd_str("20120825")
Date(2012, 8, 25)
>>> Date.from_ymd_str("2012.02.29")
Date(2012, 2, 29)

SIGNALS_EXCEPTIONS: ClassVar[Date] = Date(1111, 11, 11)¶: A placeholder Date used in Calendar to indicate that this calendar is defined exclusively using CalendarException instances. In principle, this could be any date, but only SIGNALS_EXCEPTIONS is identified by the SaveGTFS task.

class impuls.model.ExtraFieldsMixin¶

Bases: object

ExtraFieldsMixin provides helper methods for objects with a extra_fields_json Optional[str] fields.

get_extra_field(field: str) → str | None¶

get_extra_fields returns a specific of extra field stored in extra_fields_json.

Invoking this function causes an unconditional parse of extra_fields_json, which may cause a small performance penalty. Use get_extra_fields() once to avoid parsing overhead.

get_extra_fields() → dict[str, str]¶: get_extra_fields returns a dictionary of all extra fields stored in extra_fields_json.

set_extra_field(field: str, value: str | None) → None¶

set_extra_field sets a specific of extra field stored in extra_fields_json.

Invoking this function causes an unconditional parse and serialization of extra_fields_json. Use get_extra_fields() and set_extra_fields() once to avoid JSON serialization overhead.

set_extra_fields(extra_fields: Mapping[str, str] | None) → None¶: get_extra_fields sets the extra fields stored in extra_fields_json to the provided mapping.

extra_fields_json: str | None¶

final class impuls.model.ExtraTableRow(id: int, table_name: str, fields_json: str = '{}', row_sort_order: int | None = None)¶

Bases: Entity

ExtraTableRow is a special Entity which allows defining extra tables and their rows in a generic way.

Note that by default, LoadGTFS does not load unknown tables and columns.

get_field(field: str) → str | None¶

get_field returns a specific field stored in fields_json.

Invoking this function causes an unconditional parse of fields_json, which, if called repeatedly, may incur a performance penalty. Use get_fields() to avoid parsing overhead.

get_fields() → dict[str, str]¶: get_fields returns a fresh dictionary of all fields stored in the fields_json.

set_field(field: str, value: str | None) → None¶

set_field sets a specific field stored in extra_fields_json.

Invoking this function causes an unconditional parse and serialization of fields_json. Use get_fields() and set_fields() once to avoid JSON serialization overhead.

set_fields(fields: Mapping[str, str]) → None¶: set_fields sets all fields in fields_json from the provided mapping.

fields_json: str = '{}'¶

id: int¶: This field is ignored on DBConnection.create() - SQLite automatically generates an ID.

row_sort_order: int | None = None¶

table_name: str¶

final class impuls.model.FareAttribute(id: str, price: float, currency_type: str, payment_method: PaymentMethod, transfers: int | None, agency_id: str, transfer_duration: int | None = None, extra_fields_json: str | None = None)¶

Bases: Entity, ExtraFieldsMixin

FareAttributes define a single logical fare class. Due to the way FareRule is applied, there may be multiple FareAttributes representing the same “ticket”.

Equivalent to GTFS’s fare_attributes.txt entries.

class PaymentMethod(*values)¶

Bases: IntEnum

BEFORE_BOARDING = 1¶

ON_BOARD = 0¶

agency_id: str¶

currency_type: str¶

extra_fields_json: str | None = None¶

id: str¶

payment_method: PaymentMethod¶

price: float¶

transfer_duration: int | None = None¶

transfers: int | None¶

final class impuls.model.FareRule(fare_id: str, route_id: str = '', origin_id: str = '', destination_id: str = '', contains_id: str = '', id: int = 0)¶

Bases: Entity

FareRule instances restrict how FareAttribute instances can be applied.

Equivalent to GTFS’s fare_rules.txt entries.

The GTFS specification is heavily vague on how multiple rules are combined. Impuls’s author current understanding is that: rules with different route_id are logically ORed, while all rules with the same route_id are logically ANDed, both horizontally (across multiple rules) and vertically (across multiple fields). For example, the following rules:

FareRule("f", route_id="", contains_id="A")
FareRule("f", route_id="100", contains_id="A")
FareRule("f", route_id="100", contains_id="B")
FareRule("f", route_id="200", origin_id="A", destination_id="B")

Would be mean that fare f applies to (all routes if journey is completely within zone A) OR (route 100 if journey passes exactly through zones A and B) OR (route 200 if journey starts in zone A and ends in zone B (regardless if it passes through other zones)).

Thus, fare f would apply to journeys on route 100 contained within zone A (thanks to rule 1) or contained within zones A and B (thanks to rules 2 and 3), but not within zone B (as rules 2 and 3 are logically ANDed). Similarly, fare f would apply on journeys on route 200 contained within zone A (thanks to rule 1); starting in zone A, passing through zone C and ending in zone B (thanks to rule 4); but not starting in zone B and ending in zone A (as rule 4 is directional).

Essentially, origin_id and destination_id applied to the Stop.zone_id of the embarking and disembarking stops of a user’s journey leg; while contains_id applies to all Stop.zone_id between the embarking and disembarking stops, inclusive.

contains_id: str = ''¶

destination_id: str = ''¶

fare_id: str¶

id: int = 0¶

This field is ignored on DBConnection.create() - SQLite automatically generates an ID.

The GTFS primary key clause is incompatible with SQL, as it contains optional columns (in SQL PRIMARY KEY implies NOT NULL) - hence the need for a separate ID.

origin_id: str = ''¶

route_id: str = ''¶

final class impuls.model.FeedInfo(publisher_name: str, publisher_url: str, lang: str, version: str = '', contact_email: str = '', contact_url: str = '', start_date: Date | None = None, end_date: Date | None = None, extra_fields_json: str | None = None, id: int = 0)¶

Bases: Entity, ExtraFieldsMixin

FeedInfo describes metadata about the schedule dataset.

Equivalent to GTFS’s feed_info.txt.

contact_email: str = ''¶

contact_url: str = ''¶

end_date: Date | None = None¶

extra_fields_json: str | None = None¶

id: int = 0¶: id of the FeedInfo must be always 0, as there can only be at most one entry in the feed_info table.

lang: str¶

publisher_name: str¶

publisher_url: str¶

start_date: Date | None = None¶

version: str = ''¶

final class impuls.model.Frequency(trip_id: str, start_time: TimePoint, end_time: TimePoint, headway: int, exact_times: bool = False, extra_fields_json: str | None = None)¶

Bases: Entity, ExtraFieldsMixin

Frequency instances provide an alternative way of defining multiple trips in bulk. When a Trip has at least one Frequency, that trips StopTime absolute times are ignored, instead multiple trips using the relative time differences are used as a pattern for multiple trips following the same pattern.

Equivalent to GTFS’s frequencies.txt entries.

end_time: TimePoint¶

exact_times: bool = False¶

extra_fields_json: str | None = None¶

headway: int¶

start_time: TimePoint¶

trip_id: str¶

final class impuls.model.Route(id: str, agency_id: str, short_name: str, long_name: str, type: Type, color: str = '', text_color: str = '', sort_order: int | None = None, extra_fields_json: str | None = None)¶

Bases: Entity, ExtraFieldsMixin

Route instances group multiple trips operated by one Agency under a single, common identifier.

The same as a “line”; not to be confused with a “shape” or a “pattern”. For example all U2 services in Berlin should be grouped under a single route with short_name “U2” and long_name “Pankow - Ruhleben”. For agencies where lines are not commonly used in passenger information, service types may be used instead (common use case for railway operators, e.g. PKP Intercity (Poland) should represent TLK, IC, EIC and EIP train categories as routes, and Korail (South Korea) should represent KTX, ITX, Nuriro, Mungunghwa and Saemeul train categories as routes). If there’s no real distinction of services operated by an agency (common use case for long-haul coaches), a single route with agency name is sufficient.

Equivalent to GTFS’s routes.txt entries.

class Type(*values)¶

Bases: IntEnum

BUS = 3¶

CABLE_TRAM = 5¶

FERRY = 4¶

FUNICULAR = 7¶

GONDOLA = 6¶

METRO = 1¶

MONORAIL = 12¶

RAIL = 2¶

TRAM = 0¶

TROLLEYBUS = 11¶

agency_id: str¶

color: str = ''¶

extra_fields_json: str | None = None¶

id: str¶

long_name: str¶

short_name: str¶

sort_order: int | None = None¶

text_color: str = ''¶

type: Type¶

final class impuls.model.ShapePoint(shape_id: str, sequence: int, lat: float, lon: float, shape_dist_traveled: float | None = None)¶

Bases: Entity

ShapePoints describe the real path a trip takes, used for plotting journeys on a map.

Equivalent to GTFS’s shapes.txt entries.

Note that in GTFS shape_id is not a primary key in any table, rather, it’s a “virtual”, implied entity. In Impuls, this is not the case, a separate shapes table is present, with a sole primary key column, shape_id. There’s no corresponding entity for that table. Before inserting ShapePoints into a database, execute a INSERT INTO shapes (shape_id) VALUES (?) SQL statement.

lat: float¶

lon: float¶

sequence: int¶

shape_dist_traveled: float | None = None¶

shape_id: str¶

final class impuls.model.Stop(id: str, name: str, lat: float, lon: float, code: str = '', zone_id: str = '', location_type: LocationType = LocationType.STOP, parent_station: str = '', wheelchair_boarding: bool | None = None, platform_code: str = '', extra_fields_json: str | None = None)¶

Bases: Entity, ExtraFieldsMixin

Stop can represent 3 different point-like entities, depending on the location_type value, usually physical stops.

LocationType.STOP represent physical places where passengers can embark and disembark from vehicles.

LocationType.STATION represent a grouping of multiple stops and exits under a single physical structure. Note that two stops on an opposite side of a road do not form a station (as these do not form a single physical structure), but an underground bus terminus might. Stop-station structures should be only used when you want to provide exits, or different platform positions. If those details are not available, it is ok to provide single stops representing entire railway stations.

LocationType.EXIT represent an exit to a station.

Equivalent to GTFS’s stops.txt entries.

class LocationType(*values)¶

Bases: IntEnum

EXIT = 2¶

STATION = 1¶

STOP = 0¶

code: str = ''¶

extra_fields_json: str | None = None¶

id: str¶

lat: float¶

location_type: LocationType = 0¶

lon: float¶

name: str¶

parent_station: str = ''¶: parent_station references Stop.id, with empty string mapping to SQL NULL. Optional for stops, forbidden for stations and mandatory for exits.

platform_code: str = ''¶

wheelchair_boarding: bool | None = None¶

zone_id: str = ''¶

final class impuls.model.StopTime(trip_id: str, stop_id: str, stop_sequence: int, arrival_time: TimePoint, departure_time: TimePoint, pickup_type: PassengerExchange = PassengerExchange.SCHEDULED_STOP, drop_off_type: PassengerExchange = PassengerExchange.SCHEDULED_STOP, stop_headsign: str = '', shape_dist_traveled: float | None = None, platform: str = '', extra_fields_json: str | None = None)¶

Bases: Entity, ExtraFieldsMixin

StopTime represents a stoppage/passage of a Trip at/through a Stop.

Equivalent to GTFS’s stop_times.txt entries.

class PassengerExchange(*values)¶

Bases: IntEnum

MUST_PHONE = 2¶

NONE = 1¶

ON_REQUEST = 3¶

SCHEDULED_STOP = 0¶

arrival_time: TimePoint¶

departure_time: TimePoint¶

drop_off_type: PassengerExchange = 0¶

extra_fields_json: str | None = None¶

pickup_type: PassengerExchange = 0¶

platform: str = ''¶

shape_dist_traveled: float | None = None¶

stop_headsign: str = ''¶

stop_id: str¶: The referred stop_id must be of STOP location type.

stop_sequence: int¶

trip_id: str¶

class impuls.model.TimePoint¶

Bases: timedelta

TimePoint is an extension of datetime.timedelta to represent seconds since noon minus 12 hours.

The extension only provides __str__ and from_str() methods to help with conversion between TimePoints and HH:MM:SS strings.

classmethod from_str(x: str) → Self¶

Parses a TimePoint from a HH:MM:SS strings

>>> TimePoint.from_str("8:30:00").total_seconds()
30600.0
>>> TimePoint.from_str("08:30:00").total_seconds()
30600.0
>>> TimePoint.from_str("25:01:08").total_seconds()
90068.0

final class impuls.model.Transfer(from_stop_id: str = '', to_stop_id: str = '', from_route_id: str = '', to_route_id: str = '', from_trip_id: str = '', to_trip_id: str = '', type: Type = Type.RECOMMENDED, min_transfer_time: int | None = None, extra_fields_json: str | None = None, id: int = 0)¶

Bases: Entity, ExtraFieldsMixin

Transfer represent special rules for transferring between vehicles on the network.

Equivalent to GTFS’s transfers.txt entries.

class Type(*values)¶

Bases: IntEnum

IMPOSSIBLE = 3¶

IN_SEAT = 4¶

MIN_TIME_REQUIRED = 2¶

RECOMMENDED = 0¶

RE_BOARD = 5¶

TIMED = 1¶

extra_fields_json: str | None = None¶

from_route_id: str = ''¶

from_stop_id: str = ''¶

from_trip_id: str = ''¶

id: int = 0¶

This field is ignored on DBConnection.create() - SQLite automatically generates an ID.

The GTFS primary key clause is incompatible with SQL, as it contains optional columns (in SQL PRIMARY KEY implies NOT NULL) - hence the need for a separate ID.

min_transfer_time: int | None = None¶

to_route_id: str = ''¶

to_stop_id: str = ''¶

to_trip_id: str = ''¶

type: Type = 0¶

final class impuls.model.Translation(table_name: Literal['agency', 'stops', 'routes', 'trips', 'stop_times', 'feed_info', 'attributions'], field_name: str, language: str, translation: str, record_id: str = '', record_sub_id: str = '', field_value: str = '', extra_fields_json: str | None = None, id: int = 0)¶

Bases: Entity, ExtraFieldsMixin

Translation instances provide a way to translate user-facing text, URLs, emails and phone numbers in consumer apps to better serve multi-lingual regions or regions where some riders are not expected to be able to understand and read the local language.

Equivalent to GTFS’s translations.txt entries.

record_id and field_value must not be provided simultaneously. If record_sub_id is not empty, record_id must not be empty as well.

Translation entities are copied as-is to and from GTFS, and thus all of the selectors must use their GTFS equivalents. Due to the very generic nature of these entities, not all requirements are strictly enforced.

extra_fields_json: str | None = None¶

field_name: str¶: field_name defines the GTFS column name for which the translation applies. For example, to translate Trip.headsign, table_name must be set to trips and field_name must be set to trip_headsign.

field_value: str = ''¶

The original string to be translated.

An alternative way to select strings to be translated is through the record_id attribute. Exactly one of field_value or record_id must be defined - both fields can’t be empty and both fields can be simultaneously non-empty.

id: int = 0¶

This field is ignored on impuls.DBConnection.create() - SQLite automatically generates an ID.

The GTFS primary key clause is incompatible with SQL, as it contains optional columns (in SQL PRIMARY KEY implies NOT NULL) - hence the need for a separate ID.

language: str¶: An IETF language tag of the translated string.

record_id: str = ''¶

Primary key to select the appropriate record from table_name. This should be a reference to the following attributes, depending on the selected table:

Agency.id,
Stop.id,
Route.id,
Trip.id,
Attribution.id,
StopTime.trip_id.

An alternative way to select strings to be translated is through the field_value attribute. Unless the selected table is feed_info, exactly one of field_value or record_id must be defined - both fields can’t be empty and both fields can be simultaneously non-empty.

If the selected table is stop_times and record_id is not empty, record_sub_id must also be non-empty.

record_sub_id: str = ''¶

Secondary part of the primary key of the appropriate record from table_name.

This is only used for stop times, and must be a reference to StopTime.stop_sequence. record_sub_id must not be used for any other tables or when using field_value.

table_name: Literal['agency', 'stops', 'routes', 'trips', 'stop_times', 'feed_info', 'attributions']¶

table_name selects the GTFS table name of the entity type on which the translation applies:

agency for Agency,
stops for Stop,
routes for Route,
trips for Trip,
stop_times for StopTime,
feed_info for FeedInfo,
attributions for Attribution.

translation: str¶: The translated string to be shown in-place of the original string for users of the selected language.

final class impuls.model.Trip(id: str, route_id: str, calendar_id: str, headsign: str = '', short_name: str = '', direction: Direction | None = None, block_id: str = '', shape_id: str = '', wheelchair_accessible: bool | None = None, bikes_allowed: bool | None = None, exceptional: bool | None = None, extra_fields_json: str | None = None)¶

Bases: Entity, ExtraFieldsMixin

Trips represent a single journey made by a vehicle, belonging to a specific Route and Calendar, grouping multiple StopTime objects.

Equivalent to GTFS’s trips.txt entries.

class Direction(*values)¶

Bases: IntEnum

INBOUND = 1¶

OUTBOUND = 0¶

bikes_allowed: bool | None = None¶

block_id: str = ''¶

block_id is used to group multiple trips where a rider can transfer without leaving a vehicle. This should only be used for circular routes or through service between routes; grouping multiple outbound and inbound trips (from a single diagram) of a single route with block_id provides no value to riders and creates visual confusion in consumer applications.

Empty string maps to SQL NULL.

calendar_id: str¶

direction: Direction | None = None¶

exceptional: bool | None = None¶

extra_fields_json: str | None = None¶

headsign: str = ''¶

id: str¶

route_id: str¶

shape_id: str = ''¶: shape_id references Shape.id, with empty string mapping to SQL NULL.

short_name: str = ''¶

wheelchair_accessible: bool | None = None¶

impuls.model.ALL_MODEL_ENTITIES: list[Type[Entity]] = [<class 'impuls.model.agency.Agency'>, <class 'impuls.model.attribution.Attribution'>, <class 'impuls.model.calendar.Calendar'>, <class 'impuls.model.calendar_exception.CalendarException'>, <class 'impuls.model.feed_info.FeedInfo'>, <class 'impuls.model.route.Route'>, <class 'impuls.model.stop.Stop'>, <class 'impuls.model.fare_attribute.FareAttribute'>, <class 'impuls.model.fare_rule.FareRule'>, <class 'impuls.model.shape_point.ShapePoint'>, <class 'impuls.model.trip.Trip'>, <class 'impuls.model.stop_time.StopTime'>, <class 'impuls.model.frequency.Frequency'>, <class 'impuls.model.transfer.Transfer'>, <class 'impuls.model.translation.Translation'>, <class 'impuls.model.extra_table_row.ExtraTableRow'>]¶: List of all Entity classes which belong to the Impuls data model. The list is ordered to allow marshalling without KEY violations, e.g. Trip is before StopTime as the latter references Trip.id.