glenn schrooyenandClaude
5ee1061398
Stop cycle is disabled on a battery that is not cycling. Start cycle is disabled while any battery cycles and on a switched-off battery. Page only; the add-on still refuses each of these. DECISIONS notes the one-battery case. Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Bdqr7PiaHcBnzPituu2yTC
403 lines
16 KiB
Python
403 lines
16 KiB
Python
"""MAINT-11: start or stop the maintenance cycle of ONE battery.
|
|
`python3 test_maint11.py`
|
|
|
|
Graded here, at two and at three batteries: a start on a chosen battery with the
|
|
slot free (and it is that battery, not the one the bare button would pick); a
|
|
start with the slot held, in each moving phase, refused by name; an unknown
|
|
unit; a battery switched off; a battery that refuses to give power; a stop on a
|
|
battery that is not cycling; a stop on the cycling one; a request with no unit
|
|
doing what it always did; the web layer's codes; the words in both languages;
|
|
and a roster of one, whose requests, log and writes are pinned.
|
|
See DECISIONS/MAINT-11-cycle-per-battery.md.
|
|
|
|
Mutating what this file grades? Read MUTATION.md first (stale bytecode).
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import re
|
|
import socket
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
|
|
import aiohttp
|
|
from aiohttp import web as aioweb
|
|
|
|
from app import messages, packmode, web
|
|
from app.main import Controller
|
|
from app.maintenance import CHARGE, DRAIN, HOLD, IDLE, POWER_MOVING
|
|
from app.store import DEFAULTS
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
fails = []
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
class MemStore:
|
|
def __init__(self):
|
|
self.data = dict(DEFAULTS)
|
|
|
|
def set(self, k, v):
|
|
self.data[k] = v
|
|
|
|
def get_time(self, k):
|
|
raw = self.data.get(k)
|
|
if not raw:
|
|
return None
|
|
if isinstance(raw, datetime):
|
|
return raw
|
|
parsed = datetime.fromisoformat(raw)
|
|
return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
|
|
|
|
def set_time(self, k, when):
|
|
self.data[k] = when.astimezone(timezone.utc).isoformat() if when else None
|
|
|
|
def save(self):
|
|
pass
|
|
|
|
|
|
class Hass:
|
|
session = None
|
|
token = ""
|
|
|
|
def __init__(self, n):
|
|
self.numbers = {"sensor.grid": 5000.0}
|
|
self.limits_by = {}
|
|
for k in range(n):
|
|
self.numbers[f"sensor.soc_{k}"] = 80.0
|
|
self.numbers[f"sensor.batt_{k}"] = 0.0
|
|
self.limits_by[f"number.sp_{k}"] = (-5000.0, 5000.0, 1.0)
|
|
self.writes = []
|
|
|
|
async def number(self, entity, invert=False):
|
|
v = self.numbers.get(entity)
|
|
return None if v is None else (-v if invert else v)
|
|
|
|
async def limits(self, entity):
|
|
return self.limits_by.get(entity, (None, None, None))
|
|
|
|
async def set_number(self, entity, value):
|
|
self.writes.append((entity, value))
|
|
return True
|
|
|
|
async def call(self, domain, service, data):
|
|
return True
|
|
|
|
|
|
class Pub:
|
|
enabled = False
|
|
|
|
def publish(self, *a, **k):
|
|
pass
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
NAMES = ["inverter-a", "inverter-b", "inverter-c"]
|
|
OPTS = dict(meter_source="off", meter_entity="sensor.grid", gain=0.6,
|
|
slew_w=6000, deadband_w=15, target_grid_w=0, step_w=10,
|
|
heartbeat_s=10, stale_input_s=15, maintenance_discharge_w=2000,
|
|
maintenance_charge_w=2000)
|
|
|
|
|
|
def controller(n, modes=False):
|
|
opts = dict(OPTS)
|
|
opts["inverters"] = [
|
|
dict(name=NAMES[k], setpoint_entity=f"number.sp_{k}",
|
|
soc_entity=f"sensor.soc_{k}", batt_entity=f"sensor.batt_{k}",
|
|
max_w=5000.0) for k in range(n)]
|
|
if n == 1:
|
|
del opts["inverters"]
|
|
opts.update(setpoint_entity="number.sp_0", soc_entity="sensor.soc_0",
|
|
batt_entity="sensor.batt_0", max_w=5000.0)
|
|
if modes:
|
|
opts["battery_modes_enabled"] = True
|
|
hass = Hass(n)
|
|
c = Controller(opts, hass, MemStore(), Pub(), stop=asyncio.Event())
|
|
for inv in c.inverters:
|
|
inv.dev_min, inv.dev_max = -inv.max_w, inv.max_w
|
|
c.auto = True
|
|
return c, hass
|
|
|
|
|
|
def act(c, what, unit=None):
|
|
if unit is None:
|
|
return asyncio.run(c.handle_action(what))
|
|
return asyncio.run(c.handle_action(what, unit=unit))
|
|
|
|
|
|
def phases(c):
|
|
return [m.phase for m in c.maints]
|
|
|
|
|
|
def snapshot(c, hass):
|
|
return (phases(c), list(c.events), list(hass.writes),
|
|
json.dumps(c.store.data, sort_keys=True, default=str))
|
|
|
|
|
|
def settle(c):
|
|
"""Events carry a clock stamp; compare them without it."""
|
|
return [e.split(" ", 1)[1] for e in c.events]
|
|
|
|
|
|
for N in (2, 3):
|
|
print(f"\n{N} batteries")
|
|
names = NAMES[:N]
|
|
last = N - 1
|
|
|
|
# --- start with the slot free -------------------------------------------
|
|
c, hass = controller(N)
|
|
bare_pick = c._maint_button(datetime.now(timezone.utc))
|
|
check("with every battery never cycled the bare button picks the first "
|
|
"(this is the card's case)", bare_pick is c.maints[0])
|
|
r = act(c, "maint_start", names[last])
|
|
check(f"start on {names[last]}: its cycle starts, nobody else's",
|
|
r == {"unit": names[last], "phase": DRAIN}
|
|
and phases(c) == [IDLE] * last + [DRAIN])
|
|
check("...and the log says a forced start",
|
|
any("forced start" in e for e in c.events))
|
|
check("...and the machine holding the slot is the named one",
|
|
c.maint_index == last)
|
|
|
|
c, hass = controller(N)
|
|
r = act(c, "maint_start", names[0])
|
|
check("start on the first battery works the same way",
|
|
r["unit"] == names[0] and phases(c)[0] == DRAIN
|
|
and phases(c)[1:] == [IDLE] * last)
|
|
|
|
# --- start with the slot held --------------------------------------------
|
|
for phase in POWER_MOVING:
|
|
for holder in range(N):
|
|
c, hass = controller(N)
|
|
c.maints[holder].store.set("phase", phase)
|
|
before = snapshot(c, hass)
|
|
bad = []
|
|
for k in range(N):
|
|
r = act(c, "maint_start", names[k])
|
|
bad.append(not (r.get("code") == "busy"
|
|
and r["unit"] == names[holder]
|
|
and names[holder] in r["error"]
|
|
and "error" in r))
|
|
check(f"slot held by {names[holder]} in {phase}: a start on any "
|
|
f"battery (itself too) is refused, naming {names[holder]}, "
|
|
f"and nothing changes", not any(bad)
|
|
and snapshot(c, hass) == before)
|
|
|
|
# --- unknown unit --------------------------------------------------------
|
|
c, hass = controller(N)
|
|
before = snapshot(c, hass)
|
|
rs = [act(c, w, u) for w in ("maint_start", "maint_abort")
|
|
for u in ("inverter-z", "", "Inverter-A", 7, ["inverter-a"])]
|
|
check("an unknown unit (wrong name, empty, wrong case, not a string) is "
|
|
"refused for start and for stop, and nothing changes",
|
|
all(r.get("code") == "unit" and r["error"]
|
|
== messages.maint_refused("unit") for r in rs)
|
|
and snapshot(c, hass) == before)
|
|
|
|
# --- a battery switched off ---------------------------------------------
|
|
c, hass = controller(N, modes=True)
|
|
c.inverters[last].store_set(packmode.USE_KEY, False)
|
|
before = snapshot(c, hass)
|
|
r = act(c, "maint_start", names[last])
|
|
check("a battery switched off (battery modes on) is refused by name and "
|
|
"nothing changes", r.get("code") == "off" and r["unit"] == names[last]
|
|
and names[last] in r["error"] and snapshot(c, hass) == before)
|
|
r = act(c, "maint_start", names[0])
|
|
check("...while a battery that is on still starts", r.get("phase") == DRAIN
|
|
and phases(c)[0] == DRAIN)
|
|
c, hass = controller(N)
|
|
c.inverters[last].store_set(packmode.USE_KEY, False)
|
|
check("with battery modes off the stored switch is not consulted (it is "
|
|
"not shown, so it cannot refuse)",
|
|
act(c, "maint_start", names[last]).get("phase") == DRAIN)
|
|
|
|
# --- a battery that refuses to give power ---------------------------------
|
|
c, hass = controller(N)
|
|
c.inverters[last].refused = 1
|
|
r = act(c, "maint_start", names[last])
|
|
check("a battery that refuses to give power starts like the bare button "
|
|
"would start it (its drain then waits, as before)",
|
|
r == {"unit": names[last], "phase": DRAIN})
|
|
|
|
# --- stop ---------------------------------------------------------------
|
|
c, hass = controller(N)
|
|
before = snapshot(c, hass)
|
|
r = act(c, "maint_abort", names[0])
|
|
check("a stop on a battery that is not cycling is refused and nothing "
|
|
"changes (no 0 W written either)", r.get("code") == "idle"
|
|
and names[0] in r["error"] and snapshot(c, hass) == before)
|
|
c.maints[last].store.set("phase", CHARGE)
|
|
before = snapshot(c, hass)
|
|
r = act(c, "maint_abort", names[0])
|
|
check(f"a stop on {names[0]} while {names[last]} cycles is refused, and "
|
|
f"{names[last]}'s cycle is not touched", r.get("code") == "idle"
|
|
and phases(c)[last] == CHARGE and snapshot(c, hass) == before)
|
|
|
|
for phase in POWER_MOVING:
|
|
c, hass = controller(N)
|
|
c.maints[last].store.set("phase", phase)
|
|
r = act(c, "maint_abort", names[last])
|
|
check(f"a stop on the cycling battery ({phase}) ends its cycle, "
|
|
f"logs the abort and sets every inverter to 0 W",
|
|
r == {"unit": names[last], "phase": IDLE}
|
|
and phases(c) == [IDLE] * N
|
|
and any("ABORT (operator)" in e for e in c.events)
|
|
and {w for w in hass.writes}
|
|
== {(f"number.sp_{k}", 0.0) for k in range(N)}
|
|
and all(i.target == 0.0 for i in c.inverters))
|
|
|
|
# --- the old request ---------------------------------------------------
|
|
c, hass = controller(N)
|
|
r = act(c, "maint_start")
|
|
check("a start with no unit is the old call: the bare button's pick, and "
|
|
"the old answer", r == {"phase": DRAIN} and phases(c)[0] == DRAIN
|
|
and any("forced start" in e for e in c.events))
|
|
r = act(c, "maint_start")
|
|
check("...a second bare start re-enters the drain of the cycling battery, "
|
|
"as it always did", r == {"phase": DRAIN} and phases(c)[0] == DRAIN)
|
|
c.maints[last].store.set("phase", IDLE)
|
|
r = act(c, "maint_abort")
|
|
check("...and a bare stop stops the cycling battery and answers as before",
|
|
r == {"phase": IDLE} and phases(c) == [IDLE] * N
|
|
and any("ABORT (operator)" in e for e in c.events))
|
|
c, hass = controller(N)
|
|
r = act(c, "maint_abort")
|
|
check("...a bare stop with nothing cycling is still accepted, as before",
|
|
r == {"phase": IDLE} and hass.writes
|
|
and {w[1] for w in hass.writes} == {0.0})
|
|
|
|
# --- the web layer ----------------------------------------------------------
|
|
print("\nthe web layer")
|
|
|
|
|
|
async def serve(c, calls):
|
|
runner = aioweb.AppRunner(web.build_app(c, object()))
|
|
await runner.setup()
|
|
sock = socket.socket()
|
|
sock.bind(("127.0.0.1", 0))
|
|
port = sock.getsockname()[1]
|
|
sock.close()
|
|
await aioweb.TCPSite(runner, "127.0.0.1", port).start()
|
|
out = []
|
|
try:
|
|
async with aiohttp.ClientSession() as s:
|
|
for body in calls:
|
|
async with s.post(f"http://127.0.0.1:{port}/action",
|
|
json=body) as r:
|
|
out.append((r.status, await r.json(content_type=None)))
|
|
finally:
|
|
await runner.cleanup()
|
|
return out
|
|
|
|
|
|
c, hass = controller(2)
|
|
rs = asyncio.run(serve(c, [
|
|
{"action": "maint_start", "unit": "inverter-b"},
|
|
{"action": "maint_start", "unit": "inverter-a"},
|
|
{"action": "maint_abort", "unit": "inverter-a"},
|
|
{"action": "maint_start", "unit": "nope"},
|
|
{"action": "maint_abort", "unit": "inverter-b"},
|
|
{"action": "maint_start", "unit": None},
|
|
{"action": "maint_abort"},
|
|
{"action": "maint_start", "unit": ""},
|
|
]))
|
|
check("start with a unit: 200 and the unit's answer",
|
|
rs[0] == (200, {"ok": True, "result": {"unit": "inverter-b",
|
|
"phase": DRAIN}}))
|
|
check("start while the slot is held: 409, a code, the cycling battery, the "
|
|
"plain sentence", rs[1][0] == 409 and rs[1][1] == {
|
|
"ok": False, "code": "busy", "unit": "inverter-b",
|
|
"error": messages.maint_refused("busy", "inverter-b")})
|
|
check("stop on a battery that is not cycling: 409 with the sentence",
|
|
rs[2][0] == 409 and rs[2][1]["code"] == "idle"
|
|
and rs[2][1]["error"] == messages.maint_refused("idle", "inverter-a"))
|
|
check("unknown unit: 400 with the sentence", rs[3][0] == 400
|
|
and rs[3][1]["code"] == "unit"
|
|
and rs[3][1]["error"] == messages.maint_refused("unit"))
|
|
check("stop on the cycling battery: 200", rs[4] == (
|
|
200, {"ok": True, "result": {"unit": "inverter-b", "phase": IDLE}}))
|
|
check("a null unit is the bare call: 200 and the old answer",
|
|
rs[5] == (200, {"ok": True, "result": {"phase": DRAIN}}))
|
|
check("a request with no unit: 200 and the old answer", rs[6] == (
|
|
200, {"ok": True, "result": {"phase": IDLE}}))
|
|
check("an empty unit is refused (400), not taken as the bare call",
|
|
rs[7][0] == 400 and rs[7][1]["code"] == "unit"
|
|
and phases(c) == [IDLE, IDLE])
|
|
|
|
# --- the words --------------------------------------------------------------
|
|
print("\nthe words")
|
|
words = {}
|
|
for lang in ("en", "nl"):
|
|
with open(os.path.join(HERE, "app", "ui_text", f"{lang}.json"),
|
|
encoding="utf-8") as f:
|
|
words[lang] = json.load(f)["home"]["batteries"]["cycle"]
|
|
for code, sentence in messages.MAINT_REFUSED.items():
|
|
en = words["en"][code]["description"]
|
|
check(f"'{code}': the page's English is the API's sentence",
|
|
en == sentence.replace("{unit}", "{name}"))
|
|
nl = words["nl"][code]["description"]
|
|
check(f"'{code}': the Dutch has the same placeholders",
|
|
sorted(re.findall(r"\{\w+\}", nl))
|
|
== sorted(re.findall(r"\{\w+\}", en)) and nl != en)
|
|
check("both languages carry the same keys",
|
|
json.dumps(words["en"], sort_keys=True).count('":')
|
|
== json.dumps(words["nl"], sort_keys=True).count('":')
|
|
and set(words["en"]) == set(words["nl"]))
|
|
check("every code the server can answer has a page sentence",
|
|
set(messages.MAINT_REFUSED) <= set(words["en"]))
|
|
|
|
# --- the page ----------------------------------------------------------------
|
|
print("\nthe page")
|
|
JS = open(os.path.join(HERE, "app", "static", "app.js"),
|
|
encoding="utf-8").read()
|
|
check("the buttons are drawn only with two or more batteries",
|
|
"S.batteries.length > 1 ? cycleButtons(b, i) : null" in JS)
|
|
check("both buttons post the battery's name and ask first",
|
|
'JSON.stringify({ action: what, unit: b.name }) });' in JS and 'ask1("start_dialog", "maint_start")' in JS
|
|
and 'ask1("abort_dialog", "maint_abort")' in JS)
|
|
check("a refusal is shown in the page's language, by its code",
|
|
'const own = j.code && t("batteries.cycle." + j.code);' in JS
|
|
and 'toast(own ? say(own.description' in JS)
|
|
check("Stop is disabled on a battery that is not cycling",
|
|
'abort: cycling(b) };' in JS and "disabled: !can.abort" in JS)
|
|
check("Start is disabled while any battery cycles",
|
|
"start: !S.batteries.some(cycling) && " in JS
|
|
and "disabled: !can.start" in JS)
|
|
check("Start is disabled on a switched-off battery (use false, modes on)",
|
|
"&& b.use !== false, abort" in JS)
|
|
check("a battery counts as cycling when its phase is not idle",
|
|
'cycling = (x) => x.maintenance !== "idle"' in JS)
|
|
check("the old header buttons still post a bare request",
|
|
'ask("maint_start", () => simple("maint_start"))' in JS
|
|
and "body: JSON.stringify({ action: what }) });" in JS)
|
|
|
|
# --- a roster of one -----------------------------------------------------------
|
|
print("\none battery: the requests, the log and the wire stay as they were")
|
|
c, hass = controller(1)
|
|
r1 = act(c, "maint_start")
|
|
r2 = act(c, "maint_start")
|
|
r3 = act(c, "maint_abort")
|
|
r4 = act(c, "maint_abort")
|
|
check("answers", [r1, r2, r3, r4] == [{"phase": DRAIN}, {"phase": DRAIN},
|
|
{"phase": IDLE}, {"phase": IDLE}])
|
|
check("log lines (stamp removed), newest first", settle(c) == [
|
|
"maintenance: ABORT (operator)", "maintenance: ABORT (operator)",
|
|
"maintenance: forced start -> drain",
|
|
"maintenance: forced start -> drain"][:len(settle(c))]
|
|
and len(settle(c)) == 4)
|
|
check("wire: two stops, each 0 W", hass.writes == [("number.sp_0", 0.0)] * 2)
|
|
c, hass = controller(1)
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED")
|
|
sys.exit(1)
|
|
print("all passed")
|