bolster.data_sources.translink ============================== .. py:module:: bolster.data_sources.translink .. autoapi-nested-parse:: 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) .. rubric:: 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 ---------- .. toctree:: :maxdepth: 1 /autoapi/bolster/data_sources/translink/departures/index /autoapi/bolster/data_sources/translink/stops/index /autoapi/bolster/data_sources/translink/timetable/index /autoapi/bolster/data_sources/translink/vehicles/index Exceptions ---------- .. autoapisummary:: bolster.data_sources.translink.TranslinkDataError bolster.data_sources.translink.TranslinkDataNotFoundError bolster.data_sources.translink.TranslinkValidationError Functions --------- .. autoapisummary:: bolster.data_sources.translink.clear_cache bolster.data_sources.translink.find_stop_id bolster.data_sources.translink.get_departures bolster.data_sources.translink.get_departures_by_name bolster.data_sources.translink.get_departures_with_vehicles bolster.data_sources.translink.get_direct_journeys bolster.data_sources.translink.validate_departures bolster.data_sources.translink.find_stop bolster.data_sources.translink.get_stop_dataframe bolster.data_sources.translink.get_stop_lookup bolster.data_sources.translink.resolve_stop_name bolster.data_sources.translink.find_direct_trips bolster.data_sources.translink.find_services_at_stop bolster.data_sources.translink.get_trip_index bolster.data_sources.translink.get_live_vehicles bolster.data_sources.translink.validate_vehicles Package Contents ---------------- .. py:exception:: TranslinkDataError Bases: :py:obj:`Exception` Base exception for Translink data errors. Initialize self. See help(type(self)) for accurate signature. .. py:exception:: TranslinkDataNotFoundError Bases: :py:obj:`TranslinkDataError` Raised when a Translink endpoint or resource cannot be reached. Initialize self. See help(type(self)) for accurate signature. .. py:exception:: TranslinkValidationError Bases: :py:obj:`TranslinkDataError` Raised when Translink data fails validation checks. Initialize self. See help(type(self)) for accurate signature. .. py:function:: clear_cache(pattern = None) Clear cached Translink files. Returns number of files deleted. .. py:function:: find_stop_id(query) Return the Translink internal StopId for the first result matching *query*. :param query: 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. .. py:function:: get_departures(stop_id, n = 5, dt = None) 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. :param stop_id: Translink internal StopId (e.g. ``"10012778"``). Obtain via :func:`find_stop_id`. :param n: Number of departures to return (default 5). The API returns at most 8 per call; additional pages are fetched automatically if needed. :param dt: 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). :rtype: DataFrame with columns :raises TranslinkDataNotFoundError: If the API request fails. :raises TranslinkValidationError: If the API returns an unexpected response. .. py:function:: get_departures_by_name(stop_name, n = 5, dt = None) Return the next *n* departures from a stop resolved by name. Convenience wrapper that calls :func:`find_stop_id` then :func:`get_departures`. :param stop_name: Stop name or NaPTAN ATCOCode to search for. :param n: Number of departures to return (default 5). :param dt: Reference datetime (default: now, UTC). :returns: DataFrame as returned by :func:`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. .. py:function:: get_departures_with_vehicles(stop_name, n = 5, dt = None, enrich_stops = False) 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). 3. 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. :param stop_name: Stop name to search for (resolved via :func:`find_stop_id`). :param n: Number of departures to return (default 5). :param dt: Reference datetime (default: now, UTC). :param enrich_stops: 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. :rtype: DataFrame with all departure columns plus optional vehicle columns .. py:function:: get_direct_journeys(origin, destination, n = 5, dt = None) 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``). 3. Filter to trips whose scheduled origin departure is at or after *dt*, sorted by departure time. 4. Return the first *n* matching trips. :param origin: Origin stop name (resolved via :func:`~.stops.find_stop`). :param destination: Destination stop name. :param n: Maximum number of journeys to return (default 5). :param dt: Reference datetime (default: now, local Europe/London time). :returns: ``origin``, ``destination``, ``service``, ``scheduled_departure`` (HHMM str), ``scheduled_arrival`` (HHMM str), ``days``, ``direction``. :rtype: DataFrame with columns :raises TranslinkDataNotFoundError: If either stop cannot be resolved, or if no direct service runs between them at all. .. py:function:: validate_departures(df) Validate that a departures DataFrame has the expected schema and values. :param df: DataFrame as returned by :func:`get_departures`. :returns: True if validation passes. :raises TranslinkValidationError: If required columns are missing or values are invalid. .. py:function:: find_stop(query) Search for stops by name using the Translink locationApi. :param query: Partial or full stop name, e.g. ``"Cambria Street"``. :returns: ``id`` (Translink internal StopId), ``name``, ``location_type``. :rtype: List of dicts, each with keys :raises TranslinkDataNotFoundError: If the API request fails. .. py:function:: get_stop_dataframe(force_refresh = False) Return the full stop lookup as a DataFrame. Columns: ``atco_code``, ``name``, ``easting``, ``northing``, ``latitude``, ``longitude``. :param force_refresh: Re-download and rebuild the stop table. :returns: DataFrame with one row per stop, indexed by ``atco_code``. .. py:function:: get_stop_lookup(force_refresh = False) 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. :param force_refresh: 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 :rtype: Dict mapping ``ATCOCode`` (e.g. ``"700000001661"``) to a dict with keys :raises TranslinkDataNotFoundError: If the source zips cannot be downloaded. :raises TranslinkValidationError: If the resulting table is implausibly small. .. py:function:: resolve_stop_name(atco_code, fallback = True) 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. :param atco_code: NaPTAN ATCOCode, e.g. ``"700000014482"``. :param fallback: If True, attempt a live API lookup for unknown codes. :returns: Stop name string, or None if not resolvable. .. py:function:: find_direct_trips(origin_atco, dest_atco, force_refresh = False) 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. :param origin_atco: 12-digit NaPTAN ATCOCode of the origin stop. :param dest_atco: 12-digit NaPTAN ATCOCode of the destination stop. :param force_refresh: 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. .. py:function:: find_services_at_stop(stop_atco, force_refresh = False) Return all scheduled trips that call at *stop_atco*. :param stop_atco: 12-digit NaPTAN ATCOCode (e.g. ``"700000001661"``). :param force_refresh: Rebuild the trip index from source. :returns: Deduplicated list of :class:`Trip` objects, ordered by origin departure time. .. py:function:: get_trip_index(force_refresh = False) 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. :param force_refresh: 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. .. py:function:: get_live_vehicles(line = None, operator = None, enrich_stops = False) Return a snapshot of live vehicle positions from the Translink VMI feed. :param line: Optional line filter, case-insensitive (e.g. ``"11E"`` or ``"G1"``). :param operator: Optional operator filter, case-insensitive. Accepts canonical codes (``"MET"``) or VMI codes (``"TM"``). Both map to Metro. :param enrich_stops: 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). :rtype: DataFrame with one row per active vehicle. Columns :raises TranslinkDataNotFoundError: If the VMI feed cannot be reached. :raises TranslinkValidationError: If the response is malformed. .. rubric:: Example >>> vdf = get_live_vehicles(line="11E") >>> all(vdf["line"].str.upper() == "11E") True .. py:function:: validate_vehicles(df) Validate a live vehicles DataFrame. :param df: DataFrame as returned by :func:`get_live_vehicles`. :returns: True if validation passes. :raises TranslinkValidationError: If required columns are missing or coordinates are invalid.