API reference¶

Everything listed here is importable from the top-level gtfs_rt_kit package, except the modules shown with their full path.

Decoding¶

Decoding of GTFS-Realtime protobuf payloads into typed models.

parse_feed ¶

parse_feed(data: bytes) -> Feed

Decode a serialized GTFS-Realtime FeedMessage.

Raises:

Type	Description
`FeedDecodeError`	if `data` is not a valid feed message.

load_feed ¶

load_feed(path: str | PathLike[str]) -> Feed

Read and decode a GTFS-Realtime feed stored in a local file.

Models¶

Typed, immutable models for GTFS-Realtime entities.

The models mirror the GTFS-Realtime specification but expose plain Python types: optional protobuf fields become None when absent, POSIX timestamps become timezone-aware :class:~datetime.datetime objects (UTC) and enums become their specification names (for example "IN_TRANSIT_TO").

FeedHeader `dataclass` ¶

FeedHeader(
    gtfs_realtime_version: str,
    incrementality: str = "FULL_DATASET",
    timestamp: datetime | None = None,
    feed_version: str | None = None,
)

Metadata of a feed message.

TripDescriptor `dataclass` ¶

TripDescriptor(
    trip_id: str | None = None,
    route_id: str | None = None,
    direction_id: int | None = None,
    start_time: str | None = None,
    start_date: str | None = None,
    schedule_relationship: str | None = None,
)

Identifies a trip instance (TripDescriptor in the specification).

VehicleDescriptor `dataclass` ¶

VehicleDescriptor(
    id: str | None = None,
    label: str | None = None,
    license_plate: str | None = None,
)

Identifies a physical vehicle.

Position `dataclass` ¶

Position(
    latitude: float,
    longitude: float,
    bearing: float | None = None,
    odometer: float | None = None,
    speed: float | None = None,
)

Geographic position of a vehicle (WGS-84).

VehiclePosition `dataclass` ¶

VehiclePosition(
    entity_id: str,
    trip: TripDescriptor | None = None,
    vehicle: VehicleDescriptor | None = None,
    position: Position | None = None,
    current_stop_sequence: int | None = None,
    stop_id: str | None = None,
    current_status: str | None = None,
    timestamp: datetime | None = None,
    congestion_level: str | None = None,
    occupancy_status: str | None = None,
    occupancy_percentage: int | None = None,
)

Realtime position of a vehicle, optionally linked to a trip.

vehicle_id `property` ¶

vehicle_id: str

Best available identifier: vehicle id, then label, then the entity id.

trip_id `property` ¶

trip_id: str | None

Trip id from the trip descriptor, if any.

route_id `property` ¶

route_id: str | None

Route id from the trip descriptor, if any.

StopTimeEvent `dataclass` ¶

StopTimeEvent(
    delay: int | None = None,
    time: datetime | None = None,
    uncertainty: int | None = None,
    scheduled_time: datetime | None = None,
)

Predicted arrival or departure at a stop.

StopTimeUpdate `dataclass` ¶

StopTimeUpdate(
    stop_sequence: int | None = None,
    stop_id: str | None = None,
    arrival: StopTimeEvent | None = None,
    departure: StopTimeEvent | None = None,
    schedule_relationship: str = "SCHEDULED",
)

Realtime update for one stop of a trip.

TripUpdate `dataclass` ¶

TripUpdate(
    entity_id: str,
    trip: TripDescriptor,
    vehicle: VehicleDescriptor | None = None,
    stop_time_updates: tuple[StopTimeUpdate, ...] = (),
    timestamp: datetime | None = None,
    delay: int | None = None,
)

Realtime progress of a trip.

trip_id `property` ¶

trip_id: str | None

Trip id from the trip descriptor.

route_id `property` ¶

route_id: str | None

Route id from the trip descriptor.

Translation `dataclass` ¶

Translation(text: str, language: str | None = None)

A single localized text.

TranslatedString `dataclass` ¶

TranslatedString(
    translations: tuple[Translation, ...] = (),
)

A text available in one or more languages.

text ¶

text(language: str | None = None) -> str | None

Return the text in language.

Falls back to the translation without a language tag, then to the first one.

TimeRange `dataclass` ¶

TimeRange(
    start: datetime | None = None,
    end: datetime | None = None,
)

A time interval; an open end is None.

contains ¶

contains(moment: datetime) -> bool

Whether moment (timezone-aware) falls into this range.

EntitySelector `dataclass` ¶

EntitySelector(
    agency_id: str | None = None,
    route_id: str | None = None,
    route_type: int | None = None,
    direction_id: int | None = None,
    trip: TripDescriptor | None = None,
    stop_id: str | None = None,
)

Selects the agency, route, trip or stop an alert applies to.

Alert `dataclass` ¶

Alert(
    entity_id: str,
    active_periods: tuple[TimeRange, ...] = (),
    informed_entities: tuple[EntitySelector, ...] = (),
    cause: str = "UNKNOWN_CAUSE",
    effect: str = "UNKNOWN_EFFECT",
    severity_level: str = "UNKNOWN_SEVERITY",
    url: TranslatedString | None = None,
    header_text: TranslatedString | None = None,
    description_text: TranslatedString | None = None,
)

A service alert.

is_active ¶

is_active(moment: datetime) -> bool

Whether the alert is active at moment; no periods means always active.

affects_route ¶

affects_route(route_id: str) -> bool

Whether any informed entity references route_id (directly or via its trip).

affects_stop ¶

affects_stop(stop_id: str) -> bool

Whether any informed entity references stop_id.

Feed `dataclass` ¶

Feed(
    header: FeedHeader,
    vehicles: tuple[VehiclePosition, ...] = (),
    trip_updates: tuple[TripUpdate, ...] = (),
    alerts: tuple[Alert, ...] = (),
    deleted_entity_ids: tuple[str, ...] = (),
)

A decoded GTFS-Realtime FeedMessage, split by entity type.

timestamp `property` ¶

timestamp: datetime | None

Timestamp of the feed header.

vehicles_on_route ¶

vehicles_on_route(
    route_id: str,
) -> tuple[VehiclePosition, ...]

Vehicles whose trip descriptor references route_id.

trip_update_for ¶

trip_update_for(trip_id: str) -> TripUpdate | None

The trip update for trip_id, if present.

alerts_for_route ¶

alerts_for_route(route_id: str) -> tuple[Alert, ...]

Alerts that reference route_id.

Clients¶

Synchronous and asynchronous HTTP clients for GTFS-Realtime feeds.

Requires the optional http extra: pip install "gtfs-rt-kit[http]".

GtfsRtClient ¶

GtfsRtClient(
    *,
    timeout: float | Timeout = DEFAULT_TIMEOUT,
    retry: RetryPolicy | None = None,
    headers: Mapping[str, str] | None = None,
    http_client: Client | None = None,
    sleep: Callable[[float], None] = time.sleep,
)

Blocking client that downloads and decodes GTFS-Realtime feeds.

Example

with GtfsRtClient() as client: # doctest: +SKIP ... feed = client.fetch("https://example.org/vehicles.pb")

fetch_bytes ¶

fetch_bytes(url: str) -> bytes

Download the raw feed payload, retrying transient failures.

fetch ¶

fetch(url: str) -> Feed

Download and decode a feed.

vehicles ¶

vehicles(url: str) -> tuple[VehiclePosition, ...]

Vehicle positions from the feed at url.

trip_updates ¶

trip_updates(url: str) -> tuple[TripUpdate, ...]

Trip updates from the feed at url.

alerts ¶

alerts(url: str) -> tuple[Alert, ...]

Service alerts from the feed at url.

close ¶

close() -> None

Close the underlying HTTP client if this instance created it.

AsyncGtfsRtClient ¶

AsyncGtfsRtClient(
    *,
    timeout: float | Timeout = DEFAULT_TIMEOUT,
    retry: RetryPolicy | None = None,
    headers: Mapping[str, str] | None = None,
    http_client: AsyncClient | None = None,
    sleep: Callable[
        [float], Awaitable[None]
    ] = asyncio.sleep,
)

Asynchronous counterpart of :class:GtfsRtClient.

Example

async with AsyncGtfsRtClient() as client: # doctest: +SKIP ... feed = await client.fetch("https://example.org/vehicles.pb")

fetch_bytes `async` ¶

fetch_bytes(url: str) -> bytes

Download the raw feed payload, retrying transient failures.

fetch `async` ¶

fetch(url: str) -> Feed

Download and decode a feed.

vehicles `async` ¶

vehicles(url: str) -> tuple[VehiclePosition, ...]

Vehicle positions from the feed at url.

trip_updates `async` ¶

trip_updates(url: str) -> tuple[TripUpdate, ...]

Trip updates from the feed at url.

alerts `async` ¶

alerts(url: str) -> tuple[Alert, ...]

Service alerts from the feed at url.

aclose `async` ¶

aclose() -> None

Close the underlying HTTP client if this instance created it.

Retry policy shared by the synchronous and asynchronous clients.

RetryPolicy `dataclass` ¶

RetryPolicy(
    attempts: int = 3,
    backoff: float = 0.5,
    max_backoff: float = 8.0,
    retry_statuses: frozenset[
        int
    ] = _DEFAULT_RETRY_STATUSES,
)

How failed requests are retried.

Attributes:

Name	Type	Description
`attempts`	`int`	Total number of attempts, including the first one (`1` disables retries).
`backoff`	`float`	Base delay in seconds; attempt `n` waits `backoff * 2 ** (n - 1)`.
`max_backoff`	`float`	Upper bound for a single delay, also applied to `Retry-After`.
`retry_statuses`	`frozenset[int]`	HTTP status codes that are worth retrying.

delay ¶

delay(
    attempt: int, retry_after: str | None = None
) -> float

Seconds to wait after failed attempt number attempt (1-based).

should_retry_status ¶

should_retry_status(status_code: int) -> bool

Whether a response with status_code should be retried.

Static GTFS¶

Lazy loader for static GTFS archives (.zip).

Only the files you actually touch are parsed: opening a feed is cheap, reading stops parses stops.txt once and caches the result, and the potentially large stop_times.txt is only read when schedule data is requested.

Agency `dataclass` ¶

Agency(agency_id: str, name: str, url: str, timezone: str)

A row of agency.txt.

Stop `dataclass` ¶

Stop(
    stop_id: str,
    name: str,
    latitude: float | None = None,
    longitude: float | None = None,
    code: str | None = None,
    parent_station: str | None = None,
    location_type: int = 0,
)

A row of stops.txt.

Route `dataclass` ¶

Route(
    route_id: str,
    short_name: str,
    long_name: str,
    route_type: int,
    agency_id: str | None = None,
    color: str | None = None,
    text_color: str | None = None,
)

A row of routes.txt.

name `property` ¶

name: str

The short name if present, otherwise the long name.

mode `property` ¶

mode: str

A coarse transport mode ("bus", "tram", "subway", ...).

Trip `dataclass` ¶

Trip(
    trip_id: str,
    route_id: str,
    service_id: str,
    headsign: str | None = None,
    direction_id: int | None = None,
    shape_id: str | None = None,
    block_id: str | None = None,
)

A row of trips.txt.

StopTime `dataclass` ¶

StopTime(
    trip_id: str,
    stop_sequence: int,
    stop_id: str,
    arrival: int | None = None,
    departure: int | None = None,
)

A row of stop_times.txt; times are seconds after service-day midnight.

StaticGtfs ¶

StaticGtfs(source: Source)

A static GTFS feed backed by a zip archive.

Parameters:

Name	Type	Description	Default
`source`	`Source`	Path to a `.zip` file, the archive's bytes, or a binary file object.	required

Example

gtfs = StaticGtfs("warsaw.zip") # doctest: +SKIP gtfs.route("15").mode # doctest: +SKIP 'tram'

files `property` ¶

files: frozenset[str]

Names of the files contained in the archive.

agencies `cached` `property` ¶

agencies: Mapping[str, Agency]

Agencies keyed by agency_id (an empty id when the column is omitted).

timezone `cached` `property` ¶

timezone: ZoneInfo

Timezone of the feed, taken from the first agency.

stops `cached` `property` ¶

stops: Mapping[str, Stop]

Stops keyed by stop_id.

routes `cached` `property` ¶

routes: Mapping[str, Route]

Routes keyed by route_id.

trips `cached` `property` ¶

trips: Mapping[str, Trip]

Trips keyed by trip_id.

from_bytes `classmethod` ¶

from_bytes(data: bytes) -> StaticGtfs

Load a feed from the bytes of a zip archive.

has_file ¶

has_file(name: str) -> bool

Whether the archive contains name (for example "shapes.txt").

rows ¶

rows(name: str) -> Iterator[dict[str, str]]

Stream the rows of any CSV file of the archive as dictionaries.

Raises:

Type	Description
`StaticGtfsError`	if the file does not exist.

stop ¶

stop(stop_id: str) -> Stop | None

The stop with stop_id, if any.

route ¶

route(route_id: str) -> Route | None

The route with route_id, if any.

routes_by_short_name ¶

routes_by_short_name(short_name: str) -> tuple[Route, ...]

Routes whose short name equals short_name (case-insensitive).

find_route ¶

find_route(key: str) -> Route | None

Look a route up by id first, then by short name.

trip ¶

trip(trip_id: str) -> Trip | None

The trip with trip_id, if any.

trips_for_route ¶

trips_for_route(route_id: str) -> tuple[Trip, ...]

All trips of route_id.

stop_times ¶

stop_times(trip_id: str) -> tuple[StopTime, ...]

Scheduled stop times of trip_id ordered by stop_sequence.

The first call parses stop_times.txt and indexes it by trip.

parse_gtfs_time ¶

parse_gtfs_time(value: str) -> int

Convert a GTFS HH:MM:SS time (hours may exceed 23) to seconds after midnight.

Raises:

Type	Description
`ValueError`	if `value` is not a valid GTFS time.

format_gtfs_time ¶

format_gtfs_time(seconds: int) -> str

Inverse of :func:parse_gtfs_time.

Matching¶

Matching realtime vehicles to static trips and routes.

RouteHint `module-attribute` ¶

RouteHint = Callable[[VehiclePosition], 'str | None']

A callable deriving a route id or short name from a vehicle when the feed omits it.

MatchedVehicle `dataclass` ¶

MatchedVehicle(
    vehicle: VehiclePosition,
    trip: Trip | None = None,
    route: Route | None = None,
    route_key: str | None = None,
)

A vehicle position enriched with its scheduled trip and route.

line `property` ¶

line: str | None

Passenger-facing line name: route short name, else the raw route key.

mode `property` ¶

mode: str | None

Transport mode of the matched route.

match_vehicle ¶

match_vehicle(
    vehicle: VehiclePosition,
    static: StaticGtfs,
    *,
    route_hint: RouteHint | None = None,
) -> MatchedVehicle

Resolve the trip and route of vehicle.

Resolution order for the route: the trip descriptor's route_id, the route of the matched trip, then route_hint(vehicle). Each candidate is looked up as a route id first and as a route short name second.

match_vehicles ¶

match_vehicles(
    vehicles: Feed | Iterable[VehiclePosition],
    static: StaticGtfs,
    *,
    route_hint: RouteHint | None = None,
) -> list[MatchedVehicle]

Match every vehicle of a feed (or an iterable of vehicles).

filter_by_route ¶

filter_by_route(
    matched: Iterable[MatchedVehicle], route: str
) -> list[MatchedVehicle]

Keep vehicles whose route id, short name or raw route key equals route.

The comparison is case-insensitive, so "n91" matches the night line N91.

Delays¶

Delay computation: realtime predictions compared with the static schedule.

StopDelay `dataclass` ¶

StopDelay(
    stop_sequence: int | None,
    stop_id: str | None,
    event: Literal["arrival", "departure"] | None,
    scheduled: datetime | None,
    predicted: datetime | None,
    delay: int | None,
    skipped: bool = False,
)

Delay of one stop of a trip, in seconds (positive means late).

scheduled_datetime ¶

scheduled_datetime(
    service_date: date, seconds: int, tz: ZoneInfo
) -> datetime

Absolute time of a GTFS stop time.

GTFS times count from "noon minus 12 hours" of the service day, which differs from midnight on days with a daylight-saving change, and may exceed 24:00:00.

service_date_of ¶

service_date_of(
    update: TripUpdate | VehiclePosition, tz: ZoneInfo
) -> date | None

Service date from the trip's start_date or, failing that, the local timestamp date.

trip_delays ¶

trip_delays(
    update: TripUpdate,
    static: StaticGtfs,
    *,
    service_date: date | None = None,
) -> tuple[StopDelay, ...]

Per-stop delays of a trip update compared with stop_times.txt.

For each stop time update the arrival is used when present, otherwise the departure. An explicit delay in the feed wins over time - scheduled. Stops that cannot be matched to the schedule keep the feed's own delay, if any.

estimate_vehicle_delay ¶

estimate_vehicle_delay(
    vehicle: VehiclePosition,
    static: StaticGtfs,
    *,
    service_date: date | None = None,
) -> int | None

Rough delay of a vehicle: its timestamp minus the scheduled time at its current stop.

Only meaningful for vehicles reported STOPPED_AT (or with no status) at a stop; returns None when the trip, stop or timestamp is unknown.

Warsaw helpers¶

gtfs_rt_kit.warsaw ¶

Helpers for the Warsaw (ZTM) feeds published at https://mkuran.pl/gtfs/.

The Warsaw realtime feed identifies vehicles as V/<line>/<brigade> and does not fill route_id; :func:route_from_vehicle_id recovers the line from the id.

route_from_vehicle_id ¶

route_from_vehicle_id(
    vehicle: VehiclePosition,
) -> str | None

Line number encoded in a Warsaw vehicle id ("V/15/7" -> "15").

Serialization¶

Conversion of models to JSON-compatible structures.

to_jsonable ¶

to_jsonable(value: Any) -> Any

Recursively convert dataclasses, datetimes and tuples into JSON-compatible values.

Datetimes become ISO 8601 strings; dataclass properties are not included.

Errors¶

Exception hierarchy of gtfs-rt-kit.

GtfsRtKitError ¶

Bases: Exception

Base class for all errors raised by this library.

FeedDecodeError ¶

Bases: GtfsRtKitError

The payload is not a valid GTFS-Realtime FeedMessage.

FeedFetchError ¶

FeedFetchError(
    message: str,
    *,
    url: str,
    status_code: int | None = None,
)

Bases: GtfsRtKitError

A feed could not be downloaded (network error, timeout or HTTP error status).

StaticGtfsError ¶

Bases: GtfsRtKitError

A static GTFS archive is missing, malformed or lacks a required file.

API reference¶

Decoding¶

parse_feed ¶

load_feed ¶

Models¶

FeedHeader dataclass ¶

TripDescriptor dataclass ¶

VehicleDescriptor dataclass ¶

Position dataclass ¶

VehiclePosition dataclass ¶

vehicle_id property ¶

trip_id property ¶

route_id property ¶

StopTimeEvent dataclass ¶

StopTimeUpdate dataclass ¶

TripUpdate dataclass ¶

trip_id property ¶

route_id property ¶

Translation dataclass ¶

TranslatedString dataclass ¶

text ¶

TimeRange dataclass ¶

contains ¶

EntitySelector dataclass ¶

Alert dataclass ¶

is_active ¶

affects_route ¶

affects_stop ¶

Feed dataclass ¶

timestamp property ¶

vehicles_on_route ¶

trip_update_for ¶

alerts_for_route ¶

Clients¶

GtfsRtClient ¶

fetch_bytes ¶

fetch ¶

vehicles ¶

trip_updates ¶

alerts ¶

close ¶

AsyncGtfsRtClient ¶

fetch_bytes async ¶

fetch async ¶

vehicles async ¶

trip_updates async ¶

alerts async ¶

aclose async ¶

RetryPolicy dataclass ¶

delay ¶

should_retry_status ¶

Static GTFS¶

Agency dataclass ¶

Stop dataclass ¶

Route dataclass ¶

name property ¶

mode property ¶

Trip dataclass ¶

StopTime dataclass ¶

StaticGtfs ¶

files property ¶

agencies cached property ¶

timezone cached property ¶

stops cached property ¶

routes cached property ¶

trips cached property ¶

from_bytes classmethod ¶

has_file ¶

rows ¶

stop ¶

route ¶

routes_by_short_name ¶

find_route ¶

trip ¶

trips_for_route ¶

stop_times ¶

parse_gtfs_time ¶

format_gtfs_time ¶

Matching¶

RouteHint module-attribute ¶

FeedHeader `dataclass` ¶

TripDescriptor `dataclass` ¶

VehicleDescriptor `dataclass` ¶

Position `dataclass` ¶

VehiclePosition `dataclass` ¶

vehicle_id `property` ¶

trip_id `property` ¶

route_id `property` ¶

StopTimeEvent `dataclass` ¶

StopTimeUpdate `dataclass` ¶

TripUpdate `dataclass` ¶

trip_id `property` ¶

route_id `property` ¶

Translation `dataclass` ¶

TranslatedString `dataclass` ¶

TimeRange `dataclass` ¶

EntitySelector `dataclass` ¶

Alert `dataclass` ¶

Feed `dataclass` ¶

timestamp `property` ¶

fetch_bytes `async` ¶

fetch `async` ¶

vehicles `async` ¶

trip_updates `async` ¶

alerts `async` ¶

aclose `async` ¶

RetryPolicy `dataclass` ¶

Agency `dataclass` ¶

Stop `dataclass` ¶

Route `dataclass` ¶

name `property` ¶

mode `property` ¶

Trip `dataclass` ¶

StopTime `dataclass` ¶

files `property` ¶

agencies `cached` `property` ¶

timezone `cached` `property` ¶

stops `cached` `property` ¶

routes `cached` `property` ¶

trips `cached` `property` ¶

from_bytes `classmethod` ¶

RouteHint `module-attribute` ¶

MatchedVehicle `dataclass` ¶

line `property` ¶

mode `property` ¶

StopDelay `dataclass` ¶