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>
332 lines
13 KiB
Python
332 lines
13 KiB
Python
"""UX-13: stored options <-> the page model, for the keys the page derives.
|
|
|
|
`options_to_ui()` turns the stored blob into what the page edits: always a list
|
|
of inverter cards, solar rows with a "reads negative" toggle, a contract and
|
|
simple prices instead of `static_profile` JSON, and a region preset instead of
|
|
the levies figure. `ui_to_options()` turns it back, so the validator and the
|
|
Supervisor always see the real blob.
|
|
|
|
The rule that makes the round trip exact: the model carries the stored values
|
|
of the derived keys (`keep`). A derivation whose page fields still read what
|
|
`keep` decodes to writes `keep` back untouched, and only an edited one is
|
|
re-encoded. So a blob the page cannot express (a `static_profile` with odd
|
|
windows, a malformed roster) survives a save of some other field unchanged.
|
|
See DECISIONS/UX-13-one-layout-table.md.
|
|
"""
|
|
|
|
import copy
|
|
import json
|
|
|
|
from .price import FLEMISH_TAXES_COMPONENTS_EUR_KWH
|
|
from .static_tariff import ProfileInvalid, parse_static_profile
|
|
|
|
__all__ = ["options_to_ui", "ui_to_options", "REGION_PRESETS",
|
|
"arbitrage_from_batteries", "DeriveError"]
|
|
|
|
|
|
class DeriveError(ValueError):
|
|
"""The page model asks for something no options blob can say."""
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Region presets for "Grid costs, levies and VAT". Year-stamped, updated each
|
|
# January (owner, 2026-09-27). Only presets with a sourced figure are listed:
|
|
# Flanders is app/price.py's Fluvius Antwerpen sum (TARIFF-18: Eneco
|
|
# residential card, Flanders, September 2026, VAT included). Wallonia and
|
|
# Brussels have a day and a night band (config.yaml) and join with sourced
|
|
# figures; until then those sites choose "own".
|
|
# --------------------------------------------------------------------------- #
|
|
REGION_PRESETS = {
|
|
"flanders": {
|
|
"year": 2026,
|
|
"taxes_eur_kwh": round(sum(v for _n, v in
|
|
FLEMISH_TAXES_COMPONENTS_EUR_KWH), 5),
|
|
"taxes_profile": "",
|
|
},
|
|
}
|
|
OWN = "own"
|
|
|
|
_ROSTER_FLAT = ("setpoint_entity", "soc_entity", "batt_entity", "batt_invert",
|
|
"max_w", "bms_soc_floor")
|
|
_TOL = 1e-9
|
|
|
|
|
|
def _blank(v) -> bool:
|
|
return v is None or (isinstance(v, str) and not v.strip())
|
|
|
|
|
|
def _num(v):
|
|
try:
|
|
return None if _blank(v) or isinstance(v, bool) else float(v)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
# --- 1. one inverter (flat keys) or the `inverters` list -------------------- #
|
|
def _roster_decode(o):
|
|
entries = o.get("inverters")
|
|
if entries is None or entries == "" or entries == []:
|
|
card = {k: o.get(k) for k in _ROSTER_FLAT}
|
|
card.update(name="", capacity_kwh=None)
|
|
return {"inverter_shape": "flat", "inverters": [card]}
|
|
if isinstance(entries, list) and all(isinstance(e, dict) for e in entries):
|
|
return {"inverter_shape": "list",
|
|
"inverters": [dict(e) for e in entries]}
|
|
# Not something a card can show: edited in the Configuration tab only.
|
|
return {"inverter_shape": "raw", "inverters": None}
|
|
|
|
|
|
def _roster_encode(m, keep):
|
|
out = {}
|
|
shape, cards = m.get("inverter_shape"), m.get("inverters")
|
|
if shape == "raw":
|
|
raise DeriveError("the inverter list is edited in the Configuration tab")
|
|
if not cards:
|
|
raise DeriveError("at least one inverter is needed")
|
|
if shape == "flat" and len(cards) == 1:
|
|
card = cards[0]
|
|
if not _blank(card.get("name")) or not _blank(card.get("capacity_kwh")):
|
|
raise DeriveError("one inverter stored flat has no name or capacity")
|
|
for k in _ROSTER_FLAT:
|
|
if card.get(k) is not None or k in keep:
|
|
out[k] = card.get(k)
|
|
out["inverters"] = []
|
|
return out
|
|
out["inverters"] = [{k: v for k, v in c.items() if v is not None}
|
|
for c in cards]
|
|
# The flat keys beside a list are not read: they stay as stored.
|
|
out.update({k: copy.deepcopy(keep[k]) for k in _ROSTER_FLAT if k in keep})
|
|
if shape == "flat":
|
|
# CFG-04 rule 9e: a flat power request entity beside a list is refused.
|
|
out["setpoint_entity"] = ""
|
|
return out
|
|
|
|
|
|
# --- 2. solar rows: pv_entity / pv_entities / pv_invert_entities ------------ #
|
|
def _strs(v):
|
|
return [str(x).strip() for x in (v if isinstance(v, list) else [])
|
|
if not _blank(x)]
|
|
|
|
|
|
def _solar_rows(o):
|
|
listed = _strs(o.get("pv_entities"))
|
|
if listed:
|
|
return listed, True
|
|
one = "" if _blank(o.get("pv_entity")) else str(o.get("pv_entity")).strip()
|
|
return ([one] if one else []), False
|
|
|
|
|
|
def _solar_decode(o):
|
|
rows, _listed = _solar_rows(o)
|
|
flipped = set(_strs(o.get("pv_invert_entities")))
|
|
return {"ui.solar": [{"entity": e, "reversed": e in flipped}
|
|
for e in rows]}
|
|
|
|
|
|
def _solar_encode(m, keep):
|
|
out = {}
|
|
rows = [r for r in (m.get("ui.solar") or [])
|
|
if not _blank(r.get("entity"))]
|
|
entities = [str(r["entity"]).strip() for r in rows]
|
|
old_rows, was_list = _solar_rows(keep)
|
|
if len(entities) <= 1 and not was_list:
|
|
out["pv_entity"] = entities[0] if entities else ""
|
|
if "pv_entities" in keep:
|
|
out["pv_entities"] = []
|
|
else:
|
|
out["pv_entities"] = entities
|
|
if "pv_entity" in keep:
|
|
out["pv_entity"] = "" # TEL-12: the list replaces it
|
|
# Flips of entities that were never a row stay as stored (pv.stray_inverts
|
|
# reports them). Flips of removed rows go with the row.
|
|
extras = [x for x in _strs(keep.get("pv_invert_entities"))
|
|
if x not in old_rows and x not in entities]
|
|
flips = [e for e, r in zip(entities, rows) if r.get("reversed")] + extras
|
|
if flips or "pv_invert_entities" in keep:
|
|
out["pv_invert_entities"] = flips
|
|
return out
|
|
|
|
|
|
# --- 3. contract and static_profile ----------------------------------------- #
|
|
def _fixed_json(price) -> str:
|
|
return json.dumps({"periods": [
|
|
{"name": "all", "from": "00:00", "to": "24:00", "days": "mon-sun",
|
|
"eur_per_kwh": price}]})
|
|
|
|
|
|
def _daynight_json(day, night) -> str:
|
|
return json.dumps({"periods": [
|
|
{"name": "day", "from": "07:00", "to": "22:00", "days": "mon-fri",
|
|
"eur_per_kwh": day},
|
|
{"name": "night", "from": "22:00", "to": "07:00", "days": "mon-fri",
|
|
"eur_per_kwh": night},
|
|
{"name": "night", "from": "00:00", "to": "24:00", "days": "sat,sun",
|
|
"eur_per_kwh": night}]})
|
|
|
|
|
|
def _shape(text):
|
|
"""The periods of a profile without their names, for comparing, or None."""
|
|
try:
|
|
prof = parse_static_profile(text)
|
|
except ProfileInvalid:
|
|
return None
|
|
if prof is None:
|
|
return ()
|
|
return tuple(sorted((p.start_min, p.end_min, tuple(sorted(p.days)),
|
|
p.eur_per_kwh) for p in prof.periods))
|
|
|
|
|
|
def _profile_decode(raw):
|
|
"""("none"|"fixed"|"daynight"|"custom", fixed, day, night)."""
|
|
if isinstance(raw, str) and _blank(raw):
|
|
return "none", None, None, None
|
|
if not isinstance(raw, str):
|
|
return "custom", None, None, None
|
|
got = _shape(raw)
|
|
if not got:
|
|
return ("none" if got == () else "custom"), None, None, None
|
|
if len(got) == 1 and got == _shape(_fixed_json(got[0][3])):
|
|
return "fixed", got[0][3], None, None
|
|
day = next((p[3] for p in got if p[:2] == (420, 1320)), None)
|
|
night = next((p[3] for p in got if p[:2] == (1320, 420)), None)
|
|
if (day is not None and night is not None
|
|
and got == _shape(_daynight_json(day, night))):
|
|
return "daynight", None, day, night
|
|
return "custom", None, None, None
|
|
|
|
|
|
def _tariff_decode(o):
|
|
kind, fixed, day, night = _profile_decode(o.get("static_profile", ""))
|
|
provider = o.get("price_provider")
|
|
dynamic = not _blank(provider) and provider != "off"
|
|
return {"ui.contract": "dynamic" if dynamic else kind,
|
|
"price_provider": provider, "ui.fallback": kind,
|
|
"ui.fixed_price": fixed, "ui.day_price": day,
|
|
"ui.night_price": night}
|
|
|
|
|
|
def _tariff_encode(m, keep):
|
|
out = {}
|
|
contract = m.get("ui.contract")
|
|
if contract == "dynamic":
|
|
if _blank(m.get("price_provider")) or m.get("price_provider") == "off":
|
|
raise DeriveError("a dynamic contract needs a price source")
|
|
out["price_provider"] = m["price_provider"]
|
|
kind = m.get("ui.fallback")
|
|
else:
|
|
out["price_provider"] = "off"
|
|
kind = contract
|
|
if kind == "none":
|
|
out["static_profile"] = ""
|
|
elif kind == "fixed":
|
|
out["static_profile"] = _fixed_json(_price_of(m, "ui.fixed_price"))
|
|
elif kind == "daynight":
|
|
out["static_profile"] = _daynight_json(_price_of(m, "ui.day_price"),
|
|
_price_of(m, "ui.night_price"))
|
|
elif kind == "custom":
|
|
if "static_profile" not in keep:
|
|
raise DeriveError("odd tariff windows are edited in the "
|
|
"Configuration tab")
|
|
out["static_profile"] = keep["static_profile"]
|
|
else:
|
|
raise DeriveError(f"unknown contract {kind!r}")
|
|
return out
|
|
|
|
|
|
def _price_of(m, name):
|
|
value = _num(m.get(name))
|
|
if value is None:
|
|
raise DeriveError(f"{name} needs a price")
|
|
return value
|
|
|
|
|
|
# --- 4. region preset -> price_taxes_eur_kwh / price_taxes_profile ---------- #
|
|
def _region_decode(o):
|
|
taxes, profile = o.get("price_taxes_eur_kwh"), o.get("price_taxes_profile", "")
|
|
region = OWN
|
|
for rid, p in REGION_PRESETS.items():
|
|
same = _num(taxes) is not None and abs(_num(taxes) - p["taxes_eur_kwh"]) < _TOL
|
|
if same and str(profile or "").strip() == p["taxes_profile"]:
|
|
region = rid
|
|
return {"ui.region": region, "price_taxes_eur_kwh": taxes}
|
|
|
|
|
|
def _region_encode(m, keep):
|
|
region = m.get("ui.region")
|
|
if region == OWN:
|
|
out = {"price_taxes_eur_kwh": m.get("price_taxes_eur_kwh")}
|
|
if "price_taxes_profile" in keep: # "own" keeps the stored table
|
|
out["price_taxes_profile"] = keep["price_taxes_profile"]
|
|
return out
|
|
if region not in REGION_PRESETS:
|
|
raise DeriveError(f"unknown region {region!r}")
|
|
p = REGION_PRESETS[region]
|
|
out = {"price_taxes_eur_kwh": p["taxes_eur_kwh"]}
|
|
if p["taxes_profile"] or "price_taxes_profile" in keep:
|
|
out["price_taxes_profile"] = p["taxes_profile"]
|
|
return out
|
|
|
|
|
|
# --- 5. arbitrage_enabled: the OPT-11 hook ---------------------------------- #
|
|
def arbitrage_from_batteries(m):
|
|
"""OPT-11 hook. With battery modes on, `arbitrage_enabled` is true when any
|
|
battery may sell. Battery modes are not built yet, so this returns None and
|
|
the toggle is stored as the page has it: no behaviour change."""
|
|
return None
|
|
|
|
|
|
def _arbitrage_decode(o):
|
|
return {"arbitrage_enabled": o.get("arbitrage_enabled")}
|
|
|
|
|
|
def _arbitrage_encode(m, keep):
|
|
derived = arbitrage_from_batteries(m)
|
|
value = m.get("arbitrage_enabled") if derived is None else derived
|
|
return {} if value is None and "arbitrage_enabled" not in keep \
|
|
else {"arbitrage_enabled": value}
|
|
|
|
|
|
# (option keys it writes, decode, encode). Each is independent of the others.
|
|
DERIVATIONS = (
|
|
(("inverters",) + _ROSTER_FLAT, _roster_decode, _roster_encode),
|
|
(("pv_entity", "pv_entities", "pv_invert_entities"), _solar_decode,
|
|
_solar_encode),
|
|
(("price_provider", "static_profile"), _tariff_decode, _tariff_encode),
|
|
(("price_taxes_eur_kwh", "price_taxes_profile"), _region_decode,
|
|
_region_encode),
|
|
(("arbitrage_enabled",), _arbitrage_decode, _arbitrage_encode),
|
|
)
|
|
_OWNED = {k for keys, _d, _e in DERIVATIONS for k in keys}
|
|
_MODEL_ONLY = ("keep", "inverter_shape", "inverter_count", "region_year")
|
|
|
|
|
|
def options_to_ui(options: dict) -> dict:
|
|
"""The page model for a stored blob. Pure: reads nothing else."""
|
|
model = {k: copy.deepcopy(v) for k, v in options.items()
|
|
if k not in _OWNED}
|
|
for _keys, decode, _encode in DERIVATIONS:
|
|
model.update(decode(options))
|
|
model["keep"] = {k: copy.deepcopy(options[k]) for k in _OWNED
|
|
if k in options}
|
|
cards = model.get("inverters")
|
|
model["inverter_count"] = len(cards) if isinstance(cards, list) else None
|
|
preset = REGION_PRESETS.get(model["ui.region"])
|
|
model["region_year"] = preset["year"] if preset else None
|
|
return model
|
|
|
|
|
|
def ui_to_options(model: dict) -> dict:
|
|
"""The options blob for a page model. Raises DeriveError on a model no blob
|
|
can express, so the page says so instead of saving something else."""
|
|
keep = model.get("keep") or {}
|
|
out = {k: copy.deepcopy(v) for k, v in model.items()
|
|
if k not in _MODEL_ONLY and k not in _OWNED
|
|
and not k.startswith("ui.")}
|
|
for keys, decode, encode in DERIVATIONS:
|
|
was = decode(keep)
|
|
if all(model.get(name) == value for name, value in was.items()):
|
|
out.update({k: copy.deepcopy(keep[k]) for k in keys if k in keep})
|
|
else:
|
|
out.update(encode(model, keep))
|
|
return out
|