bolster.data_sources.translink

Translink (NI) Data Sources.

Provides access to live and scheduled transport data for Northern Ireland’s Translink-operated services (Ulsterbus, Metro, Glider, NI Railways).

Data sources: - Live vehicle positions: undocumented VMI feed (vpos.translinkniplanner.co.uk) - Scheduled departures: Translink journey planner API (translink.co.uk) - Stop metadata: Open Data NI ATCO-CIF timetable zips (admin.opendatani.gov.uk) - Point-to-point routing: CIF timetable trip sequences (no journey planner dependency)

Example

>>> from bolster.data_sources import translink
>>> deps = translink.get_departures_by_name("Shankill, Cambria Street", n=3)
>>> "planned_departure" in deps.columns
True

Submodules

Exceptions

`TranslinkDataError`	Base exception for Translink data errors.
`TranslinkDataNotFoundError`	Raised when a Translink endpoint or resource cannot be reached.
`TranslinkValidationError`	Raised when Translink data fails validation checks.

Functions

`clear_cache`([pattern])	Clear cached Translink files. Returns number of files deleted.
`find_stop_id`(query)	Return the Translink internal StopId for the first result matching query.
`get_departures`(stop_id[, n, dt])	Return the next n departures from a stop identified by Translink StopId.
`get_departures_by_name`(stop_name[, n, dt])	Return the next n departures from a stop resolved by name.
`get_departures_with_vehicles`(stop_name[, n, dt, ...])	Return next-N departures enriched with live vehicle positions where available.
`get_direct_journeys`(origin, destination[, n, dt])	Return the next n direct bus/rail journeys between two stops.
`validate_departures`(df)	Validate that a departures DataFrame has the expected schema and values.
`find_stop`(query)	Search for stops by name using the Translink locationApi.
`get_stop_dataframe`([force_refresh])	Return the full stop lookup as a DataFrame.
`get_stop_lookup`([force_refresh])	Return the NaPTAN ATCOCode → stop metadata lookup table.
`resolve_stop_name`(atco_code[, fallback])	Return a human-readable name for a NaPTAN ATCOCode.
`find_direct_trips`(origin_atco, dest_atco[, force_refresh])	Return all direct trips that serve origin_atco then dest_atco.
`find_services_at_stop`(stop_atco[, force_refresh])	Return all scheduled trips that call at stop_atco.
`get_trip_index`([force_refresh])	Return the stop → [(Trip, TripStop)] inverted index.
`get_live_vehicles`([line, operator, enrich_stops])	Return a snapshot of live vehicle positions from the Translink VMI feed.
`validate_vehicles`(df)	Validate a live vehicles DataFrame.

Package Contents

exception bolster.data_sources.translink.TranslinkDataError[source]

Bases: Exception

Base exception for Translink data errors.

Initialize self. See help(type(self)) for accurate signature.

exception bolster.data_sources.translink.TranslinkDataNotFoundError[source]

Bases: TranslinkDataError

Raised when a Translink endpoint or resource cannot be reached.

Initialize self. See help(type(self)) for accurate signature.

exception bolster.data_sources.translink.TranslinkValidationError[source]

Bases: TranslinkDataError

Raised when Translink data fails validation checks.

Initialize self. See help(type(self)) for accurate signature.

bolster.data_sources.translink.clear_cache(pattern=None)[source]

Clear cached Translink files. Returns number of files deleted.

bolster.data_sources.translink.find_stop_id(query)[source]

Return the Translink internal StopId for the first result matching query.

Parameters:: query (str) – Partial or full stop name, e.g. "Cambria Street" or a NaPTAN ATCOCode such as "700000014482".
Returns:: Translink internal StopId string (e.g. "10012778").
Raises:: TranslinkDataNotFoundError – If no stop matches the query.
Return type:: str

bolster.data_sources.translink.get_departures(stop_id, n=5, dt=None)[source]

Return the next n departures from a stop identified by Translink StopId.

The API returns up to 8 departures per call. If more are needed, subsequent calls advance the DepartureOrArrivalDate to fetch additional pages.

Parameters:

stop_id (str) – Translink internal StopId (e.g. "10012778"). Obtain via find_stop_id().
n (int) – Number of departures to return (default 5). The API returns at most 8 per call; additional pages are fetched automatically if needed.
dt (datetime.datetime | None) – Reference datetime for the first departure (default: now, UTC).

Returns:

planned_departure (Timestamp[UTC]), actual_departure (Timestamp[UTC]), service (str), destination (str), transport_mode (str), is_real_time (bool), is_cancelled (bool), delay_minutes (float), unique_id (str).

Return type:

DataFrame with columns

Raises:

TranslinkDataNotFoundError – If the API request fails.
TranslinkValidationError – If the API returns an unexpected response.

bolster.data_sources.translink.get_departures_by_name(stop_name, n=5, dt=None)[source]

Return the next n departures from a stop resolved by name.

Convenience wrapper that calls find_stop_id() then get_departures().

Parameters:

stop_name (str) – Stop name or NaPTAN ATCOCode to search for.
n (int) – Number of departures to return (default 5).
dt (datetime.datetime | None) – Reference datetime (default: now, UTC).

Returns:

DataFrame as returned by get_departures(), with an additional stop_name column showing the resolved stop name.

Raises:

TranslinkDataNotFoundError – If the stop cannot be found or the API fails.

Return type:

pandas.DataFrame

bolster.data_sources.translink.get_departures_with_vehicles(stop_name, n=5, dt=None, enrich_stops=False)[source]

Return next-N departures enriched with live vehicle positions where available.

Fetches departures and live VMI vehicles in parallel (two API calls), then joins on line + direction + journey time proximity (±60 minute window).

VMI vehicles are matched to departures by: 1. Line number (case-insensitive). 2. Inferred direction (inbound = destination contains “Belfast”/”CastleCourt”/

“Royal Avenue”/”City Centre”; outbound = everything else).

Journey ID (HHMM) within ±60 minutes of the actual departure time.

Not all departures will have a matched vehicle — buses that have not yet started their journey are not yet in the VMI feed.

Parameters:

stop_name (str) – Stop name to search for (resolved via find_stop_id()).
n (int) – Number of departures to return (default 5).
dt (datetime.datetime | None) – Reference datetime (default: now, UTC).
enrich_stops (bool) – If True, include current_stop_name and next_stop_name for matched vehicles.

Returns:

vehicle_id, vehicle_lat, vehicle_lon, vehicle_delay_s, current_stop, next_stop, and (if enrich_stops) current_stop_name, next_stop_name. Vehicle columns are None / NaN where no match.

Return type:

DataFrame with all departure columns plus optional vehicle columns

bolster.data_sources.translink.get_direct_journeys(origin, destination, n=5, dt=None)[source]

Return the next n direct bus/rail journeys between two stops.

Uses the CIF timetable data (Metro/Glider and Ulsterbus/GoldLine) to find services that call at both stops in order, without requiring a change. No network calls are made beyond resolving stop names; all routing is done from the locally cached timetable.

The workflow is: 1. Resolve both stop names to NaPTAN ATCOCodes via the CIF stop lookup. 2. Find all trips in the timetable that call at the origin before the

destination (find_direct_trips).

Filter to trips whose scheduled origin departure is at or after dt, sorted by departure time.
Return the first n matching trips.

Parameters:

origin (str) – Origin stop name (resolved via find_stop()).
destination (str) – Destination stop name.
n (int) – Maximum number of journeys to return (default 5).
dt (datetime.datetime | None) – Reference datetime (default: now, local Europe/London time).

Returns:

origin, destination, service, scheduled_departure (HHMM str), scheduled_arrival (HHMM str), days, direction.

Return type:

DataFrame with columns

Raises:

TranslinkDataNotFoundError – If either stop cannot be resolved, or if no direct service runs between them at all.

bolster.data_sources.translink.validate_departures(df)[source]

Validate that a departures DataFrame has the expected schema and values.

Parameters:: df (pandas.DataFrame) – DataFrame as returned by get_departures().
Returns:: True if validation passes.
Raises:: TranslinkValidationError – If required columns are missing or values are invalid.
Return type:: bool

bolster.data_sources.translink.find_stop(query)[source]

Search for stops by name using the Translink locationApi.

Parameters:: query (str) – Partial or full stop name, e.g. "Cambria Street".
Returns:: id (Translink internal StopId), name, location_type.
Return type:: List of dicts, each with keys
Raises:: TranslinkDataNotFoundError – If the API request fails.

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

Return the full stop lookup as a DataFrame.

Columns: atco_code, name, easting, northing, latitude, longitude.

Parameters:: force_refresh (bool) – Re-download and rebuild the stop table.
Returns:: DataFrame with one row per stop, indexed by atco_code.
Return type:: pandas.DataFrame

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

Return the NaPTAN ATCOCode → stop metadata lookup table.

The table is built on first call and cached in memory for the lifetime of the process. The underlying zip files are cached on disk for one week.

Parameters:

force_refresh (bool) – Re-download source zips and rebuild the lookup.

Returns:

name (str): Stop name from the CIF
easting (int): ING easting in metres
northing (int): ING northing in metres
latitude (float): WGS84 latitude
longitude (float): WGS84 longitude

Return type:

Dict mapping ATCOCode (e.g. "700000001661") to a dict with keys

Raises:

TranslinkDataNotFoundError – If the source zips cannot be downloaded.
TranslinkValidationError – If the resulting table is implausibly small.

bolster.data_sources.translink.resolve_stop_name(atco_code, fallback=True)[source]

Return a human-readable name for a NaPTAN ATCOCode.

Looks up the code in the CIF-derived table first. If not found and fallback is True, queries the Translink locationApi live.

Parameters:

atco_code (str) – NaPTAN ATCOCode, e.g. "700000014482".
fallback (bool) – If True, attempt a live API lookup for unknown codes.

Returns:

Stop name string, or None if not resolvable.

Return type:

str | None

bolster.data_sources.translink.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]]

bolster.data_sources.translink.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.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.get_live_vehicles(line=None, operator=None, enrich_stops=False)[source]

Return a snapshot of live vehicle positions from the Translink VMI feed.

Parameters:

line (str | None) – Optional line filter, case-insensitive (e.g. "11E" or "G1").
operator (str | None) – Optional operator filter, case-insensitive. Accepts canonical codes ("MET") or VMI codes ("TM"). Both map to Metro.
enrich_stops (bool) – If True, resolve current_stop / next_stop ATCOCodes to human-readable names. Requires building the stop lookup table on first call (~1.3 MB download).

Returns:

id, vehicle_id, operator, operator_raw, line, direction, journey_id, day_of_operation, longitude, latitude, longitude_prev, latitude_prev, timestamp (tz-aware), timestamp_prev (tz-aware), delay_seconds (Int64, negative = early), current_stop, next_stop, is_at_stop (bool), realtime_available (bool), mot_code (Int64).

Return type:

DataFrame with one row per active vehicle. Columns

Raises:

TranslinkDataNotFoundError – If the VMI feed cannot be reached.
TranslinkValidationError – If the response is malformed.

Example

>>> vdf = get_live_vehicles(line="11E")
>>> all(vdf["line"].str.upper() == "11E")
True

bolster.data_sources.translink.validate_vehicles(df)[source]

Validate a live vehicles DataFrame.

Parameters:: df (pandas.DataFrame) – DataFrame as returned by get_live_vehicles().
Returns:: True if validation passes.
Raises:: TranslinkValidationError – If required columns are missing or coordinates are invalid.
Return type:: bool