diff --git a/README.md b/README.md index 9b38330..2355582 100644 --- a/README.md +++ b/README.md @@ -51,6 +51,7 @@ systemd service on an always-on box, see [Running wallmonitor](https://github.co serial is pinned on first contact so a swapped or second unit can never blend into your history - Full-fidelity recording — every response stored with its complete raw JSON + (optional retention trims old raw blobs so the database stays near-flat) - Live dashboard (SSE) with rolling charts, an active-alert banner, and a live derate-forecast chart: measured handle temperature against the model's projected plateau and the trip threshold, projection drawn diff --git a/docs/recording.md b/docs/recording.md index ebc119a..d5fde57 100644 --- a/docs/recording.md +++ b/docs/recording.md @@ -7,6 +7,9 @@ around gaps, glitches, and unverified labels. The numbered points below are the recording pipeline in the order data flows through it. 1. **Records everything the charger reports, at the highest safe fidelity.** + (With the optional `--retain-raw-days` policy, samples older than the + window keep every extracted column but drop the raw JSON blob — see + [Retention](running.md#retention).) Vitals every 2 s while a vehicle is attached (5 s idle), Wi-Fi status every 30 s, lifetime counters every 60 s, firmware info every 6 h. Requests are strictly sequential and back off exponentially on failures, so the charger's diff --git a/docs/running.md b/docs/running.md index c44be65..6226c5c 100644 --- a/docs/running.md +++ b/docs/running.md @@ -45,6 +45,8 @@ is the charger's microcontroller. | `--discover [RANGE]` | — | — | Sweep the LAN for Wall Connectors and exit (own subnet, or a private CIDR) | | `--label` | `WM_LABEL` | — | Name for this charger, shown in the header, tab title and notifications | | `--peer LABEL=URL` | `WM_PEERS` | — | Link another instance from the header switcher; repeatable (env: comma-separated) | +| `--retain-raw-days` | `WM_RETAIN_RAW_DAYS` | `0` (off) | Trim raw JSON from samples older than N days (min 7); columns stay forever | +| `--compact` | — | — | One-shot VACUUM to reclaim trimmed space; run with the service stopped | | `--port` | `WM_PORT` | `8480` | Web UI port | | `--bind` | `WM_BIND` | `127.0.0.1` | Web UI bind address | | `--db` | `WM_DB` | `wallmonitor.db` | SQLite path | @@ -139,6 +141,30 @@ sudo ./deploy/install-service.sh --name right --host 192.168.1.51 --port 8481 -- A single-process, cross-device view is tracked as future work in issue #10. +## Retention + +By default nothing is ever deleted: every sample keeps its complete raw +JSON forever, and the database grows by roughly a gigabyte a month on a +busy install (about 85 % of that is the raw blobs on vitals samples). + +`--retain-raw-days N` caps that. Samples older than N days keep **every +extracted column** — charts, session pages, the thermal model, and the +degradation watch see identical history — but their raw JSON blob is +blanked by a daily background pass (chunked, so live polling never waits +more than a moment; vitals wait until the diagnostics backfill has +finished). What is given up is *re-interpretability*: extracting a field +that was never a column, the way the diagnostics columns themselves were +backfilled from raw, becomes impossible for trimmed rows. Forecast +snapshots and version info are never trimmed. + +Freed pages are reused by new inserts, so the file stops growing rather +than shrinking. To hand the space back to the filesystem once, stop the +service and run: + +```bash +uv run python -m wallmonitor --compact --db /path/to/wallmonitor.db +``` + ## Run as a service (Ubuntu / systemd) For an always-on box, `deploy/install-service.sh` installs wallmonitor as a diff --git a/tests/test_retention.py b/tests/test_retention.py new file mode 100644 index 0000000..5f6da50 --- /dev/null +++ b/tests/test_retention.py @@ -0,0 +1,86 @@ +"""Retention: raw-JSON trim, its gates and cursor, readers on trimmed rows.""" + +import time + +import pytest + +from wallmonitor.config import parse_args +from wallmonitor.db import Database + + +@pytest.fixture +def db(tmp_path): + database = Database(str(tmp_path / "test.db")) + yield database + database.close() + + +def _fill(db, days_ago, n=20): + base = time.time() - days_ago * 86400 + for i in range(n): + ts = base + i * 10 + db.insert_vitals(ts, {"handle_temp_c": 30.0, "vehicle_current_a": 16.0, + "prox_v": 1.5, "pilot_high_v": 8.9}, 1, 3680.0) + db.insert_wifi(ts, {"wifi_rssi": -60, "wifi_connected": True}) + db.insert_lifetime(ts, {"energy_wh": 1000}) + db.insert_ambient(ts, 25.0, raw={"t": 25.0}, source="test") + + +def _raw_counts(db): + return {t: db._rows(f"SELECT COUNT(*) AS n FROM {t} WHERE raw != ''")[0]["n"] + for t in Database.RAW_TRIM_TABLES} + + +def test_trim_respects_cutoff_and_keeps_columns(db): + _fill(db, days_ago=30) + _fill(db, days_ago=1) + db.set_setting("diag_backfill_done", "1") + counts = db.trim_raw(time.time() - 7 * 86400) + assert counts["vitals_samples"] == 20 and counts["wifi_samples"] == 20 + remaining = _raw_counts(db) + assert all(v == 20 for v in remaining.values()), remaining # recent rows untouched + # Columns intact and served for the trimmed era. + old = db.vitals_range(time.time() - 31 * 86400, time.time() - 29 * 86400) + assert len(old) == 20 and all(s["handle_temp_c"] == 30.0 and s["prox_v"] == 1.5 for s in old) + # Bucketed path too (must not touch the blanked raw). + bucketed = db.vitals_range(time.time() - 31 * 86400, time.time() - 29 * 86400, max_points=4) + assert bucketed and all(b["prox_v"] == pytest.approx(1.5) for b in bucketed) + + +def test_vitals_trim_waits_for_diag_backfill(db): + _fill(db, days_ago=30) + counts = db.trim_raw(time.time()) + assert "vitals_samples" not in counts # gated: backfill not done + assert counts["wifi_samples"] == 20 # other tables trim regardless + db.set_setting("diag_backfill_done", "1") + assert db.trim_raw(time.time())["vitals_samples"] == 20 + + +def test_trim_cursor_makes_second_run_cheap_and_correct(db): + _fill(db, days_ago=30) + db.set_setting("diag_backfill_done", "1") + db.trim_raw(time.time() - 7 * 86400) + # New rows age past the cutoff later; the cursor resumes, not rescans. + _fill(db, days_ago=8) + counts = db.trim_raw(time.time() - 7 * 86400) + assert counts["vitals_samples"] == 20 + assert float(db.get_setting("raw_trim_ts:vitals_samples")) > time.time() - 9 * 86400 + + +def test_vacuum_reclaims_trimmed_space(db): + for i in range(300): + db.insert_vitals(time.time() - 40 * 86400 + i, {"handle_temp_c": 30.0, "pad": "x" * 2000}, 1, 0.0) + db.set_setting("diag_backfill_done", "1") + db.trim_raw(time.time()) + before, after = db.vacuum() + # 300 rows x ~2KB of trimmed blob must come back from the file. + assert before - after > 300_000 + + +def test_retain_flag_validation(): + with pytest.raises(SystemExit): + parse_args(["--demo", "--retain-raw-days", "3"]) + cfg = parse_args(["--demo", "--retain-raw-days", "30"]) + assert cfg.retain_raw_days == 30.0 + cfg = parse_args(["--demo"]) + assert cfg.retain_raw_days == 0.0 diff --git a/wallmonitor/__main__.py b/wallmonitor/__main__.py index 45c0f1e..d9994bb 100644 --- a/wallmonitor/__main__.py +++ b/wallmonitor/__main__.py @@ -30,10 +30,13 @@ log = logging.getLogger("wallmonitor") -async def _backfill(db: Database) -> None: - """One-time, in the background: fill diagnostics columns for rows - recorded before those columns existed. Chunked, so live polling only - ever waits a moment; a settings flag makes later startups a no-op.""" +async def _maintenance(db: Database, cfg) -> None: + """Background housekeeping: the one-time diagnostics backfill, then — + when retention is enabled — a daily raw-JSON trim of samples older + than the retention window. Chunked throughout, so live polling only + ever waits a moment.""" + import time as _time + def report(done: int, total: int) -> None: if done % 200_000 < 10_000 or done >= total: log.info("diagnostics backfill: %d / %d rows", done, total) @@ -41,6 +44,16 @@ def report(done: int, total: int) -> None: touched = await asyncio.to_thread(db.backfill_diag_columns, 10_000, report) if touched: log.info("diagnostics backfill complete: %d rows", touched) + if not cfg.retain_raw_days: + return + while True: + cutoff = _time.time() - cfg.retain_raw_days * 86400.0 + counts = await asyncio.to_thread(db.trim_raw, cutoff) + total = sum(counts.values()) + if total: + log.info("retention: trimmed raw JSON from %d rows (%s)", total, + ", ".join(f"{k}={v}" for k, v in counts.items() if v)) + await asyncio.sleep(24 * 3600.0) async def run(argv: list[str] | None = None) -> None: @@ -52,6 +65,15 @@ async def run(argv: list[str] | None = None) -> None: cfg = parse_args(argv) if cfg.discover is not None: raise SystemExit(await run_discovery(cfg.discover, split_phase_hint=cfg.split_phase)) + if cfg.compact: + db = Database(cfg.db_path) + try: + before, after = db.vacuum() + finally: + db.close() + print(f"{cfg.db_path}: {before / 1e6:.1f} MB -> {after / 1e6:.1f} MB " + f"({(before - after) / 1e6:.1f} MB reclaimed)") + raise SystemExit(0) logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s %(message)s") sim_runner = None @@ -65,7 +87,7 @@ async def run(argv: list[str] | None = None) -> None: client = aiohttp.ClientSession() poller = Poller(cfg, db, bus, client) await poller.start() - backfill = asyncio.create_task(_backfill(db), name="wallmonitor-backfill") + backfill = asyncio.create_task(_maintenance(db, cfg), name="wallmonitor-maintenance") app = make_app(db, bus, poller) runner = web.AppRunner(app) diff --git a/wallmonitor/config.py b/wallmonitor/config.py index 54407f9..a2e97f3 100644 --- a/wallmonitor/config.py +++ b/wallmonitor/config.py @@ -33,6 +33,12 @@ class Config: # Human name for this charger, shown in the UI header, tab title and # notifications so several instances side by side stay attributable. label: str = "" + # Optional retention: rows older than this many days keep every extracted + # column forever but have their raw JSON blob trimmed to "". 0 = never + # trim (the default — full re-interpretability retained indefinitely). + retain_raw_days: float = 0.0 + # One-shot: VACUUM the database to reclaim trimmed space, print sizes, exit. + compact: bool = False # Sibling instances watching other chargers, as (label, url) pairs; # rendered as a switcher in the header so one browser hops between them. peers: tuple[tuple[str, str], ...] = () @@ -149,6 +155,20 @@ def parse_args(argv: list[str] | None = None) -> Config: help="Name for this charger, shown in the UI and notifications — useful when " "running one instance per Wall Connector (env: WM_LABEL)", ) + parser.add_argument( + "--retain-raw-days", + type=float, + default=_env("WM_RETAIN_RAW_DAYS", 0.0), + help="Trim the raw JSON blob from samples older than this many days (columns are " + "kept forever; 0 disables — the default). Minimum 7. (env: WM_RETAIN_RAW_DAYS)", + ) + parser.add_argument( + "--compact", + action="store_true", + default=False, + help="VACUUM the database to reclaim space freed by retention, print sizes, and exit. " + "Run it while the monitor service is stopped.", + ) parser.add_argument( "--peer", action="append", @@ -195,7 +215,10 @@ def parse_args(argv: list[str] | None = None) -> Config: except ValueError as ex: parser.error(str(ex)) - if not args.demo and not args.host and args.discover is None: + if args.retain_raw_days and args.retain_raw_days < 7: + parser.error("--retain-raw-days must be at least 7 (or 0 to disable)") + + if not args.demo and not args.host and args.discover is None and not args.compact: parser.error( "--host (or WM_WC_HOST) is required — run `wallmonitor --discover` to find " "your Wall Connector's address, or --demo for the built-in simulator" @@ -209,6 +232,8 @@ def parse_args(argv: list[str] | None = None) -> Config: demo=bool(args.demo), split_phase=bool(args.split_phase), label=args.label, + retain_raw_days=float(args.retain_raw_days), + compact=bool(args.compact), peers=peers, discover=args.discover, notify_url=args.notify_url, diff --git a/wallmonitor/db.py b/wallmonitor/db.py index 2d110bd..3304517 100644 --- a/wallmonitor/db.py +++ b/wallmonitor/db.py @@ -221,6 +221,7 @@ def __init__(self, path: str): telemetry). The schema is pure CREATE IF NOT EXISTS, re-run on every startup: "migrations" are additive statements appended here. """ + self.path = path self._conn = sqlite3.connect(path, check_same_thread=False) self._conn.row_factory = sqlite3.Row self._lock = threading.Lock() @@ -245,6 +246,73 @@ def _migrate(self) -> None: self._conn.execute(f"ALTER TABLE vitals_samples ADD COLUMN {col} REAL") self._conn.commit() + # Tables whose raw JSON blob is trimmed by the retention policy. The + # extracted columns stay forever; forecast_samples and version_info are + # exempt (tiny, and their raw payloads carry fields with no column). + RAW_TRIM_TABLES = ("vitals_samples", "wifi_samples", "lifetime_samples", "ambient_samples") + + def trim_raw(self, cutoff_ts: float, chunk: int = 10_000) -> dict[str, int]: + """Retention: blank the raw JSON on samples older than cutoff_ts. + + Columns are untouched, so charts, fits and the degradation watch see + exactly the same history — what is given up is re-interpreting old + rows for fields that were never extracted. Vitals are only trimmed + once the diagnostics backfill has finished (its json_extract source + is the raw blob). A per-table timestamp cursor in settings makes the + daily run scan only rows that newly aged past the cutoff, and the + chunked updates hold the write lock briefly each. Freed pages are + reused by new inserts, so the file stops growing; a one-shot VACUUM + (--compact) reclaims the space for the filesystem. + """ + counts: dict[str, int] = {} + for table in self.RAW_TRIM_TABLES: + if table == "vitals_samples" and self.get_setting(DIAG_BACKFILL_SETTING) != "1": + continue + cursor_key = f"raw_trim_ts:{table}" + start_ts = float(self.get_setting(cursor_key) or 0.0) + trimmed = 0 + while True: + with self._lock: + rows = self._conn.execute( + f"SELECT id, ts FROM {table} WHERE ts >= ? AND ts < ? " + "ORDER BY ts LIMIT ?", + (start_ts, cutoff_ts, chunk), + ).fetchall() + if not rows: + break + ids = [row[0] for row in rows] + placeholders = ",".join("?" for _ in ids) + cur = self._execute( + f"UPDATE {table} SET raw = '' WHERE id IN ({placeholders}) AND raw != ''", + tuple(ids), + ) + trimmed += cur.rowcount + start_ts = rows[-1][1] + self._execute( + "INSERT INTO settings(key, value) VALUES (?, ?) " + "ON CONFLICT(key) DO UPDATE SET value = excluded.value", + (cursor_key, repr(start_ts)), + ) + if len(rows) < chunk: + break + counts[table] = trimmed + return counts + + def vacuum(self) -> tuple[int, int]: + """VACUUM, returning (bytes_before, bytes_after). Exclusive — run + while nothing else is writing (the --compact command exists so this + never happens implicitly under a live poller).""" + import os + + with self._lock: + self._conn.execute("PRAGMA wal_checkpoint(TRUNCATE)") + before = os.path.getsize(self.path) + with self._lock: + self._conn.execute("VACUUM") + self._conn.execute("PRAGMA wal_checkpoint(TRUNCATE)") + after = os.path.getsize(self.path) + return before, after + def backfill_diag_columns(self, chunk: int = 10_000, progress=None) -> int: """Fill the diagnostics columns for rows recorded before they existed, from each row's raw JSON, in id-ordered chunks so the @@ -274,7 +342,7 @@ def backfill_diag_columns(self, chunk: int = 10_000, progress=None) -> int: relay_k1_v = COALESCE(json_extract(raw, '$.relay_k1_v'), json_extract(raw, '$.relay_coil_v')), relay_k2_v = json_extract(raw, '$.relay_k2_v') - WHERE id > ? AND id <= ? AND prox_v IS NULL""", + WHERE id > ? AND id <= ? AND prox_v IS NULL AND raw != ''""", (last, upto), ) touched += cur.rowcount @@ -617,13 +685,18 @@ def vitals_range(self, t_from: float, t_to: float, max_points: int = 1500) -> li if self._diag_backfilled: d = {col: col for col in VITALS_LATER_COLUMNS} else: + # CASE-guarded: a retention-trimmed row has raw = '' and + # json_extract('') is a hard error, not NULL — and SQLite does + # not promise to short-circuit COALESCE past it even when the + # column is populated. + j = "CASE WHEN raw = '' THEN NULL ELSE json_extract(raw, '{key}') END" d = { - "pilot_high_v": "COALESCE(pilot_high_v, json_extract(raw, '$.pilot_high_v'))", - "pilot_low_v": "COALESCE(pilot_low_v, json_extract(raw, '$.pilot_low_v'))", - "prox_v": "COALESCE(prox_v, json_extract(raw, '$.prox_v'))", - "relay_k1_v": "COALESCE(relay_k1_v, json_extract(raw, '$.relay_k1_v'), " - "json_extract(raw, '$.relay_coil_v'))", - "relay_k2_v": "COALESCE(relay_k2_v, json_extract(raw, '$.relay_k2_v'))", + "pilot_high_v": f"COALESCE(pilot_high_v, {j.format(key='$.pilot_high_v')})", + "pilot_low_v": f"COALESCE(pilot_low_v, {j.format(key='$.pilot_low_v')})", + "prox_v": f"COALESCE(prox_v, {j.format(key='$.prox_v')})", + "relay_k1_v": f"COALESCE(relay_k1_v, {j.format(key='$.relay_k1_v')}, " + f"{j.format(key='$.relay_coil_v')})", + "relay_k2_v": f"COALESCE(relay_k2_v, {j.format(key='$.relay_k2_v')})", } diag = ", ".join(f"{expr} AS {col}" for col, expr in d.items()) if sample_count <= max_points: diff --git a/wallmonitor/poller.py b/wallmonitor/poller.py index f951701..c58badd 100644 --- a/wallmonitor/poller.py +++ b/wallmonitor/poller.py @@ -689,6 +689,7 @@ def status(self) -> dict[str, Any]: "host": self.cfg.host, "label": self.cfg.label, "peers": [{"label": label, "url": url} for label, url in self.cfg.peers], + "retain_raw_days": self.cfg.retain_raw_days, "device_serial": self.device_serial, "serial_mismatch": self.serial_mismatch, "offline": self._offline,