diff --git a/docs/concepts/terminology.md b/docs/concepts/terminology.md index f8b3ebb5..71de5e56 100644 --- a/docs/concepts/terminology.md +++ b/docs/concepts/terminology.md @@ -16,6 +16,9 @@ Canonical names. Use these in code and docs. | **feed** | agency (in v2 UX) | A feed may carry multiple agencies; see [feeds.md](feeds.md) | | **ETA** | arrival time, time-to-arrive | "ETA" is short and unambiguous | | **bucket** | status, state | Specific term for station-view arrival classification | +| **frequency-based trip** | repeating trip, headway trip, rec-trip | GTFS `frequencies.txt` model: a trip whose stop_times are reused as offsets for N generated departures, one per `headway_secs` within the frequency window. See [vehicle.md](vehicle.md#frequency-based-trips). | +| **anchor trip** | base trip, source trip | The single `trips.txt` row whose `stop_times` carry the offsets for a frequency-based service. The one entry that physically exists in `trips` + `stop_times`; the per-departure rows are derived. | +| **generated departure** | synthetic departure, expansion row | One per-departure row produced by the `frequencyExpansion` helper from a single anchor trip. Identified by `Vehicle.id = "trip:@"` and `schedule.tripStartMin = effectiveStartMin`. | ## Technical diff --git a/docs/concepts/vehicle.md b/docs/concepts/vehicle.md index a7b3d470..6a173875 100644 --- a/docs/concepts/vehicle.md +++ b/docs/concepts/vehicle.md @@ -74,6 +74,48 @@ Orthogonal to `kind` and to [arrival-buckets](arrival-buckets.md). A the depot already broadcasting; `last` and `on-route` rows almost always do. +## Frequency-based trips + +For trips with rows in `frequencies.txt` (the [GTFS spec's +headway-based service model](https://gtfs.org/schedule/reference/#frequenciestxt)), +the active set contains one `Vehicle` per **generated departure** — +not one per anchor trip. A 15-minute headway running 05:05–22:40 +produces 71 `kind: 'scheduled'` rows, each with a distinct +`schedule.tripStartMin` (the k-th departure's effective origin time). +`tripPhase` classification still works unchanged because it's keyed +on `tripStartMin`. + +**Identification.** The `Vehicle.id` for a generated row is +`trip:@` (the anchor's id without the +suffix, `trip:`, is kept for the original anchor entry where +present). The `@` suffix is what makes the id stable across +polls and unique per generated departure. + +**Reconciler match key.** Live observations match by the composite +`(tripId, tripStartMin)` key, not by `tripId` alone. The +`enrichObservations` index uses `${tripId}|${tripStartMin}` so an +`UNSCHEDULED` observation on a frequency-based trip (which carries +the *specific* generated departure's effective start time per the +[GTFS-RT contract](specs/gtfs-rt-contract.md)) resolves to the +correct generated row. For observations missing `startTime`, a +fallback to `tripId`-only matching preserves the legacy lenient +behaviour for non-conforming producers. + +**Per-stop promotion.** The station-board merge in +[`stationBoard.ts`](../../src/lib/domain/stationBoard.ts) uses the +same composite key. A frequency-based trip's N per-stop rows each +match against the N active-set rows by `tripStartMin`; a tolerance +filter rejects per-stop rows whose `tripStartMin` doesn't match the +reconciled row's, so the same GPS position doesn't get applied to +every per-stop entry of an anchor trip. + +**Schedule-only fallback.** The publisher's `SCHEMA` adds the +`frequencies` table in +[n3ary/gtfs-publisher#252](https://github.com/n3ary/gtfs-publisher/pull/252); +cached blobs that pre-date that change report `false` from +`hasFrequenciesTable()` and the per-time query modules fall back to +schedule-only behaviour without throwing. + ## Visual rendering — kind dot The `VehicleCard` shows a small dot on the far right encoding `kind` diff --git a/docs/plan/gtfs-frequencies.md b/docs/plan/gtfs-frequencies.md new file mode 100644 index 00000000..1ae4969a --- /dev/null +++ b/docs/plan/gtfs-frequencies.md @@ -0,0 +1,222 @@ +# GTFS frequencies.txt support + +Issue: #347. Closes when the app consumes +`frequencies.txt` to expand frequency-based trips into per-departure +rows in the active set, the station board, the schedule view, the +map view, and the weekly pattern. The publisher-side prerequisite +lives in [n3ary/gtfs-publisher#252](https://github.com/n3ary/gtfs-publisher/pull/252) +(merged first; the DDL addition is what makes the table present in +the SQLite blob the app downloads). + +## What "frequency-based trip" means in this app + +A row in `frequencies.txt` says: "for this `trip_id`, run every +`headway_secs` seconds from `start_time` to `end_time`, applying the +anchor trip's `stop_times` as offsets relative to each departure +time". The cluj-napoca adapter emits these rows for `*-range` +annotations (e.g. M26 `05:05-22:40` / `10-20min` is the live case; +see +[gtfs-adapters#…/cluj-napoca/docs/known-limitations.md](https://github.com/n3ary/gtfs-adapters)). + +In the app, a frequency-based trip becomes **N `kind: 'scheduled'` +rows in the active set, one per generated departure**, each with +its own `schedule.tripStartMin` (the k-th departure's effective +origin time) and its own `Vehicle.id` +(`trip:@`). The reconciler already +matches on `(routeId, directionId, tripStartMin)`; the active set +just needs more rows. + +## Design + +### Data model + +No new domain types. The discriminated union at +`src/lib/domain/types.ts:188-217` is unchanged. A frequency-based +trip's N generated rows are N `kind: 'scheduled'` Vehicles with +distinct `schedule.tripStartMin` values and distinct `id` strings +(the `@` suffix). This is the same shape as the existing +schedule-based rows; the only difference is the id suffix and the +fact that the row's `tripStartMin` doesn't equal the anchor's +`stop_times[0].departure_time`. + +The `id` encoding choice — `trip:@` — +is a public contract: the per-stop promotion path in +`stationBoard.ts` uses Vehicle.id as a stable Svelte key, and the +reconciler's matched-scheduled index in +`stationBoard.ts:mergeReconciledIntoStationBoard` keys off the +reconciled row's `schedule.tripStartMin` (matched against the +per-stop row's `schedule.tripStartMin`) to gate the kind: 'tracked' +promotion. Without the per-row `tripStartMin` match, every per-stop +row for an anchor trip would get the same GPS position. + +### Helper module + +`src/lib/workers/gtfs/queries/frequencyExpansion.ts` is the +shared core. Five exports: + +| Export | Purpose | +|---|---| +| `hasFrequenciesTable(db)` | PRAGMA probe. Returns false on cached blobs that pre-date the publisher's DDL addition; callers fall back to schedule-only behaviour. | +| `getFrequenciesForServices(db, serviceIds)` | SQL query joining `frequencies` to `trips.service_id`. Filters out `exact_times=1` (rare, per spec). | +| `expandFrequencyToDepartures(freq, windowStart, windowEnd)` | Pure JS. One `GeneratedDeparture` per k-th departure in the window. The end-time bound is exclusive per spec ("up to but not including end_time"). | +| `expandFrequenciesToDepartures(freqs, windowStart, windowEnd)` | Convenience wrapper. Returns `Map`. | +| `getAnchorStopTimes(db, tripId)` | Per-trip stop_times rows in `stop_sequence ASC` order. Used by the per-stop expansion path to derive per-stop offset times. | + +The expansion is pure JS (no recursive SQL CTE) because (a) the +window is small (typical M26 case: 71 generated departures in a +17.5-hour window) and (b) keeping it pure makes the load-bearing +function unit-testable without a DB fixture. + +### Per-time query changes + +Every per-time query gains a `hasFrequencies: boolean` parameter +(passed in by the worker from +`state.currentFeedHasFrequencies`). When true, after the existing +SQL scan, the query calls `getFrequenciesForServices` + the +expansion helpers and merges the generated rows with the +schedule-based rows. The five queries touched: + +| Query | Expansion model | +|---|---| +| `getActiveTrips` | One `Vehicle` per generated departure. `schedule.tripStartMin` = effectiveStartMin, `schedule.scheduledArrival` = anchor.trip_end_time + k*headway. | +| `getStationArrivals` | One `ScheduleRow` per generated departure whose effective time at THIS stop falls in the query window. The effective per-stop arrival is `anchor.stop_times[stop_id].arrival_time + k*headway_min`. The row's `id` is `trip:@`. | +| `getRouteSchedule` | One `ScheduleTrip` per generated departure, with the same composite key. | +| `getActiveRouteIdsInWindow` | Boolean: a route is "active right now" iff any frequency row on it overlaps the query window. Per-row expansion is unnecessary — the route set is all we return. | +| `getRouteMapView` | Same as `getActiveTrips` but for the per-(route, direction) view; shape_id is shared per anchor (shape doesn't change per generated departure). | +| `getWeeklySchedule` | Expand each frequency row into synthetic minute slots for each matching day pattern. Headway 15 min × 17.5 h on weekdays = 67 synthetic minutes added to the `weekday` set. | + +The frequency-based rows are appended to the schedule-based rows +in the same `Vehicle[]` (or `ScheduleTrip[]`, etc.) and the rest of +the pipeline is unaware. `scanSchedule` accepts an optional +`ScheduleRow.id` override so the per-stop rows get the right +`@`-suffixed id (the default `trip:` is used when +undefined). + +### Reconciler + +No change. The reconciler at `src/lib/domain/reconcile.ts:37-203` +matches on `(routeId, directionId, tripStartMin)` — the +`tripStartMin` field is already on the active set rows, and the +generated rows have distinct values. The match tolerance +(`computeTolerance`, lines 212-235) works unchanged; a +15-minute-headway cohort's median gap is 15 min, so the median/2 +tolerance is 7-8 min, which is wider than any reasonable +observation drift. + +### `enrichObservations` composite key + +`src/lib/domain/enrichObservations.ts:10-19` previously indexed +the active set by `tripId` alone. With multiple generated rows per +tripId, that map would collapse to the last-written entry and every +observation would match the wrong generated departure. Fixed by +keying on `${tripId}|${tripStartMin}` (the primary index) and +keeping a `tripId`-only fallback for observations missing +`startTime` (preserves the legacy lenient behaviour for +non-conforming producers; for frequency-based trips, the fallback +resolves to the k=0 entry, which is the anchor's first departure). + +### `stationBoard` promotion tolerance + +`src/lib/domain/stationBoard.ts:276-297` previously promoted every +per-stop row whose `tripId` matched a reconciled row to `kind: +'tracked'`. With frequency expansion, that's wrong — the +reconciled row carries ONE specific generated departure's position; +the per-stop set has N rows (one per generated departure). Added a +`tripStartMin` equality check that gates the promotion. Trivial +change; non-frequency trips pass the check trivially because the +anchor's `tripStartMin` is identical to the active-set entry. + +### Bootstrap PRAGMA probe + +`src/lib/workers/gtfs/bootstrap.ts:385-393` runs a `sqlite_master` +probe for the `frequencies` table after the `stop_times` integrity +check. The result is stashed in `state.currentFeedHasFrequencies` +(`src/lib/workers/gtfs/state.ts:32-36`). `closeCurrent()` resets +the flag to `false`. + +### `id` encoding — public contract + +`Vehicle.id` for frequency-based rows is `trip:@`. +Anchor rows (where the trip has no frequencies row, or the row +exists but the anchor's stop_times naturally fall in the window +without expansion) keep the legacy `trip:` shape. The +per-stop promotion path's `Vehicle.id` propagation is unchanged — +the merged row inherits `v.id` from the per-stop row. + +## Stack order + +1. **n3ary/gtfs-publisher#252** — DDL addition (merged first). +2. **#347 (this)** — consumer side. Lands after #252. + +## Open design questions + +1. **Weekly view display.** A frequency-based trip with headway + 15 min / window 05:05-22:40 currently shows as 1 anchor + departure on the weekly schedule. The data layer makes + "every 15 min from 05:05 to 22:40" possible; the + rendering is a separate UX call (the data already includes 67 + synthetic minute slots in the `weekday` set; the existing + per-minute rendering just needs to handle the dense output). +2. **Per-route schedule display.** Same question for the + per-route schedule view. 67 individual rows is a lot; one + summary row with the headway is probably right. +3. **Multi-frequency rows per trip.** GTFS allows multiple + `frequencies.txt` rows per `trip_id` (e.g. "15 min 05:00-09:00" + then "30 min 09:00-22:00"). The cluj adapter only emits one + row per anchor, but the spec permits more. The + `getFrequenciesForServices` helper already returns N rows per + trip; the per-time query loops iterate them all and union the + expansions. Tested conceptually via + `frequencyExpansion.test.ts` (the `expandFrequenciesToDepartures` + test covers N rows per trip) but not end-to-end. +4. **`exact_times=1` rows.** GTFS allows `exact_times=1` + (the frequencies row exists but the trip is still + schedule-based). The cluj adapter only emits + `exact_times=0`. The helper treats + `exact_times=0` (or NULL) as the default expansion target and + `exact_times=1` as "ignore the frequencies row, use + `stop_times` directly" — the SQL `WHERE (f.exact_times IS NULL + OR f.exact_times = 0)` filters these out. Theoretical until a + feed actually needs it. + +## Verification + +- `pnpm check` — clean (svelte-check). +- `pnpm test` — 330/330 pass. +- `pnpm build` — vite build emits the production bundle. +- Unit tests cover the load-bearing expansion math + (`frequencyExpansion.test.ts`, 8 cases). +- Live data path verified at the type level: the per-time queries + accept `hasFrequencies: boolean`, the worker passes + `state.currentFeedHasFrequencies`, and the `id` encoding is + stable across polls. E2E verification (loading a real + frequencies-bearing blob and inspecting the station board) + needs the publisher's blob to be published to R2 first — that's + #252's rollout, out of scope here. + +## Out of scope + +- A `route_desc` or visual treatment of "headway 15 min" in the + route badge. That's a separate UX call. +- Per-feed opt-in / opt-out toggles in `feeds.json`. The app is + feed-agnostic per `docs/standards/feed-agnostic.md`; the + publisher's `SCHEMA` is the contract, full stop. Old blobs that + pre-date the DDL addition are handled by the bootstrap PRAGMA + probe, not by a feed flag. +- Reimplementing the weekly pattern view to summarize + frequency-based trips. The data layer supports it; the rendering + is a follow-up. +- GTFS-RT `TripUpdates` and `service_alerts` consumption. Both are + still reserved per `docs/specs/feeds-json.md:69`. +- The `gtfs-publisher-rt-reconcile` package's `parseFrequencies` + reader — has no consumer today. Left for a separate cleanup PR. + +## Related + +- Producer: `gtfs-adapters/adapters/cluj-napoca/src/assemble/derive/frequencies.ts` +- Publisher PR: [n3ary/gtfs-publisher#252](https://github.com/n3ary/gtfs-publisher/pull/252) +- GTFS spec: [frequencies.txt](https://gtfs.org/schedule/reference/#frequenciestxt) +- GTFS-RT contract: [app/docs/specs/gtfs-rt-contract.md](../specs/gtfs-rt-contract.md) +- Reconciler match key: [app/src/lib/domain/reconcile.ts](../../src/lib/domain/reconcile.ts) +- Helper: [app/src/lib/workers/gtfs/queries/frequencyExpansion.ts](../../src/lib/workers/gtfs/queries/frequencyExpansion.ts) +- Helper tests: [app/src/lib/workers/gtfs/queries/frequencyExpansion.test.ts](../../src/lib/workers/gtfs/queries/frequencyExpansion.test.ts) diff --git a/src/lib/domain/enrichObservations.ts b/src/lib/domain/enrichObservations.ts index bd16e94c..6c4046f2 100644 --- a/src/lib/domain/enrichObservations.ts +++ b/src/lib/domain/enrichObservations.ts @@ -1,37 +1,92 @@ // Resolve live observations against the static-trip index. Pure: no IO. Caller owns the active-trips snapshot. import type { LiveVehicleObservation } from '$lib/data/live/gtfsRtClient'; -import { minutesToTime } from './pipeline/timeUtils'; +import { minutesToTime, timeToMinutes } from './pipeline/timeUtils'; import type { Vehicle } from './types'; -type ActiveTripIndex = ReadonlyMap; +interface ActiveTripEntry { + directionId: 0 | 1; + tripStartMin: number; +} + +/** Composite-key index: `${tripId}|${effectiveStartMin}` → entry. The + * composite key is required for frequency-based trips — the app's + * frequency-expansion helper emits one Vehicle per generated + * departure (see #347), all sharing the anchor's tripId but with + * distinct `tripStartMin` (the effective time of the k-th departure). + * A bare tripId key would collapse the map on the last-written entry + * and the reconciler would match every observation against the wrong + * generated departure. The key is the same encoding the per-stop + * promotion path uses in `stationBoard.ts:mergeReconciledIntoStationBoard`. */ +type CompositeIndex = ReadonlyMap; -// tripId → {direction, startMin}, from the active-trips the worker fetches per tick. -export function indexActiveTripsByTripId(active: readonly Vehicle[]): ActiveTripIndex { - const out = new Map(); +/** tripId → first entry. Fallback for observations with no + * startTime. For non-frequency trips there's exactly one entry per + * tripId; for frequency-based trips, the iteration finds the k=0 + * entry (the anchor's first departure), which is the legacy + * "tripId-only match" behaviour for ambiguous cases. */ +type TripIdIndex = ReadonlyMap; + +/** Build both indices from the worker's active-trips snapshot. */ +export function indexActiveTrips(active: readonly Vehicle[]): { + byComposite: CompositeIndex; + byTripId: TripIdIndex; +} { + const byComposite: Map = new Map(); + const byTripId: Map = new Map(); for (const v of active) { if (!v.tripId) continue; const dir = v.schedule?.directionId; const start = v.schedule?.tripStartMin; if ((dir !== 0 && dir !== 1) || typeof start !== 'number') continue; - out.set(v.tripId, { directionId: dir, tripStartMin: start }); + const entry: ActiveTripEntry = { directionId: dir, tripStartMin: start }; + byComposite.set(`${v.tripId}|${start}`, entry); + if (!byTripId.has(v.tripId)) byTripId.set(v.tripId, entry); } - return out; + return { byComposite, byTripId }; +} + +/** Back-compat shim — the old API returned a single Map keyed by + * tripId. Kept for tests and any external callers; new code should + * use `indexActiveTrips` for the composite path. */ +export function indexActiveTripsByTripId(active: readonly Vehicle[]): TripIdIndex { + return indexActiveTrips(active).byTripId; } export function enrichObservations( observations: readonly LiveVehicleObservation[], active: readonly Vehicle[], ): LiveVehicleObservation[] { - const byTripId = indexActiveTripsByTripId(active); - return observations.map((obs) => enrichOne(obs, byTripId)); + const { byComposite, byTripId } = indexActiveTrips(active); + return observations.map((obs) => enrichOne(obs, byComposite, byTripId)); } function enrichOne( obs: LiveVehicleObservation, - byTripId: ActiveTripIndex, + byComposite: CompositeIndex, + byTripId: TripIdIndex, ): LiveVehicleObservation { - const sched = obs.tripId ? byTripId.get(obs.tripId) : undefined; + // Primary: composite key. For non-frequency trips the obs's + // startTime is the anchor's origin departure, identical to the + // active set's `tripStartMin`. For `UNSCHEDULED` observations + // (frequency-based trips per + // docs/specs/gtfs-rt-contract.md:89-90), the startTime is one + // specific generated departure's effective time. + const startMin = obs.startTime ? timeToMinutes(obs.startTime) : Number.NaN; + let sched: ActiveTripEntry | undefined; + if (obs.tripId && Number.isFinite(startMin)) { + sched = byComposite.get(`${obs.tripId}|${startMin}`); + } + // Fallback: tripId-only. Preserves the legacy lenient behaviour + // for non-conforming producers that don't populate startTime on + // non-frequency observations. For frequency-based trips this + // resolves to the k=0 entry, which is the anchor's first + // departure — not strictly correct (an observation without + // startTime is genuinely ambiguous), but the best we can do and + // matches pre-#347 behaviour. + if (!sched && obs.tripId) { + sched = byTripId.get(obs.tripId); + } if (sched) { return { ...obs, diff --git a/src/lib/domain/pipeline/scheduleScanner.ts b/src/lib/domain/pipeline/scheduleScanner.ts index 224baf36..f2e34de5 100644 --- a/src/lib/domain/pipeline/scheduleScanner.ts +++ b/src/lib/domain/pipeline/scheduleScanner.ts @@ -37,6 +37,13 @@ export interface ScheduleRow { trip_headsign: string | null; stop_lat: number; stop_lon: number; + /** Optional override for the emitted Vehicle.id. Defaults to + * `trip:${trip_id}` when undefined. Used by the frequency-expansion + * path to encode the generated departure's effective time, so + * two generated departures for the same anchor trip get distinct + * stable ids (and downstream consumers that key on `id` don't + * collapse them). */ + id?: string; } export interface ScheduleScannerInputs { @@ -103,7 +110,7 @@ export function scanSchedule(inputs: ScheduleScannerInputs): Vehicle[] { out.push({ kind: 'scheduled', - id: `trip:${r.trip_id}`, + id: r.id ?? `trip:${r.trip_id}`, route, type, tripId: r.trip_id, diff --git a/src/lib/domain/stationBoard.test.ts b/src/lib/domain/stationBoard.test.ts index 64da1aff..6c6c3eb8 100644 --- a/src/lib/domain/stationBoard.test.ts +++ b/src/lib/domain/stationBoard.test.ts @@ -674,7 +674,13 @@ describe('mergeReconciledIntoStationBoard', () => { lat: number, lon: number, asOf: number, + tripStartMin: number = 0, ): Vehicle { + // tripStartMin defaults to 0 for back-compat with the original + // 6-arg call sites. The frequency-aware promotion guard in + // stationBoard.ts requires the per-stop row's tripStartMin to + // match the reconciled row's; callers that test frequency-aware + // promotion must pass an explicit value. return { kind: 'tracked', id: `trip:${tripId}`, @@ -683,7 +689,7 @@ describe('mergeReconciledIntoStationBoard', () => { tripId, directionId: dir, confidence: 'medium', - schedule: { tripId, scheduledDeparture: 0, tripStartMin: 0, directionId: dir }, + schedule: { tripId, scheduledDeparture: 0, tripStartMin, directionId: dir }, position: { lat, lon, source: 'gps', asOf, speedMs: 5 }, liveSources: ['gtfs-rt'], } as Vehicle; @@ -713,7 +719,9 @@ describe('mergeReconciledIntoStationBoard', () => { it('promotes a per-stop scheduled row to reconciled when tripId matches', () => { const perStop = [perStopScheduled('T1', r24, 0, 500, 540)]; - const reconciled = [reconciledHit('T1', r24, 0, 46.78, 23.59, 12345)]; + // Pass tripStartMin=500 to match the per-stop row's value — + // the frequency-aware promotion guard requires it. + const reconciled = [reconciledHit('T1', r24, 0, 46.78, 23.59, 12345, 500)]; const out = mergeReconciledIntoStationBoard({ perStopVehicles: perStop, reconciledVehicles: reconciled, diff --git a/src/lib/domain/stationBoard.ts b/src/lib/domain/stationBoard.ts index dfc7b04e..a3cd99d5 100644 --- a/src/lib/domain/stationBoard.ts +++ b/src/lib/domain/stationBoard.ts @@ -279,6 +279,27 @@ export function mergeReconciledIntoStationBoard(inputs: StationMergeInputs): Veh if (!tid) return v; const reconciled = reconciledByTripId.get(tid); if (!reconciled || !reconciled.position) return v; + // Frequency-aware guard: for a frequency-based trip, the active + // set has N generated Vehicles (one per k-th departure), each + // with a distinct `schedule.tripStartMin` (the effective time). + // The per-stop set also has N rows, one per generated departure + // at THIS stop. The matched reconciled row carries ONE specific + // generated departure's `tripStartMin`; we must only promote the + // per-stop row whose `tripStartMin` matches. Without this check, + // every per-stop row for the anchor trip would get the same GPS + // position, which is wrong. + // + // Non-frequency trips pass the check trivially (the anchor's + // `tripStartMin` is identical to the active set's entry). + const reconciledStart = reconciled.schedule?.tripStartMin; + const thisStart = v.schedule.tripStartMin; + if ( + typeof reconciledStart === 'number' && + typeof thisStart === 'number' && + reconciledStart !== thisStart + ) { + return v; + } return { kind: 'tracked', id: v.id, diff --git a/src/lib/workers/gtfs.worker.ts b/src/lib/workers/gtfs.worker.ts index 48bb24ea..72a76a42 100644 --- a/src/lib/workers/gtfs.worker.ts +++ b/src/lib/workers/gtfs.worker.ts @@ -179,6 +179,7 @@ const api: GtfsRepo = { stopId, nowMs, windowMinutes, + state.currentFeedHasFrequencies, ); }, async getStationBoardsNear(lat, lon, radiusMeters, maxStations, nowMs, windowMinutes) { @@ -191,6 +192,7 @@ const api: GtfsRepo = { maxStations, nowMs, windowMinutes, + state.currentFeedHasFrequencies, ); }, @@ -203,6 +205,7 @@ const api: GtfsRepo = { localDate, fromMin, windowMinutes, + state.currentFeedHasFrequencies, ); }, async getActiveRouteIdsInWindow(localDate, nowMin, windowMinutes) { @@ -211,13 +214,14 @@ const api: GtfsRepo = { localDate, nowMin, windowMinutes, + state.currentFeedHasFrequencies, ); }, async getStopsAlongTrip(tripId) { return getStopsAlongTrip(await ensureDb(), tripId); }, async getWeeklySchedule(routeId, directionId) { - return getWeeklySchedule(await ensureDb(), routeId, directionId); + return getWeeklySchedule(await ensureDb(), routeId, directionId, state.currentFeedHasFrequencies); }, async getRouteDirectionEndpoints(routeId, directionId) { return getRouteDirectionEndpoints(await ensureDb(), routeId, directionId); @@ -231,6 +235,7 @@ const api: GtfsRepo = { localMin, lookbackMin, lookaheadMin, + state.currentFeedHasFrequencies, ); }, diff --git a/src/lib/workers/gtfs/bootstrap.ts b/src/lib/workers/gtfs/bootstrap.ts index 705b451e..931d9cad 100644 --- a/src/lib/workers/gtfs/bootstrap.ts +++ b/src/lib/workers/gtfs/bootstrap.ts @@ -376,6 +376,21 @@ export async function bootstrap( if (!hasRows) { throw new Error('stop_times is empty (truncated import or upstream produced an empty feed)'); } + // Soft-probe for `frequencies` (added by gtfs-publisher#252). Cached + // blobs that pre-date the DDL addition report false; the + // per-time query modules gate the expansion path on this flag so + // the app degrades to schedule-only behaviour without throwing. + // The flag is consumed by `state.currentFeedHasFrequencies`, set + // by the caller below. + const hasFrequencies = (db.selectValue( + `SELECT count(*) FROM sqlite_master WHERE type='table' AND name='frequencies'`, + ) as number) === 1; + // Lift the probe into the worker state so the per-time query + // modules can read it without re-running the PRAGMA on every + // call. Read by the frequency-expansion gate in + // `frequencyExpansion.ts` callers (activeTrips, stationArrivals, + // routeSchedule, routeMapView, weeklySchedule). + state.currentFeedHasFrequencies = hasFrequencies; } catch (e) { try { db?.close(); } catch {} try { poolUtil.unlink(opfsFile); } catch {} @@ -441,6 +456,7 @@ export function closeCurrent(): void { } state.currentFeedTz = null; state.currentFeedRtUrl = null; + state.currentFeedHasFrequencies = false; state.bootstrapping = null; // Shape polylines are feed-scoped — invalidate so the next feed // can't see stale entries from this one. diff --git a/src/lib/workers/gtfs/livePipeline.ts b/src/lib/workers/gtfs/livePipeline.ts index 0519706b..9f00b4ec 100644 --- a/src/lib/workers/gtfs/livePipeline.ts +++ b/src/lib/workers/gtfs/livePipeline.ts @@ -117,6 +117,7 @@ export async function tickLive(): Promise { nowMs, LIVE_RECONCILE_LOOKBACK_MIN, LIVE_RECONCILE_LOOKAHEAD_MIN, + state.currentFeedHasFrequencies, ); // Enrich observations with authoritative static-feed direction + // start_time via a SQL-backed lookup against `active`. Observations diff --git a/src/lib/workers/gtfs/queries/activeTrips.ts b/src/lib/workers/gtfs/queries/activeTrips.ts index 0be15cf2..905b5bab 100644 --- a/src/lib/workers/gtfs/queries/activeTrips.ts +++ b/src/lib/workers/gtfs/queries/activeTrips.ts @@ -8,6 +8,13 @@ * origin-relative (`scheduledDeparture = tripStartMin`, * `scheduledArrival = tripEndMin`). Consumers that need per-stop * ETA recompute it locally. + * + * For frequency-based trips (rows in `frequencies.txt`), each + * generated departure is emitted as its own `kind: 'scheduled'` + * Vehicle with `schedule.tripStartMin` set to the effective + * departure time and `id: trip:${tripId}@${effectiveStartMin}`. + * The reconciler matches live observations via the composite + * `(tripId, tripStartMin)` key in `enrichObservations.ts`. */ import type { Database } from '@sqlite.org/sqlite-wasm'; @@ -16,6 +23,10 @@ import { vehicleTypeFromGtfs } from '$lib/domain/types'; import { dateKeyInTz, minSinceMidnightInTz, timeToMinutes } from '$lib/domain/pipeline/timeUtils'; import { activeServicesOn } from '../activeServices'; import { selectAll } from '../sqlHelpers'; +import { + expandFrequencyToDepartures, + getFrequenciesForServices, +} from './frequencyExpansion'; import { getRoutesWithSchedule } from './routesWithSchedule'; export function getActiveTrips( @@ -24,6 +35,7 @@ export function getActiveTrips( nowMs: number, lookbackMin: number, lookaheadMin: number, + hasFrequencies: boolean, ): Vehicle[] { const localDate = dateKeyInTz(nowMs, tz); const nowMin = minSinceMidnightInTz(nowMs, tz); @@ -47,7 +59,10 @@ export function getActiveTrips( // Trip-level scan (no stop_times join in the select list): one // row per active trip with origin/end-stop times via the same // correlated subqueries getRouteMapView uses. Cheap thanks to - // stop_times_trip_seq_idx. + // stop_times_trip_seq_idx. For frequency-based trips the row + // is the anchor; we read the anchor's start/end times here and + // generate the per-departure `schedule.tripStartMin` values in + // the loop below. const rows = selectAll( db, `SELECT t.trip_id, t.trip_headsign, t.direction_id, @@ -65,12 +80,12 @@ export function getActiveTrips( const lower = nowMin - lookbackMin; const upper = nowMin + lookaheadMin; const withSchedule = getRoutesWithSchedule(db); + // Anchor trip data keyed by trip_id for the frequency-expansion + // pass below. Avoids a second trip-level scan. + const anchorByTripId = new Map(); + for (const r of rows) anchorByTripId.set(r.trip_id, r); const out: Vehicle[] = []; - for (const r of rows) { - const tripStartMin = timeToMinutes(r.trip_start_time); - const tripEndMin = timeToMinutes(r.trip_end_time); - if (tripStartMin < lower || tripStartMin > upper) continue; - if (tripEndMin < nowMin) continue; + const pushAnchor = (r: TripRow, effectiveStartMin: number, idSuffix: string): void => { const dir: 0 | 1 | -1 = r.direction_id === 0 || r.direction_id === 1 ? r.direction_id : -1; const route: Route = { @@ -83,7 +98,7 @@ export function getActiveTrips( }; out.push({ kind: 'scheduled', - id: `trip:${r.trip_id}`, + id: `trip:${r.trip_id}${idSuffix}`, route, type: route.type ?? 'unknown', tripId: r.trip_id, @@ -92,13 +107,45 @@ export function getActiveTrips( confidence: 'low', schedule: { tripId: r.trip_id, - scheduledDeparture: tripStartMin, - scheduledArrival: tripEndMin, - tripStartMin, + scheduledDeparture: effectiveStartMin, + scheduledArrival: effectiveStartMin, + tripStartMin: effectiveStartMin, headsign: r.trip_headsign ?? undefined, directionId: dir, }, }); + }; + for (const r of rows) { + const tripStartMin = timeToMinutes(r.trip_start_time); + const tripEndMin = timeToMinutes(r.trip_end_time); + if (tripStartMin < lower || tripStartMin > upper) continue; + if (tripEndMin < nowMin) continue; + pushAnchor(r, tripStartMin, ''); + } + // Frequency expansion: for each frequencies row whose anchor is + // in the active services, emit one Vehicle per generated + // departure in the window. Skipped when the loaded SQLite blob + // has no `frequencies` table (cached blobs that pre-date + // gtfs-publisher#252). + if (hasFrequencies) { + const freqs = getFrequenciesForServices(db, services); + for (const f of freqs) { + const anchor = anchorByTripId.get(f.trip_id); + if (!anchor) continue; + const anchorStartMin = timeToMinutes(anchor.trip_start_time); + const anchorEndMin = timeToMinutes(anchor.trip_end_time); + if (!Number.isFinite(anchorStartMin) || !Number.isFinite(anchorEndMin)) continue; + const headwayMin = f.headway_secs / 60; + const deps = expandFrequencyToDepartures(f, lower, upper); + for (const dep of deps) { + // Per-stop end time is anchor's last-stop time + k*headway. + // Skip departures whose effective end is already past `now` + // (the trip is fully done; nothing to display). + const effectiveEndMin = anchorEndMin + (dep.effectiveStartMin - anchorStartMin); + if (effectiveEndMin < nowMin) continue; + pushAnchor(anchor, dep.effectiveStartMin, `@${dep.effectiveStartMin}`); + } + } } return out; } diff --git a/src/lib/workers/gtfs/queries/frequencyExpansion.test.ts b/src/lib/workers/gtfs/queries/frequencyExpansion.test.ts new file mode 100644 index 00000000..de77c5ce --- /dev/null +++ b/src/lib/workers/gtfs/queries/frequencyExpansion.test.ts @@ -0,0 +1,155 @@ +/* + * Pure-function tests for the frequency-expansion helper. No DB + * required — `expandFrequencyToDepartures` is the load-bearing + * function (one bug here propagates to every per-time query) and + * it's purely arithmetic on minutes-since-midnight. + */ + +import { describe, it, expect } from 'vitest'; +import { + expandFrequencyToDepartures, + expandFrequenciesToDepartures, + type FrequencyRow, +} from './frequencyExpansion'; + +const F = ( + partial: Partial & Pick, +): FrequencyRow => ({ + trip_id: 'T_FREQ', + exact_times: 0, + ...partial, +}); + +describe('expandFrequencyToDepartures', () => { + it('emits one departure per headway within the window', () => { + // 15-min headway, 05:05 - 22:40, window 05:00 - 23:00. + // 05:05, 05:20, 05:35, ..., 22:20, 22:35 (NOT 22:50 — past 22:40 exclusive). + // First 5: 5:05, 5:20, 5:35, 5:50, 6:05. + // Last 5 in window: 21:50, 22:05, 22:20, 22:35 (and 22:50 excluded). + const deps = expandFrequencyToDepartures( + F({ start_time: '05:05:00', end_time: '22:40:00', headway_secs: 900 }), + 5 * 60, + 23 * 60, + ); + expect(deps[0]).toEqual({ effectiveStartMin: 5 * 60 + 5, k: 0 }); + expect(deps[1]).toEqual({ effectiveStartMin: 5 * 60 + 20, k: 1 }); + expect(deps[2]).toEqual({ effectiveStartMin: 5 * 60 + 35, k: 2 }); + // Last one: 22:35 (k = (22*60+35 - 5*60-5) / 15 = 17*60+30 / 15 = 70). + const last = deps[deps.length - 1]!; + expect(last.effectiveStartMin).toBe(22 * 60 + 35); + expect(last.k).toBe(70); + // Total: 71 departures (k=0..70). + expect(deps).toHaveLength(71); + }); + + it('excludes the end_time-bound departure (per spec: "up to but not including end_time")', () => { + // 10-min headway, 05:00 - 05:30. Departures: 5:00, 5:10, 5:20 (5:30 excluded). + const deps = expandFrequencyToDepartures( + F({ start_time: '05:00:00', end_time: '05:30:00', headway_secs: 600 }), + 0, + 24 * 60, + ); + expect(deps.map((d) => d.effectiveStartMin)).toEqual([5 * 60, 5 * 60 + 10, 5 * 60 + 20]); + }); + + it('clamps the start of the window (does not generate departures before windowStartMin)', () => { + // 15-min headway, 05:00 - 22:00, query window 07:30 - 09:00. + // First departure at 07:30 (k=10), last at 09:00 (k=16, inclusive + // upper bound per the same convention as `getActiveTrips`). + const deps = expandFrequencyToDepartures( + F({ start_time: '05:00:00', end_time: '22:00:00', headway_secs: 900 }), + 7 * 60 + 30, + 9 * 60, + ); + expect(deps[0]?.effectiveStartMin).toBe(7 * 60 + 30); + expect(deps[deps.length - 1]?.effectiveStartMin).toBe(9 * 60); + // 07:30, 07:45, 08:00, 08:15, 08:30, 08:45, 09:00 = 7 departures. + expect(deps).toHaveLength(7); + }); + + it('handles frequencies that cross midnight (end_time > 24:00:00)', () => { + // Night-route: 23:00 - 26:00 (i.e. 02:00 next day), 30-min headway. + // Departures: 23:00, 23:30, 00:00, 00:30, 01:00, 01:30 (02:00 excluded). + // 23:00 = 23*60 = 1380. 26:00 = 26*60 = 1560. Headway = 30 min. + // k in [0, 6), effective 1380, 1410, 1440, 1470, 1500, 1530. + const deps = expandFrequencyToDepartures( + F({ start_time: '23:00:00', end_time: '26:00:00', headway_secs: 1800 }), + 0, + 30 * 60, + ); + expect(deps.map((d) => d.effectiveStartMin)).toEqual([ + 23 * 60, 23 * 60 + 30, + 24 * 60, 24 * 60 + 30, + 25 * 60, 25 * 60 + 30, + ]); + }); + + it('returns [] when the window does not intersect the frequency window', () => { + // Frequency 10:00 - 12:00, query window 14:00 - 15:00 (no overlap). + expect(expandFrequencyToDepartures( + F({ start_time: '10:00:00', end_time: '12:00:00', headway_secs: 600 }), + 14 * 60, 15 * 60, + )).toEqual([]); + // Same freq, query window 13:00 - 14:00 — still no overlap. + expect(expandFrequencyToDepartures( + F({ start_time: '10:00:00', end_time: '12:00:00', headway_secs: 600 }), + 13 * 60, 14 * 60, + )).toEqual([]); + }); + + it('returns [] on garbage input (defence in depth — DDL CHECKs already reject these)', () => { + // Unparseable start_time. + expect(expandFrequencyToDepartures( + F({ start_time: 'not-a-time', end_time: '22:00:00', headway_secs: 900 }), + 0, 24 * 60, + )).toEqual([]); + // Unparseable end_time. + expect(expandFrequencyToDepartures( + F({ start_time: '05:00:00', end_time: 'garbage', headway_secs: 900 }), + 0, 24 * 60, + )).toEqual([]); + // end_time <= start_time. + expect(expandFrequencyToDepartures( + F({ start_time: '10:00:00', end_time: '10:00:00', headway_secs: 600 }), + 0, 24 * 60, + )).toEqual([]); + expect(expandFrequencyToDepartures( + F({ start_time: '11:00:00', end_time: '10:00:00', headway_secs: 600 }), + 0, 24 * 60, + )).toEqual([]); + // Non-positive headway. + expect(expandFrequencyToDepartures( + F({ start_time: '05:00:00', end_time: '22:00:00', headway_secs: 0 }), + 0, 24 * 60, + )).toEqual([]); + expect(expandFrequencyToDepartures( + F({ start_time: '05:00:00', end_time: '22:00:00', headway_secs: -60 }), + 0, 24 * 60, + )).toEqual([]); + }); + + it('handles a 1-minute headway (M26 case) without floating-point drift', () => { + // 60-second headway, 05:00 - 05:10. Departures: 5:00, 5:01, ..., 5:09 (10 of them). + const deps = expandFrequencyToDepartures( + F({ start_time: '05:00:00', end_time: '05:10:00', headway_secs: 60 }), + 0, 24 * 60, + ); + expect(deps).toHaveLength(10); + expect(deps[0]?.effectiveStartMin).toBe(5 * 60); + expect(deps[9]?.effectiveStartMin).toBe(5 * 60 + 9); + }); +}); + +describe('expandFrequenciesToDepartures', () => { + it('groups generated departures by trip_id, omitting trips with no in-window departures', () => { + const freqs: FrequencyRow[] = [ + F({ trip_id: 'A', start_time: '05:00:00', end_time: '22:00:00', headway_secs: 900 }), + // B's window doesn't intersect the query window — should be omitted. + F({ trip_id: 'B', start_time: '02:00:00', end_time: '04:00:00', headway_secs: 600 }), + ]; + const out = expandFrequenciesToDepartures(freqs, 5 * 60, 23 * 60); + expect(out.has('A')).toBe(true); + expect(out.has('B')).toBe(false); + expect(out.get('A')?.[0]?.effectiveStartMin).toBe(5 * 60); + }); +}); diff --git a/src/lib/workers/gtfs/queries/frequencyExpansion.ts b/src/lib/workers/gtfs/queries/frequencyExpansion.ts new file mode 100644 index 00000000..5085b8c7 --- /dev/null +++ b/src/lib/workers/gtfs/queries/frequencyExpansion.ts @@ -0,0 +1,189 @@ +/* + * GTFS frequencies.txt expansion — the consumer side of the anchor-trip + * + headway-window model. + * + * Per the GTFS spec, a row in `frequencies.txt` says: "for this + * `trip_id`, run every `headway_secs` seconds from `start_time` to + * `end_time`, applying the anchor trip's `stop_times` as offsets + * relative to each departure time". The cluj-napoca adapter emits + * these rows for `*-range` annotations (e.g. M26 `05:05-22:40` / + * `10-20min`). + * + * The publisher pipeline (gtfs-publisher#252) added the table to the + * SQLite DDL; the app-side consumer expands a frequency row into one + * `GeneratedDeparture` per departure in the active window. Each + * generated departure carries the `effectiveStartMin` (the k-th + * departure's origin time) and a `k` index (so per-stop offset times + * can be derived: anchor's stop_time at sequence N + k*headway_secs). + * + * Soft-probe via `hasFrequenciesTable(db)` — cached blobs that + * pre-date the publisher's DDL addition return false and the caller + * degrades to schedule-only behaviour without throwing. + */ + +import type { Database } from '@sqlite.org/sqlite-wasm'; +import { selectAll } from '../sqlHelpers'; +import { timeToMinutes } from '$lib/domain/pipeline/timeUtils'; + +/** Raw row from `frequencies` (anchor trip's service_id is joined via trips). */ +export interface FrequencyRow { + trip_id: string; + start_time: string; + end_time: string; + headway_secs: number; + exact_times: number | null; +} + +/** One generated departure inside a frequency row's window. */ +export interface GeneratedDeparture { + /** k-th departure's origin time, in minutes since local midnight. */ + effectiveStartMin: number; + /** 0-based departure index. Multiply by headway_secs to get the offset from start_time. */ + k: number; +} + +/** True when the open SQLite blob has a `frequencies` table. Cached + * blobs that pre-date the publisher's DDL addition (gtfs-publisher#252) + * return false; callers should treat as "no frequency-based trips". */ +export function hasFrequenciesTable(db: Database): boolean { + const row = db.selectValue( + `SELECT count(*) FROM sqlite_master WHERE type='table' AND name='frequencies'`, + ) as number | null; + return row === 1; +} + +/** All frequencies rows whose anchor trip runs on any of the given + * service_ids today. Filters out `exact_times=1` (schedule-based + * trips with a frequencies row — the spec allows it for legacy + * feeds; we expand `exact_times=0` only). No window filter — + * callers do that themselves so the window semantics stay in one + * place (`expandFrequencyToDepartures`). */ +export function getFrequenciesForServices( + db: Database, + serviceIds: readonly string[], +): FrequencyRow[] { + if (serviceIds.length === 0) return []; + const placeholders = serviceIds.map(() => '?').join(','); + type Row = { + trip_id: string; + start_time: string; + end_time: string; + headway_secs: number; + exact_times: number | null; + }; + return selectAll( + db, + `SELECT f.trip_id, f.start_time, f.end_time, f.headway_secs, f.exact_times + FROM frequencies f + JOIN trips t ON t.trip_id = f.trip_id + WHERE t.service_id IN (${placeholders}) + AND (f.exact_times IS NULL OR f.exact_times = 0);`, + serviceIds, + ); +} + +/** Pure expansion: turn one frequency row into N `GeneratedDeparture`s, + * one per departure whose effective time falls in `[windowStartMin, + * windowEndMin]`. The GTFS spec says the trip runs at `start_time, + * start_time + headway, start_time + 2*headway, …` up to but not + * including `end_time`. A departure whose effective time is exactly + * `end_time` is NOT generated (it's the exclusive end bound). Pure — + * no I/O, no Date, fully unit-testable. + * + * Returns [] on garbage input (unparseable times, non-positive + * headway, or window that doesn't intersect the frequency window). + * These guards are belt-and-suspenders — the publisher's DDL + * CHECKs (`start_time < end_time`, `headway_secs > 0`) reject the + * bad rows at INSERT time. */ +export function expandFrequencyToDepartures( + freq: FrequencyRow, + windowStartMin: number, + windowEndMin: number, +): GeneratedDeparture[] { + const startMin = timeToMinutes(freq.start_time); + const endMin = timeToMinutes(freq.end_time); + if (!Number.isFinite(startMin) || !Number.isFinite(endMin)) return []; + if (endMin <= startMin) return []; + + const headwayMin = freq.headway_secs / 60; + if (!(headwayMin > 0)) return []; + + // The frequency window may straddle midnight (cluj has 25:00+ rows + // for past-midnight trips), so we use the raw minutes-since- + // midnight arithmetic — the window argument is the same unit. + // The end bound is exclusive: a departure AT endMin is not + // generated (per the spec's "up to but not including end_time"). + const out: GeneratedDeparture[] = []; + // Clamp kStart to 0: the first departure is at startMin, never earlier. + const kStart = Math.max(0, Math.ceil((windowStartMin - startMin) / headwayMin)); + // kFreqEnd: largest k whose effective time is STRICTLY < endMin. + // `ceil - 1` handles both the integer-divisible case (e.g. start=5:00, + // end=7:30, headway=30 → ceil(150/30)-1 = 4, excludes 7:30) and the + // non-divisible case (e.g. 1055/15 = 70.33 → ceil-1 = 70, includes + // 22:35 and excludes 22:50). A `- 1` on the floor would be wrong on + // the non-divisible case. + const kFreqEnd = Math.ceil((endMin - startMin) / headwayMin) - 1; + // kWindowEnd: the query window is inclusive on both ends, matching + // the existing `getActiveTrips` window filter at + // src/lib/workers/gtfs/queries/activeTrips.ts:65-72. + const kWindowEnd = Math.floor((windowEndMin - startMin) / headwayMin); + const kEnd = Math.min(kFreqEnd, kWindowEnd); + if (kEnd < kStart) return []; + + for (let k = kStart; k <= kEnd; k++) { + out.push({ + effectiveStartMin: startMin + k * headwayMin, + k, + }); + } + return out; +} + +/** Convenience: expand many frequency rows at once. Returns a Map + * keyed by trip_id; rows with no departures in the window are + * omitted (callers can iterate the map and treat absence as + * "not active in window"). */ +export function expandFrequenciesToDepartures( + freqs: readonly FrequencyRow[], + windowStartMin: number, + windowEndMin: number, +): Map { + const out = new Map(); + for (const f of freqs) { + const deps = expandFrequencyToDepartures(f, windowStartMin, windowEndMin); + if (deps.length > 0) out.set(f.trip_id, deps); + } + return out; +} + +/** For one frequency-based trip, fetch the anchor's full stop_times + * in stop_sequence order. The expansion helper uses these to + * derive per-stop effective times (anchor's stop_time + k*headway). + * Returns rows in the same shape as a stop_times table scan (no + * stop_name / stop_lat / stop_lon — those are joined in the + * per-stop queries, not here, so this helper stays decoupled from + * the stops table). */ +export interface AnchorStopTimeRow { + trip_id: string; + stop_id: string; + /** Anchor offset time, HH:MM:SS. The k-th generated departure's + * effective time at this stop is `arrival_time + k*headway_secs`. */ + arrival_time: string; + departure_time: string; + stop_sequence: number; + pickup_type: number | null; +} + +export function getAnchorStopTimes( + db: Database, + tripId: string, +): AnchorStopTimeRow[] { + return selectAll( + db, + `SELECT trip_id, stop_id, arrival_time, departure_time, stop_sequence, pickup_type + FROM stop_times + WHERE trip_id = ? + ORDER BY stop_sequence ASC;`, + [tripId], + ); +} diff --git a/src/lib/workers/gtfs/queries/routeMapView.ts b/src/lib/workers/gtfs/queries/routeMapView.ts index 25eae198..94422ecd 100644 --- a/src/lib/workers/gtfs/queries/routeMapView.ts +++ b/src/lib/workers/gtfs/queries/routeMapView.ts @@ -18,6 +18,10 @@ import { vehicleTypeFromGtfs } from '$lib/domain/types'; import { activeServicesOn } from '../activeServices'; import { shapeCache } from '../shapeCache'; import { selectAll } from '../sqlHelpers'; +import { + expandFrequencyToDepartures, + getFrequenciesForServices, +} from './frequencyExpansion'; import { getRoutesWithSchedule } from './routesWithSchedule'; export function getRouteMapView( @@ -28,6 +32,7 @@ export function getRouteMapView( localMin: number, lookbackMin: number, lookaheadMin: number, + hasFrequencies: boolean, ): RouteMapView | null { type RouteRow = { route_id: string; @@ -56,7 +61,7 @@ export function getRouteMapView( const tables = selectAll<{ name: string }>( db, `SELECT name FROM sqlite_master WHERE type IN ('table') - AND name IN ('_route_tags', 'route_networks');`, + AND name IN ('_route_tags', 'route_networks', 'frequencies');`, ); const hasRouteTags = tables.some((t) => t.name === '_route_tags'); const hasRouteNetworks = tables.some((t) => t.name === 'route_networks'); @@ -132,6 +137,38 @@ export function getRouteMapView( })) .filter((row) => row.tripStartMin >= lowerMin && row.tripStartMin <= upperMin && row.tripEndMin >= localMin) .sort((a, b) => a.tripStartMin - b.tripStartMin); + // Frequency expansion: for each frequencies row on this + // (route, direction) in the active services, emit one + // generated departure in [lowerMin, upperMin]. Use the + // anchor's shape_id (shape doesn't change per generated + // departure). + if (hasFrequencies) { + const freqs = getFrequenciesForServices(db, services); + for (const f of freqs) { + const anchor = tripRows.find((r) => r.trip_id === f.trip_id); + if (!anchor) continue; + const anchorStartMin = timeToMinutes(anchor.trip_start_time); + const anchorEndMin = timeToMinutes(anchor.trip_end_time); + if (!Number.isFinite(anchorStartMin) || !Number.isFinite(anchorEndMin)) continue; + const deps = expandFrequencyToDepartures(f, lowerMin, upperMin); + for (const dep of deps) { + const effectiveEndMin = anchorEndMin + (dep.effectiveStartMin - anchorStartMin); + if (effectiveEndMin < localMin) continue; + activeTripRows.push({ + trip_id: anchor.trip_id, + trip_headsign: anchor.trip_headsign, + shape_id: anchor.shape_id, + trip_start_time: anchor.trip_start_time, + trip_end_time: anchor.trip_end_time, + tripStartMin: dep.effectiveStartMin, + tripEndMin: effectiveEndMin, + }); + } + } + // Re-sort after expansion so the map view's "earliest + // upcoming" pick is correct. + activeTripRows.sort((a, b) => a.tripStartMin - b.tripStartMin); + } } // 2) Stops for every active trip, in one query. Skipped when @@ -139,7 +176,7 @@ export function getRouteMapView( // below handles structure-only renders. const stopsByTrip = activeTripRows.length === 0 ? new Map() - : loadStopsForTrips(db, activeTripRows.map((t) => t.trip_id)); + : loadStopsForTrips(db, Array.from(new Set(activeTripRows.map((t) => t.trip_id)))); const trips: RouteMapTrip[] = activeTripRows.map((t) => ({ tripId: t.trip_id, diff --git a/src/lib/workers/gtfs/queries/routeSchedule.ts b/src/lib/workers/gtfs/queries/routeSchedule.ts index 5ed9c396..61c9eb64 100644 --- a/src/lib/workers/gtfs/queries/routeSchedule.ts +++ b/src/lib/workers/gtfs/queries/routeSchedule.ts @@ -4,6 +4,10 @@ * departure falls in the requested window. Caller controls the * window (rest-of-today, tomorrow until noon, night-route past * midnight) so this stays a pure window query. + * + * For frequency-based trips (rows in `frequencies.txt`), each + * generated departure is emitted as its own ScheduleTrip with + * `tripStartMin` set to the effective departure time. */ import type { Database } from '@sqlite.org/sqlite-wasm'; @@ -11,6 +15,10 @@ import type { ScheduleTrip } from '$lib/data/gtfs/types'; import { timeToMinutes } from '$lib/domain/pipeline/timeUtils'; import { activeServicesOn } from '../activeServices'; import { selectAll } from '../sqlHelpers'; +import { + expandFrequencyToDepartures, + getFrequenciesForServices, +} from './frequencyExpansion'; export function getRouteSchedule( db: Database, @@ -19,6 +27,7 @@ export function getRouteSchedule( localDate: string, fromMin: number, windowMinutes: number, + hasFrequencies: boolean, ): ScheduleTrip[] { const services = activeServicesOn(db, localDate); if (services.length === 0) return []; @@ -46,7 +55,7 @@ export function getRouteSchedule( ); const upper = fromMin + windowMinutes; - return rows + const out: ScheduleTrip[] = rows .map((r) => ({ tripId: r.trip_id, tripStartMin: timeToMinutes(r.trip_start_time), @@ -54,8 +63,33 @@ export function getRouteSchedule( headsign: r.trip_headsign, serviceId: r.service_id, })) - .filter((r) => r.tripStartMin >= fromMin && r.tripStartMin <= upper) - .sort((a, b) => a.tripStartMin - b.tripStartMin); + .filter((r) => r.tripStartMin >= fromMin && r.tripStartMin <= upper); + // Frequency expansion. For each frequencies row whose anchor is + // on this (route, direction) and in the active services, emit + // one ScheduleTrip per generated departure in the window. + if (hasFrequencies) { + const freqs = getFrequenciesForServices(db, services); + for (const f of freqs) { + const anchor = rows.find((r) => r.trip_id === f.trip_id); + if (!anchor) continue; + const anchorStartMin = timeToMinutes(anchor.trip_start_time); + const anchorEndMin = timeToMinutes(anchor.trip_end_time); + if (!Number.isFinite(anchorStartMin) || !Number.isFinite(anchorEndMin)) continue; + const deps = expandFrequencyToDepartures(f, fromMin, upper); + for (const dep of deps) { + out.push({ + tripId: f.trip_id, + tripStartMin: dep.effectiveStartMin, + // Per-trip end is anchor's last-stop time + the same delta + // we applied to the origin. + tripEndMin: anchorEndMin + (dep.effectiveStartMin - anchorStartMin), + headsign: anchor.trip_headsign, + serviceId: anchor.service_id, + }); + } + } + } + return out.sort((a, b) => a.tripStartMin - b.tripStartMin); } /** Distinct route_ids that have at least one trip departing in the @@ -73,6 +107,7 @@ export function getActiveRouteIdsInWindow( localDate: string, nowMin: number, windowMinutes: number, + hasFrequencies: boolean, ): string[] { const services = activeServicesOn(db, localDate); if (services.length === 0) return []; @@ -103,9 +138,40 @@ export function getActiveRouteIdsInWindow( ); const upper = nowMin + windowMinutes; - return Array.from(new Set( + const out = new Set( rows .filter((r) => r.trip_start_min >= nowMin && r.trip_start_min <= upper) .map((r) => r.route_id), - )); + ); + // Frequency-based contributions: a route is "active right now" if + // any frequency row on it overlaps the query window. Per-row + // expansion is unnecessary — the route set is all we return. + if (hasFrequencies) { + const freqs = getFrequenciesForServices(db, services); + if (freqs.length > 0) { + const tripIds = Array.from(new Set(freqs.map((f) => f.trip_id))); + // trip_id → route_id lookup. One tiny query; trips are + // indexed by PRIMARY KEY. + type TripRouteRow = { trip_id: string; route_id: string }; + const tripRoutes = selectAll( + db, + `SELECT trip_id, route_id FROM trips + WHERE trip_id IN (${tripIds.map(() => '?').join(',')});`, + tripIds, + ); + const tripToRoute = new Map(tripRoutes.map((r) => [r.trip_id, r.route_id])); + for (const f of freqs) { + const startMin = timeToMinutes(f.start_time); + const endMin = timeToMinutes(f.end_time); + if (!Number.isFinite(startMin) || !Number.isFinite(endMin)) continue; + // Frequency window [startMin, endMin) overlaps query window + // [nowMin, upper] iff startMin < upper AND endMin > nowMin. + if (startMin < upper && endMin > nowMin) { + const routeId = tripToRoute.get(f.trip_id); + if (routeId) out.add(routeId); + } + } + } + } + return Array.from(out); } diff --git a/src/lib/workers/gtfs/queries/stationArrivals.ts b/src/lib/workers/gtfs/queries/stationArrivals.ts index 928f5a19..d21da354 100644 --- a/src/lib/workers/gtfs/queries/stationArrivals.ts +++ b/src/lib/workers/gtfs/queries/stationArrivals.ts @@ -5,14 +5,24 @@ * "today + window" view. Everything is `kind: 'scheduled'`; the * reconciliation upgrade happens later via `mergeReconciledIntoStationBoard` * on the main thread using the worker's broadcast. + * + * For frequency-based trips (rows in `frequencies.txt`), each + * generated departure is emitted as its own `kind: 'scheduled'` + * Vehicle with `schedule.tripStartMin` set to the effective + * departure time and `id: trip:${tripId}@${effectiveStartMin}`. */ import type { Database } from '@sqlite.org/sqlite-wasm'; import type { Vehicle } from '$lib/domain/types'; import { scanSchedule, type ScheduleRow } from '$lib/domain/pipeline/scheduleScanner'; -import { dateKeyInTz, minSinceMidnightInTz } from '$lib/domain/pipeline/timeUtils'; +import { dateKeyInTz, minSinceMidnightInTz, timeToMinutes } from '$lib/domain/pipeline/timeUtils'; import { activeServicesOn } from '../activeServices'; import { selectAll } from '../sqlHelpers'; +import { + expandFrequencyToDepartures, + getAnchorStopTimes, + getFrequenciesForServices, +} from './frequencyExpansion'; export function getStationArrivals( db: Database, @@ -20,6 +30,7 @@ export function getStationArrivals( stopId: string, nowMs: number, windowMinutes: number, + hasFrequencies: boolean, ): Vehicle[] { const localDate = dateKeyInTz(nowMs, tz); const nowMinSinceMidnight = minSinceMidnightInTz(nowMs, tz); @@ -42,7 +53,7 @@ export function getStationArrivals( // a vehicle in the 'departed' bucket only // while it's still en route (not yet arrived // at its end stop). - // trip_start_time — departure_time at the trip's FIRST stop + // trip_start_time — departure_time at the FIRST stop // (origin). Surfaced for the reconciler so it // can match live observations by // (route, direction, start_time) instead of @@ -70,8 +81,122 @@ export function getStationArrivals( [stopId, ...services], ); + // Frequency expansion at this stop. For each frequency-based trip + // whose anchor passes through `stopId`, expand into one row per + // generated departure whose effective time at THIS stop falls in + // the query window. The per-stop effective arrival time is + // anchor's stop_times.arrival_time + k*headway_min. + let frequencyRows: ScheduleRow[] = []; + if (hasFrequencies) { + const freqs = getFrequenciesForServices(db, services); + const tripIds = Array.from(new Set(freqs.map((f) => f.trip_id))); + // Pull all stop_times for the frequency-based trips in one + // query so we can derive the per-stop offset for each. + const tripPh = tripIds.length === 0 ? '' : `AND st.trip_id IN (${tripIds.map(() => '?').join(',')})`; + type FreqTripRow = { + trip_id: string; + trip_headsign: string | null; + direction_id: number | null; + route_id: string; + route_short_name: string; + route_color: string | null; + route_text_color: string | null; + route_type: number | null; + }; + const tripMeta = tripIds.length === 0 ? new Map() : new Map( + selectAll( + db, + `SELECT t.trip_id, t.trip_headsign, t.direction_id, + r.route_id, r.route_short_name, r.route_color, r.route_text_color, r.route_type + FROM trips t JOIN routes r ON r.route_id = t.route_id + WHERE 1=1 ${tripPh};`, + tripIds, + ).map((r) => [r.trip_id, r]), + ); + const upper = nowMinSinceMidnight + windowMinutes; + for (const f of freqs) { + const anchor = tripMeta.get(f.trip_id); + if (!anchor) continue; + const stops = getAnchorStopTimes(db, f.trip_id); + // Pick the stop_times row for THIS stop. Most anchors have one + // row per stop_sequence; we want the matching one. + const thisStop = stops.find((s) => s.stop_id === stopId); + if (!thisStop) continue; + // Pull the stop's coords for the row (used by downstream + // consumers even though scanSchedule itself doesn't read them). + const stopCoords = selectAll<{ stop_lat: number; stop_lon: number }>( + db, + `SELECT stop_lat, stop_lon FROM stops WHERE stop_id = ?;`, + [stopId], + ); + const stopLat = stopCoords[0]?.stop_lat ?? 0; + const stopLon = stopCoords[0]?.stop_lon ?? 0; + const deps = expandFrequencyToDepartures(f, nowMinSinceMidnight - 60, upper); + const offsetMin = timeToMinutes(thisStop.arrival_time); + if (!Number.isFinite(offsetMin)) continue; + // First/last_seq for the trip (the anchor's stop_sequence + // range). Cheap: one SELECT per trip but stops is in memory + // and small. + const firstSeq = stops[0]?.stop_sequence ?? thisStop.stop_sequence; + const lastSeq = stops[stops.length - 1]?.stop_sequence ?? thisStop.stop_sequence; + const anchorStartMin = timeToMinutes(f.start_time); + const anchorEndMin = timeToMinutes(f.end_time); + if (!Number.isFinite(anchorStartMin) || !Number.isFinite(anchorEndMin)) continue; + for (const dep of deps) { + // Effective arrival at this stop = anchor's stop_times offset + (effectiveStart - anchor start). + // i.e. shift the anchor row's time by the same delta we applied to the trip origin. + const delta = dep.effectiveStartMin - anchorStartMin; + const effArrivalMin = offsetMin + delta; + const effDepartureMin = timeToMinutes(thisStop.departure_time) + delta; + const effTripEndMin = anchorEndMin + delta; + // Window: keep rows whose effective time at this stop is in [now, now+window]. + if (effArrivalMin < nowMinSinceMidnight) continue; + if (effArrivalMin > upper) continue; + // Use the effArrivalMin-formatted time so scanSchedule sees the + // effective arrival; we set trip_start_time = effectiveStartMin + // (the reconciler key) and leave arrival_time/departure_time as + // the effective per-stop times. + const pad2 = (n: number) => String(n).padStart(2, '0'); + const h = (m: number) => { + const hh = Math.floor(m / 60); + const mm = m % 60; + return `${pad2(hh)}:${pad2(mm)}:00`; + }; + // Join the trip + route data so scanSchedule can project + // without re-querying. Reuse the existing ScheduleRow + // shape: trip_id, arrival_time, departure_time, stop_sequence, + // first_seq, last_seq, trip_start_time, trip_end_time, etc. + frequencyRows.push({ + trip_id: f.trip_id, + // Stable id encodes the generated departure's effective + // origin time so each row is uniquely identifiable + // downstream (the per-stop `mergeReconciledIntoStationBoard` + // promotion path uses Vehicle.id as a stable Svelte key). + id: `trip:${f.trip_id}@${dep.effectiveStartMin}`, + arrival_time: h(effArrivalMin), + departure_time: h(effDepartureMin), + pickup_type: thisStop.pickup_type, + stop_sequence: thisStop.stop_sequence, + first_seq: firstSeq, + last_seq: lastSeq, + trip_end_time: h(effTripEndMin), + trip_start_time: h(dep.effectiveStartMin), + direction_id: anchor.direction_id, + route_id: anchor.route_id, + route_short_name: anchor.route_short_name, + route_color: anchor.route_color, + route_text_color: anchor.route_text_color, + route_type: anchor.route_type, + trip_headsign: anchor.trip_headsign, + stop_lat: stopLat, + stop_lon: stopLon, + }); + } + } + } + return scanSchedule({ - rows, + rows: [...rows, ...frequencyRows], nowMinSinceMidnight, nowMs, windowMinutes, diff --git a/src/lib/workers/gtfs/queries/stationBoards.ts b/src/lib/workers/gtfs/queries/stationBoards.ts index 617459f2..98e2da52 100644 --- a/src/lib/workers/gtfs/queries/stationBoards.ts +++ b/src/lib/workers/gtfs/queries/stationBoards.ts @@ -21,6 +21,7 @@ export function getStationBoard( stopId: string, nowMs: number, windowMinutes: number, + hasFrequencies: boolean, ): { stop: StopWithDistance; vehicles: Vehicle[] } | null { type Row = { stop_id: string; stop_name: string; stop_lat: number; stop_lon: number }; const rows = selectAll( @@ -38,7 +39,7 @@ export function getStationBoard( lon: s.stop_lon, // distance intentionally absent — no GPS context here. }, - vehicles: getStationArrivals(db, tz, stopId, nowMs, windowMinutes), + vehicles: getStationArrivals(db, tz, stopId, nowMs, windowMinutes, hasFrequencies), }; } @@ -54,10 +55,11 @@ export function getStationBoardsNear( maxStations: number, nowMs: number, windowMinutes: number, + hasFrequencies: boolean, ): { stop: StopWithDistance; vehicles: Vehicle[] }[] { const stops = getStopsNear(db, lat, lon, radiusMeters, maxStations); return stops.map((stop) => ({ stop, - vehicles: getStationArrivals(db, tz, stop.id, nowMs, windowMinutes), + vehicles: getStationArrivals(db, tz, stop.id, nowMs, windowMinutes, hasFrequencies), })); } diff --git a/src/lib/workers/gtfs/queries/weeklySchedule.ts b/src/lib/workers/gtfs/queries/weeklySchedule.ts index 4615ab44..fe8328f5 100644 --- a/src/lib/workers/gtfs/queries/weeklySchedule.ts +++ b/src/lib/workers/gtfs/queries/weeklySchedule.ts @@ -6,6 +6,12 @@ * Intentionally ignores `calendar_dates` exceptions — the weekly * table is a recurring-pattern view, not a what-runs-on-a-specific- * day view. + * + * For frequency-based trips (rows in `frequencies.txt`), the + * recurring pattern is "every `headway_secs` from `start_time` to + * `end_time`" — we expand each frequency row into one synthetic + * minute slot per generated departure and union with the + * schedule-based set. */ import type { Database } from '@sqlite.org/sqlite-wasm'; @@ -17,8 +23,10 @@ export function getWeeklySchedule( db: Database, routeId: string, directionId: 0 | 1, + hasFrequencies: boolean, ): WeeklySchedule { type Row = { + trip_id: string; departure_time: string; monday: number; tuesday: number; @@ -30,10 +38,10 @@ export function getWeeklySchedule( }; const rows = selectAll( db, - `SELECT + `SELECT t.trip_id, (SELECT departure_time FROM stop_times - WHERE trip_id = t.trip_id - ORDER BY stop_sequence ASC LIMIT 1) AS departure_time, + WHERE trip_id = t.trip_id + ORDER BY stop_sequence ASC LIMIT 1) AS departure_time, c.monday, c.tuesday, c.wednesday, c.thursday, c.friday, c.saturday, c.sunday FROM trips t @@ -45,15 +53,72 @@ export function getWeeklySchedule( const weekday = new Set(); const saturday = new Set(); const sunday = new Set(); + const addAt = (min: number, dayBits: { mon: number; tue: number; wed: number; thu: number; fri: number; sat: number; sun: number }) => { + if (dayBits.mon || dayBits.tue || dayBits.wed || dayBits.thu || dayBits.fri) { + weekday.add(min); + } + if (dayBits.sat) saturday.add(min); + if (dayBits.sun) sunday.add(min); + }; for (const r of rows) { if (!r.departure_time) continue; const m = timeToMinutes(r.departure_time); if (!Number.isFinite(m)) continue; - if (r.monday || r.tuesday || r.wednesday || r.thursday || r.friday) { - weekday.add(m); + addAt(m, { + mon: r.monday, tue: r.tuesday, wed: r.wednesday, thu: r.thursday, fri: r.friday, + sat: r.saturday, sun: r.sunday, + }); + } + // Frequency expansion. For each frequency row whose anchor is on + // this (route, direction) and runs on a service with the + // matching day bits, add start_time + k*headway_min for k=0..(N-1) + // where N = floor((end_time - start_time) / headway). This mirrors + // the JS expansion in frequencyExpansion.ts but only for the + // weekly pattern view (no window filter, full day). + if (hasFrequencies) { + type FreqRow = { + trip_id: string; + start_time: string; + end_time: string; + headway_secs: number; + monday: number; + tuesday: number; + wednesday: number; + thursday: number; + friday: number; + saturday: number; + sunday: number; + }; + const freqRows = selectAll( + db, + `SELECT t.trip_id, f.start_time, f.end_time, f.headway_secs, + c.monday, c.tuesday, c.wednesday, c.thursday, c.friday, + c.saturday, c.sunday + FROM frequencies f + JOIN trips t ON t.trip_id = f.trip_id + JOIN calendar c ON c.service_id = t.service_id + WHERE t.route_id = ? AND t.direction_id = ? + AND (f.exact_times IS NULL OR f.exact_times = 0);`, + [routeId, directionId], + ); + for (const f of freqRows) { + const startMin = timeToMinutes(f.start_time); + const endMin = timeToMinutes(f.end_time); + const headwayMin = f.headway_secs / 60; + if (!Number.isFinite(startMin) || !Number.isFinite(endMin) || !(headwayMin > 0)) continue; + // Generate all departures in [startMin, endMin) (spec: up to + // but not including end_time). Weekly view is full day so no + // window cap. + const dayBits = { + mon: f.monday, tue: f.tuesday, wed: f.wednesday, thu: f.thursday, fri: f.friday, + sat: f.saturday, sun: f.sunday, + }; + for (let k = 0; ; k++) { + const m = startMin + k * headwayMin; + if (m >= endMin) break; + addAt(Math.round(m), dayBits); + } } - if (r.saturday) saturday.add(m); - if (r.sunday) sunday.add(m); } const sorted = (s: Set) => Array.from(s).sort((a, b) => a - b); return { diff --git a/src/lib/workers/gtfs/state.ts b/src/lib/workers/gtfs/state.ts index be07f479..f83e07ea 100644 --- a/src/lib/workers/gtfs/state.ts +++ b/src/lib/workers/gtfs/state.ts @@ -28,6 +28,13 @@ class WorkerState { /** Dwell seconds per stop from the feed's _neary_config timing block. * Used by assembleLiveBoards to thread feed-specific dwell into ETA. */ currentDwellSec: number = 20; + /** True when the bound SQLite blob has a `frequencies` table. + * Set by `bootstrap()` via a `sqlite_master` PRAGMA probe. Cached + * blobs that pre-date gtfs-publisher#252 (the DDL addition) report + * false; per-time query modules gate the frequency-expansion path + * on this so the app degrades to schedule-only behaviour without + * throwing on older blobs. */ + currentFeedHasFrequencies: boolean = false; /** Promise of the in-flight bootstrap when setFeed is mid-fetch. * Used by ensureDb so the very first call can await the bind. */ bootstrapping: Promise | null = null; diff --git a/src/lib/workers/gtfs/stationSubscribers.ts b/src/lib/workers/gtfs/stationSubscribers.ts index 8fcf5869..b285ad9c 100644 --- a/src/lib/workers/gtfs/stationSubscribers.ts +++ b/src/lib/workers/gtfs/stationSubscribers.ts @@ -96,7 +96,7 @@ async function pushOne(sub: StationSub, snap: ReconciledSnapshot | null): Promis vehicles: Vehicle[]; }> = []; for (const stopId of sub.stopIds) { - const board = getStationBoard(db, tz, stopId, nowMs, DEFAULT_CONFIG.arrivalsWindowMin); + const board = getStationBoard(db, tz, stopId, nowMs, DEFAULT_CONFIG.arrivalsWindowMin, state.currentFeedHasFrequencies); if (!board) continue; scheduled.push({ stopId: board.stop.id,