diff --git a/contrib/derate_amp_control.py b/contrib/derate_amp_control.py index 4a4d0d1..3561601 100755 --- a/contrib/derate_amp_control.py +++ b/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) diff --git a/deploy/install-derate-amp-control.sh b/deploy/install-derate-amp-control.sh index 30552c8..9df279a 100755 --- a/deploy/install-derate-amp-control.sh +++ b/deploy/install-derate-amp-control.sh @@ -15,6 +15,10 @@ # sudo ./install-derate-amp-control.sh --tesla-ble http:// --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" diff --git a/tests/test_derate_amp_control.py b/tests/test_derate_amp_control.py index d00d0d0..d70b706 100644 --- a/tests/test_derate_amp_control.py +++ b/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