app.state.fleet

Per-vehicle state: telemetry buffer, arrivals log, and the features derived from them.

Fleet subscribes to the ingest pipeline. Each ping goes into its vehicle’s time-ordered buffer and marks the vehicle dirty; update() re-runs arrival detection for dirty vehicles only (cost follows the incoming data, not the fleet size) and merges the result into the vehicle’s arrivals log — stops still inside the detection window are re-detected, older ones are final.

Derived per vehicle (criterion 3 — «текущее отклонение, средняя скорость на сегменте, время простоя»):

• current deviation: delay at the latest detected arrival (this is cur_dev_s for the model);

• segment speed: distance along the plan / time between the last two arrivals of the same trip;

• dwell: time spent at the last detected stop;

• current segment: last detected stop → next planned stop.

Classes

Derived(computed_for, last_arrival, ...)
Fleet(schedules, clock, *[, buffer_s])
VehicleState(tr_id, schedule, pings, ...)

class app.state.fleet.Derived(computed_for: 'datetime', last_arrival: 'Arrival | None', cur_dev_s: 'float | None', segment_speed_kmh: 'float | None', dwell_s: 'float | None', next_pos: 'int | None')

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

__init__(computed_for: datetime, last_arrival: Arrival | None, cur_dev_s: float | None, segment_speed_kmh: float | None, dwell_s: float | None, next_pos: int | None) → None

computed_for: datetime

cur_dev_s: float | None

dwell_s: float | None

last_arrival: Arrival | None

next_pos: int | None

segment_speed_kmh: float | None

class app.state.fleet.Fleet(schedules: dict[int, VehicleSchedule], clock: DatasetClock, *, buffer_s: float = 8100)

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

__init__(schedules: dict[int, VehicleSchedule], clock: DatasetClock, *, buffer_s: float = 8100) → None

on_arrivals(fn: Callable[[int, list[Arrival]], None]) → None

Called with (tr_id, newly detected arrivals) after each update — for the history store.

on_ping(ping: Ping) → None

async run(tick_s: float = 2.0) → None

snapshot() → dict

telemetry(tr_id: int, T: datetime, span_s: float = 3600.0) → list[Ping]

The vehicle’s pings with T − span < event_time <= T (what the model may see at T).

update(T: datetime | None = None) → int

Recompute dirty vehicles as of dataset time T (default: now). Returns how many were updated.

update_ms: deque[float]

vehicles: dict[int, VehicleState]

class app.state.fleet.VehicleState(tr_id: 'int', schedule: 'VehicleSchedule | None', pings: 'list[Ping]' = <factory>, arrivals: 'dict[int, Arrival]'=<factory>, dirty: 'bool' = False, derived: 'Derived | None' = None)

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

__init__(tr_id: int, schedule: VehicleSchedule | None, pings: list[Ping] = <factory>, arrivals: dict[int, ~app.state.arrivals.Arrival]=<factory>, dirty: bool = False, derived: Derived | None = None) → None

arrivals: dict[int, Arrival]

derived: Derived | None

dirty: bool

property last_ping: Ping | None

property last_position: Ping | None

pings: list[Ping]

schedule: VehicleSchedule | None

tr_id: int