adminandClaude Opus 5.5
6862dea52a
- test_price_secret.py: the fixed-tariff note on the real /config/validate route, EN and NL, and none for a spaces-only profile (web.py lang wiring). - test_price_service.py: the start line is INFO, not WARNING; a dismiss that raises does not stop the cycle. - test_maint13/test_setup_only/test_watchdog: ignore only the dismiss of the old card, so a create with that id is still counted. - DECISIONS rule 7 quotes the settings note as it is; messages.py comment. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Bdqr7PiaHcBnzPituu2yTC
575 lines
24 KiB
Python
575 lines
24 KiB
Python
"""MAINT-13: a full pack that stops at 98-99 % is not a failed charge.
|
|
`python3 test_maint13.py`
|
|
|
|
Graded here: the stall rule, the near-full timeout rule, the 7 day back-off
|
|
after a failed cycle (and what ends it), the timeout counting only the minutes
|
|
the price gate was open, restart and old-state-file behaviour, the notification
|
|
(once), and the Controller wiring at one and two packs.
|
|
See DECISIONS/MAINT-13-full-pack-is-complete.md.
|
|
|
|
Mutating what this file grades? Read MUTATION.md first (stale bytecode).
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import tempfile
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
from app import messages
|
|
from app.maintenance import (
|
|
CHARGE, CHARGE_NEAR_FULL_PCT, CHARGE_STALL_MIN, CHARGE_TIMEOUT_MIN,
|
|
CHARGE_WALL_TIMEOUT_MIN, DRAIN, FAIL_BACKOFF_DAYS, HOLD, IDLE,
|
|
MaintConfig, Maintenance, arbitrate,
|
|
)
|
|
from app.main import Controller
|
|
from app.store import (DEFAULTS, MAINT_UNIT_KEYS, Store, migrate_maint_keys,
|
|
unit_key)
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
UTC = timezone.utc
|
|
fails = []
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
# The live site: target 100, shipped hold. Start hour 19 UTC for the schedule.
|
|
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) # B's drain ended (live)
|
|
MIN = timedelta(minutes=1)
|
|
|
|
|
|
class Mem:
|
|
"""Store's surface, no disk; times kept as the real Store keeps them."""
|
|
|
|
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_charge(store=None, cfg=CFG, at=T0):
|
|
"""A pack that has just finished its drain and is charging at `at`."""
|
|
m = Maintenance(cfg, store or Mem())
|
|
m.force_start(at)
|
|
m.tick(at, 11.0) # drain complete -> charge
|
|
assert m.phase == CHARGE
|
|
return m
|
|
|
|
|
|
def run(m, start, minutes, soc, gate=True, step=1):
|
|
"""Tick every `step` minutes for `minutes`; `soc` is a number or f(minute).
|
|
Returns (the last result, the time after the run, the first non-CHARGE result)."""
|
|
t, res, out = 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)
|
|
if out is None and res.phase != CHARGE:
|
|
out = (res, t)
|
|
break
|
|
return res, start + minutes * MIN, out
|
|
|
|
|
|
print("1. the stall rule: near full and no longer taking charge")
|
|
# Live record: 99 % at 04:00, back to 98 % at 04:15, never 100.
|
|
m = in_charge()
|
|
res, end, out = run(m, T0 + 60 * MIN, 400, lambda i: 99.0 if i < 15 else 98.0)
|
|
check("a pack that reaches 99 and falls back to 98 completes (HOLD) after "
|
|
f"{CHARGE_STALL_MIN} min without a new high",
|
|
out is not None and out[0].phase == HOLD)
|
|
check("...at 60 min after the last new high (the 99), not before",
|
|
out is not None and out[1] == T0 + 120 * MIN)
|
|
check("...with one plain log line",
|
|
out is not None and len(out[0].events) == 1
|
|
and "takes no more charge at 98 %" in out[0].events[0])
|
|
m = in_charge()
|
|
res, _, out = run(m, T0, 59, 98.0)
|
|
check("59 minutes flat is not enough", out is None and res.phase == CHARGE)
|
|
m = in_charge()
|
|
res, _, out = run(m, T0, 61, lambda i: 97.0)
|
|
check("target - 3 (97 of 100) counts as near full", out and out[0].phase == HOLD)
|
|
m = in_charge()
|
|
res, _, out = run(m, T0, 300, lambda i: 96.0)
|
|
check("target - 4 (96 of 100) does not complete however long it is flat",
|
|
out is None)
|
|
cfg99 = MaintConfig(enabled=True, start_hour=19, soc_target=99, hold_min=120)
|
|
m = in_charge(cfg=cfg99)
|
|
res, _, out = run(m, T0, 61, lambda i: 96.0)
|
|
check("the shipped target 99 works the same way (96 is near full)",
|
|
out and out[0].phase == HOLD)
|
|
m = in_charge()
|
|
res, _, out = run(m, T0, 400, lambda i: min(99.0, 94.0 + i // 30))
|
|
check("a pack still rising 1 point per 30 min is not stalled: it completes 60 "
|
|
"min after the 99 at minute 150", out and out[1] == T0 + 210 * MIN)
|
|
m = in_charge()
|
|
res, _, out = run(m, T0, 130, lambda i: 97.0 if i < 50 else (98.0 if i < 100 else 98.0))
|
|
check("a new high resets the wait: 97 flat for 50 min, 98 at 50 min -> "
|
|
"completes at 110, not 60", out and out[1] == T0 + 110 * MIN)
|
|
m = in_charge()
|
|
res, _, out = run(m, T0, 300, lambda i: 97.0, gate=False)
|
|
check("minutes with the gate closed do not count as 'no more charge'",
|
|
out is None)
|
|
m = in_charge()
|
|
_, t, _ = run(m, T0, 30, 98.0)
|
|
_, t, _ = run(m, t, 120, 98.0, gate=False)
|
|
res, _, out = run(m, t, 31, 98.0)
|
|
check("30 open + 120 closed + 31 open minutes complete (the closed ones "
|
|
"pause the count)", out and out[0].phase == HOLD)
|
|
|
|
print("2. SoC at the target completes at once")
|
|
m = in_charge()
|
|
r = m.tick(T0 + MIN, 100.0)
|
|
check("100 of 100 -> HOLD on that tick", r.phase == HOLD)
|
|
check("...with the old log line", "charge complete at 100 %" in r.events[0])
|
|
m = in_charge(cfg=cfg99)
|
|
check("99 of 99 -> HOLD", m.tick(T0 + MIN, 99.0).phase == HOLD)
|
|
|
|
print("3. a charge that runs out of time")
|
|
m = in_charge()
|
|
res, _, out = run(m, T0, CHARGE_TIMEOUT_MIN + 10,
|
|
lambda i: 60.0 + i / 55.0, step=1)
|
|
check("still rising 1 point per 55 min (never 60 flat) and near full when the "
|
|
"time is up -> counts as complete", out and out[0].phase == HOLD)
|
|
check("...with one plain line saying so",
|
|
out and len(out[0].events) == 1 and "charge time is up" in out[0].events[0]
|
|
and "counts as done" in out[0].events[0])
|
|
check("...and no failure is recorded", m.store.get_time("last_failed") is None)
|
|
m = in_charge()
|
|
res, _, out = run(m, T0, CHARGE_TIMEOUT_MIN + 10, 70.0)
|
|
check("a pack stuck at 70 % fails when the time is up",
|
|
out and out[0].phase == IDLE and out[0].failed == "charge")
|
|
check("...once: the failure is on one tick, and the pack is idle after",
|
|
m.phase == IDLE and m.tick(out[1] + MIN, 70.0).failed == "")
|
|
check("...with the old technical line and one plain line",
|
|
len(out[0].events) == 2 and "ABORT" in out[0].events[0]
|
|
and "did not finish" in out[0].events[1]
|
|
and f"{FAIL_BACKOFF_DAYS} days" in out[0].events[1])
|
|
check("...and last_failed is set, last_completed is not",
|
|
m.store.get_time("last_failed") == out[1]
|
|
and m.store.get_time("last_completed") is None)
|
|
|
|
print("4. the back-off after a failed cycle")
|
|
fail_at = out[1]
|
|
day = lambda n: fail_at.replace(hour=19, minute=0) + timedelta(days=n)
|
|
check("the next day at the start hour: no cycle (it would begin with a drain)",
|
|
not m.wants_start(day(1)) and m.tick(day(1), 70.0).phase == IDLE)
|
|
check("every day for 6 days: still no cycle",
|
|
all(not m.wants_start(day(n)) for n in range(1, 7)))
|
|
r = [m.tick(day(n), 70.0) for n in range(1, 7)]
|
|
check("...and tick() keeps the pack idle and silent",
|
|
all(x.phase == IDLE and not x.events and x.failed == "" for x in r))
|
|
until = m.backoff_until(day(1))
|
|
check("backoff_until() names the date, 7 days after the failure",
|
|
until == fail_at + timedelta(days=FAIL_BACKOFF_DAYS))
|
|
check("...it holds one minute before and is over at that minute",
|
|
m.backoff_until(until - MIN) == until and m.backoff_until(until) is None)
|
|
check("at 7 days, in the start hour, the pack wants to start again",
|
|
m.backoff_until(until) is None
|
|
and m.wants_start(until.replace(hour=19, minute=5)))
|
|
check("a forced start ignores the back-off",
|
|
m.force_start(day(1)) and m.phase == DRAIN)
|
|
m2 = in_charge()
|
|
run(m2, T0, CHARGE_TIMEOUT_MIN + 10, 70.0)
|
|
m2.force_start(day(2))
|
|
m2.tick(day(2) + MIN, 11.0)
|
|
res, _, out2 = run(m2, day(2) + 2 * MIN, 400, 100.0)
|
|
check("a forced cycle that completes ends the back-off",
|
|
out2 and out2[0].phase == HOLD)
|
|
m2.tick(out2[1] + 130 * MIN, 100.0)
|
|
check("...(the hold completes, last_completed is set)",
|
|
m2.store.get_time("last_completed") is not None
|
|
and m2.backoff_until(day(3)) is None)
|
|
m3 = in_charge()
|
|
m3.store.set_time("last_failed", T0)
|
|
m3.abort(T0 + MIN, "operator")
|
|
check("an operator abort is not a failure and starts no back-off",
|
|
m3.store.get_time("last_failed") == T0
|
|
and in_charge().store.get_time("last_failed") is None)
|
|
m4 = Maintenance(CFG, Mem())
|
|
m4.force_start(T0)
|
|
r = m4.tick(T0 + timedelta(minutes=721), 40.0)
|
|
check("a drain that times out is a failure too, with the same back-off",
|
|
r.failed == "drain" and m4.store.get_time("last_failed") is not None
|
|
and not m4.wants_start(day(1)))
|
|
|
|
print("5. only the minutes the pack was allowed to charge count")
|
|
m = in_charge()
|
|
# 5 min steps: one tick counts at most 5 min, so this is the same count as
|
|
# 1 min steps in a fifth of the ticks.
|
|
res, t, out = run(m, T0, CHARGE_TIMEOUT_MIN * 2, 70.0, gate=False, step=5)
|
|
check("72 h with the gate closed: not timed out", out is None and m.phase == CHARGE)
|
|
res, t, out = run(m, t, CHARGE_TIMEOUT_MIN - 5, 70.0, step=5)
|
|
check("...and 35 h 55 min open on top: still charging", out is None)
|
|
res, t, out = run(m, t, 10, 70.0)
|
|
check("...36 h of open time in all: the charge fails", out and out[0].failed == "charge")
|
|
m = in_charge()
|
|
WALL = T0 + CHARGE_WALL_TIMEOUT_MIN * MIN
|
|
res, t, out = run(m, T0, CHARGE_WALL_TIMEOUT_MIN + 10, 70.0, gate=False, step=5)
|
|
check("a gate that never opens still ends in a failure after 7 days (not before)",
|
|
out and out[0].failed == "charge" and WALL < out[1] <= WALL + 5 * MIN)
|
|
m = in_charge()
|
|
res, t, out = run(m, T0, CHARGE_WALL_TIMEOUT_MIN + 10, 98.0, gate=False, step=5)
|
|
check("...or in a complete charge when the pack is near full",
|
|
out and out[0].phase == HOLD and WALL < out[1] <= WALL + 5 * MIN)
|
|
m = in_charge()
|
|
res, t, out = run(m, T0, 10, 70.0)
|
|
before = m._trk["open_min"]
|
|
res = m.tick(t + 2 * timedelta(hours=1), 70.0)
|
|
check("a gap in the ticks (add-on down) counts for exactly 5 minutes (the cap)",
|
|
abs(m._trk["open_min"] - before - 5.0) < 1e-9)
|
|
m.tick(t + 2 * timedelta(hours=1) + 2 * MIN, 70.0)
|
|
check("...and a 2 minute step counts 2", abs(m._trk["open_min"] - before - 7.0) < 1e-9)
|
|
_t = t + 2 * timedelta(hours=1) + 2 * MIN
|
|
m.tick(_t - 10 * MIN, 70.0) # the clock steps back 10 min
|
|
check("...and a step back in time counts nothing (not minus)",
|
|
abs(m._trk["open_min"] - before - 7.0) < 1e-9)
|
|
|
|
print("6. restarts and old state files")
|
|
with tempfile.TemporaryDirectory() as d:
|
|
path = os.path.join(d, "state.json")
|
|
st = Store(path)
|
|
m = Maintenance(CFG, st)
|
|
m.force_start(T0)
|
|
m.tick(T0, 11.0)
|
|
_, t, out = run(m, T0, 60, 70.0, gate=False) # 60 min waiting
|
|
_, t, out = run(m, t, 40, 70.0) # 40 min charging
|
|
open_before = m._trk["open_min"]
|
|
# a fresh process on the same file
|
|
m_b = Maintenance(CFG, Store(path))
|
|
r = m_b.tick(t + MIN, 70.0)
|
|
check("a restart mid-charge keeps the open minutes (40 of 100 min; within "
|
|
"the 5 min save interval), not the 100 wall-clock minutes",
|
|
m_b.phase == CHARGE
|
|
and open_before - 6 <= m_b._trk["open_min"] <= open_before + 2
|
|
and m_b._trk["open_min"] < 60)
|
|
_, t2, out = run(m_b, t + 2 * MIN, CHARGE_TIMEOUT_MIN - 30, 70.0, step=5)
|
|
check("...so the timeout comes after 36 h in all, not 36 h from the restart",
|
|
out is not None and out[0].failed == "charge")
|
|
m = Maintenance(CFG, Store(path))
|
|
m.store.set_time("last_failed", T0)
|
|
m_c = Maintenance(CFG, Store(path))
|
|
check("a restart mid back-off keeps it (last_failed is in the file)",
|
|
m_c.backoff_until(T0 + timedelta(days=3)) == T0 + timedelta(days=7)
|
|
and not m_c.wants_start((T0 + timedelta(days=3)).replace(hour=19)))
|
|
check("...and the state file is plain JSON",
|
|
isinstance(json.load(open(path)).get("last_failed"), str))
|
|
old = {"phase": "charge",
|
|
"phase_started": (T0 - timedelta(hours=35)).isoformat(),
|
|
"last_completed": None, "last_start_attempt": None,
|
|
"auto": True}
|
|
path2 = os.path.join(d, "old.json")
|
|
json.dump(old, open(path2, "w"))
|
|
mo = Maintenance(CFG, Store(path2))
|
|
check("a state file from before this change loads with no failure and no "
|
|
"charge record", mo.backoff_until(T0) is None
|
|
and mo.store.get("charge_track") is None and mo.phase == CHARGE)
|
|
r = mo.tick(T0, 70.0)
|
|
check("...a charge in progress counts the hours it already ran (35 h)",
|
|
r.phase == CHARGE and 34 * 60 < mo._trk["open_min"] < 36 * 60)
|
|
res, _, out = run(mo, T0 + MIN, 120, 70.0)
|
|
check("...so it fails when the 36 h are up, as before",
|
|
out and out[0].failed == "charge")
|
|
# A record left by another charge (an add-on rolled back and forward
|
|
# again does not clear it) must not be read as this charge's.
|
|
stale = {"phase": "charge",
|
|
"phase_started": (T0 - timedelta(hours=2)).isoformat(),
|
|
"charge_track": {"since": (T0 - timedelta(days=30)).isoformat(),
|
|
"open_min": 2000.0, "flat_min": 59.0,
|
|
"best": 99.0, "at": None}}
|
|
ms = Maintenance(CFG, Mem(stale))
|
|
r = ms.tick(T0, 98.0)
|
|
check("a charge record from another charge is ignored: this charge counts "
|
|
"its own 2 h and its own best",
|
|
r.phase == CHARGE and 119 < ms._trk["open_min"] < 121
|
|
and ms._trk["flat_min"] == 0.0 and ms._trk["best"] == 98.0)
|
|
mi = Maintenance(CFG, Store(os.path.join(d, "none.json")))
|
|
check("no state file: due, no back-off", mi.due(T0) and mi.backoff_until(T0) is None)
|
|
|
|
print("7. two packs (N=2)")
|
|
a, b = Maintenance(CFG, Mem(), unit="inverter-a", name="inverter-a"), \
|
|
Maintenance(CFG, Mem(), unit="inverter-b", name="inverter-b")
|
|
b.force_start(T0)
|
|
b.tick(T0, 11.0)
|
|
_, t, out = run(b, T0, CHARGE_TIMEOUT_MIN + 10, 70.0)
|
|
slot = datetime(2026, 10, 7, 19, 5, tzinfo=UTC)
|
|
check("B failed: B is backed off, A (due) is not",
|
|
out and out[0].failed == "charge" and not b.wants_start(slot)
|
|
and a.wants_start(slot))
|
|
check("...so A takes the slot, and B is not named as waiting for anyone",
|
|
arbitrate([a, b], slot) == ["", ""])
|
|
a.force_start(slot)
|
|
check("...and while A cycles, B is held as before",
|
|
"inverter-a" in arbitrate([a, b], slot + 3 * MIN)[1])
|
|
check("B's failure is under B's own keys, not A's",
|
|
b.store.get_time("last_failed") is not None
|
|
and a.store.get_time("last_failed") is None)
|
|
mst = Mem({"last_failed": T0.isoformat(), "phase": "idle"})
|
|
migrate_maint_keys(mst, "inverter-a", ["inverter-a", "inverter-b"])
|
|
check("going from one pack to two moves the back-off to the first pack",
|
|
mst.data.get(unit_key("inverter-a", "last_failed")) == T0.isoformat()
|
|
and "last_failed" in MAINT_UNIT_KEYS)
|
|
|
|
|
|
print("8. the Controller: gate, notification, status row")
|
|
|
|
|
|
class Hass:
|
|
def __init__(self, numbers, limits):
|
|
self.numbers, self.limits_by, self.calls = dict(numbers), dict(limits), []
|
|
|
|
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):
|
|
return True
|
|
|
|
async def call(self, domain, service, data):
|
|
self.calls.append((domain, service, data))
|
|
return True
|
|
|
|
|
|
class Pub:
|
|
enabled = False
|
|
|
|
def publish(self, *a, **k):
|
|
pass
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
OPTS1 = dict(setpoint_entity="number.sp_a", soc_entity="sensor.soc_a",
|
|
batt_entity="sensor.batt_a", meter_source="off",
|
|
meter_entity="sensor.grid", max_w=5000.0, 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, maintenance_enabled=True,
|
|
maintenance_soc_target=100,
|
|
price_now_entity="sensor.p_now", price_avg_entity="sensor.p_avg")
|
|
OPTS2 = {k: v for k, v in OPTS1.items()
|
|
if k not in ("setpoint_entity", "soc_entity", "batt_entity")}
|
|
OPTS2["inverters"] = [
|
|
dict(name="inverter-a", setpoint_entity="number.sp_a",
|
|
soc_entity="sensor.soc_a", batt_entity="sensor.batt_a", max_w=5000.0),
|
|
dict(name="inverter-b", setpoint_entity="number.sp_b",
|
|
soc_entity="sensor.soc_b", batt_entity="sensor.batt_b", max_w=3000.0)]
|
|
|
|
|
|
def controller(opts, cheap, soc_a=70.0, soc_b=70.0):
|
|
p_now, p_avg = (0.05, 0.10) if cheap else (0.20, 0.10)
|
|
hass = Hass({"sensor.grid": 800.0, "sensor.soc_a": soc_a,
|
|
"sensor.batt_a": 0.0, "sensor.soc_b": soc_b,
|
|
"sensor.batt_b": 0.0, "sensor.p_now": p_now,
|
|
"sensor.p_avg": p_avg},
|
|
{"number.sp_a": (-5000.0, 5000.0, 1.0),
|
|
"number.sp_b": (-3000.0, 3000.0, 1.0)})
|
|
c = Controller(dict(opts), hass, Mem(), 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 one_cycle(c):
|
|
async def go():
|
|
await c.read_inputs()
|
|
await c.cycle()
|
|
asyncio.run(go())
|
|
|
|
|
|
def put_in_charge(c, which, hours_ago, open_min):
|
|
now = datetime.now(UTC)
|
|
m = c.maints[which]
|
|
m.store.set("phase", CHARGE)
|
|
m.store.set_time("phase_started", now - timedelta(hours=hours_ago))
|
|
m.store.set("charge_track", {
|
|
"since": m.store.get("phase_started"), "open_min": open_min,
|
|
"flat_min": 0.0, "best": 70.0,
|
|
"at": (now - timedelta(minutes=3)).isoformat()})
|
|
return m
|
|
|
|
|
|
def notes(h):
|
|
# Not the dismiss of PRICE-01's old fixed-tariff card, which every start sends.
|
|
# Only that dismiss: a `create` with the old id would still be counted.
|
|
return [x for x in h.calls if x[0] == "persistent_notification"
|
|
and not (x[1] == "dismiss"
|
|
and x[2].get("notification_id") == "goodwe_fixed_tariff_unused")]
|
|
|
|
|
|
for cheap in (True, False):
|
|
c, h = controller(OPTS1, cheap)
|
|
m = put_in_charge(c, 0, 5, 100.0)
|
|
one_cycle(c)
|
|
got = m._trk["open_min"] - 100.0
|
|
check(f"N=1, price gate {'open' if cheap else 'closed'}: the Controller "
|
|
f"{'adds' if cheap else 'does not add'} the 3 minutes since the last "
|
|
f"tick to the charging time (+{got:.1f})",
|
|
(2.5 < got < 3.5) if cheap else abs(got) < 0.01)
|
|
|
|
c, h = controller(OPTS1, True)
|
|
m = put_in_charge(c, 0, 40, CHARGE_TIMEOUT_MIN + 1)
|
|
one_cycle(c)
|
|
n = notes(h)
|
|
check("N=1: a charge past its time at 70 % ends in IDLE with ONE notification",
|
|
m.phase == IDLE and len(n) == 1
|
|
and n[0][2]["notification_id"] == "goodwe_maint_failed_battery"
|
|
and n[0][2]["title"] == "Battery cycle did not finish")
|
|
one_cycle(c)
|
|
one_cycle(c)
|
|
check("...and later cycles send no second one, and start no drain",
|
|
len(notes(h)) == 1 and m.phase == IDLE)
|
|
check("...the status page says when it waits until",
|
|
any(k == "Maintenance paused" and "Waits until" in v
|
|
for k, v in c.status_payload()["rows"]))
|
|
c0, _h0 = controller(OPTS1, True)
|
|
check("...and has no such row when nothing failed",
|
|
not any(k == "Maintenance paused" for k, v in c0.status_payload()["rows"]))
|
|
|
|
c, h = controller(OPTS2, True)
|
|
m = put_in_charge(c, 1, 40, CHARGE_TIMEOUT_MIN + 1)
|
|
one_cycle(c)
|
|
n = notes(h)
|
|
check("N=2: pack B fails, B alone is told, by name, under its own id",
|
|
c.maints[1].phase == IDLE and c.maints[0].phase == IDLE and len(n) == 1
|
|
and n[0][2]["notification_id"] == "goodwe_maint_failed_inverter_b"
|
|
and "inverter-b" in n[0][2]["message"])
|
|
check("...and its failure is on B's keys",
|
|
c.maints[1].store.get_time("last_failed") is not None
|
|
and c.maints[0].store.get_time("last_failed") is None)
|
|
|
|
# a site with no price entities and no fixed tariff: nothing holds the charge back
|
|
NOTARIFF = {k: v for k, v in OPTS1.items()
|
|
if k not in ("price_now_entity", "price_avg_entity")}
|
|
c, h = controller(NOTARIFF, False)
|
|
m = put_in_charge(c, 0, 5, 100.0)
|
|
one_cycle(c)
|
|
got = m._trk["open_min"] - 100.0
|
|
check("N=1, no tariff and no price entities: the gate counts as open, so the "
|
|
f"3 minutes count (+{got:.1f}) although `cheap` is false",
|
|
not c.cheap and 2.5 < got < 3.5)
|
|
c, h = controller(NOTARIFF, False)
|
|
m = put_in_charge(c, 0, 40, CHARGE_TIMEOUT_MIN + 1)
|
|
one_cycle(c)
|
|
check("...and a pack past its charging time there fails at once (not 7 days)",
|
|
m.phase == IDLE and len(notes(h)) == 1)
|
|
# a site with a price provider (TARIFF-11) and no fixed tariff: the provider's
|
|
# plan reaches only the Tariff row, never `cheap`, so no cheap window ever
|
|
# opens a grid top-up there. Like the site above, nothing gate-like holds the
|
|
# charge back, and the time counts.
|
|
PROVIDER = dict(NOTARIFF, price_provider="generic_dayahead", static_profile="")
|
|
c, h = controller(PROVIDER, False)
|
|
m = put_in_charge(c, 0, 5, 100.0)
|
|
one_cycle(c)
|
|
got = m._trk["open_min"] - 100.0
|
|
check("N=1, price provider and no fixed tariff: not a gate (`cheap` is never "
|
|
f"set from the provider), so the 3 minutes count (+{got:.1f})",
|
|
c._price_gated is False and not c.cheap
|
|
and c._charge_gate_open(c.inverters[0]) is True and 2.5 < got < 3.5)
|
|
def read(c):
|
|
asyncio.run(c.read_inputs())
|
|
|
|
|
|
c, h = controller(OPTS1, False)
|
|
read(c)
|
|
c.inverters[0].batt = -800.0
|
|
check("price gate closed but the pack is taking 800 W (sun): gate open",
|
|
c._charge_gate_open(c.inverters[0]) is True)
|
|
c.inverters[0].batt = -10.0
|
|
check("...closed and about idle: gate closed",
|
|
c._charge_gate_open(c.inverters[0]) is False)
|
|
c.inverters[0].batt = None
|
|
check("...no battery reading: gate closed", c._charge_gate_open(c.inverters[0]) is False)
|
|
c, h = controller(OPTS1, True)
|
|
read(c)
|
|
check("price gate open: gate open", c._charge_gate_open(c.inverters[0]) is True)
|
|
PROFILE = json.dumps({"periods": [
|
|
{"name": "day", "from": "07:00", "to": "22:00", "days": "all",
|
|
"eur_per_kwh": 0.29},
|
|
{"name": "night", "from": "22:00", "to": "07:00", "days": "all",
|
|
"eur_per_kwh": 0.19}]})
|
|
c, h = controller(dict(NOTARIFF, price_provider="off", static_profile=PROFILE),
|
|
False)
|
|
read(c)
|
|
check("a fixed tariff set up counts as a gate: outside its cheap hours a pack "
|
|
"that is not charging is closed, inside them it is open",
|
|
c._price_gated is True
|
|
and c._charge_gate_open(c.inverters[0]) is c.cheap)
|
|
import app.main as main_mod # noqa: E402
|
|
_real_gated, _calls = main_mod.static_gated, []
|
|
main_mod.static_gated = lambda o, now: _calls.append(now) or _real_gated(o, now)
|
|
try:
|
|
c._gated_key = None
|
|
read(c)
|
|
read(c)
|
|
read(c)
|
|
c._static_gated(c.o, datetime.now(UTC) + timedelta(hours=1))
|
|
finally:
|
|
main_mod.static_gated = _real_gated
|
|
check("...worked out once an hour, not on every tick (3 reads in one hour = 1 "
|
|
f"build; a new hour = 1 more): {len(_calls)} builds",
|
|
len(_calls) == 2 and c._price_gated is True)
|
|
|
|
# the save timer is reset on a phase change
|
|
mm = in_charge()
|
|
mm.tick(T0 + MIN, 70.0)
|
|
saved_before = mm._trk_saved
|
|
mm.abort(T0 + 2 * MIN, "operator")
|
|
check("leaving CHARGE resets the save timer along with the record",
|
|
saved_before is not None and mm._trk is None and mm._trk_saved is None)
|
|
|
|
print("9. the words")
|
|
en, nl = messages.maint_failed("charge", 7), messages.maint_failed("drain", 7)
|
|
for n_ in (en, nl):
|
|
print(" " + n_.title + " | " + n_.body + " | " + n_.body_nl)
|
|
check("EN charge text", en.body == "The battery did not fill up, so its cycle "
|
|
"stopped and tries again in 7 days.")
|
|
check("EN drain text", nl.body == "The battery did not empty, so its cycle "
|
|
"stopped and tries again in 7 days.")
|
|
check("NL texts", en.body_nl == "Batterijcyclus niet afgerond. De batterij "
|
|
"raakte niet vol, dus de cyclus is gestopt en probeert het over 7 dagen "
|
|
"opnieuw." and nl.body_nl.endswith("raakte niet leeg, dus de cyclus is "
|
|
"gestopt en probeert het over 7 dagen opnieuw."))
|
|
check("the notification carries EN, NL and the Details line, in that order",
|
|
re.fullmatch(r"The battery .*\n\nBatterijcyclus .*\n\nDetails: .*",
|
|
en.data()["message"], re.S) is not None)
|
|
check("the DOCS section and table row exist",
|
|
"### Battery cycle did not finish" in (HERE / "DOCS.md").read_text()
|
|
and "goodwe_maint_failed_<name>" in (HERE / "DOCS.md").read_text())
|
|
check("no new option: the numbers are constants",
|
|
(CHARGE_NEAR_FULL_PCT, CHARGE_STALL_MIN, FAIL_BACKOFF_DAYS)
|
|
== (3, 60, 7))
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED")
|
|
for f in fails:
|
|
print(" -", f)
|
|
sys.exit(1)
|
|
print("all passed")
|