Initial commit

Author: notsoli

.gitignore

@@ -0,0 +1,13 @@
+# Python-generated files
+__pycache__/
+*.py[oc]
+build/
+dist/
+wheels/
+*.egg-info
+
+# Virtual environments
+.venv
+
+# Data files
+data/

.python-version

@@ -0,0 +1 @@
+3.13

README.md

@@ -0,0 +1,35 @@
+# gobble-wrta
+
+gobble-wrta adapts TransitMatters' [gobble](https://github.com/transitmatters/gobble) for the Worcester Regional Transit Authority bus network. WRTA doesn't expose a standard real-time GTFS feed, so this project reads their CAD/AVL API instead and republishes bus arrivals in formats that existing transit tooling already understands:
+
+- CSV output matching the format used by the [TransitMatters Data Dashboard](https://github.com/transitmatters/t-performance-dash)
+- A GTFS Realtime stream
+
+## How it works
+
+WRTA does publish an official GTFS Realtime feed, but it only carries service alerts -- no trip updates or vehicle positions. gobble-wrta fills that gap by polling WRTA's CAD/AVL API for live vehicle positions and next-stop info, matching each vehicle against its GTFS route/trip/stop, and recording arrival and departure events from the result.
+
+### GTFS Realtime coverage
+
+Support for the spec is partial so far:
+
+**Trip updates** -- trip info (trip/route/direction), and a single stop time update for the immediate next stop.
+
+**Vehicle positions** -- vehicle ID, latitude/longitude, speed, current stop sequence, stop ID, timestamp, occupancy, and trip info.
+
+Not yet supported: trip start time, schedule relationship, vehicle license plate, vehicle odometer, congestion level, and occupancy status.
+
+## Requirements to develop locally
+
+- [`uv`](https://docs.astral.sh/uv/) with Python 3.13
+  - Ensure `uv` is using the correct Python version by running `uv venv --python 3.13`
+
+## Development Instructions
+
+1. In the root directory, run `uv sync` to install dependencies.
+2. Run `uv run src/gobble.py` to start.
+3. Output will be in `data/` in your current working directory.
+
+## Support TransitMatters
+
+If you've found this app helpful or interesting, please consider [donating](https://transitmatters.org/donate) to TransitMatters to help support our mission to provide data-driven advocacy for a more reliable, sustainable, and equitable transit system in Metropolitan Boston.

pyproject.toml

@@ -0,0 +1,17 @@
+[project]
+name = "gobble-wrta"
+version = "0.1.0"
+description = "Process WRTA events into a format that can be consumed by the Data Dashboard and an unofficial GTFS Realtime feed"
+authors = ["TransitMatters Labs <labs@transitmatters.org>"]
+readme = "README.md"
+requires-python = ">=3.13"
+license = "MIT"
+dependencies = [
+    "gtfs-realtime-bindings>=2.1.0",
+    "pandas>=3.0.3",
+]
+
+[dependency-groups]
+dev = [
+    "pandas-stubs>=3.0.3.260530",
+]

src/disk.py

@@ -0,0 +1,63 @@
+import csv
+import pathlib
+from typing import Any
+
+import pandas as pd
+
+import util
+from logger import set_up_logging
+
+logger = set_up_logging(__name__)
+
+MAIN_DIR = pathlib.Path("./data/")
+
+# column order matches the data dashboard's expected CSV schema
+CSV_FIELDS = [
+    "service_date",
+    "route_id",
+    "trip_id",
+    "direction_id",
+    "stop_id",
+    "stop_sequence",
+    "vehicle_id",
+    "vehicle_label",
+    "event_type",
+    "event_time",
+    "scheduled_headway",
+    "scheduled_tt",
+    "vehicle_consist",
+    "occupancy_status",
+    "occupancy_percentage",
+]
+
+
+def write_events(events_df: pd.DataFrame) -> None:
+    for row in events_df.itertuples():
+        _write_row(row)
+
+
+def _write_row(row) -> None:
+    dir_path = MAIN_DIR / util.output_dir_path(
+        row.route_id, row.direction_id, row.stop_id, row.event_time
+    )
+    dir_path.mkdir(parents=True, exist_ok=True)
+    file_path = dir_path / "events.csv"
+
+    write_header = not file_path.exists()
+    with open(file_path, "a", newline="") as f:
+        writer = csv.DictWriter(f, fieldnames=CSV_FIELDS)
+        if write_header:
+            writer.writeheader()
+        writer.writerow(_row_to_csv_dict(row))
+
+
+def _row_to_csv_dict(row) -> dict[str, Any]:
+    d: dict[str, Any] = {}
+    for field in CSV_FIELDS:
+        value = getattr(row, field)
+        if field == "event_time" and isinstance(value, pd.Timestamp):
+            value = value.isoformat()
+        elif pd.isna(value):
+            value = ""
+        d[field] = value
+    return d

src/event.py

@@ -0,0 +1,150 @@
+from datetime import datetime
+
+import pandas as pd
+
+import disk
+import gtfs
+import lookups
+import trip_state
+import util
+import vehicle_snapshot
+import vehicles
+from logger import set_up_logging
+from trip_state import VehicleState
+
+logger = set_up_logging(__name__)
+
+
+def process_vehicle_update(
+    vehicle: vehicles.Vehicle,
+    current_lookups: lookups.Lookups,
+    gtfs_data: gtfs.GTFS,
+) -> None:
+    """Advances one vehicle's tracked trip state by a single poll's worth of new data: detects
+    whether it has passed its previously-seen next stop, emits an ARR/DEP event pair if so, and
+    refreshes its live snapshot for the GTFS-RT feed either way."""
+    route_id = current_lookups.route_id_by_ligne.get(vehicle.id_ligne)
+    if route_id is None:
+        return
+    direction_id = lookups.direction_id_for_destination(route_id, vehicle.destination)
+
+    now = datetime.now(util.EASTERN_TIME)
+    state = trip_state.get_state(vehicle.numero_equipement)
+
+    if (
+        state is None
+        or state.route_id != route_id
+        or state.direction_id != direction_id
+        or util.service_date(state.updated_at) != util.service_date(now)
+    ):
+        state = VehicleState(
+            trip_id=None,
+            trip_id_is_synthesized=False,
+            route_id=route_id,
+            direction_id=direction_id,
+            stop_sequence=0,
+            prev_stop_name=None,
+            updated_at=now,
+        )
+
+    current_name = vehicle.arret_suiv_name
+    if (
+        state.prev_stop_name is not None
+        and current_name is not None
+        and state.prev_stop_name != current_name
+    ):
+        stop_id = lookups.resolve_stop_id(
+            current_lookups,
+            route_id,
+            direction_id,
+            state.prev_stop_name,
+            vehicle.lat,
+            vehicle.lng,
+        )
+        if stop_id is not None:
+            _emit_event(
+                current_lookups, gtfs_data, state, route_id, direction_id, stop_id, vehicle, now
+            )
+
+    if current_name is not None:
+        state.prev_stop_name = current_name
+    state.updated_at = now
+    trip_state.set_state(vehicle.numero_equipement, state)
+
+    vehicle_snapshot.update(vehicle, route_id, direction_id, state, current_lookups, now)
+
+
+def _synthesize_trip_id(
+    vehicle: vehicles.Vehicle, route_id: str, direction_id: int, event_time: datetime
+) -> str:
+    return (
+        f"{vehicle.numero_equipement}-{route_id}-{direction_id}-"
+        f"{event_time:%Y%m%dT%H%M%S}"
+    )
+
+
+def _emit_event(
+    current_lookups: lookups.Lookups,
+    gtfs_data: gtfs.GTFS,
+    state: VehicleState,
+    route_id: str,
+    direction_id: int,
+    stop_id: str,
+    vehicle: vehicles.Vehicle,
+    event_time: datetime,
+) -> None:
+    if state.trip_id is None:
+        matched_trip_id = lookups.match_trip_id(
+            current_lookups, route_id, direction_id, stop_id, event_time
+        )
+        if matched_trip_id is not None:
+            state.trip_id = matched_trip_id
+            state.trip_id_is_synthesized = False
+        else:
+            state.trip_id = _synthesize_trip_id(vehicle, route_id, direction_id, event_time)
+            state.trip_id_is_synthesized = True
+            logger.warning(
+                f"No GTFS trip match for vehicle={vehicle.numero_equipement} "
+                f"route={route_id} direction={direction_id} stop={stop_id}; "
+                f"using synthesized trip_id={state.trip_id}"
+            )
+
+    if state.trip_id_is_synthesized:
+        state.stop_sequence += 1
+    else:
+        resolved_sequence = lookups.stop_sequence_for(
+            current_lookups, state.trip_id, stop_id, state.stop_sequence
+        )
+        state.stop_sequence = (
+            resolved_sequence if resolved_sequence is not None else state.stop_sequence + 1
+        )
+
+    base_row = {
+        "service_date": util.service_date_iso8601(event_time),
+        "route_id": route_id,
+        "trip_id": state.trip_id,
+        "direction_id": direction_id,
+        "stop_id": stop_id,
+        "stop_sequence": state.stop_sequence,
+        "vehicle_id": "0",
+        "vehicle_label": vehicle.numero_equipement,
+        "event_time": event_time,
+        "vehicle_consist": vehicle.numero_equipement,
+        "occupancy_status": None,
+        "occupancy_percentage": vehicle.taux_remplissage,
+    }
+    event_df = pd.DataFrame(
+        [
+            {**base_row, "event_type": "ARR"},
+            {**base_row, "event_type": "DEP"},
+        ]
+    )
+
+    enriched = lookups.add_gtfs_headways(event_df, gtfs_data.trips, gtfs_data.stop_times)
+    disk.write_events(enriched)
+
+    logger.info(
+        f"{event_time.isoformat()} ARR/DEP vehicle={vehicle.numero_equipement} "
+        f"route={route_id} direction={direction_id} stop={stop_id} "
+        f"trip_id={state.trip_id!r} stop_sequence={state.stop_sequence}"
+    )

src/gobble.py

@@ -0,0 +1,15 @@
+from poll_loop import poll_vehicles_forever
+from gtfs import start_watching_gtfs
+from rt_server import start_server
+from topology import start_watching_topology
+
+
+def main():
+    start_watching_gtfs()
+    start_watching_topology()
+    start_server()
+    poll_vehicles_forever()
+
+
+if __name__ == "__main__":
+    main()