diff --git a/README.md b/README.md index 3443f2e..395cf25 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,13 @@ This is a repository for attempting to make my home ceiling fans [smart like thi There is a [plan for dealing with the fans](docs/plans/2026-03-08-fan-control-phase1.md). +## TPMS ingest (issue #7) + +Slice-1 lives under `src/tpms/`: read rtl_433 JSON from stdin, write to SQLite, +expose a small JSON API for a future home dashboard. Scope for slice-1 is +parked-only, any-tire-low alerting for the 2013 Mazda CX-9 (315 MHz, decoder +r156). Systemd units + install ritual: [`deploy/plexpi/`](deploy/plexpi/README.md). + ## What We Are Working With ### Software diff --git a/deploy/plexpi/README.md b/deploy/plexpi/README.md new file mode 100644 index 0000000..ddb83e5 --- /dev/null +++ b/deploy/plexpi/README.md @@ -0,0 +1,76 @@ +# TPMS slice-1 deploy on plexpi + +Manual install ritual — plexpi does not have `itguy` deployment. Bluelinky and +fuelbot follow the same pattern (clone + venv + systemd unit under `pi`). + +Both units are `systemctl --user` units, so they inherit the `pi` user +identity automatically — no `User=`/`Group=` directives (which systemd rejects +in `--user` units). + +## Prereqs + +1. RTL-SDR dongle plugged into plexpi USB, antenna facing the driveway. +2. `rtl_433` installed (`sudo apt install rtl-433`). +3. Confirm reception near the parked CX-9: + + ```bash + rtl_433 -M utc -f 315000000 -R 156 -M level + ``` + + You should see periodic decoded events tagged `Abarth-124Spider`. If nothing + arrives after a few minutes, walk closer to the vehicle or drive the car + briefly to wake the sensors. `-M utc` is important — without it, rtl_433 + emits system-local timestamps and the ingest daemon will skew every reading + by the local UTC offset. + +## Install + +```bash +ssh -i ~/.ssh/id_claude pi@192.168.68.54 + +# Data and log directories live under $HOME so no sudo is needed and +# systemd --user units can write freely. +mkdir -p ~/.local/share/tpms ~/tpms-logs + +cd ~ +git clone https://github.com/tclancy/radiofrequency tpms +cd tpms +uv sync + +mkdir -p ~/.config/systemd/user +cp deploy/plexpi/tpms-capture.service ~/.config/systemd/user/ +cp deploy/plexpi/tpms-api.service ~/.config/systemd/user/ +systemctl --user daemon-reload +systemctl --user enable --now tpms-api.service +systemctl --user enable --now tpms-capture.service +sudo loginctl enable-linger pi # so --user units survive logout +``` + +## Verify + +```bash +# Ingest capturing something +journalctl --user -u tpms-capture -f + +# API responding +curl http://192.168.68.54:8090/api/vehicles +curl http://192.168.68.54:8090/api/vehicles/mazda-cx9/tpms/latest +curl http://192.168.68.54:8090/api/health +``` + +The first `latest` response is likely `{"readings": [], ...}` until the CX-9's +sensors wake up (parked sensors transmit ~once per 60 s, sometimes much less). +`/api/health` returns `receiver_ok: false` until the first reading lands. + +## Sensor IDs on record + +Slice-1 does not map sensor→corner (any-tire-low alerting only). The `sensors` +table populates automatically as new sensor IDs are seen — no manual seed. + +## Env vars + +| Var | Default | Purpose | +|-----|---------|---------| +| `TPMS_DB` | `~/.local/share/tpms/tpms.sqlite3` | SQLite path (both services share) | +| `TPMS_LOW_PSI` | `30.0` | Threshold that sets `any_low: true` in `/latest` | +| `TPMS_STALE_SECONDS` | `900` | Reading is `stale: true` after this many seconds; `receiver_ok` flips false past the same window | diff --git a/deploy/plexpi/tpms-api.service b/deploy/plexpi/tpms-api.service new file mode 100644 index 0000000..831a997 --- /dev/null +++ b/deploy/plexpi/tpms-api.service @@ -0,0 +1,18 @@ +[Unit] +Description=TPMS JSON API — reads tpms.sqlite3, serves :8090 +Documentation=https://github.com/tclancy/radiofrequency/issues/7 +After=network.target + +[Service] +Type=simple +Environment=TPMS_DB=%h/.local/share/tpms/tpms.sqlite3 +Environment=TPMS_LOW_PSI=30.0 +Environment=TPMS_STALE_SECONDS=900 +WorkingDirectory=%h/tpms +ExecStart=%h/tpms/.venv/bin/python -m src.tpms.api +Restart=on-failure +RestartSec=15 +StandardError=append:%h/tpms-logs/api.log + +[Install] +WantedBy=default.target diff --git a/deploy/plexpi/tpms-capture.service b/deploy/plexpi/tpms-capture.service new file mode 100644 index 0000000..0271343 --- /dev/null +++ b/deploy/plexpi/tpms-capture.service @@ -0,0 +1,20 @@ +[Unit] +Description=TPMS capture — rtl_433 -> ingest daemon (Mazda CX-9, 315 MHz decoder r156) +Documentation=https://github.com/tclancy/radiofrequency/issues/7 +After=network.target + +[Service] +Type=simple +Environment=TPMS_DB=%h/.local/share/tpms/tpms.sqlite3 +# rtl_433 streams JSON on stdout; the ingest daemon reads its stdin. Piping via +# systemd keeps this to a single unit — no MQTT broker required for slice-1. +# `-M utc` forces UTC timestamps in the JSON output — without it rtl_433 uses +# system-local time and the ingest daemon would silently skew every reading. +ExecStart=/bin/sh -c '/usr/bin/rtl_433 -M utc -f 315000000 -R 156 -F json 2>>%h/tpms-logs/rtl_433.log | %h/tpms/.venv/bin/python -m src.tpms.ingest' +Restart=on-failure +RestartSec=15 +WorkingDirectory=%h/tpms +StandardError=append:%h/tpms-logs/capture.log + +[Install] +WantedBy=default.target diff --git a/pyproject.toml b/pyproject.toml index ed82217..5757aa9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,6 +6,7 @@ readme = "README.md" requires-python = ">=3.13" dependencies = [ "click>=8.1", + "flask>=3.1.3", "httpx>=0.27", "matplotlib>=3.10", "numpy>=1.26", diff --git a/src/tpms/__init__.py b/src/tpms/__init__.py new file mode 100644 index 0000000..690a6b6 --- /dev/null +++ b/src/tpms/__init__.py @@ -0,0 +1,7 @@ +"""TPMS ingest slice-1 — parked-only tire-pressure capture for a home dashboard. + +Scope: one vehicle (2013 Mazda CX-9), one frequency (315 MHz), rtl_433 decoder r156. +Reads JSON events from rtl_433 (stdin), writes to SQLite, exposes a small JSON API. + +See docs/tpms-slice-1.md for the deployment plan on plexpi. +""" diff --git a/src/tpms/api.py b/src/tpms/api.py new file mode 100644 index 0000000..0626b74 --- /dev/null +++ b/src/tpms/api.py @@ -0,0 +1,166 @@ +"""Flask JSON API for the home dashboard to read TPMS state. + +Endpoints: + GET /api/vehicles + List known vehicles (slice-1: mazda-cx9 only). + GET /api/vehicles//tpms/latest + Latest reading per sensor, one entry per sensor_id. Empty list is + legitimate (no readings yet); the client uses `/api/health` to tell + empty-because-parked from empty-because-broken. + GET /api/vehicles//tpms/history?since=&until=&limit= + Raw readings within the window. Defaults: last 7 days, limit 5000. + GET /api/health + Overall health — last event timestamp, receiver_ok flag. + +Design notes: +- No auth. LAN-only service, mounted behind whatever the plexpi already fronts. +- One SQLite connection per request keeps thread-safety trivial and lets us + swap `TPMS_DB` in tests without touching Flask internals. +- Pressure is stored in kPa; response payloads carry both kPa and PSI so the + dashboard doesn't have to know the conversion. +""" + +from __future__ import annotations + +import sqlite3 +import time +from pathlib import Path + +from flask import Flask, jsonify, request + +from . import db +from .config import Config + +_KPA_TO_PSI = 0.14503773773 + + +def kpa_to_psi(kpa: float | None) -> float | None: + return None if kpa is None else round(kpa * _KPA_TO_PSI, 1) + + +def _serialize_reading(row) -> dict: + kpa = row["pressure_kpa"] + return { + "ts": row["ts"], + "sensor_id": row["sensor_id"], + "pressure_kpa": kpa, + "pressure_psi": kpa_to_psi(kpa), + "temperature_c": row["temperature_c"], + "battery_ok": row["battery_ok"], + } + + +def create_app(config: Config | None = None) -> Flask: + cfg = config or Config.from_env() + app = Flask(__name__) + app.config["TPMS_CONFIG"] = cfg + + # Init schema once at app startup so per-request connects can skip the + # `CREATE TABLE IF NOT EXISTS` script and vehicle upsert. The API is a + # concurrent reader alongside the ingest daemon — extra write-lock traffic + # per request would contend needlessly on the plexpi SQLite file. + db.connect(cfg.db_path).close() + + @app.get("/api/vehicles") + def list_vehicles(): + with _connect(cfg.db_path) as conn: + rows = conn.execute( + "SELECT slug, make, model, year, vin, frequency_hz, decoder " + "FROM vehicles ORDER BY slug" + ).fetchall() + return jsonify([dict(r) for r in rows]) + + @app.get("/api/vehicles//tpms/latest") + def latest(slug: str): + with _connect(cfg.db_path) as conn: + if not _vehicle_exists(conn, slug): + return jsonify({"error": "unknown vehicle"}), 404 + now = int(time.time()) + sensor_ids = db.vehicle_sensor_ids(conn, slug) + readings = [] + any_low = False + all_stale = bool(sensor_ids) + for sid in sensor_ids: + row = db.latest_reading_for_sensor(conn, sid) + if row is None: + continue + item = _serialize_reading(row) + item["stale"] = (now - row["ts"]) > cfg.stale_seconds + if not item["stale"]: + all_stale = False + if ( + item["pressure_psi"] is not None + and item["pressure_psi"] < cfg.low_psi + ): + any_low = True + readings.append(item) + return jsonify( + { + "slug": slug, + "readings": readings, + "any_low": any_low, + "all_stale": all_stale, + "low_psi_threshold": cfg.low_psi, + } + ) + + @app.get("/api/vehicles//tpms/history") + def history(slug: str): + now = int(time.time()) + since = int(request.args.get("since", now - 7 * 86400)) + until = int(request.args.get("until", now)) + limit = min(int(request.args.get("limit", 5000)), 20000) + with _connect(cfg.db_path) as conn: + if not _vehicle_exists(conn, slug): + return jsonify({"error": "unknown vehicle"}), 404 + rows = db.history_for_vehicle(conn, slug, since, until, limit) + return jsonify( + { + "slug": slug, + "since": since, + "until": until, + "count": len(rows), + "readings": [_serialize_reading(r) for r in rows], + } + ) + + @app.get("/api/health") + def health(): + with _connect(cfg.db_path) as conn: + last = db.last_event_ts(conn) + now = int(time.time()) + receiver_ok = last is not None and (now - last) <= cfg.stale_seconds + return jsonify( + { + "last_event_ts": last, + "receiver_ok": receiver_ok, + "stale_seconds": cfg.stale_seconds, + "now": now, + } + ) + + return app + + +def _connect(db_path: Path) -> sqlite3.Connection: + """Lightweight read-side connection. Assumes create_app() already ran + `db.connect` once to init the schema. + """ + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + return conn + + +def _vehicle_exists(conn, slug: str) -> bool: + row = conn.execute("SELECT 1 FROM vehicles WHERE slug = ?", (slug,)).fetchone() + return row is not None + + +def main() -> int: # pragma: no cover + app = create_app() + app.run(host="0.0.0.0", port=8090) + return 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/src/tpms/config.py b/src/tpms/config.py new file mode 100644 index 0000000..f430643 --- /dev/null +++ b/src/tpms/config.py @@ -0,0 +1,32 @@ +"""Env-driven config for the TPMS ingest + API services. + +Both services read the same config so a shared DB path and vehicle set stay in sync +without hand-wiring flags. `TPMS_DB` is the only required env var — everything else +falls back to sensible defaults for a plexpi single-vehicle install. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from pathlib import Path + +DEFAULT_DB_PATH = Path("/var/lib/tpms/tpms.sqlite3") +DEFAULT_STALE_SECONDS = 15 * 60 +DEFAULT_LOW_PSI = 30.0 + + +@dataclass(frozen=True) +class Config: + db_path: Path + stale_seconds: int + low_psi: float + + @classmethod + def from_env(cls, env: dict[str, str] | None = None) -> "Config": + e = env if env is not None else os.environ + return cls( + db_path=Path(e.get("TPMS_DB", str(DEFAULT_DB_PATH))), + stale_seconds=int(e.get("TPMS_STALE_SECONDS", str(DEFAULT_STALE_SECONDS))), + low_psi=float(e.get("TPMS_LOW_PSI", str(DEFAULT_LOW_PSI))), + ) diff --git a/src/tpms/db.py b/src/tpms/db.py new file mode 100644 index 0000000..be0eabc --- /dev/null +++ b/src/tpms/db.py @@ -0,0 +1,218 @@ +"""SQLite schema + connection helpers for TPMS readings. + +Three tables: +- vehicles: one row per vehicle (slug, freq, decoder). Slice-1 seeds Mazda CX-9 only. +- sensors: one row per known physical sensor. Position is nullable — slice-1 does not + map sensor→corner (Tom's answer to Q4: any-tire-low is sufficient). +- readings: one row per rtl_433 event. Primary key (ts, sensor_id) drops perfect dupes + without a UNIQUE index round-trip. + +`connect` is idempotent: safe to re-call on every ingest event. +""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +SCHEMA_VERSION = 1 + +_SCHEMA_SQL = """ +CREATE TABLE IF NOT EXISTS vehicles ( + id INTEGER PRIMARY KEY, + slug TEXT UNIQUE NOT NULL, + make TEXT NOT NULL, + model TEXT NOT NULL, + year INTEGER NOT NULL, + vin TEXT, + frequency_hz INTEGER NOT NULL, + decoder TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS sensors ( + vehicle_id INTEGER NOT NULL REFERENCES vehicles(id), + sensor_id TEXT NOT NULL, + position TEXT, + PRIMARY KEY (vehicle_id, sensor_id) +); + +CREATE TABLE IF NOT EXISTS readings ( + ts INTEGER NOT NULL, + sensor_id TEXT NOT NULL, + pressure_kpa REAL, + temperature_c REAL, + battery_ok INTEGER, + raw_json TEXT NOT NULL, + PRIMARY KEY (ts, sensor_id) +); + +CREATE INDEX IF NOT EXISTS readings_by_sensor_ts + ON readings(sensor_id, ts DESC); + +CREATE INDEX IF NOT EXISTS readings_by_ts + ON readings(ts DESC); + +CREATE TABLE IF NOT EXISTS schema_version ( + version INTEGER PRIMARY KEY +); +""" + +# Slice-1 vehicle set. Slug "mazda-cx9" is stable and used by the API path. +# rtl_433 register 156 (Abarth-124Spider) is confirmed working at 315 MHz for the +# Mazda VDO sensors per issue #1 research. +_SEED_VEHICLES = [ + { + "slug": "mazda-cx9", + "make": "Mazda", + "model": "CX-9", + "year": 2013, + "vin": "JM3TB3CV0D0420001", + "frequency_hz": 315_000_000, + "decoder": "r156", + }, +] + + +def connect(db_path: Path) -> sqlite3.Connection: + """Open a connection, ensure schema, seed the vehicle table. + + Runs on every ingest event — SQLite's IF NOT EXISTS keeps this cheap. WAL mode + lets the API reader concurrent-read while ingest writes. + """ + db_path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA foreign_keys=ON") + conn.executescript(_SCHEMA_SQL) + _seed_vehicles(conn) + _record_version(conn) + conn.commit() + return conn + + +def _seed_vehicles(conn: sqlite3.Connection) -> None: + for v in _SEED_VEHICLES: + conn.execute( + """ + INSERT INTO vehicles (slug, make, model, year, vin, frequency_hz, decoder) + VALUES (:slug, :make, :model, :year, :vin, :frequency_hz, :decoder) + ON CONFLICT(slug) DO UPDATE SET + make = excluded.make, + model = excluded.model, + year = excluded.year, + vin = excluded.vin, + frequency_hz = excluded.frequency_hz, + decoder = excluded.decoder + """, + v, + ) + + +def _record_version(conn: sqlite3.Connection) -> None: + conn.execute( + "INSERT INTO schema_version (version) VALUES (?) " + "ON CONFLICT(version) DO NOTHING", + (SCHEMA_VERSION,), + ) + + +def upsert_reading( + conn: sqlite3.Connection, + ts: int, + sensor_id: str, + pressure_kpa: float | None, + temperature_c: float | None, + battery_ok: int | None, + raw_json: str, +) -> bool: + """Insert a reading; return True if new, False if a duplicate was skipped.""" + cur = conn.execute( + """ + INSERT INTO readings (ts, sensor_id, pressure_kpa, temperature_c, battery_ok, raw_json) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(ts, sensor_id) DO NOTHING + """, + (ts, sensor_id, pressure_kpa, temperature_c, battery_ok, raw_json), + ) + conn.commit() + return cur.rowcount > 0 + + +def register_sensor( + conn: sqlite3.Connection, vehicle_slug: str, sensor_id: str +) -> None: + """Ensure a `sensors` row exists so the API can enumerate vehicle→sensors. + + Positions stay NULL — slice-1 does not map sensor→corner. + """ + conn.execute( + """ + INSERT INTO sensors (vehicle_id, sensor_id, position) + SELECT id, ?, NULL FROM vehicles WHERE slug = ? + ON CONFLICT(vehicle_id, sensor_id) DO NOTHING + """, + (sensor_id, vehicle_slug), + ) + conn.commit() + + +def vehicle_sensor_ids(conn: sqlite3.Connection, slug: str) -> list[str]: + rows = conn.execute( + """ + SELECT s.sensor_id + FROM sensors s + JOIN vehicles v ON v.id = s.vehicle_id + WHERE v.slug = ? + ORDER BY s.sensor_id + """, + (slug,), + ).fetchall() + return [r["sensor_id"] for r in rows] + + +def latest_reading_for_sensor( + conn: sqlite3.Connection, sensor_id: str +) -> sqlite3.Row | None: + return conn.execute( + """ + SELECT ts, sensor_id, pressure_kpa, temperature_c, battery_ok + FROM readings + WHERE sensor_id = ? + ORDER BY ts DESC + LIMIT 1 + """, + (sensor_id,), + ).fetchone() + + +def history_for_vehicle( + conn: sqlite3.Connection, + slug: str, + since: int, + until: int, + limit: int = 5000, +) -> list[sqlite3.Row]: + # JOIN sensors on (vehicle_id, sensor_id) not sensor_id alone: TPMS sensor + # IDs aren't globally unique, so if a second vehicle is ever seeded, an + # overlapping ID would otherwise cross-contaminate the history results. + return conn.execute( + """ + SELECT r.ts, r.sensor_id, r.pressure_kpa, r.temperature_c, r.battery_ok + FROM readings r + JOIN vehicles v ON v.slug = ? + JOIN sensors s + ON s.vehicle_id = v.id + AND s.sensor_id = r.sensor_id + WHERE r.ts >= ? + AND r.ts <= ? + ORDER BY r.ts DESC + LIMIT ? + """, + (slug, since, until, limit), + ).fetchall() + + +def last_event_ts(conn: sqlite3.Connection) -> int | None: + row = conn.execute("SELECT MAX(ts) AS ts FROM readings").fetchone() + return row["ts"] if row and row["ts"] is not None else None diff --git a/src/tpms/ingest.py b/src/tpms/ingest.py new file mode 100644 index 0000000..e81c578 --- /dev/null +++ b/src/tpms/ingest.py @@ -0,0 +1,201 @@ +"""Ingest daemon — reads rtl_433 JSON events from stdin, writes to SQLite. + +rtl_433 emits one JSON object per line when invoked with `-F json`. This module +parses each line, filters to the configured vehicle's decoder, normalizes the +fields we care about, and upserts a reading. + +Non-matching lines (other decoders, malformed JSON, log messages) are logged and +skipped — the daemon must not exit on a bad line or the whole capture pipeline +stops. +""" + +from __future__ import annotations + +import json +import logging +import sqlite3 +import sys +from collections.abc import Iterable +from datetime import datetime, timezone +from typing import IO + +from . import db +from .config import Config + +log = logging.getLogger(__name__) + +_VEHICLE_SLUG = "mazda-cx9" + + +def parse_event(line: str) -> dict | None: + """Return the parsed dict for a rtl_433 JSON line, or None if unusable. + + Returning None (not raising) is intentional — the daemon must survive bad + lines without dropping the pipe. + """ + line = line.strip() + if not line: + return None + try: + return json.loads(line) + except json.JSONDecodeError: + log.warning("skipping non-JSON line: %.120s", line) + return None + + +def is_target_vehicle(event: dict) -> bool: + """Match rtl_433 model to the Mazda VDO decoder. + + The comparison is case-insensitive because the model string varies across + rtl_433 versions (e.g. `Abarth-124Spider`, `Abarth 124Spider`). + """ + model = ( + str(event.get("model", "")) + .lower() + .replace("-", "") + .replace(" ", "") + .replace("_", "") + ) + return model == "abarth124spider" + + +def _psi_to_kpa(psi: float) -> float: + return psi * 6.8947572932 + + +def normalize(event: dict, now: datetime | None = None) -> dict | None: + """Extract the fields we store from an rtl_433 event. + + Returns a dict of {ts, sensor_id, pressure_kpa, temperature_c, battery_ok, raw_json} + or None if the event lacks the identifying fields we need. + """ + sensor_id = event.get("id") + if sensor_id is None: + return None + sensor_id = str(sensor_id) + + ts = _parse_time(event, now) + + pressure_kpa: float | None + if "pressure_kPa" in event: + pressure_kpa = float(event["pressure_kPa"]) + elif "pressure_kpa" in event: + pressure_kpa = float(event["pressure_kpa"]) + elif "pressure_PSI" in event: + pressure_kpa = _psi_to_kpa(float(event["pressure_PSI"])) + else: + pressure_kpa = None + + temperature_c: float | None + if "temperature_C" in event: + temperature_c = float(event["temperature_C"]) + elif "temperature_c" in event: + temperature_c = float(event["temperature_c"]) + else: + temperature_c = None + + battery_ok = _parse_battery(event) + + return { + "ts": ts, + "sensor_id": sensor_id, + "pressure_kpa": pressure_kpa, + "temperature_c": temperature_c, + "battery_ok": battery_ok, + "raw_json": json.dumps(event, sort_keys=True), + } + + +def _parse_time(event: dict, now: datetime | None) -> int: + """Parse rtl_433's `time` string into unix seconds UTC. + + rtl_433 emits `time` as `YYYY-MM-DD HH:MM:SS` — **assumed to be UTC** because + the capture unit invokes rtl_433 with `-M utc` (see deploy/plexpi/tpms- + capture.service). Without `-M utc`, rtl_433 emits system-local time with no + tz suffix, which would silently skew every reading by the local UTC offset. + If the field is missing or unparseable, fall back to injected `now` (or the + current wall clock). + """ + raw = event.get("time") + if isinstance(raw, str): + for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S"): + try: + dt = datetime.strptime(raw, fmt).replace(tzinfo=timezone.utc) + return int(dt.timestamp()) + except ValueError: + continue + if now is None: + now = datetime.now(tz=timezone.utc) + return int(now.timestamp()) + + +def _parse_battery(event: dict) -> int | None: + """Battery arrives as `battery_ok: 1` in newer rtl_433, `battery: "OK"` in older.""" + if "battery_ok" in event: + return 1 if event["battery_ok"] else 0 + if "battery" in event: + return 1 if str(event["battery"]).lower() == "ok" else 0 + return None + + +def process_line(conn: sqlite3.Connection, line: str) -> bool: + """Parse + persist one line. Return True if a new reading was written.""" + event = parse_event(line) + if event is None: + return False + if not is_target_vehicle(event): + return False + normalized = normalize(event) + if normalized is None: + return False + db.register_sensor(conn, _VEHICLE_SLUG, normalized["sensor_id"]) + written = db.upsert_reading( + conn, + ts=normalized["ts"], + sensor_id=normalized["sensor_id"], + pressure_kpa=normalized["pressure_kpa"], + temperature_c=normalized["temperature_c"], + battery_ok=normalized["battery_ok"], + raw_json=normalized["raw_json"], + ) + if written: + log.info( + "tpms reading id=%s kpa=%s c=%s", + normalized["sensor_id"], + normalized["pressure_kpa"], + normalized["temperature_c"], + ) + return written + + +def run( + stream: IO[str] | Iterable[str] = sys.stdin, + config: Config | None = None, +) -> int: + """Consume the stream to EOF. Returns the number of new readings written.""" + cfg = config or Config.from_env() + conn = db.connect(cfg.db_path) + try: + written = 0 + for line in stream: + try: + if process_line(conn, line): + written += 1 + except Exception: # pragma: no cover - defensive + log.exception("failed to process line: %.120s", line.strip()) + return written + finally: + conn.close() + + +def main() -> int: # pragma: no cover - thin CLI wrapper + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(name)s %(levelname)s %(message)s", + ) + run() + return 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/tests/tpms/__init__.py b/tests/tpms/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/tpms/conftest.py b/tests/tpms/conftest.py new file mode 100644 index 0000000..d09f52c --- /dev/null +++ b/tests/tpms/conftest.py @@ -0,0 +1,14 @@ +from __future__ import annotations + +import pytest + +from src.tpms.config import Config + + +@pytest.fixture +def config(tmp_path) -> Config: + return Config( + db_path=tmp_path / "tpms.sqlite3", + stale_seconds=900, + low_psi=30.0, + ) diff --git a/tests/tpms/fixtures.py b/tests/tpms/fixtures.py new file mode 100644 index 0000000..5a7fd59 --- /dev/null +++ b/tests/tpms/fixtures.py @@ -0,0 +1,49 @@ +"""rtl_433 JSON fixtures for TPMS tests. + +Shape modeled on real Abarth-124Spider decoder output. Field names track what +rtl_433 emits at CLI (`-F json`) — case matters (`pressure_kPa`, `temperature_C`). +""" + +from __future__ import annotations + +import json + +SAMPLE_EVENT = { + "time": "2026-07-12 19:30:00", + "model": "Abarth-124Spider", + "type": "TPMS", + "id": "6ec8f7a1", + "flags": "00", + "pressure_kPa": 220.0, + "temperature_C": 28.5, + "battery_ok": 1, + "mic": "CRC", +} + + +def sample_line(**overrides) -> str: + """Return a single-line JSON event string with optional overrides.""" + event = dict(SAMPLE_EVENT) + event.update(overrides) + return json.dumps(event) + "\n" + + +def sample_stream(*event_overrides: dict) -> list[str]: + if not event_overrides: + return [sample_line()] + return [sample_line(**o) for o in event_overrides] + + +NON_TARGET_LINE = ( + json.dumps( + { + "time": "2026-07-12 19:30:05", + "model": "Somfy-RTS", + "id": "12345", + "code": "0x1234", + } + ) + + "\n" +) + +MALFORMED_LINE = "this is not JSON\n" diff --git a/tests/tpms/test_api.py b/tests/tpms/test_api.py new file mode 100644 index 0000000..426a697 --- /dev/null +++ b/tests/tpms/test_api.py @@ -0,0 +1,116 @@ +from __future__ import annotations + +import io +import time + +import pytest + +from src.tpms import db, ingest +from src.tpms.api import create_app + +from . import fixtures + + +@pytest.fixture +def client(config): + ingest.run(io.StringIO(fixtures.sample_line()), config=config) + app = create_app(config) + app.testing = True + return app.test_client() + + +@pytest.fixture +def empty_client(config): + conn = db.connect(config.db_path) + conn.close() + app = create_app(config) + app.testing = True + return app.test_client() + + +def test_list_vehicles(client): + resp = client.get("/api/vehicles") + assert resp.status_code == 200 + body = resp.get_json() + assert body[0]["slug"] == "mazda-cx9" + assert body[0]["frequency_hz"] == 315_000_000 + + +def test_latest_returns_reading_and_low_flag(config): + ingest.run( + io.StringIO(fixtures.sample_line(**{"pressure_kPa": 150.0})), + config=config, + ) + app = create_app(config) + app.testing = True + client = app.test_client() + resp = client.get("/api/vehicles/mazda-cx9/tpms/latest") + assert resp.status_code == 200 + body = resp.get_json() + assert body["slug"] == "mazda-cx9" + assert body["any_low"] is True + assert body["low_psi_threshold"] == 30.0 + assert len(body["readings"]) == 1 + reading = body["readings"][0] + assert reading["pressure_kpa"] == 150.0 + assert reading["pressure_psi"] is not None + assert reading["pressure_psi"] < 30.0 + + +def test_latest_reports_all_stale_when_over_threshold(config): + old_ts = int(time.time()) - 4000 + ingest.run( + io.StringIO(fixtures.sample_line(time=_fmt(old_ts), **{"pressure_kPa": 220.0})), + config=config, + ) + app = create_app(config) + app.testing = True + resp = app.test_client().get("/api/vehicles/mazda-cx9/tpms/latest") + body = resp.get_json() + assert body["all_stale"] is True + assert body["readings"][0]["stale"] is True + + +def test_latest_unknown_vehicle_is_404(empty_client): + resp = empty_client.get("/api/vehicles/nope/tpms/latest") + assert resp.status_code == 404 + + +def test_history_respects_since_until(config): + events = [ + {"time": _fmt(1000), "id": "6ec8f7a1", "pressure_kPa": 200.0}, + {"time": _fmt(2000), "id": "6ec8f7a1", "pressure_kPa": 210.0}, + {"time": _fmt(3000), "id": "6ec8f7a1", "pressure_kPa": 220.0}, + ] + ingest.run(io.StringIO("".join(fixtures.sample_stream(*events))), config=config) + app = create_app(config) + app.testing = True + resp = app.test_client().get( + "/api/vehicles/mazda-cx9/tpms/history?since=1500&until=2500" + ) + body = resp.get_json() + assert body["count"] == 1 + assert body["readings"][0]["ts"] == 2000 + + +def test_health_no_data(empty_client): + resp = empty_client.get("/api/health") + body = resp.get_json() + assert body["last_event_ts"] is None + assert body["receiver_ok"] is False + + +def test_health_receiver_ok_when_recent(client): + resp = client.get("/api/health") + body = resp.get_json() + assert body["last_event_ts"] is not None + # Sample event's time is 2026-07-12 19:30:00 which will be in the past by the + # time this runs, so receiver_ok depends on stale threshold vs. clock. + assert isinstance(body["receiver_ok"], bool) + + +def _fmt(ts: int) -> str: + """Convert a unix ts back to rtl_433's `YYYY-MM-DD HH:MM:SS` UTC format.""" + from datetime import datetime, timezone + + return datetime.fromtimestamp(ts, tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S") diff --git a/tests/tpms/test_db.py b/tests/tpms/test_db.py new file mode 100644 index 0000000..3d45af3 --- /dev/null +++ b/tests/tpms/test_db.py @@ -0,0 +1,70 @@ +from __future__ import annotations + +from src.tpms import db + + +def test_connect_creates_schema_and_seeds_vehicle(config): + conn = db.connect(config.db_path) + try: + row = conn.execute( + "SELECT slug, make, model, year, frequency_hz, decoder FROM vehicles" + ).fetchone() + assert row["slug"] == "mazda-cx9" + assert row["make"] == "Mazda" + assert row["frequency_hz"] == 315_000_000 + assert row["decoder"] == "r156" + finally: + conn.close() + + +def test_connect_is_idempotent(config): + for _ in range(3): + conn = db.connect(config.db_path) + conn.close() + conn = db.connect(config.db_path) + try: + count = conn.execute("SELECT COUNT(*) AS n FROM vehicles").fetchone()["n"] + assert count == 1 + finally: + conn.close() + + +def test_upsert_reading_returns_false_on_duplicate(config): + conn = db.connect(config.db_path) + try: + assert db.upsert_reading(conn, 100, "abc", 220.0, 25.0, 1, "{}") is True + assert db.upsert_reading(conn, 100, "abc", 220.0, 25.0, 1, "{}") is False + finally: + conn.close() + + +def test_register_sensor_is_idempotent(config): + conn = db.connect(config.db_path) + try: + db.register_sensor(conn, "mazda-cx9", "sensor-A") + db.register_sensor(conn, "mazda-cx9", "sensor-A") + ids = db.vehicle_sensor_ids(conn, "mazda-cx9") + assert ids == ["sensor-A"] + finally: + conn.close() + + +def test_history_bounded_by_since_until(config): + conn = db.connect(config.db_path) + try: + db.register_sensor(conn, "mazda-cx9", "s1") + db.upsert_reading(conn, 1000, "s1", 200.0, 20.0, 1, "{}") + db.upsert_reading(conn, 2000, "s1", 210.0, 22.0, 1, "{}") + db.upsert_reading(conn, 3000, "s1", 220.0, 24.0, 1, "{}") + rows = db.history_for_vehicle(conn, "mazda-cx9", since=1500, until=2500) + assert [r["ts"] for r in rows] == [2000] + finally: + conn.close() + + +def test_last_event_ts_empty_returns_none(config): + conn = db.connect(config.db_path) + try: + assert db.last_event_ts(conn) is None + finally: + conn.close() diff --git a/tests/tpms/test_ingest.py b/tests/tpms/test_ingest.py new file mode 100644 index 0000000..0f6859c --- /dev/null +++ b/tests/tpms/test_ingest.py @@ -0,0 +1,93 @@ +from __future__ import annotations + +import io + +from src.tpms import db, ingest + +from . import fixtures + + +def test_parse_event_survives_non_json(): + assert ingest.parse_event(fixtures.MALFORMED_LINE) is None + assert ingest.parse_event("") is None + assert ingest.parse_event(" \n") is None + + +def test_is_target_vehicle_matches_variants(): + for model in ("Abarth-124Spider", "Abarth 124Spider", "abarth_124spider"): + assert ingest.is_target_vehicle({"model": model}) + assert not ingest.is_target_vehicle({"model": "Somfy-RTS"}) + assert not ingest.is_target_vehicle({}) + + +def test_normalize_extracts_pressure_and_temperature(): + event = fixtures.SAMPLE_EVENT.copy() + normalized = ingest.normalize(event) + assert normalized is not None + assert normalized["sensor_id"] == "6ec8f7a1" + assert normalized["pressure_kpa"] == 220.0 + assert normalized["temperature_c"] == 28.5 + assert normalized["battery_ok"] == 1 + assert normalized["ts"] > 0 + + +def test_normalize_converts_psi_to_kpa(): + event = { + "time": "2026-07-12 19:30:00", + "model": "Abarth-124Spider", + "id": "abc", + "pressure_PSI": 32.0, + } + normalized = ingest.normalize(event) + assert normalized is not None + assert normalized["pressure_kpa"] == 32.0 * 6.8947572932 + + +def test_normalize_missing_id_returns_none(): + assert ingest.normalize({"model": "Abarth-124Spider"}) is None + + +def test_normalize_handles_old_battery_field(): + event = fixtures.SAMPLE_EVENT.copy() + del event["battery_ok"] + event["battery"] = "OK" + normalized = ingest.normalize(event) + assert normalized is not None + assert normalized["battery_ok"] == 1 + + event["battery"] = "LOW" + normalized = ingest.normalize(event) + assert normalized is not None + assert normalized["battery_ok"] == 0 + + +def test_run_writes_matching_events(config): + stream = io.StringIO( + "".join( + [ + fixtures.sample_line(), + fixtures.NON_TARGET_LINE, + fixtures.MALFORMED_LINE, + fixtures.sample_line(id="7fa22b03", **{"pressure_kPa": 210.0}), + ] + ) + ) + written = ingest.run(stream, config=config) + assert written == 2 + + conn = db.connect(config.db_path) + try: + rows = conn.execute( + "SELECT sensor_id, pressure_kpa FROM readings ORDER BY sensor_id" + ).fetchall() + assert [r["sensor_id"] for r in rows] == ["6ec8f7a1", "7fa22b03"] + ids = db.vehicle_sensor_ids(conn, "mazda-cx9") + assert ids == ["6ec8f7a1", "7fa22b03"] + finally: + conn.close() + + +def test_duplicate_events_are_deduped(config): + line = fixtures.sample_line() + stream = io.StringIO(line + line + line) + assert ingest.run(stream, config=config) == 1 diff --git a/uv.lock b/uv.lock index 8a75f78..39c7dd4 100644 --- a/uv.lock +++ b/uv.lock @@ -243,6 +243,23 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/81/47/dd9a212ef6e343a6857485ffe25bba537304f1913bdbed446a23f7f592e1/filelock-3.29.0-py3-none-any.whl", hash = "sha256:96f5f6344709aa1572bbf631c640e4ebeeb519e08da902c39a001882f30ac258", size = 39812, upload-time = "2026-04-19T15:39:08.752Z" }, ] +[[package]] +name = "flask" +version = "3.1.3" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "blinker" }, + { name = "click" }, + { name = "itsdangerous" }, + { name = "jinja2" }, + { name = "markupsafe" }, + { name = "werkzeug" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/26/00/35d85dcce6c57fdc871f3867d465d780f302a175ea360f62533f12b27e2b/flask-3.1.3.tar.gz", hash = "sha256:0ef0e52b8a9cd932855379197dd8f94047b359ca0a78695144304cb45f87c9eb", size = 759004, upload-time = "2026-02-19T05:00:57.678Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/7f/9c/34f6962f9b9e9c71f6e5ed806e0d0ff03c9d1b0b2340088a0cf4bce09b18/flask-3.1.3-py3-none-any.whl", hash = "sha256:f4bcbefc124291925f1a26446da31a5178f9483862233b23c0c96a20701f670c", size = 103424, upload-time = "2026-02-19T05:00:56.027Z" }, +] + [[package]] name = "fonttools" version = "4.62.1" @@ -364,6 +381,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" }, ] +[[package]] +name = "itsdangerous" +version = "2.2.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/9c/cb/8ac0172223afbccb63986cc25049b154ecfb5e85932587206f42317be31d/itsdangerous-2.2.0.tar.gz", hash = "sha256:e0050c0b7da1eea53ffaf149c0cfbb5c6e2e2b69c4bef22c81fa6eb73e5f6173", size = 54410, upload-time = "2024-04-16T21:28:15.614Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/04/96/92447566d16df59b2a776c0fb82dbc4d9e07cd95062562af01e408583fc4/itsdangerous-2.2.0-py3-none-any.whl", hash = "sha256:c6242fc49e35958c8b15141343aa660db5fc54d4f13a1db01a3f5891b98700ef", size = 16234, upload-time = "2024-04-16T21:28:14.499Z" }, +] + [[package]] name = "jinja2" version = "3.1.6" @@ -947,6 +973,7 @@ version = "0.1.0" source = { virtual = "." } dependencies = [ { name = "click" }, + { name = "flask" }, { name = "httpx" }, { name = "matplotlib" }, { name = "numpy" }, @@ -964,6 +991,7 @@ dev = [ [package.metadata] requires-dist = [ { name = "click", specifier = ">=8.1" }, + { name = "flask", specifier = ">=3.1.3" }, { name = "httpx", specifier = ">=0.27" }, { name = "matplotlib", specifier = ">=3.10" }, { name = "numpy", specifier = ">=1.26" }, @@ -1238,3 +1266,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/db/d9/c495884c6e548fce18a8f40568ff120bc3a4b7b99813081c8ac0c936fa64/watchdog-6.0.0-py3-none-win_amd64.whl", hash = "sha256:cbafb470cf848d93b5d013e2ecb245d4aa1c8fd0504e863ccefa32445359d680", size = 79070, upload-time = "2024-11-01T14:07:10.686Z" }, { url = "https://files.pythonhosted.org/packages/33/e8/e40370e6d74ddba47f002a32919d91310d6074130fe4e17dabcafc15cbf1/watchdog-6.0.0-py3-none-win_ia64.whl", hash = "sha256:a1914259fa9e1454315171103c6a30961236f508b9b623eae470268bbcc6a22f", size = 79067, upload-time = "2024-11-01T14:07:11.845Z" }, ] + +[[package]] +name = "werkzeug" +version = "3.1.8" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "markupsafe" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/dd/b2/381be8cfdee792dd117872481b6e378f85c957dd7c5bca38897b08f765fd/werkzeug-3.1.8.tar.gz", hash = "sha256:9bad61a4268dac112f1c5cd4630a56ede601b6ed420300677a869083d70a4c44", size = 875852, upload-time = "2026-04-02T18:49:14.268Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/93/8c/2e650f2afeb7ee576912636c23ddb621c91ac6a98e66dc8d29c3c69446e1/werkzeug-3.1.8-py3-none-any.whl", hash = "sha256:63a77fb8892bf28ebc3178683445222aa500e48ebad5ec77b0ad80f8726b1f50", size = 226459, upload-time = "2026-04-02T18:49:12.72Z" }, +]