Production-readiness: telemetry, run lifecycle, GTFS-RT, and full cleanup - #61
Open
dotjae wants to merge 70 commits into
Open
Production-readiness: telemetry, run lifecycle, GTFS-RT, and full cleanup#61dotjae wants to merge 70 commits into
dotjae wants to merge 70 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).
Generate a valid GTFS Schedule feed.zip from the feed app's ORM models (loaded from feed/fixtures/gtfs.json) and serve it at /feed/schedule/feed.zip, mirroring the GTFS-RT pipeline. - feed/schedule/exporter.py: build_gtfs_zip() serializes one Feed's rows into GTFS .txt files (columns derived by model introspection) and publish_gtfs_zip() writes feed/files/gtfs.zip atomically. - schedule_engine.tasks.build_schedule: Celery task + daily beat entry. - feed export_gtfs management command for on-demand/boot generation. - docker-entrypoint.sh: export on boot only if gtfs.zip is absent. - feed.views.schedule: serve gtfs.zip, 404 when not yet generated. - Minimal exporter test; pytest-django settings in pyproject.toml.
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.
… feat/fetch-telemetry # Conflicts: # backend/databus/celery.py
Keep project_point_to_polyline in progression/shapes.py with a noqa: compute.py calls it via module-attribute access, which ruff cannot see.
Replace the vendored backend/gtfs-eta/ uv workspace member with a path dependency on the sibling simovilab/gtfs-eta checkout, which now ships the gtfs_eta alias namespace, estimate_stop_times(shape=), and the seed_baseline_model entry point that runs/domain/progression/stop_times.py already lazy-imports. uv's path-source normalization rejects a literal "../../gtfs-eta" once resolved inside the container (/app sits one level below the container's filesystem root), so the same relative path can't work identically on host and in-container. Instead backend/gtfs-eta is a committed symlink to ../../gtfs-eta: uv only normalizes the bare "gtfs-eta" segment, and the OS resolves the symlink's own ".." traversal, clamping at root instead of erroring. compose.dev.yml bind-mounts ../gtfs-eta:/gtfs-eta (read-write, since uv's setuptools editable build touches gtfs_eta.egg-info/ in the source tree) into the four backend-based services so the mount target matches what the symlink resolves to. uv sync runs at container start (docker-entrypoint.sh), after that mount exists, so no build-context change was needed. Also: add xgboost as a direct databus dependency (models/__init__.py in gtfs-eta unconditionally imports it), and exclude the gtfs-eta symlink from ruff/pytest collection since it now points at gtfs-eta's own full codebase and test suite rather than a trimmed vendored subset.
The old artifacts were seeded by the vendored gtfs-eta and referenced module paths that no longer exist; the new registry stores paths relative to the registry dir so the same files work in-container.
…to WARNING Lifecycle detectors (RunStartedDetector, RunTrackingStartedDetector, RunCompletedDetector, RunTrackingRestoredDetector, RunTrackingLostDetector, RunTrackingExpiredDetector) already gate on the run's current run_lifecycle_state before returning a DetectionResult, so a detection cannot itself re-fire an event the run has already outgrown. The observed flood instead comes from a race between concurrent detections for the same run: two telemetry pings can both read the pre-transition state before the first one's run_lifecycle_event task lands, so both enqueue the same event. The second is a harmless no-op by the time it runs — the run already reached the event's target state — but RunLifecycleService.process_event rejects it the same way it would reject a genuinely invalid transition, and the task logged every rejection at ERROR with a full traceback. Add RunLifecycleService's RunLifecycleError.errors to include the run's current_state, and a target_state_for_event() lookup over TRANSITIONS. run_lifecycle_event now compares the two: if the rejection's current_state already equals the fired event's own target state, it's an idempotent re-fire and logs a concise WARNING instead of ERROR+traceback. Genuine invalid transitions are unaffected.
…run_id
_load_run now raises RunLifecycleError("...missing run_id") instead of
letting a missing run_id reach Run.objects.get(id=None), which never
matches and surfaces as a confusing Run.DoesNotExist. Existing
RunLifecycleError handlers in api/views.py and
realtime_engine/tasks.py already catch it, so no unhandled 500s.
Fills ruff D1xx gaps (module/class/method docstrings) across the fleet/operator/vehicle/equipment domain models and their admin/apps/views scaffolding. Adds full return-type hints to __str__/save methods and an explicit ManyToManyField[Agency, Any] annotation on Company.linked_agency, resolving the one mypy var-annotated error the django-stubs plugin raises for that field (its implicit through-model type parameter isn't otherwise inferable). No runtime behavior changes; full pytest suite still 432 passed.
Fills ruff D1xx gaps (module/class/function docstrings) across the website app's admin/apps/models/urls/views scaffolding and adds full param/return type hints to the index view. No runtime behavior changes; full pytest suite still 432 passed.
Fills ruff D1xx gaps in celery.py (module docstring documenting the beat schedule -- fetch-positions every 10s with expires=10, the two GTFS-RT feed builders every 15s, the stale-run scan every 30s, and the daily schedule rebuild -- plus a docstring on debug_task). Fixes both mypy arg-type errors in urls.py by giving urlpatterns an explicit list[URLPattern | URLResolver] annotation, since static()'s return type (list[URLPattern]) differs from the include()-based path() entries' inferred list[URLResolver]; both are valid urlpatterns entries at runtime. No runtime behavior changes, and settings.py values are untouched; full pytest suite still 432 passed.
Fills ruff D1xx gaps (module/function docstrings) across cleanup_runs.py's connection helpers, DB/Redis purge functions, and CLI driver. Fixes all 14 mypy findings: adds psycopg2.extensions.connection/ cursor annotations (get_db, _count, _delete, db_purge_*) and redis-py Awaitable-union cast helpers (_get/_hgetall/_smembers/_rkeys/_srem, mirroring realtime_engine/tasks.py's _get/_smembers/_hgetall but taking the client as a parameter since this script builds one per invocation rather than sharing a module-level client) for the hgetall/get/smembers/ keys/srem calls in the Redis purge helpers; adds `assert conn/r is not None` narrowing in main()'s dispatch block, since both are only ever None when their connect step already called sys.exit(1). Also drops two extraneous f-string prefixes (F541) on plain error-tip strings. No runtime behavior changes; full pytest suite still 432 passed.
WhichShapesView queried RouteStop.route/.shape, which don't exist (the model's fields are linked_route/linked_shape), and asked GeoShape for a direction_id it doesn't have. FindTripsView queried TripTime.trip_time, which doesn't exist (the field is departure_time), and matched TripTime to Trip by the bare trip_id string instead of the linked_trip FK, which can cross-match same-numbered trips across feeds. Both endpoints back the run-registration UI cascade (pick a route -> its shapes -> candidate trips with run lifecycle state) and had never worked.
CompanySerializer declared a PrimaryKeyRelatedField for "agency", but the Company model has no such field -- it's linked_agency (M2M to Agency). Dropping the stale explicit field declaration (plus a duplicated Meta.model line) lets HyperlinkedModelSerializer introspect the model correctly, fixing the live /api/company/ endpoint.
is_run_validated was a no-op returning True, leaving a race window between VALIDATE_RUN and INITIALIZE_RUN: a vehicle/trip/operator could be claimed by another run, or the nightly build_schedule feed rotation could drop the run's trip, in between the two events. The guard now re-delegates to is_vehicle_available/is_trip_available/is_operator_available (idempotent re-fires where this run already holds its own claims still pass) and re-confirms the trip exists in whichever feed is current now, raising RunLifecycleError with field-keyed detail on failure. Transition table and guard wiring are unchanged -- all fixed inside the guard itself.
…ainst current code
Delete never-collected test artifacts (pytest only collects test_*.py,
so none of these ever ran):
- operations/tests.py, runs/tests.py, schedule_engine/tests.py,
website/tests.py: 1-line empty Django app stubs
- api/tests.py: stale test asserting a "Hello world!" response from a
reverse("runs") route that no longer exists
- realtime_engine/tests.py: broken import of end_run, initialize_run,
register_run, validate_run from realtime_engine.tasks — none of
which exist anymore
- tests/: entire directory removed (api_tester.py was a one-off manual
script, backup.md was old scratch notes)
tests/fake_trip.py contained a hardcoded API token committed to git
history. That token must be rotated.
feed/tests.py is untouched — it holds real exporter tests.
LoginSerializer and RunSerializer have zero references outside this file (grep-verified); CreateRunSerializer and RunUpdateSerializer remain in use and are untouched. PositionSerializer had a duplicated vehicle field declaration, a duplicated "vehicle" entry in fields, and a stale commented-out create() block. VehicleStopStatusSerializer, CongestionLevelSerializer, and OccupancyStatusSerializer each had fields = "__all__" written twice in Meta. Also dropped the now-unused Run import. No behavior changes beyond removing dead/duplicate declarations.
realtime_engine.tasks (module and process_position_update) were imported mid-file for the task-level test section, triggering two E402 violations. Moved both imports to the top-level import block; the section comment marking the task-level tests stays in place.
pytest only collects test_*.py, so backend/feed/tests.py (smoke tests for build_gtfs_zip) was never run. Move it to backend/feed/tests/test_schedule_exporter.py following the tests/ package convention already used by other apps.
compose.prod.yml starts Redis with --requirepass, but no client in the codebase passed a password, so production auth would fail. Introduce databus/redis_client.py, a small dependency-light factory (plain os.getenv, no Django imports) that builds every Redis client from REDIS_HOST/REDIS_PORT/REDIS_PASSWORD, treating an empty password as unset so dev without auth keeps working. Route all inline redis.Redis(...) call sites through it, including the two lifecycle modules that hardcoded host="state", and wire the same password into Channels' CHANNEL_LAYERS via the redis:// URL form.
Replace the warning about REDIS_PASSWORD not being consumed by any Redis client with an accurate statement: all clients now go through databus/redis_client.py and honor the password when set.
Both are annotation-level, behavior-neutral: shapes.py's stop_lat/lon DecimalField(null=True) columns are read via .values() with no null filter, so float() already assumed non-null at runtime; annotate that assumption explicitly rather than changing the query. guards.py's short-turn terminal-stop check already guarantees .last() is non-None via the preceding .exists() check; cast makes that explicit instead of leaving mypy to flag a false positive.
dotjae
changed the title
Add an hourly Celery task that keeps the stored GTFS Schedule in sync with each active provider's upstream feed, mirroring infobús's get_schedule flow. - feed/schedule/importer.py: import_schedule_if_changed(provider) HEAD-checks the provider's schedule_url ETag against the current Feed.http_etag, skips when unchanged, and otherwise imports the zip's 9 core GTFS tables via bulk_create. The whole DB mutation (flipping the prior current feed, creating the new Feed, importing every table) runs in a single transaction.atomic() so a failure rolls back cleanly. Missing/blank non-nullable columns are filled with each field's empty default (0/""), covering feeds that omit optional columns (e.g. pickup_type/drop_off_type); Stop.stop_point is built inline from lat/lon; malformed rows are skipped. - schedule_engine/tasks.py: fetch_schedule() iterates active GTFSProviders with per-provider error isolation and returns an updated/unchanged/errored summary. - databus/celery.py: fetch-schedule-hourly-at-30 beat entry (crontab minute=30). - feed/tests/test_schedule_importer.py: 12 tests (new import, unchanged ETag, missing-header fallback, atomic rollback, lean-feed missing columns, etc.). Verified end-to-end against the live bUCR feed: full lossless import (1 route, 122 trips, 22 stops, 1043 stop_times), with re-run correctly detecting the unchanged ETag. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BMNbeYrs2LKHpRoF5UFeYD
…ncy and test the GTFS Schedule importer
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
This branch brings Databús to a production-ready state for the upcoming public release. It consolidates the outstanding feature work (HTTP telemetry ingestion, the run-lifecycle FSM + MQTT/AMQP pipeline, GTFS Schedule zip publishing, and ETA stop-time predictions) onto a single branch, then applies a six-phase cleanup so the repository is documented, typed, dead-code-free, and enforceably clean.
Scope: 110 commits · 282 files · +21,954 / −8,371.
Feature work (consolidated)
fetch_positionspolls in-service sensors (gated onvehicle:<id>:current_run), republishes onto thetransit/vehicle/<id>/positionMQTT topic; starvation fix (pre-fetch filter,expires,soft_time_limit) took the task from blocking all pool slots to ~0.02 s.runs/domain/lifecycle/, detection/progression layers, an in-worker MQTT consumer bootstep, and idempotent re-fire handling.databus.events, routing keysruns.lifecycle.<event>, versioned JSON envelope, fire-and-forget/log-and-drop so telemetry never blocks on the broker.feed.schedule.exporter.build_gtfs_zipservesschedule/feed.zip; realtime feeds (vehicle_positions/trip_updates, JSON + protobuf) served via DjangoFileResponse.gtfs-etaintegrated through the lazy-import seam inruns/domain/progression/stop_times.py; vendored copy replaced with an editable path dependency on the sibling repo.Production-readiness cleanup (Phases 1–6)
tests.pystubs and scratch scripts, unused serializers.D1(pep257) + mypy (django-stubs) wired intomake lint/make typecheck, so docstring and type-hint coverage is objectively gated.docs/content/pages audited against the code; the AMQP publisher, beat schedule, telemetry paths, and data model brought current; productionsite_urlset; site builds with no issues.databus/redis_client.pyfactory that honorsREDIS_PASSWORDat every client site and inCHANNEL_LAYERS(URL-encoded), and the last in-repo mypy stragglers cleared.Test plan
make lint(ruff incl. D rules) — clean (run on host).make typecheck(mypy in docker) — clean in tracked code (only vendoredgtfs-django/errors remain).docker compose -f compose.dev.yml run --rm orchestrator uv run pytest -q— 448 passed./feed/*endpoints serve valid GTFS-RT; beat tasks cycle with zero errors; an MQTT position flows through the new Redis factory and is gated correctly.Known items to resolve before merge
These are tracked and intentionally not blocking the PR from being opened for review:
backend/tests/fake_trip.py— it remains recoverable from git history.compose.prod.ymlhas no mount and there is no PyPI package yet; the production import path does not exist until the packaging decision lands. (MODEL_REGISTRY_DIR/ETA_*prod env vars are pending the same decision.)RunStartedDetectordepends on speed telemetry — a vehicle whose source never reports speed can never leave TRACKING → IN_PROGRESS; worth confirming against real telemetry sources.runs:in_progress— a silent vehicle keeps appearing in GTFS-RT output until a terminal transition; documented, behavior to confirm.