diff --git a/contrib/backtest_forecast.py b/contrib/backtest_forecast.py index 252799e..b654ce3 100644 --- a/contrib/backtest_forecast.py +++ b/contrib/backtest_forecast.py @@ -499,6 +499,7 @@ def main(argv: list[str] | None = None) -> int: now = time.time() fits = thermal.fit_sessions(db, now, lookback_days=args.lookback_days) params = thermal.fit_history(db, now, fits=fits) + fits = thermal.forecast_fits(fits) # what the deployed forecast learns from idle_model = thermal.load_idle_offset(db) sessions = [ sess for sess in db.sessions_range(now - args.lookback_days * 86400, now) diff --git a/docs/thermal-model.md b/docs/thermal-model.md index e685d00..42b6673 100644 --- a/docs/thermal-model.md +++ b/docs/thermal-model.md @@ -285,6 +285,11 @@ row. More sessions either confirm it or dissolve it. ### What it compares +- **Every session in the lookback.** The forecast learns from the newest + 40 sessions; the watch does not. A count window makes the baseline + whatever happened ~40 charges ago. On one install a single odd week at + that edge read as +8 °C of "confirmed" drift, while the 80 days since + the verified anchor were flat (−0.04 °C). - **Only free-running fits.** Windows the charger was regulating are excluded outright — their plateau is a setpoint, not an equilibrium. `regulated_n` says how many sat out, and the Alerts page says so too, diff --git a/tests/test_wallmonitor.py b/tests/test_wallmonitor.py index 761c5b7..3a8f917 100644 --- a/tests/test_wallmonitor.py +++ b/tests/test_wallmonitor.py @@ -2,6 +2,7 @@ import asyncio import math +import threading import time from statistics import median @@ -9,6 +10,7 @@ import pytest from wallmonitor import thermal +from wallmonitor import web as web_module from wallmonitor.config import Config from wallmonitor.db import Database from wallmonitor.poller import EventBus, Poller @@ -1300,6 +1302,29 @@ async def test_thermal_drift_ambient_confound_caps_verdict_at_lead(db): assert drift["drifting"] is False and drift["lead"] is True +async def test_thermal_drift_baseline_is_whole_lookback_not_newest_40(db): + # Regression: fit_sessions once kept only the newest 40 sessions, so the + # watch's baseline was whatever happened ~40 charges ago. A flat install + # with one low stretch sitting at that edge read as confirmed drift + # (seen live: +8 C over 34 days; flat over the 80 days since the anchor). + now = time.time() + rises = [36.0] * 12 + [29.0] * 6 + [36.0] * 34 # the dip lands at the 40th-newest + for i, rise in enumerate(rises): + _seed_thermal_session(db, now - (len(rises) - i) * 7200, ambient_c=25.0, + rise_ref_c=rise + (0.3 if i % 2 else -0.3)) + fits = thermal.fit_sessions(db, now) + assert len(fits) == len(rises), "every session in the lookback is fitted" + # The forecast still learns from the newest FORECAST_SESSIONS only. + params = thermal.fit_history(db, now, fits=fits) + assert params.tau_fits == thermal.FORECAST_SESSIONS + assert thermal.forecast_fits(fits) == fits[-thermal.FORECAST_SESSIONS:] + drift = thermal.detect_drift(fits) + assert drift is not None and drift["drifting"] is False + assert abs(drift["delta_c"]) < thermal.DRIFT_WARN_C + # The window the cap used to impose does alarm on the same data. + assert thermal.detect_drift(fits[-40:])["drifting"] is True + + async def test_thermal_baseline_anchor(db): now = time.time() rises = [36.0, 36.5, 35.8, 36.2, 42.0, 41.5, 42.3] @@ -1649,6 +1674,35 @@ async def test_thermal_api_endpoint(db): await client.close() +async def test_thermal_api_serves_stale_fit_while_refitting(db, monkeypatch): + # The history fit can take tens of seconds and the amp controller polls + # with a 10 s timeout: a stale fit is served at once while one refit + # runs in the background, instead of every poll waiting on it. + app = make_app(db, EventBus(), None) + async with TestClient(TestServer(app)) as client: + assert (await client.get("/api/thermal")).status == 200 # cold: waits for the first fit + release, finished, calls = threading.Event(), threading.Event(), [] + real = thermal.fit_sessions + + def slow_fit(*args, **kwargs): + calls.append(1) + release.wait(10) + try: + return real(*args, **kwargs) + finally: + finished.set() + + monkeypatch.setattr(thermal, "fit_sessions", slow_fit) + monkeypatch.setattr(web_module, "THERMAL_REFIT_S", 0.0) # every request finds it stale + t0 = time.monotonic() + for _ in range(3): + assert (await client.get("/api/thermal")).status == 200 + assert time.monotonic() - t0 < 2.0, "the stale fit is served without waiting" + assert len(calls) == 1, "one background refit, not one per request" + release.set() + await asyncio.to_thread(finished.wait, 10) + + async def test_backoff_on_unreachable_host(db, unused_tcp_port): cfg = Config( host=f"127.0.0.1:{unused_tcp_port}", diff --git a/wallmonitor/thermal.py b/wallmonitor/thermal.py index 09e7699..93d8899 100644 --- a/wallmonitor/thermal.py +++ b/wallmonitor/thermal.py @@ -276,6 +276,14 @@ def ambient_from_idle_handle(handle_c: float, model: IdleOffset = BUILTIN_IDLE_O # this split segments; each segment's opening ramp is a fit candidate. SEGMENT_SPLIT_GAP_S = 300.0 MAX_SEGMENTS_PER_SESSION = 4 +# The forecast's tau, rise and current exponent come from the newest this +# many sessions; the degradation watch uses every session in the lookback. +# The cap predates any scoring. Lifting it for the forecast was backtested +# (one install, 114 sessions): the median tau fell 11.5 -> 11.25 min and the +# in-run trajectory forecast at cold starts moved 0.4-0.5 C optimistic, +# the direction that trips the charger, for small gains elsewhere. So the +# forecast keeps it until a deliberate window is scored on its own. +FORECAST_SESSIONS = 40 @dataclass @@ -677,12 +685,22 @@ def fit_sessions(db: Database, now: float, lookback_days: float = 120.0) -> list back as the handle warmed, so the fitted plateau is a setpoint it held rather than the connector's own equilibrium. Such fits still serve the forecast — they describe what the handle actually did — but the - degradation watch compares only free-running ones.""" + degradation watch compares only free-running ones. + + Every session in the lookback is fitted, not just the newest few. The + degradation watch regresses on whatever this returns, so a count cap + makes its baseline whatever happened ~40 charges ago: on one install a + single odd week at the edge of a 40-session window read as +8 °C of + confirmed drift, where the full history since the verified anchor was + flat (-0.04 °C over 80 days). The time window, not a session count, + decides what the baseline is. Fits from the newest FORECAST_SESSIONS + carry forecast=True; only those teach the forecast (forecast_fits).""" sessions = [ session for session in db.sessions_range(now - lookback_days * 86400, now) if session.get("end_ts") and (session.get("charging_s") or 0) >= MIN_SEGMENT_S - ][:40] + ] + forecast_ids = {session["id"] for session in sessions[:FORECAST_SESSIONS]} fits: list[dict] = [] idle_model = load_idle_offset(db) for sess in sessions: @@ -784,6 +802,7 @@ def fit_sessions(db: Database, now: float, lookback_days: float = 120.0) -> list fits.append( { "session_id": sess["id"], + "forecast": sess["id"] in forecast_ids, "start_ts": seg[0][0], "tau_min": round(tau_s / 60.0, 2), "rise_ref_c": round(rise, 2) if rise is not None else None, @@ -810,6 +829,12 @@ def fit_sessions(db: Database, now: float, lookback_days: float = 120.0) -> list return fits +def forecast_fits(fits: list[dict]) -> list[dict]: + """The fits the forecast's parameters come from: the newest + FORECAST_SESSIONS sessions' (see there for why not all of them).""" + return [fit for fit in fits if fit.get("forecast", True)] + + def _fit_current_exponent(fits: list[dict]) -> tuple[float, int, float | None]: """The exponent n in rise = rise_ref * (I/48)^n, from free-running fits: (n, fits used, standard error). The prior with no fits used when the @@ -848,16 +873,21 @@ def fit_history(db: Database, now: float, lookback_days: float = 120.0, degradation watch, and the rise_ref_c median taken here, all share one current law. With the I^2 prior, off-reference fits carry a bias that the watch's current term then has to absorb; with a fitted exponent that - term has nothing left to explain.""" + term has nothing left to explain. + + The parameters, exponent included, come from forecast_fits only; the + re-normalization reaches every fit, so the watch's older fits share the + forecast's current law too.""" if fits is None: fits = fit_sessions(db, now, lookback_days) - exponent, exp_fits, exp_se = _fit_current_exponent(fits) + recent = forecast_fits(fits) + exponent, exp_fits, exp_se = _fit_current_exponent(recent) for fit in fits: if fit.get("rise_c") is not None and fit.get("current_a"): fit["rise_ref_c"] = round(fit["rise_c"] * (REF_CURRENT_A / fit["current_a"]) ** exponent, 2) - taus = [fit["tau_min"] for fit in fits] - rises = [fit["rise_ref_c"] for fit in fits if fit["rise_ref_c"] is not None] - rmses = [fit["rmse_c"] for fit in fits] + taus = [fit["tau_min"] for fit in recent] + rises = [fit["rise_ref_c"] for fit in recent if fit["rise_ref_c"] is not None] + rmses = [fit["rmse_c"] for fit in recent] return ThermalParams( tau_min=median(taus) if taus else DEFAULT_TAU_MIN, rise_ref_c=median(rises) if rises else DEFAULT_RISE_REF_C, diff --git a/wallmonitor/web.py b/wallmonitor/web.py index 1763f86..44df66e 100644 --- a/wallmonitor/web.py +++ b/wallmonitor/web.py @@ -22,6 +22,9 @@ STATIC_PKG = "wallmonitor.static" +# How long /api/thermal serves a history fit before refitting in the background. +THERMAL_REFIT_S = 6 * 3600 + def _float_q(request: web.Request, name: str, default: float) -> float: """Float query parameter, silently falling back on absent/garbage input @@ -224,27 +227,60 @@ async def api_event_ingest(request: web.Request) -> web.Response: # Model parameters change only as new sessions land, so the (SQLite-heavy) # history fit is cached; the live prediction is computed on every call. - thermal_fit: dict = {"params": None, "fits": [], "ts": 0.0, "idle_offset": None} + # The fit covers the whole drift lookback and takes tens of seconds on a + # long history (27 s for 72 fits on a 4-core Intel N100), while the amp + # controller polls with a 10 s timeout. So a stale fit is served at once + # and refitted in the background; only a cold cache or ?refit waits. + thermal_fit: dict = {"params": None, "fits": [], "ts": 0.0, "idle_offset": None, "task": None} + + async def refit_thermal() -> None: + started = time.time() + idle_offset = await asyncio.to_thread(db.get_setting, thermal.IDLE_OFFSET_SETTING) + fits = await asyncio.to_thread(thermal.fit_sessions, db, started) + params = await asyncio.to_thread(thermal.fit_history, db, started, fits=fits) + if started >= thermal_fit["ts"]: # an older refit finishing late must not win + thermal_fit.update(fits=fits, params=params, ts=started, idle_offset=idle_offset) + + def _log_refit_failure(task: asyncio.Task) -> None: + if not task.cancelled() and task.exception() is not None: + log.error("thermal refit failed", exc_info=task.exception()) + + def refit_in_background() -> asyncio.Task: + """The running refit, or a new one: never two at once.""" + task = thermal_fit["task"] + if task is None or task.done(): + task = asyncio.create_task(refit_thermal(), name="thermal-refit") + task.add_done_callback(_log_refit_failure) + thermal_fit["task"] = task + return task + + async def warm_thermal(_app: web.Application) -> None: + refit_in_background() + + async def stop_thermal(_app: web.Application) -> None: + task = thermal_fit["task"] + if task is not None and not task.done(): + task.cancel() + + app.on_startup.append(warm_thermal) + app.on_cleanup.append(stop_thermal) async def api_thermal(request: web.Request) -> web.Response: """The full thermal picture: fitted model, live forecast, per-segment fits, drift verdict, baseline anchor. Also what the BLE amp - controller polls every 30 s. ?refit busts the 6 h fit cache.""" + controller polls every 30 s. ?refit waits for a fresh fit.""" now = time.time() # A newly adopted idle-offset model reinterprets every proxy-tier # fit, so the cache is keyed on the stored model too. idle_offset = await asyncio.to_thread(db.get_setting, thermal.IDLE_OFFSET_SETTING) - if ( - thermal_fit["params"] is None - or now - thermal_fit["ts"] > 6 * 3600 - or thermal_fit["idle_offset"] != idle_offset - or "refit" in request.query - ): - fits = await asyncio.to_thread(thermal.fit_sessions, db, now) - thermal_fit["fits"] = fits - thermal_fit["params"] = thermal.fit_history(db, now, fits=fits) - thermal_fit["ts"] = now - thermal_fit["idle_offset"] = idle_offset + if "refit" in request.query: + await refit_thermal() + elif thermal_fit["params"] is None: + # Cold (just started): nothing to serve yet. Shielded so a caller + # that gives up doesn't cancel the refit everyone else is waiting on. + await asyncio.shield(refit_in_background()) + elif now - thermal_fit["ts"] > THERMAL_REFIT_S or thermal_fit["idle_offset"] != idle_offset: + refit_in_background() result = await asyncio.to_thread(thermal.predict, db, now, thermal_fit["params"]) anchor = await asyncio.to_thread(thermal.baseline_anchor, db) return web.json_response(