jetstan-agent 0.1.20

Production-grade OTA agent for Linux-based vehicles — MQTT communication backbone.

Jetstan Agent

A production-grade OTA agent for Linux-based vehicles. It currently runs on a Raspberry Pi, but the architecture is designed to move to automotive Linux hardware without major changes.

It connects securely to HiveMQ Cloud, subscribes to the update topic, routes incoming commands, and — for ota.available — upgrades itself via pip and restarts. See INSTALL.md for running it as a real systemd service (boot-start, crash/hang recovery, the OTA update loop).

Project layout

jetstan_agent/
    __init__.py      Package version.
    __main__.py      Enables `python -m jetstan_agent`.
    cli.py            argparse entrypoint (`jetstan-agent` console script).
    config.py         All tunable values and environment-variable loading.
    mqtt.py           TLS connection, subscribe, publish, auto-reconnect.
    agent.py          Wires MQTT to message routing; keeps the process alive.
    message_router.py Parses JSON commands and dispatches by "type".
    updater.py         Self-upgrade via pip + restart-to-apply.
    sd_notify.py       Minimal systemd readiness/watchdog notifications.
    logger.py          Central logging configuration.
    system_checks.py   Read-only root/systemd/distro checks for the installer.
    service_unit.py    Renders the packaged systemd unit template.
    service_cli.py     systemctl/journalctl wrappers.
    installer.py        `jetstan-agent-install` entrypoint.
    uninstaller.py      `jetstan-agent-uninstall` entrypoint.
    templates/          Packaged systemd unit + default agent.env.
tests/
    (one test module per file above)

Why each file exists

• config.py — the single source of truth for connection settings. Non-secret values (port, keepalive, topic) have defaults; the broker host, username, and password have none and must come from environment variables, so credentials never live in source control.
• mqtt.py — everything specific to talking to the broker (TLS handshake, subscribe, reconnect, systemd readiness/watchdog signaling). It knows nothing about OTA — it only hands raw (topic, payload) pairs to a callback.
• message_router.py — parses each payload as JSON and dispatches by "type". Unknown types and invalid JSON are logged and swallowed, never raised — a single bad message can't crash the agent.
• updater.py — the actual OTA logic: pip install --upgrade jetstan-agent==<version>, then a deliberate process exit so systemd's Restart=always relaunches into the newly installed code. No custom download/signing pipeline — PyPI already does atomic package installs.
• agent.py — the composition root. It owns the MQTT client and hands incoming payloads to message_router.
• logger.py — configures the root logger once, at startup.
• cli.py — run is the in-process foreground agent (what systemd's ExecStart= invokes); status/start/stop/restart/logs are a separate concern — they operate on the systemd service itself via service_cli.py, so "start" never means two different things.
• installer.py / uninstaller.py / service_unit.py / system_checks.py — turn pip install jetstan-agent into a managed, hardened systemd service. See INSTALL.md for the full flow and design rationale.
• __main__.py — enables python -m jetstan_agent (defaults to run).

Configuration

All configuration is via environment variables (read by config.py).

Variable	Required	Default
MQTT_HOST	Yes	—
MQTT_USERNAME	Yes	—
MQTT_PASSWORD	Yes	—
MQTT_PORT	No	8883
MQTT_TOPIC	No	jetstan/demo/update
JETSTAN_MQTT_CLIENT_ID	No	jetstan-agent
JETSTAN_LOG_LEVEL	No	INFO

In production (systemd on the vehicle, CI, etc.) set these as real environment variables — e.g. a systemd unit's EnvironmentFile= — never hardcode them in a file that gets committed.

Local development: .env file

For local development, config.py automatically loads a .env file from the project root (via python-dotenv) before reading any configuration — no manual export needed, and no code changes.

cp .env.example .env
# then edit .env and fill in MQTT_HOST / MQTT_USERNAME / MQTT_PASSWORD

.env is git-ignored, so credentials never get committed. If a real environment variable is already set, it takes precedence over .env — so this has no effect on production deployments where .env won't even exist.

Arduino firmware updates

software/arduino_updates/ extends the OTA pipeline to also compile and flash an Arduino connected to the Pi over USB — fully automatically, as part of every agent restart (which includes right after every self-update, since that ends in a restart). Once the one-time host setup below is done, updating the Arduino side is just: edit the .ino, bump the version, python -m twine upload dist/* — no SSH, no manual arduino-cli/avrdude commands, ever again.

Agent update finishes -> Compile sketch (if source changed) -> Flash (if firmware changed) -> Verify -> Continue startup

This runs through software/deployment/, a small generic dependency-DAG task runner (see below) — jetstan_agent itself has no idea this is "Arduino," it just asks "does software/ have a deployment pipeline to run?", the same way it already asks whether software/ has custom MQTT commands. Today's pipeline is two tasks (compile_arduino_firmware -> flash_arduino_firmware); a future board (e.g. a custom PCB) means registering more tasks in software/deployment/pipeline.py, not touching the OTA plumbing again.

One-time host setup vs. every-push automation

This is the important split to understand:

• One-time, human-run over SSH (per Pi, ever): install avrdude and arduino-cli, and grant the service account USB-serial access. The OTA pipeline deliberately never automates this — letting a remote MQTT message trigger apt install/system-tool installation would mean anyone who can publish to the topic gets root-equivalent code execution on the Pi, which defeats the entire point of running the agent as an unprivileged, capability-stripped systemd service. See INSTALL.md for the exact one-time commands.
• Every push, fully automatic: with those tools already present, the already-unprivileged agent can compile and flash on its own — no new privilege needed, just running tools that already exist against paths it already has write access to (/var/lib/jetstan, via the systemd unit's StateDirectory=).

Configuration

Opt-in and off by default. Add to /etc/jetstan/agent.env (or the local .env for development):

Variable	Required	Default
ARDUINO_ENABLED	No	false
ARDUINO_BOARD	No	uno (also: nano, mega2560)
ARDUINO_PORT	No	auto (or an explicit /dev/tty...)
ARDUINO_BAUD	No	115200
ARDUINO_AUTO_COMPILE	No	true
ARDUINO_FIRMWARE_SOURCE	No	the packaged sample_firmware/motor_test sketch
ARDUINO_FIRMWARE_PATH	No	{ARDUINO_STATE_DIR}/firmware.hex (compiled output)
ARDUINO_CLI_PATH	No	arduino-cli
ARDUINO_COMPILE_TIMEOUT_SECONDS	No	60
ARDUINO_STATE_DIR	No	/var/lib/jetstan
ARDUINO_FLASH_RETRIES	No	3
ARDUINO_FLASH_TIMEOUT_SECONDS	No	60

Set ARDUINO_AUTO_COMPILE=false to fall back to the old workflow (manually compile elsewhere and drop a .hex at ARDUINO_FIRMWARE_PATH yourself) — useful if you'd rather not compile on-device, e.g. for a future board with a different/heavier build system.

Both compiling and flashing skip themselves (with a clean log line, not an error) when nothing changed: the sketch source is fingerprinted (SHA-256 over its files) and compared against the last-compiled source before invoking arduino-cli at all, and the compiled .hex output is separately fingerprinted and compared against the last-flashed one before invoking avrdude — so a routine Python-only OTA push doesn't re-compile or re-flash anything, and an actual sketch change is always picked up, with no manually-tracked version number to forget to bump.

Module layout

software/arduino_updates/ (Arduino/AVR-specific), matching the rest of this repo's one-file-one-concern style:

• exceptions.py — one ArduinoUpdateError subclass per failure mode (board missing, permission denied, timeout, compile failure, verification mismatch, ...), so callers/logs are specific about what went wrong.
• config.py — ArduinoConfig.from_env(), plus the board-name -> avrdude/arduino-cli-parameters registry (BoardProfile).
• discovery.py — finds /dev/ttyUSB*//dev/ttyACM* device nodes and resolves which one to use; refuses to guess if there's more than one and none is configured explicitly.
• compiler.py — compiles ARDUINO_FIRMWARE_SOURCE via arduino-cli (never installs it — raises a clear error if missing, pointing at the one-time setup step); skips compiling via a source fingerprint, same atomic-write state pattern as everything else here.
• firmware.py — content-fingerprints the compiled .hex and persists the last-flashed fingerprint (atomic write, same pattern as installer.py).
• flasher.py — an abstract BoardFlasher interface plus AVRDudeFlasher. Adding STM32/ESP32/RP2040 support later means adding another BoardFlasher subclass and a FLASHERS entry — nothing in the OTA pipeline changes.
• verifier.py — runs the board-specific verify step and turns a mismatch into a VerificationError.
• __init__.py — check_and_flash() (raises on failure) and sync_firmware_safe() (never raises).
• sample_firmware/motor_test/ — a minimal sketch (drives a motor driver's RPWM/LPWM pins at ~15% duty cycle from boot, nothing else) for validating the compile -> flash -> verify pipeline end-to-end without needing real application firmware yet.

software/deployment/ (generic — knows nothing about Arduino):

• dag.py — Task(name, run, depends_on, critical) plus run_dag(): real topological ordering (Kahn's algorithm), cycle/unknown-dependency detection before anything runs, per-task failure isolation (a critical=True task's failure skips everything downstream of it; a bug in one task never crashes the runner).
• pipeline.py — this project's concrete graph: compile_arduino_firmware -> flash_arduino_firmware. A future board gets its own tasks registered here.
• __init__.py — run_deployment_dag(), never raises — what jetstan_agent/cli.py actually calls before starting the MQTT loop.

Install locally (editable)

python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install -e ".[dev]"

Run

Either export the variables directly:

export MQTT_HOST="xxxxxxxx.s1.eu.hivemq.cloud"
export MQTT_USERNAME="your-username"
export MQTT_PASSWORD="your-password"

or, for local development, create a .env file as shown above and skip the export lines entirely. Then:

python -m jetstan_agent
# or, equivalently:
jetstan-agent run

On startup the agent connects over TLS, subscribes to jetstan/demo/update, and logs every message it receives. If the connection drops (or the broker is briefly unreachable), it reconnects automatically with backoff — no restart needed.

Test connectivity

1. Run the agent as shown above and leave it running.

2. From the HiveMQ Cloud console (or any MQTT client, e.g. mosquitto_pub), publish a message to jetstan/demo/update:

   mosquitto_pub -h <your-cluster-host> -p 8883 --cafile /etc/ssl/certs/ca-certificates.crt \
     -u <username> -P <password> -t jetstan/demo/update -m '{"version": "1.0.1"}'
   

3. Confirm the agent's log output prints the received message.

Run the test suite

pytest