Fleet Telemetry

1"""Fleet Telemetry API built with Pixeltable.
2
3    pxt schema update app.py fleet
4    pxt service run app.py fleet
5"""
6import pixeltable as pxt
7import pixeltable.functions as pxtf
8from pixeltable.serving import FastAPIRouter
9
10from udfs import alert_label, severity_score, vehicle_tag
11
12# ---- tables ----
13TableModel = pxt.model_base()
14
15
16class Readings(TableModel, name='readings', has_default_idxs=False):
17    id = pxt.Column(value=pxtf.uuid.uuid7(), primary_key=True)
18    vehicle_id: pxt.String
19    fleet: pxt.String
20    status: pxt.String              # moving / idle / parked
21    recorded_at: pxt.String         # ISO-8601, sortable
22    speed_kph: pxt.Float
23    engine_temp_c: pxt.Float
24    fuel_pct: pxt.Float | None
25
26    severity = severity_score(speed_kph, engine_temp_c, fuel_pct)
27    alert = alert_label(severity)   # chained on another computed column
28    tag = vehicle_tag(fleet, vehicle_id)
29
30    __indexes__ = [pxt.BtreeIndex(vehicle_id), pxt.BtreeIndex(fleet),
31                   pxt.BtreeIndex(status), pxt.BtreeIndex(recorded_at)]
32
33
34# ---- queries ----
35@pxt.query
36def vehicle_history(vehicle_id: str):
37    """All readings for one vehicle, newest first (uses the vehicle_id index)."""
38    return Readings.where(Readings.vehicle_id == vehicle_id).select(
39        Readings.recorded_at, Readings.speed_kph, Readings.engine_temp_c, Readings.severity, Readings.alert
40    ).order_by(Readings.recorded_at, asc=False)
41
42
43@pxt.query
44def alerts_since(since: str):
45    """Non-ok readings recorded at or after a timestamp (uses the recorded_at index)."""
46    return Readings.where((Readings.recorded_at >= since) & (Readings.alert != 'ok')).select(
47        Readings.tag, Readings.recorded_at, Readings.severity, Readings.alert
48    ).order_by(Readings.severity, asc=False)
49
50
51@pxt.query
52def fleet_board(fleet: str, status: str):
53    """Vehicles in a fleet with a given status (fleet + status indexes)."""
54    return Readings.where((Readings.fleet == fleet) & (Readings.status == status)).select(
55        Readings.vehicle_id, Readings.recorded_at, Readings.alert
56    ).order_by(Readings.vehicle_id)
57
58
59# ---- routes ----
60telemetry_api = FastAPIRouter(name='telemetry_api')
61telemetry_api.add_insert_route(
62    Readings, path='/readings',
63    inputs=[Readings.vehicle_id, Readings.fleet, Readings.status, Readings.recorded_at, Readings.speed_kph,
64            Readings.engine_temp_c, Readings.fuel_pct],
65    outputs=[Readings.id, Readings.severity, Readings.alert],
66)
67telemetry_api.add_update_route(Readings, path='/readings/correct',
68                               inputs=[Readings.speed_kph, Readings.engine_temp_c, Readings.fuel_pct],
69                               outputs=[Readings.id, Readings.severity, Readings.alert])
70telemetry_api.add_delete_route(Readings, path='/readings/delete')
71telemetry_api.add_compute_route(Readings, path='/score',
72                                inputs=[Readings.speed_kph, Readings.engine_temp_c, Readings.fuel_pct],
73                                outputs=[Readings.severity, Readings.alert])
74telemetry_api.add_query_route(path='/vehicles/history', query=vehicle_history, method='get')
75telemetry_api.add_query_route(path='/alerts', query=alerts_since, method='get')
76telemetry_api.add_query_route(path='/fleets/board', query=fleet_board, method='get')