feat(realtime): poll HTTP telemetry sources and publish vehicle positions - #61
Open
dotjae wants to merge 17 commits into
Open
feat(realtime): poll HTTP telemetry sources and publish vehicle positions#61dotjae wants to merge 17 commits into
dotjae wants to merge 17 commits into
Conversation
Vendored inference half of the ETA model lifecycle (canonical training source lives in gtfs-django/eta_prediction): estimator, feature engineering, and the model registry loader, with heavy deps slimmed (xgboost is an optional extra). The registry resolves model/metadata paths relative to its directory, so it is relocatable: bind-mountable, checked-in placeholders, or models written by an external retraining suite all load regardless of MODEL_REGISTRY_DIR. Wired into databus as an editable uv workspace member. See backend/gtfs-eta/README.md for provenance and extraction intent.
Deterministic synthetic baseline (polyreg_distance, global) so the stop-time producer runs end-to-end on a fresh checkout without a trained model. Kept as a standalone commit so it is trivial to drop once the retraining suite supplies real models. Regenerate with: python -m gtfs_eta.seed_baseline_model.
Replace the fake_stop_times placeholder with a real producer that calls gtfs_eta.estimate_stop_times. Pure/impure split: compute_stop_time_updates derives contract entries from run state + shape geometry; produce_stop_times does the Redis I/O. Upcoming stops come from the monotonic shape geometry and distances feed the estimator's precomputed-distance hook, fixing duplicate stop_sequences and the non-decreasing upcoming count. Builder sorts/dedups StopTimeUpdate entries defensively. New config: MODEL_REGISTRY_DIR, ETA_MAX_STOPS, ETA_DEFAULT_UNCERTAINTY_S.
Cover the pure helper, both bug regressions (no duplicate stop_sequence; non-increasing upcoming count), output-adapter edges, and the impure producer's Redis read/skip/write guards.
Superseded by the real ETA producer. Removes the fabricated stop-time generator and its static route_stops.csv (whose 0-based sequences surfaced an off-by-one).
Generates the pending schema for the Sensor model (moved provides_* flags off Equipment/EquipmentLog) plus the Company.linked_agency ManyToMany conversion and Vehicle.status SOLD removal already present in models.py.
The telemetry publisher/consumer read these from the environment but they were never declared in the env files.
Introduces realtime_engine/sources/ — a registry of source adapters that fetch vehicle position data and normalize it into the shared position telemetry contract: - transforms.py: unit conversions (km/h->m/s, km->m) and Costa Rica local timestamp parsing ported from navsat-bridge, plus a dotted-path getter. - base.py: SourceAdapter protocol and a string-keyed adapter registry. - http_json.py: a generic HTTP+JSON adapter (kind="http") driven entirely by a sensor's source_http_url/source_json_mapping, so new HTTP feeds (e.g. NavSat) need no per-provider code. - publisher.py: paho-mqtt v2 publishing of normalized positions to transit/vehicle/<id>/position, matching the existing ingestion consumer.
DB-free unit tests for realtime_engine/sources/: unit conversions and timestamp parsing, the HTTP+JSON adapter (array and single-object bodies, unit conversion, malformed-record skipping, vehicle_id resolution from the mapped path vs. sensor fallback, and contract compliance with position.validate_for_write), and the MQTT publisher (topic/payload shape, paho v2 client setup, batch publish/disconnect, per-message error handling). Sensors are faked with SimpleNamespace and requests.get/paho are monkeypatched — no HTTP, MQTT, or database access.
Polls ACTIVE Sensor rows configured for HTTP position feeds (source_type "http" or "both"), fetches readings via the pluggable adapter registry, keeps only readings for vehicles currently running (per runs:in_progress), and publishes the survivors on transit/vehicle/<id>/position. Per-sensor failures are caught and logged so one bad source can't sink the poll.
Wires the new HTTP telemetry poll into beat alongside the other realtime_engine periodic tasks.
fetch_position, fetch_and_publish, and update_gtfs_realtime were an early sketch superseded by the pluggable HTTP source adapters and the new realtime_engine.tasks.fetch_positions task. Removes the now-unused requests/paho/Vehicle/chord/group imports along with them; the legitimate builder tasks are untouched.
operations/migrations was force-committed against the repo's convention of gitignoring migrations/ and regenerating them at container start, and it was missing from that regen list. Untracks the committed migration (it stays on disk, now gitignored like its siblings) and adds "operations" to APPS_TO_MIGRATE in docker-entrypoint.sh.
Building the in-service set from runs:in_progress deadlocked HTTP-only vehicles: a run only enters IN_PROGRESS once telemetry proves the vehicle is moving, but that telemetry is exactly what fetch_positions delivers, and it refused to publish until the run was already IN_PROGRESS. Gate instead on vehicle:<id>:current_run presence -- the same signal the MQTT consumer uses to accept telemetry -- so a CONFIRMED run bootstraps forward.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
realtime_engine/sources/) with a generic HTTP+JSON adapter driven entirely by a Sensor'ssource_http_url+source_json_mapping— no per-provider code needed for feeds that fit the mapping schema.fetch_positionsCelery task: polls ACTIVE HTTP/both sensors every 10s, filters to in-service vehicles, and republishes positions ontotransit/vehicle/<id>/position(the same MQTT topic/consumer used by direct MQTT telemetry).operations.Sensorwithsource_type/source_http_url/source_json_mappingfields to configure per-sensor HTTP sources.vehicle:<id>:current_run(same signal the MQTT consumer uses) rather thanruns:in_progress, which deadlocked HTTP-only vehicles — a run only reaches IN_PROGRESS once telemetry proves movement, and that telemetry is exactly what this task delivers.operations/migrationsis now regenerated at container start like the repo's other apps (was previously force-committed, missing from the regen list).Dependency note
This branch is stacked on the still-open #60 (
feat/eta-stop-times), so the diff currently includes those commits too. Please merge #60 first — once it lands onmain, this PR's diff will automatically collapse to just the fetch-telemetry commits.Test plan
pytest realtime_engine runs operations schedule_engine— 369 passed (run inside the dev container)