fix: probe plan survives a reboot

contrib/derate_amp_control.py

@@ -106,6 +106,7 @@
 import json
 import os
 import sys
+import time
 import urllib.error
 import urllib.request
 from dataclasses import asdict, dataclass, field, fields, replace
@@ -200,6 +201,11 @@ class State:
     probe_cable: str | None = None
     probe_session_ts: float | None = None
     probes_done: list[dict] = field(default_factory=list)
+    # Whether an empty probe history has been rebuilt from wallmonitor's
+    # event log. A lost state file (a reboot clearing /tmp, a reinstall)
+    # otherwise reads as "never probed": the next session probes at once,
+    # off-cadence, and repeats the first condition of the plan.
+    probe_history_checked: bool = False
     # Session bookkeeping the cable conditions are judged on: the last tick
     # seen charging (so the gap before a session is known), when this
     # session's charging began, that gap, and how long the current has held
@@ -694,6 +700,66 @@ def _cable_status(
     return "impossible"
 
 
+def probe_history_from_events(
+    events: list[dict], sessions: list[dict], now_ts: float, cfg: Config
+) -> tuple[list[dict], float | None]:
+    """Rebuild (probes_done, last_probe_ts) from wallmonitor's amp events.
+
+    Every probe start is logged as an amp_capped event carrying
+    detail.probe (older ones only say "calibration probe" in the reason);
+    completion is not always logged — a probe that ends on a sustainable
+    current no higher than its own holds silently. So a probe counts as
+    completed when the controller's next amp event, or now if there is
+    none yet, comes at least a full hold after its start: an abandoned
+    probe (a lower thermal cap, an unplug restoring normal rate) always
+    logs something sooner. The cadence anchor is the plug-in time of the
+    probe's session, as decide() anchors it; a few minutes earlier than the
+    daemon's own charging start, which only ever makes the next one due
+    sooner, never a cold slot too late."""
+    amp_events = []
+    for event in events:
+        detail = event.get("detail")
+        if isinstance(detail, str):
+            try:
+                detail = json.loads(detail)
+            except ValueError:
+                detail = None
+        amp_events.append((float(event["ts"]), event.get("kind"), detail if isinstance(detail, dict) else {}))
+    amp_events.sort(key=lambda item: item[0])
+
+    done: list[dict] = []
+    last_anchor = None
+    hold_s = cfg.probe_hold_min * 60.0
+    for i, (ts, kind, detail) in enumerate(amp_events):
+        if kind != "amp_capped":
+            continue
+        probe = detail.get("probe")
+        if isinstance(probe, dict):
+            amps, cable = float(probe.get("amps") or 0.0), probe.get("cable") or "any"
+        elif "calibration probe" in (detail.get("reason") or ""):
+            amps, cable = float(detail.get("to_a") or 0.0), "any"
+        else:
+            continue
+        end_ts = amp_events[i + 1][0] if i + 1 < len(amp_events) else now_ts
+        if end_ts - ts < hold_s - 60.0:  # 60 s: the log stamps on receipt, the hold clock on the tick
+            continue
+        done.append({"ts": ts + hold_s, "amps": amps, "cable": cable})
+        anchor = next(
+            (float(s["start_ts"]) for s in sessions
+             if s.get("start_ts") is not None and s["start_ts"] <= ts and (s.get("end_ts") is None or ts <= s["end_ts"])),
+            ts,
+        )
+        last_anchor = anchor if last_anchor is None else max(last_anchor, anchor)
+    return done[-PROBE_LOG_KEEP:], last_anchor
+
+
+def needs_probe_history(state: State, cfg: Config) -> bool:
+    return (
+        bool(cfg.probe_amps) and not state.probe_history_checked and state.probe_started_ts is None
+        and state.last_probe_ts is None and not state.probes_done
+    )
+
+
 def decide(thermal: dict, state: State, cfg: Config) -> tuple[Action, State, str]:
     """What to do, the state to persist, and why: the derate decision with
     the calibration probe layered over it."""
@@ -711,6 +777,22 @@ def fetch_thermal(base_url: str) -> dict:
         return json.load(resp)
 
 
+def recover_probe_history(base_url: str, state: State, cfg: Config, now_ts: float) -> State:
+    """An empty probe history, refilled from wallmonitor's event log (see
+    probe_history_from_events). A year back is far longer than any plan
+    takes to collect; the cadence after that only needs the latest probe."""
+    since = now_ts - 400 * 86400
+    query = f"from={since:.0f}&to={now_ts:.0f}"
+    with urllib.request.urlopen(
+        f"{base_url}/api/events?{query}&kinds=amp_capped,amp_restored,amp_adjust_failed", timeout=10
+    ) as resp:
+        events = json.load(resp).get("events") or []
+    with urllib.request.urlopen(f"{base_url}/api/sessions?{query}", timeout=10) as resp:
+        sessions = json.load(resp).get("sessions") or []
+    done, last_probe_ts = probe_history_from_events(events, sessions, now_ts, cfg)
+    return replace(state, probes_done=done, last_probe_ts=last_probe_ts, probe_history_checked=True)
+
+
 def set_charging_amps(tesla_ble_url: str, amps: float) -> None:
     """Apply an amp value through the ESPHome tesla-ble bridge's
     charging_amps number entity (the least-privilege BLE pairing)."""
@@ -868,7 +950,10 @@ def main(argv: list[str] | None = None) -> int:
     parser.add_argument(
         "--state-file",
         default="/tmp/derate_amp_control.state.json",
-        help="remembers cap state and debounce streaks between runs",
+        help=(
+            "remembers cap state, debounce streaks and the probe plan's progress between runs; the "
+            "default is cleared on reboot, so the installer points this at /var/lib (default: %(default)s)"
+        ),
     )
     parser.add_argument(
         "--probe-amps",
@@ -970,6 +1055,15 @@ def main(argv: list[str] | None = None) -> int:
         probe_warm_min=args.probe_warm_min,
     )
     state = load_state(args.state_file)
+    if needs_probe_history(state, cfg):
+        try:
+            state = recover_probe_history(args.wallmonitor, state, cfg, time.time())
+            print(f"probe history rebuilt from the event log: {len(state.probes_done)} completed")
+        except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, KeyError, TypeError, ValueError) as exc:
+            # Without the history a probe would look overdue; hold the plan
+            # this tick rather than restart it. Thermal capping is unaffected.
+            print(f"warn: cannot rebuild probe history ({exc}); no probe this tick", file=sys.stderr)
+            cfg = replace(cfg, probe_amps=())
 
     try:
         thermal = fetch_thermal(args.wallmonitor)

deploy/install-derate-amp-control.sh

@@ -15,6 +15,10 @@
 #   sudo ./install-derate-amp-control.sh --tesla-ble http://<esp32-host> --probe-amps 32,40 --probe-cable cold,warm
 #   sudo ./install-derate-amp-control.sh --uninstall
 #
+# State (the active cap, debounce streaks, the probe plan's progress) lives
+# in /var/lib/derate-amp-control/state.json unless --state-file says
+# otherwise; the daemon's own default is under /tmp, which a reboot clears.
+#
 # The ESP32 host/IP lands only in the local systemd unit — never commit it.
 # The paired device should be configured with the least-privilege
 # CHARGING_MANAGER role; this script has no opinion on that, it only talks
@@ -40,6 +44,8 @@ PROBE_INTERVAL_DAYS=""
 PROBE_PLAN_INTERVAL_DAYS=""
 PROBE_REPLICATES=""
 PROBE_HOLD_MIN=""
+STATE_DIR="/var/lib/${SERVICE_NAME}"
+LEGACY_STATE_FILE="/tmp/derate_amp_control.state.json"
 STATE_FILE=""
 INTERVAL="30"
 DRY_RUN="0"
@@ -111,7 +117,8 @@ DAEMON_ARGS="--tesla-ble ${TESLA_BLE} --wallmonitor ${WALLMONITOR}"
 [[ -n "$PROBE_PLAN_INTERVAL_DAYS" ]] && DAEMON_ARGS+=" --probe-plan-interval-days ${PROBE_PLAN_INTERVAL_DAYS}"
 [[ -n "$PROBE_REPLICATES" ]] && DAEMON_ARGS+=" --probe-replicates ${PROBE_REPLICATES}"
 [[ -n "$PROBE_HOLD_MIN" ]] && DAEMON_ARGS+=" --probe-hold-min ${PROBE_HOLD_MIN}"
-[[ -n "$STATE_FILE" ]] && DAEMON_ARGS+=" --state-file ${STATE_FILE}"
+[[ -z "$STATE_FILE" ]] && STATE_FILE="${STATE_DIR}/state.json"
+DAEMON_ARGS+=" --state-file ${STATE_FILE}"
 [[ "$DRY_RUN" == "1" ]] && DAEMON_ARGS+=" --dry-run"
 [[ -n "$EXTRA_ARGS" ]] && DAEMON_ARGS+=" ${EXTRA_ARGS}"
 
@@ -122,6 +129,7 @@ DAEMON_ARGS="--tesla-ble ${TESLA_BLE} --wallmonitor ${WALLMONITOR}"
   echo "[Service]"
   echo "Type=oneshot"
   echo "User=${RUN_USER}"
+  echo "StateDirectory=${SERVICE_NAME}"
   echo "ExecStart=/usr/bin/env python3 ${DAEMON} ${DAEMON_ARGS}"
 } > "$UNIT_PATH"
 
@@ -137,6 +145,15 @@ DAEMON_ARGS="--tesla-ble ${TESLA_BLE} --wallmonitor ${WALLMONITOR}"
   echo "WantedBy=timers.target"
 } > "$TIMER_PATH"
 
+# Carry an existing install's state over from /tmp, so the move does not
+# itself reset the probe plan (the daemon would rebuild it from the event
+# log, but the active cap and streaks would be lost mid-session).
+install -d -o "$RUN_USER" -m 0755 "$STATE_DIR"
+if [[ "$STATE_FILE" != "$LEGACY_STATE_FILE" && -f "$LEGACY_STATE_FILE" && ! -e "$STATE_FILE" ]]; then
+  install -o "$RUN_USER" -m 0644 "$LEGACY_STATE_FILE" "$STATE_FILE"
+  echo "moved state from ${LEGACY_STATE_FILE} to ${STATE_FILE}"
+fi
+
 systemctl daemon-reload
 systemctl enable --now "${SERVICE_NAME}.timer"
 

tests/test_derate_amp_control.py

@@ -16,6 +16,7 @@
 trusted."""
 
 import importlib.util
+import json
 import pathlib
 import sys
 
@@ -830,3 +831,88 @@ def test_probe_state_survives_an_old_state_file(tmp_path):
     assert state.probes_done == [] and state.probe_amps is None
     action, new, reason = dac.decide(_thermal(will_trip=False), state, _probe_cfg())
     assert action.kind == "none" and "probe holding 32A" in reason and new.cap_value == 32.0
+
+
+def _probe_events(start_ts, amps=32.0, cable="cold", next_after_min=41.0, next_reason="probe complete"):
+    events = [{"ts": start_ts, "kind": "amp_capped",
+               "detail": json.dumps({"to_a": amps, "reason": "calibration probe due", "probe": {"amps": amps, "cable": cable}})}]
+    if next_after_min is not None:
+        events.append({"ts": start_ts + next_after_min * 60, "kind": "amp_restored",
+                       "detail": json.dumps({"to_a": 48.0, "reason": next_reason})})
+    return events
+
+
+def test_probe_history_rebuilt_from_events_keeps_the_plan_on_cadence():
+    # The 2026-09-24 incident: a reboot cleared /tmp three days after a
+    # 32 A cold probe, and the next session probed 32 A cold again. Rebuilt
+    # from the event log, that session is not due; a week on, the plan moves
+    # to a condition with no replicates rather than repeating the first.
+    t0 = 1_000_000.0
+    last_charge = t0 - 12 * 3600
+    for days_ago, expected in ((3, []), (7, [(30, 32.0, "warm")])):
+        session_start = t0 - days_ago * 86400
+        sessions = [{"start_ts": session_start - 5, "end_ts": session_start + 14 * 3600}]
+        done, last = dac.probe_history_from_events(_probe_events(session_start + 45), sessions, t0 - 3600, _plan_cfg())
+        assert done == [{"ts": session_start + 45 + 2400, "amps": 32.0, "cable": "cold"}]
+        assert last == session_start - 5  # anchored at the plug-in, as decide() anchors it
+        state = dac.State(probes_done=done, last_probe_ts=last, probe_history_checked=True, last_charging_ts=last_charge)
+        starts, _ = _simulate(_plan_cfg(), days=1, start_state=state)
+        assert [(m, a, c) for _, m, a, c in starts] == expected
+
+
+def test_probe_history_counts_only_holds_that_ran_their_full_length():
+    cfg = _plan_cfg()
+    now = 2_000_000.0
+    abandoned = _probe_events(1_000_000.0, next_after_min=12.0, next_reason="session ended: restoring normal rate")
+    # a probe that ends on a sustainable current no higher than its own logs nothing at completion
+    silent = _probe_events(1_100_000.0, amps=40.0, cable="warm", next_after_min=None)
+    legacy = [{"ts": 1_200_000.0, "kind": "amp_capped",
+               "detail": {"to_a": 32.0, "reason": "calibration probe due: capping to 32A for 40min"}},
+              {"ts": 1_200_000.0 + 2460, "kind": "amp_restored", "detail": None}]
+    thermal_cap = [{"ts": 1_300_000.0, "kind": "amp_capped", "detail": json.dumps({"to_a": 44.0, "reason": "plateau"})}]
+    events = list(reversed(abandoned + silent + legacy + thermal_cap))  # the API returns newest first
+    done, last = dac.probe_history_from_events(events, [], now, cfg)
+    assert [(p["amps"], p["cable"]) for p in done] == [(40.0, "warm"), (32.0, "any")]
+    assert last == 1_200_000.0  # no session known: anchored at the start itself
+    # ...but a probe still inside its hold is not yet complete
+    done, last = dac.probe_history_from_events(silent, [], 1_100_000.0 + 600, cfg)
+    assert done == [] and last is None
+
+
+def test_main_rebuilds_an_empty_probe_history_once(tmp_path, monkeypatch, capsys):
+    state_file = tmp_path / "state.json"
+    args = ["--tesla-ble", "http://127.0.0.1:1", "--wallmonitor", "http://wm", "--state-file", str(state_file),
+            "--probe-amps", "32,40", "--probe-cable", "cold,warm"]
+    snap = _thermal(will_trip=False, ts=1_000_000.0, current_a=48.0)
+    monkeypatch.setattr(dac, "fetch_thermal", lambda url: snap)
+    calls = []
+
+    def recovered(url, state, cfg, now_ts):
+        calls.append(url)
+        return dac.replace(state, probes_done=[{"ts": 1.0, "amps": 32.0, "cable": "cold"}],
+                           last_probe_ts=snap["ts"] - 3 * 86400, probe_history_checked=True)
+
+    monkeypatch.setattr(dac, "recover_probe_history", recovered)
+    assert dac.main(args) == 0
+    assert calls == ["http://wm"] and "rebuilt from the event log: 1" in capsys.readouterr().out
+    assert dac.load_state(str(state_file)).probe_history_checked
+    dac.main(args)
+    assert len(calls) == 1  # never again once checked
+
+
+def test_main_holds_the_plan_when_the_history_cannot_be_rebuilt(tmp_path, monkeypatch, capsys):
+    # Brand-new state with a cold session starting: without the fallback
+    # this is exactly the tick that would probe off-cadence.
+    state_file = tmp_path / "state.json"
+    dac.save_state(str(state_file), dac.State(last_charging_ts=1_000_000.0 - 12 * 3600))
+    monkeypatch.setattr(dac, "fetch_thermal", lambda url: _thermal(will_trip=False, ts=1_000_000.0, current_a=48.0))
+
+    def unreachable(*a):
+        raise dac.urllib.error.URLError("refused")
+
+    monkeypatch.setattr(dac, "recover_probe_history", unreachable)
+    args = ["--tesla-ble", "http://127.0.0.1:1", "--wallmonitor", "http://wm", "--state-file", str(state_file),
+            "--probe-amps", "32", "--probe-cable", "cold", "--dry-run"]
+    assert dac.main(args) == 0
+    out = capsys.readouterr()
+    assert "no probe this tick" in out.err and "would cap" not in out.out