glenn schrooyenandClaude
c015230452
Owner decision 2026-11-07: clients see nothing new until SMA-4. config.yaml, translations, labels and the page layout are back to 23ff9d0's. The validation of the keys stays for callers that skip the schema. Co-Authored-By: Claude <noreply@anthropic.com>
751 lines
33 KiB
Python
751 lines
33 KiB
Python
"""SMA-2: the inverter driver seam. `python3 test_sma2_driver.py`
|
|
|
|
Part 1 grades the GoodWe driver against a Home Assistant that logs every call:
|
|
it must make exactly the calls the code it replaced made, in the same order.
|
|
Part 2 grades what `Inverter` reads from its driver. Part 3 drives a REAL
|
|
Controller with a roster of one fake-driver unit and one GoodWe unit and
|
|
checks that it writes, reads, limits, grades, drops and publishes each by its
|
|
own driver, with a Home Assistant that records every entity it is asked about
|
|
so a call site that goes round the seam is seen. Part 4 is the SAFETY-28
|
|
census and rule 9 on the new keys. Part 5 is the schema.
|
|
|
|
Mutating a module this file grades? Read MUTATION.md first (purge __pycache__).
|
|
The seam mutants this file was run against are listed at the end.
|
|
"""
|
|
|
|
import asyncio
|
|
import io
|
|
import sys
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
from app import drivers
|
|
from app.drivers import (DEFAULT_DRIVER, REGISTRY, SELECTABLE, Driver,
|
|
DriverNotBound, GoodWeEsphomeDriver, Probe,
|
|
UnavailableDriver, build_driver, census_key)
|
|
from app.freshness import BOUNDS, Bound
|
|
from app.inverter import LINK_STALE_S, Inverter, build_roster
|
|
from app.main import Controller
|
|
from app.store import DEFAULTS
|
|
from app.validate import (actuator_writers, validate_startup_config,
|
|
validate_unit_ranges)
|
|
|
|
fails = []
|
|
HERE = Path(__file__).resolve().parent
|
|
|
|
|
|
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, *a):
|
|
return None
|
|
|
|
def set_time(self, *a):
|
|
pass
|
|
|
|
|
|
class Pub:
|
|
"""Keeps the last published values, like test_mqtt.py's."""
|
|
enabled = False
|
|
|
|
def __init__(self):
|
|
self.last = {}
|
|
|
|
def publish(self, values):
|
|
self.last = dict(values)
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
class LogHass:
|
|
"""Home Assistant for the GoodWe unit. EVERY call is logged with its entity,
|
|
so a path that talks to Home Assistant about anything else is on record."""
|
|
|
|
def __init__(self, numbers=None, limits=None, accept=True):
|
|
self.numbers = dict(numbers or {})
|
|
self.limits_by = dict(limits or {})
|
|
self.accept = accept
|
|
self.log = [] # (method, entity, extra)
|
|
|
|
async def number(self, entity, invert=False):
|
|
self.log.append(("number", entity, invert))
|
|
v = self.numbers.get(entity)
|
|
return None if v is None else (-v if invert else v)
|
|
|
|
async def limits(self, entity):
|
|
self.log.append(("limits", entity, None))
|
|
return self.limits_by.get(entity, (None, None, None))
|
|
|
|
async def set_number(self, entity, value):
|
|
self.log.append(("set_number", entity, value))
|
|
if self.accept:
|
|
self.numbers[entity.replace("number.sp", "sensor.batt")] = value
|
|
return self.accept
|
|
|
|
async def state(self, entity):
|
|
self.log.append(("state", entity, None))
|
|
return None
|
|
|
|
async def call(self, *a):
|
|
self.log.append(("call", repr(a), None))
|
|
return True
|
|
|
|
def entities(self):
|
|
"""Every entity name Home Assistant was asked about (service calls
|
|
like notifications name none)."""
|
|
return {e for m, e, _x in self.log if m != "call"}
|
|
|
|
|
|
def run(coro):
|
|
return asyncio.run(coro)
|
|
|
|
|
|
def raises(exc, make):
|
|
"""Does `run(make())` raise `exc`? `make` builds the coroutine."""
|
|
try:
|
|
run(make())
|
|
except exc:
|
|
return True
|
|
except Exception:
|
|
return False
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# a fake driver: a unit reached by host:port:unit, no Home Assistant at all
|
|
# ---------------------------------------------------------------------------
|
|
FAKE_BOUND = Bound(signal="fake", unchanged_s=11.0, move_w=100.0,
|
|
release_s=5.0, wired=True, derivation="test")
|
|
|
|
|
|
class FakeDriver(Driver):
|
|
name = "fake"
|
|
writer_key = "host"
|
|
brand = "Fake"
|
|
link_stale_s = 7.0
|
|
batt_bound = FAKE_BOUND
|
|
settle_s = 3.0
|
|
words = {"idle": "handed back"}
|
|
|
|
def __init__(self, host="", port=502, unit_id=1, rating=1000.0, soc=60.0):
|
|
self.host, self.port, self.unit_id = host, port, unit_id
|
|
self.rating, self.soc, self.power = rating, soc, 0.0
|
|
self.accept = True
|
|
self.calls = [] # (what, arg)
|
|
|
|
@classmethod
|
|
def from_entry(cls, entry):
|
|
return cls(entry.get("host", ""), entry.get("port", 502),
|
|
entry.get("unit_id", 1), rating=entry.get("_rating", 1000.0))
|
|
|
|
@classmethod
|
|
def actuator_id_of(cls, entry):
|
|
host = str(entry.get("host") or "").strip()
|
|
return (f"{host}:{entry.get('port', 502)}:{entry.get('unit_id', 1)}"
|
|
if host else "")
|
|
|
|
def configured(self):
|
|
return bool(self.host)
|
|
|
|
def actuator_id(self):
|
|
return f"{self.host}:{self.port}:{self.unit_id}"
|
|
|
|
def soc_source(self):
|
|
return "fake-soc"
|
|
|
|
def batt_source(self):
|
|
return "fake-power"
|
|
|
|
async def limits(self):
|
|
self.calls.append(("limits", None))
|
|
return -self.rating, self.rating
|
|
|
|
async def read(self):
|
|
self.calls.append(("read", None))
|
|
return self.soc, self.power, {"state": "ok"}
|
|
|
|
async def write(self, w):
|
|
self.calls.append(("write", w))
|
|
if self.accept:
|
|
self.power = w # the battery follows its command
|
|
return self.accept
|
|
|
|
async def release(self):
|
|
self.calls.append(("release", None))
|
|
return True
|
|
|
|
async def probe(self):
|
|
self.calls.append(("probe", None))
|
|
return Probe(-self.rating, self.rating, self.soc, self.power, {})
|
|
|
|
|
|
def calls(drv, what):
|
|
return [a for w, a in drv.calls if w == what]
|
|
|
|
|
|
print("SMA-2 - the inverter driver seam")
|
|
print()
|
|
# ==============================================================================
|
|
print("1. the GoodWe driver makes the calls the code it replaced made")
|
|
|
|
GW = dict(setpoint_entity="number.sp_a", soc_entity="sensor.soc_a",
|
|
batt_entity="sensor.batt_a")
|
|
h1 = LogHass({"sensor.soc_a": 55.0, "sensor.batt_a": 700.0},
|
|
{"number.sp_a": (-5000.0, 4000.0, 1.0)})
|
|
d1 = GoodWeEsphomeDriver(**GW).bind(h1)
|
|
check("a driver built from an entry names its three entities and is configured",
|
|
GoodWeEsphomeDriver.from_entry(dict(GW)).setpoint_entity == "number.sp_a"
|
|
and d1.configured() and d1.actuator_id() == "number.sp_a"
|
|
and (d1.soc_source(), d1.batt_source())
|
|
== ("sensor.soc_a", "sensor.batt_a"))
|
|
check("...and one with no setpoint entity is not",
|
|
not GoodWeEsphomeDriver().configured()
|
|
and not GoodWeEsphomeDriver(soc_entity="sensor.soc_a").configured())
|
|
check("limits() is hass.limits(setpoint) and hands back (min, max)",
|
|
run(d1.limits()) == (-5000.0, 4000.0)
|
|
and h1.log == [("limits", "number.sp_a", None)])
|
|
h1.log.clear()
|
|
check("read() asks for the SoC first, then the power, with the invert flag, "
|
|
"and nothing else",
|
|
run(d1.read()) == (55.0, 700.0, {})
|
|
and h1.log == [("number", "sensor.soc_a", False),
|
|
("number", "sensor.batt_a", False)])
|
|
h1.log.clear()
|
|
d1i = GoodWeEsphomeDriver(batt_invert=True, **GW).bind(h1)
|
|
check("...and a reversed power sensor is reversed by hass.number, as before",
|
|
run(d1i.read()) == (55.0, -700.0, {})
|
|
and h1.log[1] == ("number", "sensor.batt_a", True))
|
|
h1.log.clear()
|
|
check("write(w) is hass.set_number(setpoint, w) and says whether it took it",
|
|
run(d1.write(-250.0)) is True and h1.log == [("set_number", "number.sp_a",
|
|
-250.0)])
|
|
h1.log.clear()
|
|
h1.accept = False
|
|
check("...and False when Home Assistant refuses",
|
|
run(d1.write(10.0)) is False)
|
|
h1.accept = True
|
|
h1.log.clear()
|
|
check("release() is write(0): GoodWe is handed back by its board's own watchdog",
|
|
run(d1.release()) is True
|
|
and h1.log == [("set_number", "number.sp_a", 0.0)])
|
|
h1.log.clear()
|
|
h1.numbers.update({"sensor.soc_a": 55.0, "sensor.batt_a": 700.0})
|
|
pr = run(d1.probe())
|
|
check("probe() reads the range, the SoC and the power and WRITES NOTHING",
|
|
pr == Probe(-5000.0, 4000.0, 55.0, 700.0, {})
|
|
and all(m != "set_number" for m, _e, _x in h1.log))
|
|
check("a driver with no Home Assistant is loud, not a silent 'no answer'",
|
|
raises(DriverNotBound, GoodWeEsphomeDriver(**GW).limits)
|
|
and raises(DriverNotBound, GoodWeEsphomeDriver(**GW).read)
|
|
and raises(DriverNotBound, lambda: GoodWeEsphomeDriver(**GW).write(1.0))
|
|
and not GoodWeEsphomeDriver(**GW).bound())
|
|
_swap = {"h": h1}
|
|
d1g = GoodWeEsphomeDriver(**GW).bind_getter(lambda: _swap["h"])
|
|
h1b = LogHass({"sensor.soc_a": 1.0, "sensor.batt_a": 2.0})
|
|
_swap["h"] = h1b
|
|
check("bind_getter() looks Home Assistant up at each use, so a Controller whose "
|
|
".hass is replaced afterwards is followed",
|
|
d1g.bound() and run(d1g.read()) == (1.0, 2.0, {}) and h1b.log)
|
|
check("its numbers are the ones the code it replaced used",
|
|
(d1.link_stale_s, d1.batt_bound, d1.brand)
|
|
== (LINK_STALE_S, BOUNDS["batt"], "GoodWe") and LINK_STALE_S == 30.0
|
|
and d1.settle_s == 18.0 and d1.words == {})
|
|
|
|
# ==============================================================================
|
|
print()
|
|
print("2. what Inverter reads from its driver")
|
|
|
|
NOW = datetime(2026, 11, 7, 12, 0, tzinfo=timezone.utc)
|
|
fk = FakeDriver(host="10.9.9.9", unit_id=3)
|
|
inv_f = Inverter("fake-a", driver=fk, max_w=2000.0)
|
|
inv_g = Inverter("inverter-b", max_w=3000.0, **{
|
|
"setpoint_entity": "number.sp_b", "soc_entity": "sensor.soc_b",
|
|
"batt_entity": "sensor.batt_b"})
|
|
check("an Inverter built from the old keywords gets the GoodWe driver, with "
|
|
"the same entities on the same attributes",
|
|
isinstance(inv_g.driver, GoodWeEsphomeDriver)
|
|
and (inv_g.setpoint_entity, inv_g.soc_entity, inv_g.batt_entity,
|
|
inv_g.batt_invert) == ("number.sp_b", "sensor.soc_b",
|
|
"sensor.batt_b", False))
|
|
check("...and one given another driver has none of those entities",
|
|
(inv_f.setpoint_entity, inv_f.soc_entity, inv_f.batt_entity) == ("", "", ""))
|
|
check("has_writer() is the driver's configured()",
|
|
inv_f.has_writer() and inv_g.has_writer()
|
|
and not Inverter("x", driver=FakeDriver(host="")).has_writer()
|
|
and not Inverter("y").has_writer())
|
|
check("the freshness gate is built on the DRIVER's bound",
|
|
inv_f.batt_fresh.bound is FAKE_BOUND
|
|
and inv_g.batt_fresh.bound is BOUNDS["batt"])
|
|
for u in (inv_f, inv_g):
|
|
u.note_write(False, NOW - timedelta(seconds=10))
|
|
check("is_live() takes its window from the driver: 10 s of rejected writes "
|
|
"drops a 7 s unit and not a GoodWe one",
|
|
inv_f.is_live(NOW)[0] is False and inv_g.is_live(NOW)[0] is True)
|
|
check("...and an explicit window still wins",
|
|
inv_f.is_live(NOW, stale_s=60.0)[0] is True
|
|
and inv_g.is_live(NOW, stale_s=5.0)[0] is False)
|
|
inv_f5 = Inverter("fake-5", driver=FakeDriver(host="h"), max_w=2000.0)
|
|
inv_f5.note_write(False, NOW - timedelta(seconds=5))
|
|
check("link_note() says the driver's window: 'after 7 seconds' for the fake "
|
|
"unit, 'after 30 seconds' for the GoodWe one, and nothing for a healthy unit",
|
|
"after 7 seconds" in inv_f5.link_note(NOW)
|
|
and "after 30 seconds" in inv_g.link_note(NOW)
|
|
and "cannot be controlled" in inv_f.link_note(NOW)
|
|
and Inverter("z", driver=FakeDriver(host="h")).link_note(NOW) == "")
|
|
check("an unconfigured unit gets the no-setpoint row, as before",
|
|
Inverter("x", driver=FakeDriver(host="")).link_note(NOW).startswith("x: ")
|
|
and Inverter("x", driver=FakeDriver(host="")).link_note(NOW)
|
|
== Inverter("x").link_note(NOW) != "")
|
|
check("an entry that names a driver this build lacks is an unconfigured unit, "
|
|
"not a crash and not GoodWe",
|
|
isinstance(build_driver({"driver": "nope", "setpoint_entity": "number.x"}),
|
|
UnavailableDriver)
|
|
and not build_driver({"driver": "nope"}).configured()
|
|
and build_roster({"inverters": [{"name": "n", "driver": "nope",
|
|
"setpoint_entity": "number.x",
|
|
"max_w": 1000}]}, None)[0].has_writer()
|
|
is False)
|
|
|
|
# ==============================================================================
|
|
print()
|
|
print("3. a roster of one fake unit and one GoodWe unit, in a real Controller")
|
|
|
|
REGISTRY["fake"] = FakeDriver
|
|
OPTS = dict(meter_source="off", meter_entity="sensor.grid", max_w=5000,
|
|
gain=0.6, slew_w=0, deadband_w=15, target_grid_w=0, step_w=10,
|
|
heartbeat_s=10, stale_input_s=15, shadow_mode="off",
|
|
inverters=[
|
|
dict(name="fake-a", driver="fake", host="10.9.9.9", port=502,
|
|
unit_id=3, max_w=2000, bms_soc_floor=10),
|
|
dict(name="inverter-b", setpoint_entity="number.sp_b",
|
|
soc_entity="sensor.soc_b", batt_entity="sensor.batt_b",
|
|
max_w=3000, bms_soc_floor=15)])
|
|
B_ENT = {"number.sp_b", "sensor.soc_b", "sensor.batt_b"}
|
|
|
|
|
|
def site(rating_f=1000.0, **over):
|
|
hass = LogHass({"sensor.grid": 3000.0, "sensor.soc_b": 50.0,
|
|
"sensor.batt_b": 0.0},
|
|
{"number.sp_b": (-3000.0, 3000.0, 1.0)})
|
|
pub = Pub()
|
|
c = Controller(dict(OPTS, **over), hass, MemStore(), pub,
|
|
stop=asyncio.Event())
|
|
f = c.inverters[0].driver
|
|
f.rating = rating_f
|
|
c.inverters[0].dev_min, c.inverters[0].dev_max = -rating_f, rating_f
|
|
c.inverters[1].dev_min, c.inverters[1].dev_max = -3000.0, 3000.0
|
|
c.auto = True
|
|
return c, hass, pub, f
|
|
|
|
|
|
async def cycles(c, n=6):
|
|
for _ in range(n):
|
|
await c.read_inputs()
|
|
await c.cycle()
|
|
|
|
|
|
c3, h3, p3, f3 = site()
|
|
check("the roster is built through the registry: a fake unit and a GoodWe unit",
|
|
isinstance(c3.inverters[0].driver, FakeDriver)
|
|
and isinstance(c3.inverters[1].driver, GoodWeEsphomeDriver)
|
|
and f3.actuator_id() == "10.9.9.9:502:3")
|
|
run(cycles(c3))
|
|
fa, fb = c3.inverters
|
|
print(f" after 6 cycles: fake-a wrote {calls(f3, 'write')}, GoodWe wrote "
|
|
f"{[x for m, e, x in h3.log if m == 'set_number']}, shares "
|
|
f"{c3.allocation.shares if c3.allocation else None}")
|
|
check("READ: the unit's SoC and power come from its driver, the GoodWe unit's "
|
|
"from its entities",
|
|
calls(f3, "read") and fa.soc == 60.0 and fb.soc == 50.0
|
|
and ("number", "sensor.soc_b", False) in h3.log)
|
|
check("WRITE: the fake unit's share went to ITS driver and the GoodWe unit's to "
|
|
"ITS entity, each non-zero, the same direction",
|
|
calls(f3, "write") and calls(f3, "write")[-1] > 0
|
|
and [x for m, e, x in h3.log if m == "set_number" and e == "number.sp_b"][-1] > 0)
|
|
check("ALLOCATE: each unit holds a share inside its own rating, and the shares "
|
|
"are the site's (3000 W of grid, fake capped at 1000 W by its driver)",
|
|
c3.allocation is not None
|
|
and set(c3.allocation.shares) == {"fake-a", "inverter-b"}
|
|
and 0 < c3.allocation.shares["fake-a"] <= 1000.0
|
|
and 0 < c3.allocation.shares["inverter-b"] <= 3000.0
|
|
and calls(f3, "write")[-1] == c3.allocation.shares["fake-a"])
|
|
check("GRADE: the fake unit's rating is the driver's range (1000 W), the "
|
|
"GoodWe unit's is its entry's (3000 W), each from its own source",
|
|
(fa.rating_w(), fb.rating_w()) == (1000.0, 3000.0))
|
|
check("the fake's power follows its command, so the freshness gate (the fake's "
|
|
"own bound) lets the reading through",
|
|
fa.batt == calls(f3, "write")[-1] and fa.batt_fresh.bound is FAKE_BOUND)
|
|
run(c3.send_heartbeats(10_000.0, 10.0))
|
|
n_before = len(calls(f3, "write"))
|
|
check("HEARTBEAT: it re-emits the fake unit's standing command through its driver",
|
|
n_before >= 2 and calls(f3, "write")[-1] == calls(f3, "write")[-2])
|
|
run(c3.command_site_zero())
|
|
check("STOP: command_site_zero() reaches the fake unit through its driver, as "
|
|
"0 W (the hand-back is SMA-10's, and release() has no caller yet)",
|
|
calls(f3, "write")[-1] == 0.0 and not calls(f3, "release"))
|
|
c3.publish()
|
|
print(f" published unit rows: "
|
|
f"{sorted(k for k in p3.last if k.startswith('unit/'))}")
|
|
check("PUBLISH: one set of rows per unit, with that unit's own numbers",
|
|
p3.last.get("unit/fake_a/soc") == 60.0
|
|
and p3.last.get("unit/inverter_b/soc") == 50.0
|
|
and "unit/fake_a/battery" in p3.last and "unit/inverter_b/battery" in p3.last)
|
|
rows = c3.checks()
|
|
check("STATUS: the fake unit's sensor rows are not 'unset' (its driver has a "
|
|
"source), and it has the control row",
|
|
not [r for r in rows if r.get("id") == "sensor" and not r["ok"]
|
|
and "fake" in r["text"]])
|
|
# liveness: the fake's 7 s window, the GoodWe unit's 30 s
|
|
c4, h4, p4, f4 = site()
|
|
run(cycles(c4, 2))
|
|
f4.accept = False # the fake's link is down; GoodWe's is fine
|
|
_now = datetime.now(timezone.utc)
|
|
for u in c4.inverters:
|
|
u.note_write(False, _now - timedelta(seconds=10))
|
|
check("LIVENESS: after 10 s of rejected writes the fake unit is left out of the "
|
|
"split and the GoodWe unit stays in it, per its driver's window",
|
|
c4.live_units() == ("inverter-b",))
|
|
run(cycles(c4, 4))
|
|
check("...and the survivor carries the site: the allocation names only it",
|
|
c4.allocation is not None and set(c4.allocation.shares) == {"inverter-b"}
|
|
and c4.allocation.shares["inverter-b"] > 1000.0)
|
|
# the tripwire: nothing in the world was asked about the fake unit
|
|
asked = h3.entities() | h4.entities()
|
|
print(f" Home Assistant was asked about: {sorted(asked)}")
|
|
check("TRIPWIRE: Home Assistant was only ever asked about the GoodWe unit's "
|
|
"entities and the meter - no empty name, no host - on read, write, "
|
|
"heartbeat, stop and limits alike",
|
|
asked <= B_ENT | {"sensor.grid"} and "" not in asked
|
|
and not any("10.9.9.9" in e or "fake" in e for e in asked))
|
|
check("...and the GoodWe unit was never touched through the fake",
|
|
not [1 for w, a in f3.calls + f4.calls
|
|
if isinstance(a, str) and "number.sp_b" in a])
|
|
|
|
# limits and the per-unit range check, through run_control()
|
|
async def drive(c, secs):
|
|
try:
|
|
await asyncio.wait_for(c.run_control(), timeout=secs)
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
|
|
|
|
c5, h5, p5, f5 = site(rating_f=2500.0)
|
|
f5.rating = 2500.0
|
|
c5.inverters[0].dev_min = c5.inverters[0].dev_max = None
|
|
c5.inverters[1].dev_min = c5.inverters[1].dev_max = None
|
|
run(drive(c5, 3.0))
|
|
check("LIMITS: run_control() reads the fake unit's range from its driver and "
|
|
"the GoodWe unit's from its entity",
|
|
calls(f5, "limits")
|
|
and (c5.inverters[0].dev_min, c5.inverters[0].dev_max) == (-2500.0, 2500.0)
|
|
and (c5.inverters[1].dev_min, c5.inverters[1].dev_max) == (-3000.0, 3000.0)
|
|
and [e for m, e, x in h5.log if m == "limits"] == ["number.sp_b"])
|
|
c6, h6, p6, f6 = site()
|
|
f6.rating = 1500.0 # the entry says 2000 W
|
|
c6.inverters[0].dev_min = c6.inverters[0].dev_max = None
|
|
run(drive(c6, 10.0))
|
|
check("RULE 2 PER UNIT: a fake unit rated above what its driver says it accepts "
|
|
"is refused by name, as a GoodWe one is",
|
|
c6.refused and any("fake-a" in p and "10.9.9.9:502:3" in p
|
|
for p in c6.refused)
|
|
and not calls(f6, "write"))
|
|
|
|
# the settings panel has no Controller: it binds each unit's driver itself
|
|
from app import selfconfig # noqa: E402
|
|
hs = LogHass({}, {"number.sp_b": (-3000.0, 3000.0, 1.0)})
|
|
_ps, _unread = run(selfconfig.problems_and_ranges(
|
|
dict(OPTS, inverters=[dict(OPTS["inverters"][0], _rating=1500.0),
|
|
OPTS["inverters"][1]]), hs))
|
|
check("PANEL: the settings panel reads the fake unit's range from its driver "
|
|
"and the GoodWe unit's from its entity, and grades the fake unit by name",
|
|
any("fake-a" in p and "10.9.9.9:502:3 accepts at most 1500" in p
|
|
for p in _ps)
|
|
and [e for m, e, x in hs.log if m == "limits"] == ["number.sp_b"]
|
|
and _unread == [])
|
|
|
|
# the seam is the only way in: a fake-only roster touches Home Assistant for
|
|
# nothing but the meter
|
|
hz = LogHass({"sensor.grid": 800.0})
|
|
cz = Controller(dict(OPTS, inverters=[OPTS["inverters"][0]]), hz, MemStore(),
|
|
Pub(), stop=asyncio.Event())
|
|
cz.inverters[0].dev_min, cz.inverters[0].dev_max = -1000.0, 1000.0
|
|
cz.auto = True
|
|
run(cycles(cz, 3))
|
|
check("A FAKE-ONLY ROSTER never asks Home Assistant for anything but the "
|
|
"meter, and still reads and writes (nothing on the control path is "
|
|
"keyed on a GoodWe entity)",
|
|
hz.entities() <= {"sensor.grid"} and calls(cz.inverters[0].driver, "read")
|
|
and calls(cz.inverters[0].driver, "write"))
|
|
|
|
# ==============================================================================
|
|
print()
|
|
print("4. SAFETY-28's census counts actuator ids; rule 9 on the new keys")
|
|
|
|
def entry(**kw):
|
|
return dict(name="u", max_w=3000, bms_soc_floor=10, **kw)
|
|
|
|
|
|
check("GoodWe entries are counted as before: one entity named twice is one "
|
|
"actuator with two writers, whatever its case",
|
|
{k: len(v) for k, v in actuator_writers({"inverters": [
|
|
entry(setpoint_entity="number.X"),
|
|
entry(setpoint_entity=" number.x ")]}).items()} == {"number.x": 2}
|
|
and actuator_writers({"setpoint_entity": "number.a", "inverters": [
|
|
entry(setpoint_entity="number.a")]})["number.a"]
|
|
== ["setpoint_entity", "inverters[0].setpoint_entity"])
|
|
cen = actuator_writers({"inverters": [
|
|
entry(driver="fake", host="gw", port=502, unit_id=1),
|
|
entry(driver="fake", host="gw", port=502, unit_id=2),
|
|
entry(driver="fake", host="gw", port=503, unit_id=1),
|
|
entry(driver="fake", host="GW", port=502, unit_id=1)]})
|
|
print(f" census {cen}")
|
|
check("another driver's actuator is host:port:unit: two units behind one "
|
|
"gateway are two actuators, and one unit named twice is one",
|
|
set(cen) == {"gw:502:1", "gw:502:2", "gw:503:1"}
|
|
and cen["gw:502:1"] == ["inverters[0].host", "inverters[3].host"])
|
|
check("a GoodWe key on a fake entry STILL declares a writer (the pre-SMA-2 "
|
|
"census kept, the driver's actuator added to it), and a `host` on a "
|
|
"GoodWe entry declares none",
|
|
actuator_writers({"inverters": [entry(driver="fake", host="gw",
|
|
setpoint_entity="number.x")]})
|
|
== {"number.x": ["inverters[0].setpoint_entity"],
|
|
"gw:502:1": ["inverters[0].host"]}
|
|
and actuator_writers({"inverters": [entry(setpoint_entity="number.x",
|
|
host="gw")]})
|
|
== {"number.x": ["inverters[0].setpoint_entity"]})
|
|
check("an entry naming a driver this build lacks still counts its setpoint "
|
|
"entity - refusing too much is the safe direction",
|
|
actuator_writers({"inverters": [entry(driver="nope",
|
|
setpoint_entity="number.q")]})
|
|
== {"number.q": ["inverters[0].setpoint_entity"]})
|
|
check("census_key() folds case and space",
|
|
census_key(" Number.X ") == "number.x")
|
|
|
|
GOOD = dict(entry(setpoint_entity="number.sp_a", soc_entity="sensor.soc_a",
|
|
batt_entity="sensor.batt_a"))
|
|
BASEOPTS = {"meter_source": "off", "meter_entity": "sensor.grid"}
|
|
base_problems = validate_startup_config(dict(BASEOPTS, inverters=[dict(GOOD)]))
|
|
print(f" baseline problems for a plain GoodWe roster: {base_problems}")
|
|
|
|
|
|
def prob(**kw):
|
|
return validate_startup_config(dict(BASEOPTS, inverters=[dict(GOOD, **kw)]))
|
|
|
|
|
|
check("an existing entry (no new key) is graded exactly as before",
|
|
prob() == base_problems)
|
|
check("...and so is one that says driver goodwe_esphome, or carries every new "
|
|
"key in a valid shape",
|
|
prob(driver="goodwe_esphome") == base_problems
|
|
and prob(driver="", host="", maintenance=False) == base_problems
|
|
and prob(host="inv.local", port=502, unit_id=3, maintenance=True)
|
|
== base_problems)
|
|
def named(problems, key):
|
|
return [p for p in problems if f"inverters.{key}" in p or f"({key})" in p]
|
|
|
|
|
|
check("only goodwe_esphome can be selected for now",
|
|
named(prob(driver="sma_modbus"), "driver")
|
|
and named(prob(driver="fake"), "driver")
|
|
and SELECTABLE == (DEFAULT_DRIVER,))
|
|
check("a host with a space, a bad port or unit number and a non-boolean "
|
|
"maintenance are each refused by key; the right shapes are not",
|
|
named(prob(host="a b"), "host")
|
|
and named(prob(port=0), "port") and named(prob(port=70000), "port")
|
|
and named(prob(port="x"), "port") and named(prob(port=True), "port")
|
|
and named(prob(unit_id=248), "unit_id")
|
|
and named(prob(unit_id=1.5), "unit_id")
|
|
and named(prob(maintenance="yes"), "maintenance")
|
|
and not named(prob(port=65535, unit_id=0, host="10.0.0.1"), "port"))
|
|
check("a GoodWe entry with no setpoint entity is refused with the same sentence "
|
|
"as before",
|
|
any("has no" in p and "setpoint entity" in p.lower()
|
|
for p in validate_startup_config(dict(BASEOPTS, inverters=[
|
|
dict(GOOD, setpoint_entity="")]))))
|
|
fu = FakeDriver(host="gw")
|
|
check("rule 2 grades any unit by its driver's actuator id, so a unit with no "
|
|
"setpoint entity is no longer skipped",
|
|
validate_unit_ranges([Inverter("a", driver=fu, max_w=2000.0)])
|
|
== [] and validate_unit_ranges([_u := Inverter("a", driver=fu,
|
|
max_w=2000.0)]) == []
|
|
and (setattr(_u, "dev_max", 1000.0) or True)
|
|
and len(validate_unit_ranges([_u])) == 1
|
|
and "gw:502:1" in validate_unit_ranges([_u])[0])
|
|
|
|
# ==============================================================================
|
|
print()
|
|
print("5. the schema")
|
|
cfg = io.open(HERE / "config.yaml", encoding="utf-8").read()
|
|
sch = cfg[cfg.index("\nschema:"):]
|
|
ent = sch[sch.index("\n inverters:"):]
|
|
ent = ent[:ent.index("\n\n")]
|
|
decl = dict(ln.strip().lstrip("- ").split(": ", 1)
|
|
for ln in ent.splitlines()[1:] if ": " in ln)
|
|
NEW_KEYS = ("driver", "host", "port", "unit_id", "maintenance")
|
|
check("HIDDEN FROM CLIENTS (owner, 2026-11-07): config.yaml's per-entry schema "
|
|
"is what it was on 23ff9d0 - the five driver keys are not in it, and "
|
|
"setpoint_entity is still required - so the Supervisor form shows "
|
|
"nothing new",
|
|
not set(NEW_KEYS) & set(decl) and not decl["setpoint_entity"].endswith("?")
|
|
and set(decl) == {"name", "setpoint_entity", "max_w", "bms_soc_floor",
|
|
"soc_entity", "batt_entity", "batt_invert",
|
|
"capacity_kwh"})
|
|
_tr = {lg: io.open(HERE / "translations" / f"{lg}.yaml", encoding="utf-8").read()
|
|
for lg in ("en", "nl")}
|
|
from app import messages as _msgs # noqa: E402
|
|
check("...and no translation, label or description was added for them",
|
|
not any(f" {k}:\n" in t.split(" estop_fitted:")[0].split(" inverters:")[1]
|
|
for t in _tr.values() for k in NEW_KEYS)
|
|
and not set(NEW_KEYS) & (set(_msgs.LABELS) | set(_msgs.LABELS_NL)))
|
|
|
|
# ==============================================================================
|
|
print()
|
|
print("6. round 2: census parity, setters, rule 9(d), NaN ports, hidden keys")
|
|
|
|
import copy, itertools, math # noqa: E402
|
|
from app import uimodel # noqa: E402
|
|
from app.validate import ACTUATOR_WRITE_KEYS # noqa: E402
|
|
|
|
|
|
def old_census(opts):
|
|
"""23ff9d0's census, copied: every ACTUATOR_WRITE_KEYS key, wherever it is."""
|
|
found = []
|
|
|
|
def walk(node, path):
|
|
if isinstance(node, dict):
|
|
for k, v in node.items():
|
|
here = f"{path}.{k}" if path else str(k)
|
|
if k in ACTUATOR_WRITE_KEYS:
|
|
e = str(v or "").strip()
|
|
if e:
|
|
found.append((here, e))
|
|
continue
|
|
walk(v, here)
|
|
elif isinstance(node, (list, tuple)):
|
|
for i, v in enumerate(node):
|
|
walk(v, f"{path}[{i}]")
|
|
walk(opts, "")
|
|
cen = {}
|
|
for where, e in found:
|
|
cen.setdefault(e.lower(), []).append(where)
|
|
return cen
|
|
|
|
|
|
check("ACTUATOR_WRITE_KEYS is exactly ('setpoint_entity',): the parity below "
|
|
"is about that key", tuple(ACTUATOR_WRITE_KEYS) == ("setpoint_entity",))
|
|
SP = ["number.a", "Number.A", " number.b ", "", None, 5, ["x"]]
|
|
shapes = []
|
|
for flat, e0, e1 in itertools.product(SP, SP, SP):
|
|
shapes.append({"setpoint_entity": flat, "inverters": [
|
|
{"name": "a", "setpoint_entity": e0}, {"name": "b", "setpoint_entity": e1}]})
|
|
shapes += [{"x": {"y": [{"setpoint_entity": "number.z"}]}, "setpoint_entity": "n.q"},
|
|
{"inverters": "junk"}, {"inverters": [{"driver": "nope",
|
|
"setpoint_entity": "n.k"}]},
|
|
{"inverters": [{"driver": "goodwe_esphome", "setpoint_entity": "n.k"}]},
|
|
{}, {"inverters": []}]
|
|
bad = [i for i, o in enumerate(shapes) if actuator_writers(o) != old_census(o)]
|
|
check(f"CENSUS PARITY: on {len(shapes)} flat/roster/odd option shapes the "
|
|
"census equals 23ff9d0's, keys, order and all", not bad)
|
|
mixed = {"setpoint_entity": "number.a", "inverters": [
|
|
{"name": "a", "setpoint_entity": "number.a"},
|
|
{"name": "f", "driver": "fake", "host": "gw", "port": 502, "unit_id": 1,
|
|
"setpoint_entity": "number.ignored"},
|
|
{"name": "g", "setpoint_entity": "number.g"}]}
|
|
new_c, old_c = actuator_writers(mixed), old_census(mixed)
|
|
check("...and on a mixed config (a fake-driver entry too) the census is a "
|
|
"SUPERSET of the pre-SMA-2 one: every old writer is still counted, "
|
|
"plus the driver's own actuator",
|
|
all(new_c.get(k) and set(v) <= set(new_c[k]) for k, v in old_c.items())
|
|
and "gw:502:1" in new_c)
|
|
|
|
gi = Inverter("s", setpoint_entity="number.s", soc_entity="sensor.a",
|
|
batt_entity="sensor.b")
|
|
gi.setpoint_entity, gi.soc_entity, gi.batt_entity, gi.batt_invert = (
|
|
"number.t", "sensor.c", "sensor.d", True)
|
|
check("SETTERS: assigning setpoint_entity/soc_entity/batt_entity/batt_invert "
|
|
"on an Inverter writes through to its GoodWe driver, as it did before",
|
|
(gi.driver.setpoint_entity, gi.driver.soc_entity, gi.driver.batt_entity,
|
|
gi.driver.batt_invert) == ("number.t", "sensor.c", "sensor.d", True)
|
|
and gi.has_writer() and gi.driver.actuator_id() == "number.t")
|
|
|
|
|
|
def _assign():
|
|
inv_f.setpoint_entity = "x"
|
|
|
|
|
|
try:
|
|
_assign()
|
|
_refused = False
|
|
except AttributeError:
|
|
_refused = True
|
|
check("...and on a driver that has no such field it refuses, not loses the value",
|
|
_refused)
|
|
|
|
nogood = validate_startup_config(dict(BASEOPTS, inverters=[
|
|
{k: v for k, v in GOOD.items() if k != "setpoint_entity"}]))
|
|
check("RULE 9(d): a GoodWe entry with no setpoint_entity is refused, by name "
|
|
"and by position, with the sentence it always had",
|
|
any(p.startswith("inverters[0] (u) has no") and "setpoint entity" in p.lower()
|
|
and "Set its ESPHome number entity, or remove the entry." in p
|
|
for p in nogood)
|
|
and nogood != base_problems)
|
|
check("...also with `driver: goodwe_esphome` said out loud",
|
|
any("has no" in p for p in validate_startup_config(dict(
|
|
BASEOPTS, inverters=[{k: v for k, v in dict(GOOD, driver="goodwe_esphome").items()
|
|
if k != "setpoint_entity"}]))))
|
|
|
|
nan, inf = float("nan"), float("inf")
|
|
for key in ("port", "unit_id"):
|
|
for bad_v, nm in ((nan, "NaN"), (inf, "inf"), (-inf, "-inf")):
|
|
try:
|
|
got = prob(**{key: bad_v})
|
|
ok_ = bool(named(got, key))
|
|
except Exception as e: # noqa: BLE001
|
|
got, ok_ = repr(e), False
|
|
check(f"{key} = {nm} is refused by key and does not raise", ok_)
|
|
|
|
ent = dict(name="a", setpoint_entity="number.a", max_w=3000, bms_soc_floor=10,
|
|
soc_entity="sensor.s", batt_entity="sensor.b", driver="goodwe_esphome",
|
|
host="h", port=502, unit_id=3, maintenance=False)
|
|
opts_rt = {"inverters": [ent, dict(ent, name="b", setpoint_entity="number.b")]}
|
|
back = uimodel.ui_to_options(uimodel.options_to_ui(copy.deepcopy(opts_rt)))
|
|
check("an entry that carries the driver keys (from a caller that skips the "
|
|
"schema) comes back from a save through the page model unchanged",
|
|
back.get("inverters") == opts_rt["inverters"])
|
|
|
|
del REGISTRY["fake"]
|
|
|
|
# THE SEAM MUTANTS THIS FILE WAS RUN AGAINST (twelve, each one edit to a copy of
|
|
# the tree; all twelve turn it red):
|
|
# write_unit calls hass.set_number itself -> part 3 WRITE (and the tripwire)
|
|
# _read_unit calls hass.number(soc/batt entity) -> part 3 READ
|
|
# run_control's limits keyed on setpoint_entity -> part 3 LIMITS
|
|
# the panel's range check keyed on setpoint_entity -> part 3 PANEL
|
|
# has_writer() back to bool(setpoint_entity) -> part 2 has_writer
|
|
# is_live() window back to LINK_STALE_S -> part 2 is_live
|
|
# link_note() window back to LINK_STALE_S -> part 2 link_note
|
|
# Freshness built on BOUNDS["batt"] -> part 2 freshness bound
|
|
# _write_unit's guard keyed on setpoint_entity -> part 3 WRITE
|
|
# the census back to the GoodWe key only -> part 4 census
|
|
# rule 2 per unit back to .setpoint_entity -> part 3 RULE 2 PER UNIT
|
|
# the status rows' source test back to the entities -> part 3 STATUS
|
|
# NOT COVERED: the call in amain() that reads the flat setpoint entity's range
|
|
# before any Controller exists. It runs only inside amain(), which no check
|
|
# starts; it is two lines and the GoodWe driver's own limits(), graded in part 1.
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED")
|
|
sys.exit(1)
|
|
print("all passed")
|