bolster.data_sources.translink.timetable

Translink timetable — trip stop sequences from ATCO-CIF archives.

Parses the QD/QS/QO/QI/QT records from the Metro/Glider and Ulsterbus/GoldLine CIF zips to build a queryable trip database: for any stop you can find all services that call there, and for any pair of stops you can find direct services that serve both in order.

Record formats parsed: - QD Journey header: operator, line number, description - QS Journey schedule: departure time, date range, days of week, direction - QO7 Origin timing point: first stop ATCO, departure time - QI7 Intermediate timing point: stop ATCO, arrival, departure times - QT7 Terminus timing point: last stop ATCO, arrival time

ATCO code mapping: trip records store 11-digit codes (e.g. 00000009264) while QL/QB stop records use 12-digit codes starting with 7 (e.g. 700000009264). The relation is stop_atco = '7' + trip_atco[0:11].

Example

>>> trips = get_trip_index()
>>> trips["700000001661"]  # stops serving Victoria Square
[TripStop(...), ...]

Attributes

logger

Classes

`TripStop`	A single stop within a trip's stop sequence.
`Trip`	A single scheduled trip (one journey on one day pattern).

Functions

`get_trip_index`([force_refresh])	Return the stop → [(Trip, TripStop)] inverted index.
`find_services_at_stop`(stop_atco[, force_refresh])	Return all scheduled trips that call at stop_atco.
`find_direct_trips`(origin_atco, dest_atco[, force_refresh])	Return all direct trips that serve origin_atco then dest_atco.

Module Contents

bolster.data_sources.translink.timetable.logger[source]

class bolster.data_sources.translink.timetable.TripStop[source]

A single stop within a trip’s stop sequence.

atco: str[source]

arrive: str[source]

depart: str[source]

seq: int[source]

class bolster.data_sources.translink.timetable.Trip[source]

A single scheduled trip (one journey on one day pattern).

operator: str[source]

line: str[source]

description: str[source]

depart_hhmm: str[source]

date_from: str[source]

date_to: str[source]

days: str[source]

direction: str[source]

stops: list[TripStop] = [][source]

bolster.data_sources.translink.timetable.get_trip_index(force_refresh=False)[source]

Return the stop → [(Trip, TripStop)] inverted index.

Built on first call and cached in-process for the lifetime of the Python process. The underlying CIF zips are cached on disk for one week.

Parameters:: force_refresh (bool) – Re-download CIF zips and rebuild the index.
Returns:: Dict mapping ATCOCode → list of (Trip, TripStop) tuples for every scheduled trip that calls at that stop.
Return type:: dict[str, list[tuple[Any, Any]]]

bolster.data_sources.translink.timetable.find_services_at_stop(stop_atco, force_refresh=False)[source]

Return all scheduled trips that call at stop_atco.

Parameters:

stop_atco (str) – 12-digit NaPTAN ATCOCode (e.g. "700000001661").
force_refresh (bool) – Rebuild the trip index from source.

Returns:

Deduplicated list of Trip objects, ordered by origin departure time.

Return type:

list[Trip]

bolster.data_sources.translink.timetable.find_direct_trips(origin_atco, dest_atco, force_refresh=False)[source]

Return all direct trips that serve origin_atco then dest_atco.

A trip is considered direct if both stops appear in its stop sequence and the origin stop appears before the destination stop.

Parameters:

origin_atco (str) – 12-digit NaPTAN ATCOCode of the origin stop.
dest_atco (str) – 12-digit NaPTAN ATCOCode of the destination stop.
force_refresh (bool) – Rebuild the trip index from source.

Returns:

List of (Trip, origin_TripStop, dest_TripStop) tuples, ordered by the trip’s origin departure time. Empty list if no direct service runs between the two stops.

Return type:

list[tuple[Trip, TripStop, TripStop]]