adminandClaude Opus 5.5
badea3cb2d
- uimodel: editing a listed inverter keeps the stored flat keys, and a region of "own" keeps a stored price_taxes_profile (both were dropped). - haentities: a failed read also waits CACHE_S, so polls while Home Assistant is down do not queue one 10 s read each. - ui_text: the Flanders preset names its DSO, and the help says Wallonia and Brussels have no figures yet. REGION_PRESETS cites the card. - tests: decode of a reversed solar row, both kept keys, the backoff. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
225 lines
10 KiB
Python
225 lines
10 KiB
Python
"""UX-13 P0: stored options <-> page model, exactly. `python3 test_uimodel.py`
|
|
|
|
app/uimodel.py derives the keys the page does not show (the flat inverter,
|
|
the solar keys, static_profile, the levies figure, arbitrage_enabled). Part 1
|
|
round-trips every options blob the test suite hands the validator
|
|
(scenarios/ux13/, recorded by tools/ux13_validator_calls.py at 652097a), the
|
|
TEST-23 option files and config.yaml's defaults: options -> model -> JSON ->
|
|
options must be identical. Parts 2-6 edit one page field and check the blob.
|
|
Mutating a module this file grades? Read MUTATION.md first (purge __pycache__).
|
|
"""
|
|
|
|
import glob
|
|
import gzip
|
|
import io
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
logging.disable(logging.WARNING) # the validator's own log lines
|
|
|
|
from app import uimodel as U # noqa: E402
|
|
from app.price import (FLEMISH_TAXES_COMPONENTS_EUR_KWH, # noqa: E402
|
|
PRICE_OPTION_DEFAULTS)
|
|
from app.static_tariff import parse_static_profile # noqa: E402
|
|
from app.validate import validate_startup_config # noqa: E402
|
|
|
|
fails = []
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
def trip(options):
|
|
return U.ui_to_options(json.loads(json.dumps(U.options_to_ui(options))))
|
|
|
|
|
|
def edit(options, **fields):
|
|
model = json.loads(json.dumps(U.options_to_ui(options)))
|
|
for k, v in fields.items():
|
|
model[k.replace("__", ".")] = v
|
|
return U.ui_to_options(model)
|
|
|
|
|
|
def raises(fn):
|
|
try:
|
|
fn()
|
|
except U.DeriveError:
|
|
return True
|
|
return False
|
|
|
|
|
|
def defaults():
|
|
"""config.yaml's `options:` block, read without PyYAML (not in CI)."""
|
|
text = io.open(os.path.join(HERE, "config.yaml"), encoding="utf-8").read()
|
|
block = text[text.index("\noptions:"):text.index("\nschema:")]
|
|
out = {}
|
|
for key, raw in re.findall(r"^ ([a-z_0-9]+):(.*)$", block, re.M):
|
|
raw = raw.split(" #")[0].strip()
|
|
try:
|
|
out[key] = json.loads(raw)
|
|
except ValueError:
|
|
out[key] = raw
|
|
return out
|
|
|
|
|
|
DEFAULTS = defaults()
|
|
SITE = dict(DEFAULTS, setpoint_entity="number.goodwe_setpoint",
|
|
soc_entity="sensor.soc", batt_entity="sensor.batt")
|
|
|
|
print("1. every known blob survives the round trip unchanged")
|
|
_calls = [json.loads(line) for line in gzip.open(os.path.join(
|
|
HERE, "scenarios", "ux13", "validator-calls-652097a.jsonl.gz"), "rt")]
|
|
_test23 = sorted(glob.glob(os.path.join(HERE, "scenarios", "ux13", "test23",
|
|
"*.json")))
|
|
_blobs = ([("config.yaml", DEFAULTS)]
|
|
+ [(c["source"], c["opts"]) for c in _calls
|
|
if isinstance(c["opts"], dict)]
|
|
+ [(os.path.basename(f), json.load(open(f))) for f in _test23])
|
|
_bad = []
|
|
for _src, _o in _blobs:
|
|
try:
|
|
_back = trip(_o)
|
|
except Exception as e: # noqa: BLE001
|
|
_bad.append(f"{_src}: {type(e).__name__} {e}")
|
|
continue
|
|
if json.dumps(_back, sort_keys=True) != json.dumps(_o, sort_keys=True):
|
|
_bad.append(f"{_src}: {sorted(set(_back) ^ set(_o)) or 'values'}")
|
|
print(f" {len(_calls)} validator calls, {len(_test23)} TEST-23 files, "
|
|
f"config.yaml defaults")
|
|
check(f"all {len(_blobs)} are identical after options -> model -> JSON -> "
|
|
f"options (not: {_bad[:3]})", not _bad and len(_calls) >= 200
|
|
and len(_test23) >= 5 and len(DEFAULTS) >= 78)
|
|
check("config.yaml was read whole: the defaults hold every option",
|
|
"inverters" in DEFAULTS and DEFAULTS["inverters"] == []
|
|
and DEFAULTS["price_taxes_eur_kwh"] == 0.11736)
|
|
_m = U.options_to_ui(DEFAULTS)
|
|
check("the page model never shows a derived key: it keeps them aside",
|
|
not {"setpoint_entity", "pv_entity", "pv_entities", "static_profile",
|
|
"price_taxes_profile"} & set(_m)
|
|
and set(_m["keep"]) >= {"static_profile", "pv_entity", "max_w"})
|
|
check("a blob that lacks keys gains none", trip({"max_w": 900}) == {"max_w": 900})
|
|
|
|
print("\n2. one inverter or a list")
|
|
_one = edit(SITE, inverters=[dict(U.options_to_ui(SITE)["inverters"][0],
|
|
max_w=1500)])
|
|
check("editing the one inverter writes the flat keys and keeps the list empty",
|
|
_one["max_w"] == 1500 and _one["inverters"] == []
|
|
and _one["setpoint_entity"] == "number.goodwe_setpoint")
|
|
_card = U.options_to_ui(SITE)["inverters"][0]
|
|
_two = edit(SITE, inverters=[
|
|
dict(_card, name="inverter-a", capacity_kwh=None),
|
|
dict(_card, name="inverter-b", setpoint_entity="number.goodwe_b",
|
|
soc_entity="sensor.soc_b", batt_entity="sensor.batt_b")])
|
|
check("a second inverter writes the list and clears the flat power request "
|
|
"entity, which the validator refuses beside a list",
|
|
len(_two["inverters"]) == 2 and _two["setpoint_entity"] == ""
|
|
and "capacity_kwh" not in _two["inverters"][0]
|
|
and not validate_startup_config(_two))
|
|
_listed = dict(SITE, setpoint_entity="", inverters=[
|
|
{"name": "garage", "setpoint_entity": "number.g", "max_w": 2000,
|
|
"bms_soc_floor": 10}])
|
|
_l1 = edit(_listed, inverters=[dict(_listed["inverters"][0], max_w=1800)])
|
|
check("a list of one stays a list, so the inverter keeps its name and stored "
|
|
"state", _l1["inverters"] == [dict(_listed["inverters"][0], max_w=1800)])
|
|
check("editing a listed inverter keeps the unread flat keys as stored",
|
|
all(_l1[k] == _listed[k] for k in ("setpoint_entity", "soc_entity",
|
|
"batt_entity", "max_w")))
|
|
check("a name on the flat inverter is refused rather than dropped",
|
|
raises(lambda: edit(SITE, inverters=[dict(_card, name="garage")])))
|
|
_raw = dict(SITE, inverters="garage")
|
|
check("a roster no card can show is kept when another field is saved, and "
|
|
"cannot be edited here",
|
|
edit(_raw, gain=0.5)["inverters"] == "garage"
|
|
and raises(lambda: edit(_raw, inverters=[_card])))
|
|
|
|
print("\n3. solar rows")
|
|
check("a stored flip reads as a reversed row",
|
|
U.options_to_ui(dict(DEFAULTS, pv_entity="sensor.pv",
|
|
pv_invert_entities=["sensor.pv"]))["ui.solar"]
|
|
== [{"entity": "sensor.pv", "reversed": True}])
|
|
check("one row writes pv_entity",
|
|
edit(DEFAULTS, ui__solar=[{"entity": "sensor.pv", "reversed": False}])
|
|
["pv_entity"] == "sensor.pv")
|
|
_pv2 = edit(dict(DEFAULTS, pv_entity="sensor.a"), ui__solar=[
|
|
{"entity": "sensor.a", "reversed": False},
|
|
{"entity": "sensor.b", "reversed": True}])
|
|
check("two rows write pv_entities, clear pv_entity and flip the reversed one",
|
|
_pv2["pv_entities"] == ["sensor.a", "sensor.b"] and _pv2["pv_entity"] == ""
|
|
and _pv2["pv_invert_entities"] == ["sensor.b"])
|
|
_stray = dict(DEFAULTS, pv_entities=["sensor.a", "sensor.b"],
|
|
pv_invert_entities=["sensor.b", "sensor.old"])
|
|
_pv1 = edit(_stray, ui__solar=[{"entity": "sensor.a", "reversed": False}])
|
|
check("removing a reversed row drops its flip, and a flip that was never a row "
|
|
"stays as stored", _pv1["pv_entities"] == ["sensor.a"]
|
|
and _pv1["pv_invert_entities"] == ["sensor.old"])
|
|
|
|
print("\n4. contract and prices")
|
|
_fixed = edit(DEFAULTS, ui__contract="fixed", ui__fixed_price=0.31)
|
|
check("a fixed price writes a profile the tariff parser reads as one price",
|
|
_fixed["price_provider"] == "off"
|
|
and [p.eur_per_kwh for p in parse_static_profile(
|
|
_fixed["static_profile"]).periods] == [0.31]
|
|
and U.options_to_ui(_fixed)["ui.contract"] == "fixed")
|
|
_dn = edit(SITE, ui__contract="daynight", ui__day_price=0.34,
|
|
ui__night_price=0.26)
|
|
check("day and night write a profile that starts and reads back the same",
|
|
not validate_startup_config(_dn)
|
|
and U.options_to_ui(_dn)["ui.day_price"] == 0.34
|
|
and U.options_to_ui(_dn)["ui.night_price"] == 0.26)
|
|
_dyn = edit(_dn, ui__contract="dynamic", price_provider="generic_dayahead")
|
|
check("a dynamic contract keeps the day and night profile as its fallback",
|
|
_dyn["price_provider"] == "generic_dayahead"
|
|
and _dyn["static_profile"] == _dn["static_profile"])
|
|
check("a dynamic contract without a price source is refused here",
|
|
raises(lambda: edit(DEFAULTS, ui__contract="dynamic")))
|
|
_odd = json.dumps({"periods": [
|
|
{"from": "00:00", "to": "12:00", "days": "mon-sun", "eur_per_kwh": 0.2},
|
|
{"from": "12:00", "to": "24:00", "days": "mon-sun", "eur_per_kwh": 0.3}]})
|
|
_om = U.options_to_ui(dict(DEFAULTS, static_profile=_odd))
|
|
check("odd windows show as 'custom' and survive saving another field",
|
|
_om["ui.contract"] == "custom"
|
|
and edit(dict(DEFAULTS, static_profile=_odd), gain=0.5)
|
|
["static_profile"] == _odd)
|
|
|
|
print("\n5. region presets")
|
|
_fl = U.REGION_PRESETS["flanders"]
|
|
check("the Flanders preset is the sum of app/price.py's components, and is the "
|
|
"shipped default", _fl["taxes_eur_kwh"] == round(sum(
|
|
v for _n, v in FLEMISH_TAXES_COMPONENTS_EUR_KWH), 5)
|
|
== PRICE_OPTION_DEFAULTS["price_taxes_eur_kwh"] == DEFAULTS[
|
|
"price_taxes_eur_kwh"])
|
|
check("every preset carries its year", all(
|
|
isinstance(p["year"], int) and p["year"] >= 2026
|
|
for p in U.REGION_PRESETS.values()))
|
|
_own = dict(DEFAULTS, price_taxes_eur_kwh=0.06)
|
|
check("a figure that is no preset reads as 'own', and choosing Flanders "
|
|
"writes the preset", U.options_to_ui(_own)["ui.region"] == U.OWN
|
|
and edit(_own, ui__region="flanders")["price_taxes_eur_kwh"] == 0.11736)
|
|
_prof = dict(DEFAULTS, price_taxes_eur_kwh=0.12,
|
|
price_taxes_profile='{"day": 0.14, "night": 0.1}')
|
|
check("'own' keeps a stored levies table when the figure is edited",
|
|
edit(_prof, price_taxes_eur_kwh=0.13)["price_taxes_profile"]
|
|
== _prof["price_taxes_profile"])
|
|
check("'own' stores the typed figure",
|
|
edit(DEFAULTS, ui__region=U.OWN, price_taxes_eur_kwh=0.13)
|
|
["price_taxes_eur_kwh"] == 0.13)
|
|
|
|
print("\n6. arbitrage_enabled: the OPT-11 hook changes nothing yet")
|
|
check("the hook answers None while battery modes do not exist",
|
|
U.arbitrage_from_batteries(U.options_to_ui(DEFAULTS)) is None)
|
|
check("so the toggle is stored as the page has it",
|
|
edit(DEFAULTS, arbitrage_enabled=True)["arbitrage_enabled"] is True)
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED")
|
|
sys.exit(1)
|
|
print("all passed")
|