Middleman

The middleman (main.py) is the core process: eight threads sharing four queues, coordinated by a stop_event and a sink_control object. Every thread checks stop_event at least once a second, so SIGINT/SIGTERM bring the whole process down cleanly.

Threads

Thread	Module	Job
activity-monitor	transport/activity.py	Declares the stream active/inactive based on packet arrival
udp-receiver	transport/udp_receiver.py	Parses CSV, pushes to Grafana inline, fans out to 3 queues
influx-writer	sinks/influx_writer.py	Batches influx_q into the telemetry bucket
marple-sender	sinks/marple_sender.py	Flushes marple_q to Marple once a second
influx-stats	sinks/influx_stats_writer.py	Writes middleman health metrics to the stats bucket
stats-reporter	stats/reporter.py	Prints a status table every STATS_INTERVAL seconds
lap-tracker	analysis/lap_tracker.py	Detects laps from GPS, see Lap Tracking
web-api	web/api.py	FastAPI dashboard and control endpoints

All eight are daemon threads started from main(). On shutdown, main() joins each with an 8 second timeout and logs a warning for any that don't stop cleanly.

CSV parsing

schema.py is the single source of truth for the wire format, mirroring the firmware's CSV output. Each field entry declares its name, type, which message type(s) it belongs to, and which sinks it's destined for.

One file to edit

If the firmware's CSV layout changes, schema.py is the only file that needs to change here. Column order must match the CSV exactly for each message type.

transport/udp_receiver.py parses each datagram:

1. Split on commas. If the first token is a valid message type (1 or 2), consume it as the prefix; otherwise infer the type from the field count.
2. Cast every remaining field to its declared type (int, float, or str).
3. Require time_ms and time_unix to be present. If either is missing, the packet is discarded and a parse error is counted.
4. Wrap the result in a Packet, a dataclass(slots=True) chosen for cheap allocation at 100 Hz.

Session tracking

sd_filename (the field marked is_session in schema.py) only travels in message type 1. Type 2 packets inherit whatever session was last seen on a type-1 packet, since they don't carry their own filename.

Packet.seq_id is currently always 0

packet.py builds seq_id from a field named id or seq, but schema.py defines neither. Until one of those fields is added to the schema, every packet's seq_id is 0. It isn't used by any sink today, but don't rely on it for ordering or deduplication.

Fan-out

Every parsed packet is pushed, in this order:

1. Grafana, inline, synchronously, no queue (see Grafana).
2. influx_q, if the Influx sink is enabled.
3. marple_q, if the Marple sink is enabled.
4. analysis_q, if the analysis (lap tracker) sink is enabled.

Queues are ring buffers

influx_q, marple_q, and analysis_q are bounded queue.Queues used as ring buffers: when full, _ring_put() discards the oldest pending packet to make room for the newest, rather than blocking the receiver or dropping the new one. stats_q is the exception: it's a small bounded queue that simply drops the newest stats point on overflow, since a missed summary point doesn't matter.

Freshness over completeness

Under sustained backpressure (a slow sink, a network stall) the receiver never blocks and never falls behind: it keeps the newest data and sacrifices the oldest queued packets. This matches live telemetry's priority, the same trade the firmware makes on its SD ring buffer and UDP send path.

Activity and sink control

ActivityMonitor marks the stream active on every packet and goes inactive after UDP_INACTIVITY_TIMEOUT seconds of silence, checked once a second by run_activity_monitor. Every sink loop calls activity.wait_for_active(timeout=1.0) before doing any work, and closes its connection (Influx client, Marple client, Grafana session) when the stream goes inactive, reopening it automatically once packets resume.

sink_control is four independent threading.Events (grafana, influx, marple, analysis), seeded from .env at startup and toggled at runtime by POST /api/system_state. Every sink checks its own event on each loop iteration, so a toggle takes effect within about a second.

Stats

stats/counters.py keeps sliding-window rates (a 2 second window) and lifetime totals behind a single lock. stats/reporter.py prints a formatted table to the log every STATS_INTERVAL seconds and pushes the same numbers to InfluxDB (see InfluxDB), so middleman health is queryable, not just visible in the container logs.

The web dashboard

web/api.py runs a FastAPI app on port 8000 inside the container (8005 on the host), serving a static dashboard (web/index.html) at / plus:

Method	Endpoint	Does
GET/POST	/api/system_state	Sink toggles and lap thresholds
GET/POST	/api/settings	Finish line coordinates
GET	/api/last_lap_snapshot	Last completed lap's time and distance
GET	/api/latest	Latest value of every field for the current session
GET	/api/track	Downsampled GPS track for the current session

/api/track and /api/latest query InfluxDB directly with Flux (see InfluxDB); the rest read and write the same in-memory analyzer_settings dict and sink_control object the other threads use, so a change from the dashboard takes effect immediately, no restart needed.

Adding an analyzer

The lap tracker is the only analyzer today, but analysis_q is generic. To add another:

1. Create analysis/your_analyzer.py.
2. Implement run_your_analyzer(analysis_q, influx_client, settings, stop_event, activity), matching the shape in analysis/analyzer_protocol.py.
3. Swap run_lap_tracker for it in the thread_defs list in main.py.