app.ingest.pipeline

Ingest pipeline: NDTP fixes and replayed rows become Ping objects on one timeline, in one stream.

Both sources feed the same pipeline, arbitrated per vehicle: while a vehicle has sent NDTP within ndtp_fresh_s, replayed rows for it are dropped. When its live feed goes quiet, replay takes over; when NDTP returns, it takes back over (criterion 5: degrade to historical data, recover after reconnect).

Classes

Ingest(clock, unit_to_tr, int], *, ...)
IngestStats([ndtp_pings, replay_pings, ...])

class app.ingest.pipeline.Ingest(clock: ~app.clock.DatasetClock, unit_to_tr: dict[int, int], *, ndtp_fresh_s: float = 60.0, max_skew_s: float = 300.0, wall: ~collections.abc.Callable[[], float] = <built-in function time>)

Базовые классы: object

__init__(clock: ~app.clock.DatasetClock, unit_to_tr: dict[int, int], *, ndtp_fresh_s: float = 60.0, max_skew_s: float = 300.0, wall: ~collections.abc.Callable[[], float] = <built-in function time>) → None

accept_ndtp(fix: NdtpFix) → Ping | None

accept_replay(ping: Ping) → bool

async consume_ndtp(fixes: Queue[NdtpFix]) → None

live_vehicles() → list[int]

Vehicles currently fed by NDTP (not replay).

snapshot() → dict

subscribe(fn: Callable[[Ping], None]) → None

class app.ingest.pipeline.IngestStats(ndtp_pings: 'int' = 0, replay_pings: 'int' = 0, replay_suppressed: 'int' = 0, unknown_unit_fixes: 'int' = 0, clock_skew_fallbacks: 'int' = 0, subscriber_errors: 'int' = 0)

Базовые классы: object

__init__(ndtp_pings: int = 0, replay_pings: int = 0, replay_suppressed: int = 0, unknown_unit_fixes: int = 0, clock_skew_fallbacks: int = 0, subscriber_errors: int = 0) → None

clock_skew_fallbacks: int

ndtp_pings: int

replay_pings: int

replay_suppressed: int

subscriber_errors: int

unknown_unit_fixes: int