adminandClaude Opus 5.5
cd1bf2a694
HOLD's time check moves into Maintenance._hold(), used by step() and
_no_soc(). A hold that loses its SoC now ends at hold_min, writes
last_completed and logs the same COMPLETE line. The 15 min HOLD line says
the hold still ends at its normal time. Behaviour with a reading is
unchanged (differential trace against db9e231).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
256 lines
9.7 KiB
Python
256 lines
9.7 KiB
Python
"""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, 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
|
|
|
|
|
|
|
|
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)
|
|
m = in_hold()
|
|
res, _, out, _ = run(m, H0, 30, 100.0)
|
|
check("HOLD with a reading is unchanged: 0 W, ends at hold_min", out is None
|
|
and res.setpoint_w == 0.0)
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED:")
|
|
for f in fails:
|
|
print(" -", f)
|
|
sys.exit(1)
|
|
print("all ok")
|