From 478a9c91d6a87009d7b8862a1d356276d68a7b7f Mon Sep 17 00:00:00 2001 From: Petrus Pennanen Date: Wed, 7 Oct 2026 17:25:41 +0300 Subject: [PATCH] Radiation page: RadiaCode readings from a JSON bridge, radwatch or HA /radiation shows dose rate, count rate, accumulated dose, battery, temperature, reading age and radwatch's watch alarm; /api/radiation serves the same JSON. Served like /nerd (HTML read from disk per request) behind the same auth. carwatch/radiation.py reads one of three sources chosen by radiation.source in config.json: - json: a bridge's plain-GET JSON object (recommended, no token) - radwatch: `radwatch.py status` on this box, read-only; the alarm verdict is radwatch's own, never recomputed here - ha: HA REST /api/states/, token from config or env, only sent to a private host (mercedesme._is_private_ha) Five states that never look alike: unconfigured, error (with reason), empty, stale (older than stale_seconds, default 60, or age unknown), live. Fields a source has not sent yet read "not yet reported". status() never raises; an exception becomes an error state with its text. tests/test_radiation.py: 47 tests (HA 401/404/unreachable, bridge bad JSON/missing fields/stale ts, empty radwatch DB, an older radwatch without status) plus a negative control that collapses the stale and error distinctions and checks that the scenario table catches it. Co-Authored-By: Claude Opus 5.5 --- FEATURES.md | 5 + README.md | 27 ++ carwatch/radiation.html | 88 +++++++ carwatch/radiation.py | 328 +++++++++++++++++++++++ carwatch/webchat.py | 17 +- config.example.json | 17 ++ tests/test_radiation.py | 562 ++++++++++++++++++++++++++++++++++++++++ 7 files changed, 1043 insertions(+), 1 deletion(-) create mode 100644 carwatch/radiation.html create mode 100644 carwatch/radiation.py create mode 100644 tests/test_radiation.py diff --git a/FEATURES.md b/FEATURES.md index ebef8f8..6dbc909 100644 --- a/FEATURES.md +++ b/FEATURES.md @@ -29,6 +29,11 @@ - `bench/bench.sh`: one command on a fresh Pi 5 builds llama.cpp and whisper.cpp, downloads a quantized Gemma 4 E2B, benchmarks real tokens/sec, probes the dashcam API and the mirror's ADB - systemd service + idempotent installer; config lives in `/etc/carwatch`, never in the repo +**Radiation page** +- `/radiation`: RadiaCode 10x dose rate, count rate, accumulated dose, battery, temperature, reading age and radwatch's alarm verdict +- Three sources by config: a JSON bridge (recommended), radwatch's SQLite on the same box, or Home Assistant REST +- Live / stale / source error / no reading yet / not configured are five different screens, never an empty page + ## Probe-gated (wired the day the hardware answers) - Parked-impact clip posts (needs the camera's event flags from the probe) diff --git a/README.md b/README.md index a01e87f..e173e2d 100644 --- a/README.md +++ b/README.md @@ -368,6 +368,33 @@ through `carwatch/config.py`, so there is exactly one place to edit: brain); without it the car says it has not been described yet - `home_ssids` — wifi networks that mean "parked at home" - `wolfbox` — dashcam AP name/password and poll interval +- `radiation` — where the `/radiation` page reads a RadiaCode 10x from (see below) + +### Radiation page (`/radiation`) + +Dose rate (µSv/h), count rate (CPS), accumulated dose, battery, temperature, +the age of the last reading and, with radwatch, its `watch` alarm verdict. +`/api/radiation` serves the same as JSON. `radiation.source` picks one of: + +- `json` (recommended) — a bridge on another machine serves one object by + plain GET, no auth: `{"dose_rate", "count_rate", "ts", ...}` with + `accumulated_dose`, `battery`, `temperature`, `hardness`, `serial`, + `firmware` optional. `ts` is unix seconds or ISO 8601 (give an offset). + Set `radiation.json.url`, e.g. `http://bridge.example:8099/radiacode.json`. +- `radwatch` — [radwatch](https://github.com/ThinkOffApp/radwatch) logs on + this box; the page runs `radwatch.py status` (read-only) with + `radiation.radwatch.python` / `script` / `db`. Needs a radwatch with the + `status` subcommand. +- `ha` — Home Assistant REST, `GET /api/states/`. + `radiation.ha.url` and a long-lived token in `radiation.ha.token`, or + `$CARWATCH_RADIATION_HA_URL` / `$CARWATCH_RADIATION_HA_TOKEN` (env wins). + The token is only sent to a LAN / `.local` / Tailscale host. + +`radiation.stale_seconds` (default 60) is when a reading stops counting as +live. The page shows five states that never look alike: no source +configured, source error (with the reason), connected but no reading yet, +stale, and live. A field the source has not sent yet reads "not yet +reported", not an error. ## Bench-day probe diff --git a/carwatch/radiation.html b/carwatch/radiation.html new file mode 100644 index 0000000..9377a65 --- /dev/null +++ b/carwatch/radiation.html @@ -0,0 +1,88 @@ + + + + + + CarWatch Radiation + + + +

☢ Radiation

RadiaCode readings · 💬 Chat with the car · all PIDs

+
+ +
+ +
+ + + diff --git a/carwatch/radiation.py b/carwatch/radiation.py new file mode 100644 index 0000000..7014a7d --- /dev/null +++ b/carwatch/radiation.py @@ -0,0 +1,328 @@ +"""Radiation readings for the /radiation page - a RadiaCode 10x, read from +whichever of three sources the config names. CarWatch only READS here: the +statistics (dose conversion, the watch alarm's sigma) belong to radwatch and +are shown as radwatch computed them, never recomputed. + +Config block (config.json, see config.example.json): + + "radiation": { + "source": "json" | "radwatch" | "ha" | "", empty or absent = off + "stale_seconds": 60, + "json": {"url": ""}, + "radwatch": {"python": "python3", "script": "~/radwatch/radwatch.py", + "db": "~/radwatch/radwatch.sqlite"}, + "ha": {"url": "", "token": "", "entity_prefix": "sensor.radiacode_"} + } + + json a bridge on another machine serves one JSON object over plain GET + (no auth) every few seconds. The recommended source: it needs no + token and no radwatch on this box. + radwatch radwatch runs on THIS box; `radwatch.py status` reads its SQLite + read-only and prints the latest reading plus the watch verdict. + ha Home Assistant REST, GET /api/states/, with a long-lived + token. $CARWATCH_RADIATION_HA_URL / $CARWATCH_RADIATION_HA_TOKEN + beat the config. The token is only ever sent to a private / LAN / + Tailscale host (the same rule as mercedesme, issue #14). + +Five states, and the page must never let two of them look alike (an empty +page for an exception is the bug this module exists to not have): + + unconfigured no source chosen + error source chosen but misconfigured, unreachable or refusing; + `error` says which + empty source answered, holds no reading yet + stale a reading exists but is older than stale_seconds, or its age + cannot be known (an unknown age is never shown as live) + live a reading younger than stale_seconds +""" + +from __future__ import annotations + +import json +import os +import subprocess +import time +import urllib.error +import urllib.request +from datetime import datetime + +from carwatch.config import load_raw + +DEFAULT_STALE_S = 60 +_HTTP_TIMEOUT = 4.0 # fail fast: the page polls, a hung source must not hang it +_RADWATCH_TIMEOUT = 10.0 + +# key, label, unit for the sources that do not report their own units +FIELDS = [ + ("dose_rate", "Dose rate", "µSv/h"), + ("count_rate", "Count rate", "CPS"), + ("accumulated_dose", "Accumulated dose", "µSv"), + ("battery", "Battery", "%"), + ("temperature", "Temperature", "°C"), +] + + +NOT_YET = "not yet reported" +UNAVAILABLE = "unavailable" + + +class SourceError(Exception): + """The source could not be read. The message is shown on the page.""" + + +def _num(v): + """A finite float, or None. HA states are strings; "unavailable" and + "unknown" are not numbers and must not become 0.""" + if isinstance(v, bool) or v is None: + return None + try: + f = float(v) + except (TypeError, ValueError): + return None + return f if f == f and f not in (float("inf"), float("-inf")) else None + + +def _age_from_ts(ts, now: float) -> float | None: + """Seconds since ts, which may be unix seconds or ISO 8601. A naive ISO + stamp is read as this box's local time (radwatch's convention); give an + offset to be unambiguous. None when it does not parse.""" + n = _num(ts) # unix seconds, as a number or a numeric string + if n is not None: + return round(now - n, 1) + if not isinstance(ts, str) or not ts.strip(): + return None + try: + t = datetime.fromisoformat(ts.strip().replace("Z", "+00:00")) + except ValueError: + return None + return round(now - t.timestamp(), 1) + + +def _expand(p: str) -> str: + return os.path.expanduser(str(p or "")) + + +# ── sources ───────────────────────────────────────────────────────────── +# Each returns a dict: values {key: number|None}, units {key: str}, age_s, +# age_basis, and optionally notes {key: why the value is None} / extra / +# alarm / alarm_note / empty_note. Each raises SourceError for anything the +# owner has to fix or wait out. A field the source has simply not sent yet +# is a note ("not yet reported"), never an error. + +def _get_json(url: str, headers: dict | None = None): + req = urllib.request.Request(url, headers=headers or {}) + with urllib.request.urlopen(req, timeout=_HTTP_TIMEOUT) as r: + return r.read() + + +def read_json(c: dict, now: float) -> dict: + url = str((c.get("json") or {}).get("url") or "").strip() + if not url: + raise SourceError("radiation.json.url is not set") + if not url.startswith(("http://", "https://")): + raise SourceError("radiation.json.url must start with http:// or https://") + try: + raw = _get_json(url) + except urllib.error.HTTPError as e: + raise SourceError(f"bridge answered HTTP {e.code}") + except Exception as e: # URLError, timeout, refused, DNS + raise SourceError(f"bridge unreachable: {getattr(e, 'reason', e)}") + try: + d = json.loads(raw.decode("utf-8", "replace")) + except ValueError as e: + raise SourceError(f"bridge returned invalid JSON: {e}") + if not isinstance(d, dict): + raise SourceError(f"bridge returned JSON {type(d).__name__}, expected an object") + known = [k for k, _, _ in FIELDS] + ["ts"] + if not any(k in d for k in known): + raise SourceError("bridge JSON has none of the expected fields " + "(dose_rate, count_rate, accumulated_dose, battery, temperature, ts)") + # accumulated_dose / battery / temperature / hardness arrive about once a + # minute and are ABSENT for the first minute after a bridge restart. + extra = {k: d[k] for k in ("hardness", "serial", "firmware") if d.get(k) not in (None, "")} + if "served_at" in d: + extra["bridge_lag_s"] = _num(_age_from_ts(d.get("ts"), _num(d.get("served_at")) or now)) + notes = {k: NOT_YET if k not in d else UNAVAILABLE + for k, _, _ in FIELDS if _num(d.get(k)) is None} + age = _age_from_ts(d.get("ts"), now) + return {"values": {k: _num(d.get(k)) for k, _, _ in FIELDS}, "notes": notes, + "units": {}, "extra": extra, "age_s": age, + "alarm_note": "the bridge does not serve the radwatch watch alarm", + "age_basis": "bridge ts" if age is not None else + ("bridge sent no ts" if d.get("ts") in (None, "") else "bridge ts did not parse")} + + +def read_radwatch(c: dict, now: float) -> dict: + rc = c.get("radwatch") or {} + script = _expand(rc.get("script") or "~/radwatch/radwatch.py") + db = _expand(rc.get("db") or "~/radwatch/radwatch.sqlite") + py = _expand(rc.get("python") or "python3") + if not os.path.isfile(script): + raise SourceError(f"radwatch not found at {script} (radiation.radwatch.script)") + try: + p = subprocess.run([py, script, "status", "--db", db], capture_output=True, + text=True, timeout=_RADWATCH_TIMEOUT) + except subprocess.TimeoutExpired: + raise SourceError(f"radwatch status took over {_RADWATCH_TIMEOUT:.0f}s") + except OSError as e: + raise SourceError(f"cannot run radwatch: {e}") + try: + d = json.loads(p.stdout) + except ValueError: + # An older radwatch has no `status` subcommand: argparse exits 2 and + # says so on stderr. Show that, it names the fix. + tail = (p.stderr or p.stdout or "").strip().splitlines()[-1:] or ["no output"] + raise SourceError(f"radwatch status exited {p.returncode}: {tail[0][:200]}") + if not isinstance(d, dict) or not d.get("ok"): + raise SourceError(str((d or {}).get("error") or f"radwatch status exited {p.returncode}")) + r = d.get("reading") or {} + dev = d.get("device") or {} + out = {"values": {"dose_rate": _num(r.get("dose_rate_usv_h")), + "count_rate": _num(r.get("count_rate_cps")), + "accumulated_dose": _num(dev.get("accumulated_dose_usv")), + "battery": _num(dev.get("battery_pct")), + "temperature": _num(dev.get("temperature_c"))}, + "units": {}, "extra": {}, "age_s": _num(r.get("age_s")), + "age_basis": "radwatch reading ts", + "alarm": d.get("watch"), "alarm_note": d.get("watch_note")} + out["notes"] = {k: (NOT_YET if (k in ("accumulated_dose", "battery", "temperature") + and not dev) or not r else UNAVAILABLE) + for k, v in out["values"].items() if v is None} + if dev: + out["extra"]["device_age_s"] = _num(dev.get("age_s")) + elif r: + out["extra"]["device_note"] = "radwatch has logged no battery/temperature record yet" + if not r: + out["empty_note"] = f"radwatch database has no readings yet ({d.get('readings', 0)} rows)" + return out + + +def _ha_settings(c: dict) -> tuple[str, str, str]: + h = c.get("ha") or {} + url = (os.environ.get("CARWATCH_RADIATION_HA_URL") or h.get("url") or "").strip().rstrip("/") + token = (os.environ.get("CARWATCH_RADIATION_HA_TOKEN") or h.get("token") or "").strip() + prefix = (h.get("entity_prefix") or "sensor.radiacode_").strip() + return url, token, prefix + + +def read_ha(c: dict, now: float) -> dict: + from carwatch.mercedesme import _is_private_ha + url, token, prefix = _ha_settings(c) + if not url: + raise SourceError("radiation.ha.url is not set") + if not token: + raise SourceError("no Home Assistant token: set radiation.ha.token " + "or $CARWATCH_RADIATION_HA_TOKEN") + if not _is_private_ha(url): + raise SourceError("refusing to send the HA token to a non-private host " + "(LAN, .local or Tailscale only)") + hdr = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} + states, absent = {}, [] + for key in [k for k, _, _ in FIELDS] + ["last_reading"]: + eid = prefix + key + try: + states[key] = json.loads(_get_json(f"{url}/api/states/{eid}", hdr).decode()) + except urllib.error.HTTPError as e: + if e.code in (401, 403): + raise SourceError(f"Home Assistant rejected the token (HTTP {e.code})") + if e.code == 404: + absent.append(eid) + continue + raise SourceError(f"Home Assistant answered HTTP {e.code} for {eid}") + except ValueError as e: + raise SourceError(f"Home Assistant returned invalid JSON for {eid}: {e}") + except Exception as e: # first unreachable call ends the poll, no 6x timeout + raise SourceError(f"Home Assistant unreachable at {url}: {getattr(e, 'reason', e)}") + if not states: + raise SourceError(f"none of the {prefix}* entities exist in Home Assistant") + values = {k: _num((states.get(k) or {}).get("state")) for k, _, _ in FIELDS} + notes = {k: ("no such entity" if k not in states else + str(states[k].get("state") or UNAVAILABLE)) + for k, v in values.items() if v is None} + units = {k: (states[k].get("attributes") or {}).get("unit_of_measurement") + for k, _, _ in FIELDS if k in states} + age, basis = None, "no last_reading sensor and no dose rate timestamp" + lr = (states.get("last_reading") or {}).get("state") + if lr not in (None, "", "unknown", "unavailable"): + age = _age_from_ts(lr, now) + basis = "last_reading sensor" if age is not None else "last_reading sensor did not parse" + if age is None and "dose_rate" in states: + s = states["dose_rate"] + stamp = s.get("last_reported") or s.get("last_updated") + age = _age_from_ts(stamp, now) if stamp else None + if age is not None: + basis = "HA last_reported of the dose rate entity" + out = {"values": values, "notes": notes, "units": {k: u for k, u in units.items() if u}, + "extra": {"missing_entities": absent} if absent else {}, + "age_s": age, "age_basis": basis, + "alarm_note": "the radwatch watch alarm is not available through Home Assistant"} + if all(v is None for v in values.values()): + out["empty_note"] = "Home Assistant reports every radiacode sensor unavailable or unknown" + return out + + +SOURCES = {"json": read_json, "radwatch": read_radwatch, "ha": read_ha} + + +# ── classification ────────────────────────────────────────────────────── + +def classify(reading: dict, stale_s: float) -> tuple[str, str]: + """(state, message) for a reading a source returned without error.""" + if all(v is None for v in (reading.get("values") or {}).values()): + return "empty", reading.get("empty_note") or "source answered, no reading yet" + age = reading.get("age_s") + if age is None: + return "stale", f"reading age unknown ({reading.get('age_basis')}), not claimed live" + if age < -stale_s: + return "stale", f"reading is {-age:.0f}s in the future: clock skew between boxes" + if age > stale_s: + return "stale", f"last reading {age:.0f}s ago, older than {stale_s:.0f}s" + return "live", f"last reading {max(age, 0):.0f}s ago" + + +def _stale_s(c: dict) -> float: + v = _num(c.get("stale_seconds")) + if not v or v <= 0: + return DEFAULT_STALE_S + return int(v) if v.is_integer() else v + + +def status(cfg: dict | None = None, now: float | None = None) -> dict: + """The page's whole payload. Never raises: any exception becomes an + `error` state carrying the exception text.""" + now = time.time() if now is None else now + out = {"state": "error", "source": None, "message": "", "error": None, + "stale_seconds": DEFAULT_STALE_S, "age_s": None, "age_basis": None, + "values": {}, "extra": {}, "alarm": None, "alarm_note": None, + "checked_at": round(now, 1)} + try: + c = ((load_raw() if cfg is None else cfg).get("radiation") or {}) + if not isinstance(c, dict): + raise SourceError("radiation in config.json must be an object") + out["stale_seconds"] = stale_s = _stale_s(c) + src = str(c.get("source") or "").strip().lower() + if not src: + out.update(state="unconfigured", + message='no radiation source configured: set radiation.source to ' + '"json", "radwatch" or "ha" in config.json') + return out + out["source"] = src + reader = SOURCES.get(src) + if reader is None: + raise SourceError(f'unknown radiation.source "{src}" (json, radwatch or ha)') + r = reader(c, now) + units, notes = r.get("units") or {}, r.get("notes") or {} + out["values"] = {k: {"label": label, "value": r["values"].get(k), + "unit": units.get(k) or unit, + "note": None if r["values"].get(k) is not None + else notes.get(k, UNAVAILABLE)} + for k, label, unit in FIELDS} + out.update(age_s=r.get("age_s"), age_basis=r.get("age_basis"), + extra=r.get("extra") or {}, alarm=r.get("alarm"), + alarm_note=r.get("alarm_note")) + out["state"], out["message"] = classify(r, stale_s) + except SourceError as e: + out.update(state="error", error=str(e), message="source error") + except Exception as e: # a bug, shown as one, never as an empty page + out.update(state="error", error=f"{type(e).__name__}: {e}", message="internal error") + return out diff --git a/carwatch/webchat.py b/carwatch/webchat.py index 2b7d6e7..c19a559 100644 --- a/carwatch/webchat.py +++ b/carwatch/webchat.py @@ -482,7 +482,7 @@ def _switch_later(name: str) -> None: -
trust wifiPi vitalsall PIDsstreamsjournal
+
trust wifiPi vitalsall PIDsradiationstreamsjournal
📊Read
@@ -2946,6 +2946,21 @@ def run(cmd): "application/json") return self._send(200, json.dumps(prov.status()), "application/json") + elif self.path.startswith("/api/radiation"): + # RadiaCode readings for /radiation from the source config names + # (carwatch.radiation). status() never raises: an exception comes + # back as state "error" with its text, never as an empty page. + from carwatch import radiation as _rad + return self._send(200, json.dumps(_rad.status()), "application/json") + elif self.path.startswith("/radiation"): + # Radiation page (carwatch/radiation.html), read from disk per + # request like /nerd so a git pull updates it without a restart. + try: + with open(os.path.join(REPO, "carwatch", "radiation.html"), + encoding="utf-8") as f: + return self._send(200, f.read()) + except Exception as e: + return self._send(404, f"radiation.html missing: {e}") elif self.path.startswith("/cloudcar"): # Minimal setup + status page for the cloud provider: email -> # vendor emails a one-time code -> code -> tokens on the Pi. diff --git a/config.example.json b/config.example.json index 73c24fe..5bb26b4 100644 --- a/config.example.json +++ b/config.example.json @@ -25,5 +25,22 @@ "host": "", "led_count": 30, "brightness": 128 + }, + "radiation": { + "source": "", + "stale_seconds": 60, + "json": { + "url": "" + }, + "radwatch": { + "python": "python3", + "script": "~/radwatch/radwatch.py", + "db": "~/radwatch/radwatch.sqlite" + }, + "ha": { + "url": "http://homeassistant.local:8123", + "token": "", + "entity_prefix": "sensor.radiacode_" + } } } \ No newline at end of file diff --git a/tests/test_radiation.py b/tests/test_radiation.py new file mode 100644 index 0000000..e940f70 --- /dev/null +++ b/tests/test_radiation.py @@ -0,0 +1,562 @@ +"""carwatch.radiation: three sources, five states that must never look alike. + +Every source is exercised against a real local HTTP server or a stand-in +radwatch script, never the network: the JSON bridge, Home Assistant (200, +401, 404, unreachable) and `radwatch.py status` (missing script, missing +DB, empty DB, stale, live, an older radwatch without `status`). + +The negative control at the bottom collapses the stale/error distinction in +a copy of the classifier and checks that the scenario table catches it, so +the table cannot pass vacuously. + +Set RADWATCH_SCRIPT=/path/to/radwatch.py to also run the real radwatch +`status` against a temp database (skipped otherwise).""" + +import json +import os +import socket +import sqlite3 +import sys +import tempfile +import threading +import time +import unittest +import urllib.request +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from unittest import mock + +from carwatch import radiation + +NOW = 1791382470.0 + +# Shape of a real bridge reading (read 7 Oct 2026). serial and address are +# scrubbed: this repo is public. The optional fields (accumulated_dose, +# battery, temperature, hardness) are ABSENT, as for the first minute after +# a bridge restart. +BRIDGE_SAMPLE = {"dose_rate": 0.14134200682747178, "count_rate": 9.889473915100098, + "serial": "RC-10X-TEST", "firmware": "4.14", + "ts": NOW - 13.1, "served_at": NOW - 8.8} +BRIDGE_FULL = dict(BRIDGE_SAMPLE, accumulated_dose=12.5, battery=81, temperature=24.5, + hardness=1.02) + + +def _free_port(): + s = socket.socket(); s.bind(("127.0.0.1", 0)); p = s.getsockname()[1]; s.close(); return p + + +class _Fake: + """One local server; each test sets `routes` {path: (code, body)}.""" + routes: dict = {} + seen_auth: list = [] + + class H(BaseHTTPRequestHandler): + def do_GET(self): + _Fake.seen_auth.append(self.headers.get("Authorization")) + code, body = _Fake.routes.get(self.path, (404, {"message": "Entity not found."})) + data = body if isinstance(body, bytes) else json.dumps(body).encode() + self.send_response(code) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def log_message(self, *a): + pass + + +def setUpModule(): + import socketserver + + class _S(ThreadingHTTPServer): + def server_bind(self): # skip getfqdn(), see test_webchat_ask + socketserver.TCPServer.server_bind(self) + self.server_name, self.server_port = "localhost", self.server_address[1] + global SRV, BASE + port = _free_port() + SRV = _S(("127.0.0.1", port), _Fake.H) + threading.Thread(target=SRV.serve_forever, daemon=True).start() + BASE = f"http://127.0.0.1:{port}" + + +def tearDownModule(): + SRV.shutdown() + + +def _dead_url(): + return f"http://127.0.0.1:{_free_port()}" # nothing listens there + + +# ── no source / bad config ───────────────────────────────────────────── + +class TestConfig(unittest.TestCase): + def test_no_block_is_unconfigured_not_error(self): + d = radiation.status({}, now=NOW) + self.assertEqual(d["state"], "unconfigured") + self.assertIsNone(d["error"]) + self.assertIn("radiation.source", d["message"]) + + def test_empty_source_is_unconfigured(self): + self.assertEqual(radiation.status({"radiation": {"source": ""}}, now=NOW)["state"], + "unconfigured") + + def test_unknown_source_is_an_error_naming_it(self): + d = radiation.status({"radiation": {"source": "geiger"}}, now=NOW) + self.assertEqual(d["state"], "error") + self.assertIn("geiger", d["error"]) + + def test_stale_seconds_default_and_override(self): + self.assertEqual(radiation.status({}, now=NOW)["stale_seconds"], 60) + d = radiation.status({"radiation": {"stale_seconds": 15}}, now=NOW) + self.assertEqual(d["stale_seconds"], 15) + d = radiation.status({"radiation": {"stale_seconds": "nonsense"}}, now=NOW) + self.assertEqual(d["stale_seconds"], 60) + + def test_an_unexpected_exception_is_shown_not_swallowed(self): + with mock.patch.dict(radiation.SOURCES, {"json": lambda c, now: 1 / 0}): + d = radiation.status({"radiation": {"source": "json"}}, now=NOW) + self.assertEqual(d["state"], "error") + self.assertIn("ZeroDivisionError", d["error"]) + + def test_reads_config_file_when_no_cfg_given(self): + with mock.patch.object(radiation, "load_raw", return_value={}): + self.assertEqual(radiation.status(now=NOW)["state"], "unconfigured") + + +# ── JSON bridge (the recommended source) ─────────────────────────────── + +class TestJsonBridge(unittest.TestCase): + def _cfg(self, url=None, **kw): + return {"radiation": {"source": "json", "json": {"url": url or BASE + "/radiacode.json"}, **kw}} + + def _serve(self, code, body): + _Fake.routes = {"/radiacode.json": (code, body)} + + def test_live_sample(self): + self._serve(200, BRIDGE_SAMPLE) + d = radiation.status(self._cfg(), now=NOW) + self.assertEqual(d["state"], "live", d) + self.assertAlmostEqual(d["values"]["dose_rate"]["value"], 0.1413, places=3) + self.assertEqual(d["values"]["dose_rate"]["unit"], "µSv/h") + self.assertAlmostEqual(d["values"]["count_rate"]["value"], 9.889, places=2) + self.assertEqual(d["age_s"], 13.1) + self.assertEqual(d["extra"]["firmware"], "4.14") + self.assertEqual(d["extra"]["bridge_lag_s"], 4.3) + + def test_absent_optional_fields_are_not_yet_reported_not_error(self): + self._serve(200, BRIDGE_SAMPLE) + d = radiation.status(self._cfg(), now=NOW) + self.assertEqual(d["state"], "live") + for k in ("accumulated_dose", "battery", "temperature"): + self.assertIsNone(d["values"][k]["value"]) + self.assertEqual(d["values"][k]["note"], "not yet reported") + self.assertNotIn("hardness", d["extra"]) + + def test_full_payload(self): + self._serve(200, BRIDGE_FULL) + d = radiation.status(self._cfg(), now=NOW) + self.assertEqual(d["values"]["battery"]["value"], 81) + self.assertEqual(d["values"]["accumulated_dose"]["value"], 12.5) + self.assertEqual(d["values"]["temperature"]["value"], 24.5) + self.assertEqual(d["extra"]["hardness"], 1.02) + self.assertIsNone(d["values"]["battery"]["note"]) + + def test_null_field_is_unavailable_not_not_yet(self): + self._serve(200, dict(BRIDGE_SAMPLE, battery=None)) + d = radiation.status(self._cfg(), now=NOW) + self.assertEqual(d["values"]["battery"]["note"], "unavailable") + + def test_stale_ts(self): + self._serve(200, dict(BRIDGE_SAMPLE, ts=NOW - 300)) + d = radiation.status(self._cfg(), now=NOW) + self.assertEqual(d["state"], "stale") + self.assertIn("300s ago", d["message"]) + self.assertIsNone(d["error"]) + self.assertIsNotNone(d["values"]["dose_rate"]["value"]) # last known still shown + + def test_stale_seconds_is_configurable(self): + self._serve(200, dict(BRIDGE_SAMPLE, ts=NOW - 30)) + self.assertEqual(radiation.status(self._cfg(), now=NOW)["state"], "live") + self.assertEqual(radiation.status(self._cfg(stale_seconds=20), now=NOW)["state"], "stale") + + def test_iso_ts_with_offset(self): + from datetime import datetime, timezone + iso = datetime.fromtimestamp(NOW - 5, timezone.utc).isoformat().replace("+00:00", "Z") + self._serve(200, dict(BRIDGE_SAMPLE, ts=iso)) + d = radiation.status(self._cfg(), now=NOW) + self.assertEqual(d["state"], "live") + self.assertEqual(d["age_s"], 5.0) + + def test_missing_ts_is_stale_never_live(self): + body = {k: v for k, v in BRIDGE_SAMPLE.items() if k not in ("ts", "served_at")} + self._serve(200, body) + d = radiation.status(self._cfg(), now=NOW) + self.assertEqual(d["state"], "stale") + self.assertIn("age unknown", d["message"]) + + def test_future_ts_is_clock_skew_not_live(self): + self._serve(200, dict(BRIDGE_SAMPLE, ts=NOW + 600)) + d = radiation.status(self._cfg(), now=NOW) + self.assertEqual(d["state"], "stale") + self.assertIn("future", d["message"]) + + def test_no_readings_in_object_is_empty(self): + self._serve(200, {"ts": NOW, "serial": "RC-10X-TEST"}) + d = radiation.status(self._cfg(), now=NOW) + self.assertEqual(d["state"], "empty") + + def test_unreachable(self): + d = radiation.status(self._cfg(url=_dead_url() + "/x.json"), now=NOW) + self.assertEqual(d["state"], "error") + self.assertIn("unreachable", d["error"]) + + def test_bad_json(self): + self._serve(200, b"not json") + d = radiation.status(self._cfg(), now=NOW) + self.assertEqual(d["state"], "error") + self.assertIn("invalid JSON", d["error"]) + + def test_json_array_is_error(self): + self._serve(200, [1, 2]) + d = radiation.status(self._cfg(), now=NOW) + self.assertEqual(d["state"], "error") + self.assertIn("expected an object", d["error"]) + + def test_object_with_no_known_fields_is_error(self): + self._serve(200, {"hello": "world"}) + d = radiation.status(self._cfg(), now=NOW) + self.assertEqual(d["state"], "error") + self.assertIn("none of the expected fields", d["error"]) + + def test_http_error(self): + self._serve(500, {"error": "boom"}) + d = radiation.status(self._cfg(), now=NOW) + self.assertEqual(d["state"], "error") + self.assertIn("HTTP 500", d["error"]) + + def test_url_not_set(self): + d = radiation.status({"radiation": {"source": "json"}}, now=NOW) + self.assertEqual(d["state"], "error") + self.assertIn("radiation.json.url", d["error"]) + + +# ── Home Assistant ───────────────────────────────────────────────────── + +def _ha_state(eid, state, unit=None, **kw): + a = {"friendly_name": eid} + if unit: + a["unit_of_measurement"] = unit + return {"entity_id": eid, "state": state, "attributes": a, **kw} + + +def _ha_routes(prefix="sensor.radiacode_", last_reading=None, skip=()): + from datetime import datetime, timezone + lr = last_reading or datetime.fromtimestamp(NOW - 7, timezone.utc).isoformat() + vals = {"dose_rate": ("0.12", "µSv/h"), "count_rate": ("8.4", "cps"), + "accumulated_dose": ("3.2", "µSv"), "battery": ("77", "%"), + "temperature": ("22.0", "°C"), "last_reading": (lr, None)} + return {f"/api/states/{prefix}{k}": (200, _ha_state(prefix + k, v, u)) + for k, (v, u) in vals.items() if k not in skip} + + +class TestHomeAssistant(unittest.TestCase): + TOKEN = "test-token-not-a-secret-0123456789" + + def _cfg(self, url=None, token=None, **kw): + return {"radiation": {"source": "ha", "ha": {"url": url or BASE, + "token": self.TOKEN if token is None else token, + **kw}}} + + def setUp(self): + _Fake.seen_auth = [] + for k in ("CARWATCH_RADIATION_HA_URL", "CARWATCH_RADIATION_HA_TOKEN"): + p = mock.patch.dict(os.environ, {}, clear=False) + p.start(); self.addCleanup(p.stop) + os.environ.pop(k, None) + + def test_live(self): + _Fake.routes = _ha_routes() + d = radiation.status(self._cfg(), now=NOW) + self.assertEqual(d["state"], "live", d) + self.assertEqual(d["values"]["dose_rate"]["value"], 0.12) + self.assertEqual(d["values"]["count_rate"]["unit"], "cps") # HA's own unit + self.assertEqual(d["age_basis"], "last_reading sensor") + self.assertEqual(d["age_s"], 7.0) + self.assertIn("not available through Home Assistant", d["alarm_note"]) + self.assertIn(f"Bearer {self.TOKEN}", _Fake.seen_auth) + + def test_401(self): + _Fake.routes = {p: (401, {"message": "Unauthorized"}) for p in _ha_routes()} + d = radiation.status(self._cfg(), now=NOW) + self.assertEqual(d["state"], "error") + self.assertIn("rejected the token (HTTP 401)", d["error"]) + self.assertNotIn(self.TOKEN, json.dumps(d)) + + def test_unreachable(self): + t0 = time.time() + d = radiation.status(self._cfg(url=_dead_url()), now=NOW) + self.assertEqual(d["state"], "error") + self.assertIn("unreachable", d["error"]) + self.assertLess(time.time() - t0, 5) + + def test_stale_last_reading(self): + from datetime import datetime, timezone + _Fake.routes = _ha_routes(last_reading=datetime.fromtimestamp(NOW - 3600, timezone.utc).isoformat()) + d = radiation.status(self._cfg(), now=NOW) + self.assertEqual(d["state"], "stale") + + def test_missing_entity_is_a_note_not_an_error(self): + _Fake.routes = _ha_routes(skip=("battery",)) + d = radiation.status(self._cfg(), now=NOW) + self.assertEqual(d["state"], "live") + self.assertEqual(d["values"]["battery"]["note"], "no such entity") + self.assertIn("sensor.radiacode_battery", d["extra"]["missing_entities"]) + + def test_unavailable_state_is_not_zero(self): + r = _ha_routes() + r["/api/states/sensor.radiacode_temperature"] = (200, _ha_state("sensor.radiacode_temperature", "unavailable")) + _Fake.routes = r + d = radiation.status(self._cfg(), now=NOW) + self.assertIsNone(d["values"]["temperature"]["value"]) + self.assertEqual(d["values"]["temperature"]["note"], "unavailable") + + def test_no_entities_at_all_is_error(self): + _Fake.routes = {} + d = radiation.status(self._cfg(), now=NOW) + self.assertEqual(d["state"], "error") + self.assertIn("none of the sensor.radiacode_* entities", d["error"]) + + def test_falls_back_to_last_reported(self): + from datetime import datetime, timezone + r = _ha_routes(skip=("last_reading",)) + eid = "sensor.radiacode_dose_rate" + r[f"/api/states/{eid}"] = (200, _ha_state(eid, "0.12", "µSv/h", + last_reported=datetime.fromtimestamp(NOW - 9, timezone.utc).isoformat())) + _Fake.routes = r + d = radiation.status(self._cfg(), now=NOW) + self.assertEqual(d["state"], "live") + self.assertIn("last_reported", d["age_basis"]) + + def test_no_token_is_error_naming_both_places(self): + d = radiation.status(self._cfg(token=""), now=NOW) + self.assertEqual(d["state"], "error") + self.assertIn("CARWATCH_RADIATION_HA_TOKEN", d["error"]) + + def test_env_beats_config(self): + _Fake.routes = _ha_routes() + with mock.patch.dict(os.environ, {"CARWATCH_RADIATION_HA_TOKEN": "env-token-0123456789abcdef", + "CARWATCH_RADIATION_HA_URL": BASE}): + d = radiation.status(self._cfg(url="http://homeassistant.local:8123"), now=NOW) + self.assertEqual(d["state"], "live") + self.assertIn("Bearer env-token-0123456789abcdef", _Fake.seen_auth) + + def test_token_never_sent_to_a_public_host(self): + with mock.patch("carwatch.mercedesme._is_private_ha", return_value=False), \ + mock.patch.object(radiation, "_get_json") as g: + d = radiation.status(self._cfg(url="http://ha.example.com:8123"), now=NOW) + self.assertEqual(d["state"], "error") + self.assertIn("non-private", d["error"]) + g.assert_not_called() + + +# ── radwatch on this box ─────────────────────────────────────────────── + +def _fake_radwatch(tmp, payload=None, exit_code=0, stderr=""): + """A stand-in for `radwatch.py status`: prints payload, exits exit_code.""" + p = os.path.join(tmp, "radwatch.py") + with open(p, "w") as f: + f.write("import sys, json\n" + f"assert sys.argv[1:3] == ['status', '--db'], sys.argv\n" + f"sys.stderr.write({stderr!r})\n" + f"print({json.dumps(payload) if payload is not None else ''!r})\n" + f"sys.exit({exit_code})\n") + return p + + +def _rw_reading(age, **kw): + return {"ok": True, "readings": 700, + "reading": {"ts": "2026-10-07T16:00:00", "age_s": age, "dose_rate_usv_h": 0.11, + "dose_rate_err_pct": 9.0, "count_rate_cps": 9.5, "count_rate_err_pct": 3.0}, + "device": {"ts": "2026-10-07T15:59:30", "age_s": age + 30, "accumulated_dose_usv": 4.2, + "accumulated_over_s": 3600, "temperature_c": 23.0, "battery_pct": 64.0}, + "watch": {"ts": "2026-10-07T16:00:00", "baseline_cps": 9.4, "window_cps": 9.6, + "sigma": 0.8, "alert": False, "threshold_calibrated": False}, + "watch_note": None, **kw} + + +class TestRadwatch(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp() + + def _cfg(self, script): + return {"radiation": {"source": "radwatch", + "radwatch": {"python": sys.executable, "script": script, + "db": os.path.join(self.tmp, "radwatch.sqlite")}}} + + def test_live_with_alarm(self): + d = radiation.status(self._cfg(_fake_radwatch(self.tmp, _rw_reading(3.0))), now=NOW) + self.assertEqual(d["state"], "live", d) + self.assertEqual(d["values"]["dose_rate"]["value"], 0.11) + self.assertEqual(d["values"]["battery"]["value"], 64.0) + self.assertEqual(d["alarm"]["sigma"], 0.8) # radwatch's number, not recomputed + self.assertFalse(d["alarm"]["alert"]) + + def test_stale(self): + d = radiation.status(self._cfg(_fake_radwatch(self.tmp, _rw_reading(120.0))), now=NOW) + self.assertEqual(d["state"], "stale") + self.assertIsNone(d["error"]) + + def test_empty_db(self): + payload = {"ok": True, "readings": 0, "reading": None, "device": None, + "watch": None, "watch_note": "not enough readings yet: 0 of 630"} + d = radiation.status(self._cfg(_fake_radwatch(self.tmp, payload)), now=NOW) + self.assertEqual(d["state"], "empty") + self.assertIn("no readings yet", d["message"]) + self.assertIsNone(d["error"]) + self.assertEqual(d["alarm_note"], "not enough readings yet: 0 of 630") + + def test_no_device_record_yet_is_not_yet_reported(self): + payload = _rw_reading(3.0, device=None) + d = radiation.status(self._cfg(_fake_radwatch(self.tmp, payload)), now=NOW) + self.assertEqual(d["state"], "live") + self.assertEqual(d["values"]["battery"]["note"], "not yet reported") + + def test_missing_db_is_error_with_radwatchs_reason(self): + payload = {"ok": False, "error": "no radwatch database at /x/radwatch.sqlite"} + d = radiation.status(self._cfg(_fake_radwatch(self.tmp, payload, exit_code=2)), now=NOW) + self.assertEqual(d["state"], "error") + self.assertIn("no radwatch database", d["error"]) + + def test_script_missing(self): + d = radiation.status(self._cfg(os.path.join(self.tmp, "nope.py")), now=NOW) + self.assertEqual(d["state"], "error") + self.assertIn("radwatch not found", d["error"]) + + def test_old_radwatch_without_status(self): + p = _fake_radwatch(self.tmp, None, exit_code=2, + stderr="radwatch.py: error: argument cmd: invalid choice: 'status'\n") + d = radiation.status(self._cfg(p), now=NOW) + self.assertEqual(d["state"], "error") + self.assertIn("invalid choice", d["error"]) + + +@unittest.skipUnless(os.environ.get("RADWATCH_SCRIPT"), "set RADWATCH_SCRIPT to run against real radwatch") +class TestRealRadwatch(unittest.TestCase): + """The same states through the real `radwatch.py status`, on a temp DB.""" + + def _cfg(self, db): + return {"radiation": {"source": "radwatch", "radwatch": { + "python": sys.executable, "script": os.environ["RADWATCH_SCRIPT"], "db": db}}} + + def _db(self, rows): + db = os.path.join(tempfile.mkdtemp(), "r.sqlite") + c = sqlite3.connect(db) + c.execute("CREATE TABLE reading(ts TEXT PRIMARY KEY, count_rate REAL, count_rate_err REAL, " + "dose_rate REAL, dose_rate_err REAL, flags INT)") + c.executemany("INSERT INTO reading VALUES(?,?,?,?,?,0)", rows) + c.commit(); c.close() + return db + + def test_missing_empty_stale_live(self): + from datetime import datetime, timedelta + missing = os.path.join(tempfile.mkdtemp(), "typo.sqlite") + self.assertEqual(radiation.status(self._cfg(missing))["state"], "error") + self.assertFalse(os.path.exists(missing)) + self.assertEqual(radiation.status(self._cfg(self._db([])))["state"], "empty") + old = (datetime.now() - timedelta(minutes=10)).isoformat() + self.assertEqual(radiation.status(self._cfg(self._db([(old, 9, 3, 1e-7, 9)])))["state"], "stale") + new = datetime.now().isoformat() + d = radiation.status(self._cfg(self._db([(new, 9, 3, 1e-7, 9)]))) + self.assertEqual(d["state"], "live", d) + self.assertEqual(d["values"]["dose_rate"]["value"], 0.1) + + +# ── negative control ─────────────────────────────────────────────────── + +def _scenario_failures(status_fn): + """Run the four must-differ scenarios through status_fn; list what went wrong.""" + bad = [] + _Fake.routes = {"/b.json": (200, dict(BRIDGE_SAMPLE, ts=NOW - 600))} + cfg = {"radiation": {"source": "json", "json": {"url": BASE + "/b.json"}}} + expect = [ + ({}, "unconfigured"), + ({"radiation": {"source": "json", "json": {"url": _dead_url() + "/b.json"}}}, "error"), + (cfg, "stale"), + ] + for c, want in expect: + got = status_fn(c, now=NOW)["state"] + if got != want: + bad.append(f"{want} rendered as {got}") + _Fake.routes = {"/b.json": (200, dict(BRIDGE_SAMPLE, ts=NOW - 2))} + if status_fn(cfg, now=NOW)["state"] != "live": + bad.append("live not live") + return bad + + +class TestNegativeControl(unittest.TestCase): + def test_real_classifier_keeps_every_state_distinct(self): + self.assertEqual(_scenario_failures(radiation.status), []) + + def test_collapsed_stale_is_caught(self): + """A classifier that drops the age check (stale shown as live) must fail the table.""" + def no_stale(reading, stale_s): + return "live", "collapsed" + with mock.patch.object(radiation, "classify", no_stale): + self.assertIn("stale rendered as live", _scenario_failures(radiation.status)) + + def test_collapsed_error_is_caught(self): + """A status() that swallows source errors into 'empty' must fail the table.""" + real = radiation.status + + def swallow(cfg, now=None): + d = real(cfg, now) + return dict(d, state="empty") if d["state"] == "error" else d + self.assertIn("error rendered as empty", _scenario_failures(swallow)) + + +# ── the routes, through a real server ────────────────────────────────── + +class TestRoutes(unittest.TestCase): + @classmethod + def setUpClass(cls): + import socketserver + from carwatch import webchat + cls.webchat = webchat + + class _S(webchat.ThreadingHTTPServer): + def server_bind(self): + socketserver.TCPServer.server_bind(self) + self.server_name, self.server_port = "localhost", self.server_address[1] + cls.port = _free_port() + cls.srv = _S(("127.0.0.1", cls.port), webchat.Handler) + threading.Thread(target=cls.srv.serve_forever, daemon=True).start() + + @classmethod + def tearDownClass(cls): + cls.srv.shutdown() + + def _get(self, path): + with mock.patch.object(self.webchat.Handler, "_authorised", lambda self: True): + with urllib.request.urlopen(f"http://127.0.0.1:{self.port}{path}", timeout=10) as r: + return r.status, r.headers.get("Content-Type"), r.read().decode() + + def test_api_returns_state_json(self): + with mock.patch.object(radiation, "load_raw", return_value={}): + code, ctype, body = self._get("/api/radiation") + self.assertEqual(code, 200) + self.assertIn("application/json", ctype) + self.assertEqual(json.loads(body)["state"], "unconfigured") + + def test_page_serves_and_knows_every_state(self): + code, ctype, body = self._get("/radiation?t=x") + self.assertEqual(code, 200) + self.assertIn("text/html", ctype) + for st in ("live", "stale", "error", "empty", "unconfigured"): + self.assertIn(f"{st}:", body.split("const TITLES=")[1].split("};")[0]) + self.assertIn("/api/radiation", body) + + def test_unified_page_links_here(self): + self.assertIn("href=/radiation", self.webchat.UNIFIED_PAGE) + + +if __name__ == "__main__": + unittest.main()