glenn schrooyenandClaude Opus 5.5
f39e52efd3
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
357 lines
14 KiB
Python
357 lines
14 KiB
Python
"""MAINT-14: maintenance with no SoC reading.
|
|
`python3 test_maint14.py`
|
|
|
|
Graded here: charge and drain with SoC None at one and two packs - the charge
|
|
timeout and 7 day backstop still run (failure, or done when the last SoC was
|
|
>= target - 3), the drain holds at 0 W and times out, SoC coming back resumes
|
|
normally with no back-off, the log lines (once per episode), a restart while the
|
|
SoC is missing, and one pack's missing SoC leaving the other pack alone.
|
|
See DECISIONS/MAINT-14-no-soc-in-maintenance.md.
|
|
|
|
Mutating what this file grades? Read MUTATION.md first (stale bytecode).
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from app.maintenance import (
|
|
CHARGE, CHARGE_NEAR_FULL_PCT, CHARGE_TIMEOUT_MIN, CHARGE_WALL_TIMEOUT_MIN,
|
|
DRAIN, DRAIN_TIMEOUT_MIN, FAIL_BACKOFF_DAYS, HOLD, IDLE, NOSOC_NOTE_MIN,
|
|
MaintConfig, Maintenance, arbitrate,
|
|
)
|
|
from app.store import DEFAULTS, Store
|
|
|
|
UTC = timezone.utc
|
|
fails = []
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
CFG = MaintConfig(enabled=True, interval_days=28, start_hour=19,
|
|
discharge_w=2000, charge_w=2000, soc_floor=11,
|
|
soc_target=100, hold_min=120)
|
|
T0 = datetime(2026, 10, 4, 18, 24, tzinfo=UTC)
|
|
MIN = timedelta(minutes=1)
|
|
|
|
|
|
class Mem:
|
|
def __init__(self, data=None):
|
|
self.data = dict(DEFAULTS, **(data or {}))
|
|
|
|
def set(self, k, v):
|
|
self.data[k] = v
|
|
|
|
def set_time(self, k, when):
|
|
self.data[k] = when.astimezone(UTC).isoformat() if when else None
|
|
|
|
def get_time(self, k):
|
|
raw = self.data.get(k)
|
|
return datetime.fromisoformat(raw) if raw else None
|
|
|
|
def save(self):
|
|
pass
|
|
|
|
|
|
def in_drain(store=None, unit=None, at=T0):
|
|
m = Maintenance(CFG, store or Mem(), unit=unit, name=unit or "")
|
|
m.force_start(at)
|
|
assert m.phase == DRAIN
|
|
return m
|
|
|
|
|
|
def in_charge(store=None, unit=None, at=T0):
|
|
m = in_drain(store, unit, at)
|
|
m.tick(at, 11.0)
|
|
assert m.phase == CHARGE
|
|
return m
|
|
|
|
|
|
def run(m, start, minutes, soc, gate=True, step=1):
|
|
"""Tick every `step` min; returns (last result, end time, first result that
|
|
is not the starting phase with its time, all events)."""
|
|
ph, t, res, out, ev = m.phase, start, None, None, []
|
|
for i in range(0, minutes, step):
|
|
t = start + i * MIN
|
|
s = soc(i) if callable(soc) else soc
|
|
res = m.tick(t, s, gate_open=gate)
|
|
ev += res.events
|
|
if out is None and res.phase != ph:
|
|
out = (res, t)
|
|
break
|
|
return res, start + minutes * MIN, out, ev
|
|
|
|
|
|
print("1. charge with no SoC")
|
|
m = in_charge()
|
|
res, _, out, _ = run(m, T0, 600, None)
|
|
check("it keeps charging: the loop owns the setpoint, no start of a new phase",
|
|
out is None and res.phase == CHARGE and res.charge_only
|
|
and not res.owns_setpoint)
|
|
|
|
m = in_charge()
|
|
run(m, T0, 60, 60.0)
|
|
res, _, out, ev = run(m, T0 + 60 * MIN, CHARGE_TIMEOUT_MIN + 10, None)
|
|
check("the charge timeout still runs: it fails once the time is up (last SoC 60)",
|
|
out and out[0].failed == "charge" and out[0].phase == IDLE)
|
|
check("...with the MAINT-13 back-off and its one plain line",
|
|
m.store.get_time("last_failed") is not None
|
|
and sum("did not finish" in e for e in ev) == 1
|
|
and f"{FAIL_BACKOFF_DAYS} days" in ev[-1])
|
|
check("...it took 36 h of allowed time from the charge start, not more",
|
|
out and out[1] <= T0 + timedelta(minutes=CHARGE_TIMEOUT_MIN + 10))
|
|
check("...and the pack does not want a start in the back-off",
|
|
not m.wants_start((T0 + timedelta(days=3)).replace(hour=19)))
|
|
|
|
m = in_charge()
|
|
res, _, out, _ = run(m, T0, CHARGE_TIMEOUT_MIN + 10, None)
|
|
check("never saw a SoC: the timeout is a failure, not 'done'",
|
|
out and out[0].failed == "charge")
|
|
|
|
for last in (97.0, 99.0):
|
|
m = in_charge()
|
|
run(m, T0, 10, last)
|
|
res, _, out, ev = run(m, T0 + 10 * MIN, CHARGE_TIMEOUT_MIN + 20, None)
|
|
check(f"last SoC {last:.0f} (>= target - {CHARGE_NEAR_FULL_PCT}): time up counts as done",
|
|
out and out[0].phase == HOLD and out[0].failed == ""
|
|
and m.store.get_time("last_failed") is None)
|
|
check("...one plain line says so",
|
|
sum("counts as done" in e for e in ev) == 1
|
|
and f"{last:.0f} %" in " ".join(ev))
|
|
m = in_charge()
|
|
run(m, T0, 10, 96.0)
|
|
res, _, out, _ = run(m, T0 + 10 * MIN, CHARGE_TIMEOUT_MIN + 20, None)
|
|
check("last SoC 96 (below target - 3): time up is a failure",
|
|
out and out[0].failed == "charge")
|
|
m = in_charge()
|
|
run(m, T0, 10, 99.0)
|
|
run(m, T0 + 10 * MIN, 10, 50.0)
|
|
res, _, out, _ = run(m, T0 + 20 * MIN, CHARGE_TIMEOUT_MIN + 20, None)
|
|
check("'last seen' is the last reading, not the best (99 then 50: failure)",
|
|
out and out[0].failed == "charge")
|
|
|
|
print("2. the 7 day backstop, with the gate shut")
|
|
m = in_charge()
|
|
run(m, T0, 10, 70.0)
|
|
res, _, out, _ = run(m, T0 + 10 * MIN, CHARGE_WALL_TIMEOUT_MIN + 30, None,
|
|
gate=False, step=5)
|
|
check("gate never opens, last SoC 70: failure at 7 days",
|
|
out and out[0].failed == "charge"
|
|
and out[1] >= T0 + timedelta(minutes=CHARGE_WALL_TIMEOUT_MIN)
|
|
and out[1] <= T0 + timedelta(minutes=CHARGE_WALL_TIMEOUT_MIN + 10))
|
|
m = in_charge()
|
|
run(m, T0, 10, 98.0)
|
|
res, _, out, _ = run(m, T0 + 10 * MIN, CHARGE_WALL_TIMEOUT_MIN + 30, None,
|
|
gate=False, step=5)
|
|
check("gate never opens, last SoC 98: done at 7 days",
|
|
out and out[0].phase == HOLD and out[0].failed == ""
|
|
and out[1] >= T0 + timedelta(minutes=CHARGE_WALL_TIMEOUT_MIN))
|
|
m = in_charge()
|
|
run(m, T0, 10, 70.0)
|
|
res, _, out, _ = run(m, T0 + 10 * MIN, 3000, None, gate=False, step=5)
|
|
check("gate shut, no SoC: the open-time clock does not run (no early end)",
|
|
out is None)
|
|
|
|
print("3. SoC comes back mid-charge")
|
|
m = in_charge()
|
|
run(m, T0, 30, 80.0)
|
|
run(m, T0 + 30 * MIN, 600, None)
|
|
res, _, out, _ = run(m, T0 + 630 * MIN, 5, 80.0)
|
|
check("normal behaviour resumes: still charging, nothing recorded",
|
|
out is None and res.phase == CHARGE
|
|
and m.store.get_time("last_failed") is None)
|
|
res, _, out, _ = run(m, T0 + 640 * MIN, 10, 100.0)
|
|
check("...and it completes at the target as usual",
|
|
out and out[0].phase == HOLD and m.store.get_time("last_failed") is None)
|
|
m = in_charge()
|
|
run(m, T0, 30, 98.0)
|
|
run(m, T0 + 30 * MIN, 600, None)
|
|
res, _, out, _ = run(m, T0 + 630 * MIN, 30, 98.0)
|
|
check("the stall wait paused while the SoC was missing (30 min flat after the "
|
|
"gap is not 60)", out is None)
|
|
res, _, out, _ = run(m, T0 + 660 * MIN, 40, 98.0)
|
|
check("...and completes once 60 min flat are in", out and out[0].phase == HOLD)
|
|
m = in_charge()
|
|
run(m, T0, 10, 70.0)
|
|
run(m, T0 + 10 * MIN, CHARGE_TIMEOUT_MIN - 100, None)
|
|
res, _, out, _ = run(m, T0 + (CHARGE_TIMEOUT_MIN - 90) * MIN, 5, 70.0)
|
|
check("a gap just before the limit then a reading: no failure, no back-off",
|
|
out is None and m.store.get_time("last_failed") is None)
|
|
|
|
print("4. drain with no SoC")
|
|
m = in_drain()
|
|
run(m, T0, 30, 60.0)
|
|
res, _, out, ev = run(m, T0 + 30 * MIN, 60, None)
|
|
check("DRAIN holds at 0 W, owned by maintenance, the same as HOLD",
|
|
out is None and res.phase == DRAIN and res.owns_setpoint
|
|
and res.setpoint_w == 0.0 and not res.charge_only)
|
|
hold = in_charge()
|
|
hold.tick(T0 + MIN, 100.0)
|
|
hr = hold.tick(T0 + 2 * MIN, None)
|
|
check("...the same answer as HOLD gives",
|
|
(hr.owns_setpoint, hr.setpoint_w) == (res.owns_setpoint, res.setpoint_w))
|
|
check("never above 0 W on any missing tick",
|
|
all(m.tick(T0 + (100 + i) * MIN, None).setpoint_w == 0.0 for i in range(30)))
|
|
check("one plain line that the drain holds, once, not per tick",
|
|
sum("drain holds at 0 W" in e for e in ev) == 1)
|
|
check("one plain line after the threshold, once",
|
|
sum("no battery level for" in e for e in ev) == 1
|
|
and f"{NOSOC_NOTE_MIN} min" in " ".join(ev))
|
|
m = in_drain()
|
|
res, _, out, ev = run(m, T0, NOSOC_NOTE_MIN - 1, None)
|
|
check("under the threshold the 'no level for long' line is not said",
|
|
not any("no battery level for" in e for e in ev))
|
|
m = in_drain()
|
|
res, _, out, ev = run(m, T0, DRAIN_TIMEOUT_MIN + 10, None)
|
|
check("the 12 h drain timeout still runs: failure with back-off",
|
|
out and out[0].failed == "drain" and out[0].phase == IDLE
|
|
and out[1] > T0 + timedelta(minutes=DRAIN_TIMEOUT_MIN)
|
|
and m.store.get_time("last_failed") is not None)
|
|
m = in_drain()
|
|
run(m, T0, 60, None)
|
|
res, _, out, _ = run(m, T0 + 60 * MIN, 5, 60.0)
|
|
check("SoC back: the drain continues at the drain rate",
|
|
out is None and res.phase == DRAIN and res.owns_setpoint
|
|
and res.setpoint_w == CFG.discharge_w)
|
|
res, _, out, _ = run(m, T0 + 70 * MIN, 5, 10.0)
|
|
check("...and ends at the floor as usual",
|
|
out and out[0].phase == CHARGE and m.store.get_time("last_failed") is None)
|
|
m = in_drain()
|
|
run(m, T0, 60, None)
|
|
run(m, T0 + 60 * MIN, 5, 60.0)
|
|
res, _, out, ev = run(m, T0 + 70 * MIN, 5, None)
|
|
check("a second episode says its lines again",
|
|
any("drain holds at 0 W" in e for e in ev))
|
|
|
|
print("5. the log is once per episode")
|
|
m = in_charge()
|
|
res, _, out, ev = run(m, T0, 300, None)
|
|
check("a 5 h charge episode: exactly one 'no level for long' line, none per tick",
|
|
len(ev) == 1 and "no battery level for" in ev[0])
|
|
run(m, T0 + 300 * MIN, 3, 70.0)
|
|
res, _, out, ev = run(m, T0 + 303 * MIN, 60, None)
|
|
check("...and one more for a new episode", len(ev) == 1)
|
|
m = in_charge()
|
|
res, _, out, ev = run(m, T0, 5, None)
|
|
check("a 5 min gap says nothing", ev == [])
|
|
|
|
print("6. restart while the SoC is missing")
|
|
with tempfile.TemporaryDirectory() as d:
|
|
path = os.path.join(d, "state.json")
|
|
m = Maintenance(CFG, Store(path))
|
|
m.force_start(T0)
|
|
m.tick(T0, 11.0)
|
|
run(m, T0, 97, lambda i: 90.0 if i < 96 else 98.0) # 98 seen 1 min ago
|
|
run(m, T0 + 97 * MIN, 3, None) # SoC lost; state saved at once
|
|
t = T0 + 100 * MIN # restart 3 min after the loss, before the next 5 min save
|
|
m2 = Maintenance(CFG, Store(path))
|
|
r = m2.tick(t, None)
|
|
check("a fresh process reads charge, and still charges", m2.phase == CHARGE
|
|
and r.phase == CHARGE and r.charge_only)
|
|
check("...with the open minutes and the last SoC seen from the file",
|
|
m2._trk is not None and m2._trk["last"] == 98.0
|
|
and 92 <= m2._trk["open_min"] <= 105)
|
|
res, _, out, _ = run(m2, t + MIN, CHARGE_TIMEOUT_MIN, None, step=5)
|
|
check("...so the timeout counts from the first charge minute and, with the "
|
|
"last SoC 98, completes (no failure)",
|
|
out and out[0].phase == HOLD and out[0].failed == ""
|
|
and out[1] < T0 + timedelta(minutes=CHARGE_TIMEOUT_MIN + 60))
|
|
# drain: restart keeps it at 0 W and keeps the clock
|
|
path2 = os.path.join(d, "s2.json")
|
|
md = Maintenance(CFG, Store(path2))
|
|
md.force_start(T0)
|
|
md.tick(T0 + 100 * MIN, None)
|
|
md2 = Maintenance(CFG, Store(path2))
|
|
r = md2.tick(T0 + 101 * MIN, None)
|
|
check("a drain restarted with no SoC holds at 0 W", md2.phase == DRAIN
|
|
and r.setpoint_w == 0.0 and r.owns_setpoint)
|
|
res, _, out, _ = run(md2, T0 + 102 * MIN, DRAIN_TIMEOUT_MIN, None, step=5)
|
|
check("...and the 12 h clock is from the drain start, not the restart",
|
|
out and out[0].failed == "drain"
|
|
and out[1] <= T0 + timedelta(minutes=DRAIN_TIMEOUT_MIN + 10))
|
|
# a charge restarted with no record of any SoC (old state file)
|
|
old = {"phase": "charge", "phase_started": (T0 - timedelta(hours=35)).isoformat(),
|
|
"last_completed": None, "last_start_attempt": None, "auto": True}
|
|
import json
|
|
path3 = os.path.join(d, "s3.json")
|
|
json.dump(old, open(path3, "w"))
|
|
mo = Maintenance(CFG, Store(path3))
|
|
res, _, out, _ = run(mo, T0, 120, None)
|
|
check("an old state file with no SoC seen: the charge fails at 36 h",
|
|
out and out[0].failed == "charge")
|
|
|
|
print("7. two packs")
|
|
a, b = (Maintenance(CFG, Mem(), unit="inverter-a", name="inverter-a"),
|
|
Maintenance(CFG, Mem(), unit="inverter-b", name="inverter-b"))
|
|
a.force_start(T0)
|
|
b_alone = Maintenance(CFG, Mem(), unit="inverter-b", name="inverter-b")
|
|
same = True
|
|
for i in range(0, 120, 10):
|
|
t = T0 + i * MIN
|
|
plan = arbitrate([a, b], t)
|
|
ra = a.tick(t, None, blocked_by=plan[0])
|
|
rb = b.tick(t, 55.0, blocked_by=plan[1])
|
|
rb0 = b_alone.tick(t, 55.0, blocked_by=plan[1])
|
|
same = same and (rb.phase, rb.setpoint_w, rb.deferred) == \
|
|
(rb0.phase, rb0.setpoint_w, rb0.deferred)
|
|
check("A drains with no SoC at 0 W while B (SoC 55, idle) is as it is when A "
|
|
"is healthy", ra.setpoint_w == 0.0 and ra.owns_setpoint and rb.phase == IDLE
|
|
and rb.setpoint_w is None and not rb.owns_setpoint)
|
|
check("...B's answers are unchanged tick by tick", same)
|
|
res = None
|
|
for i in range(0, DRAIN_TIMEOUT_MIN + 30, 10):
|
|
t = T0 + 120 * MIN + i * MIN
|
|
plan = arbitrate([a, b], t)
|
|
res = a.tick(t, None, blocked_by=plan[0])
|
|
b.tick(t, 55.0, blocked_by=plan[1])
|
|
if res.failed:
|
|
break
|
|
check("A's drain times out and A backs off (A's keys only)",
|
|
res.failed == "drain" and a.store.get_time("last_failed") is not None
|
|
and b.store.get_time("last_failed") is None)
|
|
slot = (T0 + timedelta(days=1)).replace(hour=19, minute=5)
|
|
check("...then B, with its SoC, takes the slot as usual",
|
|
arbitrate([a, b], slot) == ["", ""] and b.wants_start(slot)
|
|
and not a.wants_start(slot))
|
|
b.force_start(slot)
|
|
rb = b.tick(slot + 5 * MIN, 50.0)
|
|
check("...and B's drain runs at the drain rate",
|
|
rb.phase == DRAIN and rb.setpoint_w == CFG.discharge_w)
|
|
# the other way: B charges with no SoC, A has a SoC
|
|
a = Maintenance(CFG, Mem(), unit="inverter-a", name="inverter-a")
|
|
b = Maintenance(CFG, Mem(), unit="inverter-b", name="inverter-b")
|
|
b.force_start(T0)
|
|
b.tick(T0, 11.0)
|
|
run(b, T0, 5, 97.0)
|
|
a_alone = Maintenance(CFG, Mem(), unit="inverter-a", name="inverter-a")
|
|
same, t = True, T0 + 10 * MIN
|
|
while t < T0 + timedelta(minutes=CHARGE_TIMEOUT_MIN + 30):
|
|
plan = arbitrate([a, b], t)
|
|
rb = b.tick(t, None, blocked_by=plan[1])
|
|
ra = a.tick(t, 60.0, blocked_by=plan[0])
|
|
ra0 = a_alone.tick(t, 60.0, blocked_by=plan[0])
|
|
same = same and (ra.phase, ra.setpoint_w, ra.deferred) == \
|
|
(ra0.phase, ra0.setpoint_w, ra0.deferred)
|
|
if rb.phase != CHARGE:
|
|
break
|
|
t += 5 * MIN
|
|
check("B charges with no SoC (last 97): A (SoC 60) is unchanged tick by tick",
|
|
same)
|
|
check("...B ends as done (HOLD), A untouched, no failure on either",
|
|
rb.phase == HOLD and rb.failed == ""
|
|
and b.store.get_time("last_failed") is None
|
|
and a.store.get_time("last_failed") is None and a.phase == IDLE)
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED")
|
|
for f in fails:
|
|
print(" -", f)
|
|
sys.exit(1)
|
|
print("all passed")
|