"""MAINT-15: the HOLD ends on time with no SoC reading. `python3 test_maint15.py` Graded here: a HOLD that loses its SoC (at the start, mid-hold, just before hold_min) ends on time, writes last_completed and says the COMPLETE line; SoC back mid-hold changes nothing; HOLD stays 0 W; one pack's loss leaves the other pack alone; a restart in HOLD with no SoC and an old state file. See DECISIONS/MAINT-15-hold-ends-without-soc.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, DRAIN, HOLD, IDLE, NOSOC_NOTE_MIN, MaintConfig, Maintenance, ) 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 HOLD_MIN = CFG.hold_min def in_hold(store=None, unit=None, at=T0): """A pack in HOLD that started at `at` + 1 min (charge counted as done).""" m = in_charge(store, unit, at) m.tick(at + MIN, 100.0) assert m.phase == HOLD return m H0 = T0 + MIN # the hold's start print("1. SoC lost in HOLD: the hold ends on time") for label, lost_at in (("at the start", 0), ("mid-hold", 60), ("just before hold_min", HOLD_MIN - 1)): m = in_hold() run(m, H0, lost_at, 100.0) if lost_at else None res, _, out, ev = run(m, H0 + lost_at * MIN, HOLD_MIN + 5 - lost_at, None) check(f"lost {label}: ends at hold_min, to IDLE, not failed", out is not None and out[0].phase == IDLE and out[0].failed == "" and out[1] == H0 + HOLD_MIN * MIN) check("...last_completed is written at that time", m.store.get_time("last_completed") == H0 + HOLD_MIN * MIN) check("...the same COMPLETE line, once, and no last_failed", sum("cycle COMPLETE (120 min hold)" in e for e in ev) == 1 and m.store.get_time("last_failed") is None) check("...the pack is idle and hands the setpoint back", m.phase == IDLE and not out[0].owns_setpoint) m = in_hold() res, _, out, _ = run(m, H0, HOLD_MIN - 1, None) check("before hold_min: still HOLD at 0 W, owned, nothing written", out is None and res.phase == HOLD and res.owns_setpoint and res.setpoint_w == 0.0 and m.store.get_time("last_completed") is None) check("...the pack is not due again (next due date moved after the end)", (run(m, H0 + (HOLD_MIN - 1) * MIN, 3, None), m.due(H0 + 130 * MIN))[1] is False) print("2. the text") m = in_hold() _, _, out, ev = run(m, H0, HOLD_MIN + 5, None) note = [e for e in ev if "no battery level for" in e] check("one 15 min line, saying the hold ends at its normal time", len(note) == 1 and "normal time" in note[0] and "waits" not in note[0] and f"{NOSOC_NOTE_MIN} min" in note[0]) check("...and the other events are the one COMPLETE line only", len(ev) == 2) m = in_hold() _, _, _, ev = run(m, H0, NOSOC_NOTE_MIN - 2, None) check("under the threshold the line is not said", ev == []) print("3. SoC comes back mid-hold") ends = {} for label, soc in (("no SoC", lambda i: None if i < 40 else 100.0), ("SoC the whole time", 100.0), ("a low SoC after the gap", lambda i: None if i < 40 else 55.0)): m = in_hold() res, _, out, ev = run(m, H0, HOLD_MIN + 5, soc) ends[label] = (out[1], m.store.get_time("last_completed")) if out else None check(f"{label}: ends at hold_min", out is not None and out[1] == H0 + HOLD_MIN * MIN) check("...the end time is the same with or without the gap", len(set(ends.values())) == 1) m = in_hold() run(m, H0, 40, None) res, _, out, _ = run(m, H0 + 40 * MIN, 30, 100.0) check("back mid-hold: still HOLD at 0 W until the time is up", out is None and res.phase == HOLD and res.setpoint_w == 0.0) print("4. restart in HOLD with no SoC") 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) m.tick(H0, 100.0) run(m, H0, 50, None) m2 = Maintenance(CFG, Store(path)) r = m2.tick(H0 + 51 * MIN, None) check("a fresh process reads HOLD and holds 0 W", m2.phase == HOLD and r.phase == HOLD and r.owns_setpoint and r.setpoint_w == 0.0) res, _, out, _ = run(m2, H0 + 52 * MIN, HOLD_MIN, None) check("...and the hold ends at hold_min from the hold start, not the restart", out and out[0].phase == IDLE and out[1] == H0 + HOLD_MIN * MIN and m2.store.get_time("last_completed") == H0 + HOLD_MIN * MIN) import json old = {"phase": "hold", "phase_started": H0.isoformat(), "last_completed": None, "last_start_attempt": None, "auto": True} p2 = os.path.join(d, "old.json") json.dump(old, open(p2, "w")) mo = Maintenance(CFG, Store(p2)) r = mo.tick(H0 + 10 * MIN, None) check("an old state file in HOLD loads and holds 0 W", mo.phase == HOLD and r.setpoint_w == 0.0) res, _, out, _ = run(mo, H0 + 11 * MIN, HOLD_MIN, None) check("...and ends on time, writing last_completed", out and out[1] == H0 + HOLD_MIN * MIN and mo.store.get_time("last_completed") == H0 + HOLD_MIN * MIN) mr = Maintenance(CFG, Store(p2)) check("...the written file reloads as a finished cycle", mr.phase == IDLE and mr.store.get_time("last_completed") is not None) print("5. two packs") a, b = (Maintenance(CFG, Mem(), unit="inverter-a", name="inverter-a"), Maintenance(CFG, Mem(), unit="inverter-b", name="inverter-b")) a_ok, b_ok = (Maintenance(CFG, Mem(), unit="inverter-a", name="inverter-a"), Maintenance(CFG, Mem(), unit="inverter-b", name="inverter-b")) for x in (a, a_ok): x.force_start(T0) x.tick(T0, 11.0) x.tick(H0, 100.0) # B is idle; A's hold loses its SoC (twin site: A keeps it) same, a_end = True, None for i in range(0, HOLD_MIN + 10): t = H0 + i * MIN ra = a.tick(t, None) rb = b.tick(t, 60.0) ra2 = a_ok.tick(t, 100.0) rb2 = b_ok.tick(t, 60.0) same = same and (rb.phase, rb.owns_setpoint, rb.setpoint_w) == \ (rb2.phase, rb2.owns_setpoint, rb2.setpoint_w) and b.phase == b_ok.phase if a_end is None and a.phase == IDLE: a_end = t check("A loses SoC in HOLD: A ends on time (as the twin with a reading)", a_end == H0 + HOLD_MIN * MIN and a_ok.phase == IDLE and a.store.get_time("last_completed") == a_ok.store.get_time("last_completed")) check("...B is the same as in the twin site where A kept its SoC", same) # B in its own HOLD with SoC, A in HOLD without a, b = (Maintenance(CFG, Mem(), unit="inverter-a", name="inverter-a"), Maintenance(CFG, Mem(), unit="inverter-b", name="inverter-b")) a.force_start(T0); a.tick(T0, 11.0); a.tick(H0, 100.0) b_start = H0 + 30 * MIN b.force_start(b_start - MIN); b.tick(b_start - MIN, 11.0); b.tick(b_start, 100.0) b_end = None for i in range(0, HOLD_MIN + 40): t = H0 + i * MIN a.tick(t, None) if t >= b_start: b.tick(t, 100.0) if b_end is None and b.phase == IDLE: b_end = t check("B (SoC present) ends at its own hold_min, A's loss changes nothing", b_end == b_start + HOLD_MIN * MIN and a.phase == IDLE and b.store.get_time("last_completed") == b_start + HOLD_MIN * MIN) check("...A's end is not delayed by B", a.store.get_time("last_completed") == H0 + HOLD_MIN * MIN) print("6. other phases unchanged") m = in_drain() res, _, out, _ = run(m, T0, 60, None) check("DRAIN with no SoC still holds 0 W and does not end", out is None and res.phase == DRAIN and res.setpoint_w == 0.0) m = in_charge() res, _, out, _ = run(m, T0, 600, None) check("CHARGE with no SoC still keeps charging", out is None and res.phase == CHARGE and res.charge_only) def old_hold_branch(phase_started, last_completed, now, hold_min): """FROZEN copy of release/1.0 (db9e231) HOLD branch of step(), with a reading: returns (phase, owns_setpoint, setpoint_w, events, last_completed). Kept as a reference for the differential check below.""" elapsed = (now - phase_started).total_seconds() / 60.0 if elapsed >= hold_min: return (IDLE, False, None, [f"maintenance: cycle COMPLETE ({elapsed:.0f} min hold) " "- handing back to the loop"], now) return (HOLD, True, 0.0, [], last_completed) print("7. with a reading: the new HOLD against the frozen release/1.0 branch") for hm in (0, 5, 60, 120): cfg = MaintConfig(enabled=True, interval_days=28, start_hour=19, discharge_w=2000, charge_w=2000, soc_floor=11, soc_target=100, hold_min=hm) m = Maintenance(cfg, Mem()) m.force_start(T0) m.tick(T0, 11.0) m.tick(H0, 100.0) started, last, same, end = H0, None, True, None for i in range(0, hm + 15): t = H0 + i * MIN if m.phase != HOLD: break want = old_hold_branch(started, last, t, hm) r = m.tick(t, 100.0 - (i % 3)) got = (r.phase, r.owns_setpoint, r.setpoint_w, r.events, m.store.get_time("last_completed")) same = same and got == want last = want[4] if r.phase == IDLE: end = t check(f"hold_min {hm}: every tick identical (phase, owns, W, events, " "last_completed)", same) check("...ends at hold_min, last_completed = that tick, one COMPLETE line", end == H0 + hm * MIN and m.store.get_time("last_completed") == end and m.phase == IDLE) m = in_hold() res, _, out, ev = run(m, H0, HOLD_MIN + 5, 100.0) check("HOLD with a reading: 0 W until the end, then IDLE at hold_min", out is not None and out[1] == H0 + HOLD_MIN * MIN and out[0].phase == IDLE) check("...last_completed is that tick and the COMPLETE line is said once", m.store.get_time("last_completed") == H0 + HOLD_MIN * MIN and ev == ["maintenance: cycle COMPLETE (120 min hold) " "- handing back to the loop"]) print() if fails: print(f"{len(fails)} FAILED:") for f in fails: print(" -", f) sys.exit(1) print("all ok")