"""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 or time windows (P5) 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 import re from .price import FLEMISH_TAXES_COMPONENTS_EUR_KWH, MAX_PLAUSIBLE_EUR_KWH from .static_tariff import (DAY_NAMES, MINUTES_PER_DAY, Period, ProfileInvalid, parse_static_profile) __all__ = ["options_to_ui", "ui_to_options", "ui_to_options_partial", "REGION_PRESETS", "PERIOD_DAYS", "arbitrage_from_batteries", "DeriveError", "COUNT", "GROWN"] class DeriveError(ValueError): """The page model asks for something no options blob can say. `code` names the sentence the page shows (app/ui_text `derive.`), and `field` the page field it is shown beside.""" def __init__(self, text, code="", field="", **values): super().__init__(text) self.code, self.field = code, field self.values = values # the sentence's {placeholders} (P5) self.keys = () # the option keys the failed derivation writes # The page model's inverter count, which show conditions read, and what adding # a card sets: a second inverter turns the one flat inverter into a list. COUNT = "inverter_count" GROWN = {"inverter_shape": "list"} # --------------------------------------------------------------------------- # # 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", "raw_roster", "inverters") if not cards: raise DeriveError("at least one inverter is needed", "no_inverter", "inverters") 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", "flat_named", "inverters") 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" or _roster_decode(keep)["inverter_shape"] == "flat": # CFG-04 rule 9e: a flat power request entity beside a list is refused, # so a list that replaces the flat inverter clears it. 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 # --- 3b. the time-window editor (UX-13 P5) ---------------------------------- # # Each row is {"from", "to", "days", "price"} and, when read from a stored # profile, its "name". `days` is one of these; a profile with any other day set # is edited in the Configuration tab only (`ui.periods` is None). PERIOD_DAYS = ("mon-sun", "mon-fri", "sat,sun") + DAY_NAMES _DAYSETS = {d: frozenset(range(7)) if d == "mon-sun" else frozenset(range(5)) if d == "mon-fri" else frozenset((5, 6)) if d == "sat,sun" else frozenset((DAY_NAMES.index(d),)) for d in PERIOD_DAYS} _TIME = re.compile(r"^\s*(\d{1,2}):(\d{2})\s*$") def _hhmm(minute) -> str: return f"{minute // 60:02d}:{minute % 60:02d}" def _periods_decode(raw): """The rows for a stored profile: [] when unset, None when it cannot be shown as rows (unparsable, or days the editor has no choice for).""" try: prof = parse_static_profile(raw) if isinstance(raw, str) else None except ProfileInvalid: return None if prof is None: return [] if isinstance(raw, str) else None rows = [] for p in prof.periods: days = next((d for d, s in _DAYSETS.items() if s == p.days), None) if days is None: return None rows.append({"from": _hhmm(p.start_min), "to": _hhmm(p.end_min), "days": days, "price": p.eur_per_kwh, "name": p.name}) return rows def _minute(text, *, end): """"07:00" -> 420, "24:00" -> 1440 as an end, else None.""" m = _TIME.match(str(text or "")) if not m: return None h, mi = int(m.group(1)), int(m.group(2)) if mi > 59 or h > 24 or (h == 24 and (mi or not end)): return None return h * 60 + mi def _periods_json(rows) -> str: """The static_profile for the editor's rows, or DeriveError naming the first problem: a bad time, an empty price, an overlap or a gap.""" if not rows: raise DeriveError("add at least one time window", "no_periods", "ui.periods") periods = [] for i, row in enumerate(rows, start=1): row = row if isinstance(row, dict) else {} start = _minute(row.get("from"), end=False) end = _minute(row.get("to"), end=True) if start is None or end is None: raise DeriveError(f"window {i} has a bad time", "period_time", "ui.periods", row=i) end = end or MINUTES_PER_DAY # "00:00" as an end if start == end: raise DeriveError(f"window {i} is empty", "period_zero", "ui.periods", row=i) price = _num(row.get("price")) if price is None: raise DeriveError(f"window {i} needs a price", "period_price", "ui.periods", row=i) if abs(price) > MAX_PLAUSIBLE_EUR_KWH: raise DeriveError(f"window {i}'s price is out of range", "period_price_range", "ui.periods", row=i) if row.get("days") not in _DAYSETS: raise DeriveError(f"window {i} has no days", "unknown", "ui.periods") name = str(row.get("name") or "").strip() or f"window{i}" periods.append(Period(name=name, start_min=start, end_min=end, days=_DAYSETS[row["days"]], eur_per_kwh=price)) _tiling(periods) return json.dumps({"periods": [ {"name": p.name, "from": _hhmm(p.start_min), "to": _hhmm(p.end_min), "days": row["days"], "eur_per_kwh": p.eur_per_kwh} for p, row in zip(periods, rows)]}) def _tiling(periods): """Each minute of the week in exactly one window, as static_tariff.py requires. The first overlap or gap, by day and time, is the error.""" for day in range(7): start, kind, rows = None, None, None for minute in range(MINUTES_PER_DAY + 1): here = [i for i, p in enumerate(periods, start=1) if minute < MINUTES_PER_DAY and p.covers(day, minute)] now = ("gap" if not here else "overlap" if len(here) > 1 else None) if minute < MINUTES_PER_DAY else "end" if kind is not None and (now != kind or (kind == "overlap" and here != rows)): values = {"day": DAY_NAMES[day], "from": _hhmm(start), "to": _hhmm(minute)} if kind == "gap": raise DeriveError("a time has no window", "period_gap", "ui.periods", **values) raise DeriveError("two windows cover one time", "period_overlap", "ui.periods", a=rows[0], b=rows[1], **values) if now in ("gap", "overlap") and kind is None: start, kind, rows = minute, now, here def _tariff_decode(o): raw = o.get("static_profile", "") kind, fixed, day, night = _profile_decode(raw) 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, "ui.periods": _periods_decode(raw)} 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", "no_source", "price_provider") 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": rows = m.get("ui.periods") if rows is not None: out["static_profile"] = _periods_json(rows) elif "static_profile" not in keep: raise DeriveError("odd tariff windows are edited in the " "Configuration tab", "custom_profile", "ui.contract") else: # Days the editor cannot show: kept as stored. out["static_profile"] = keep["static_profile"] else: raise DeriveError(f"unknown contract {kind!r}", "unknown", "ui.contract") return out def _price_of(m, name): value = _num(m.get(name)) if value is None: raise DeriveError(f"{name} needs a price", "no_price", name) 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}", "unknown", "ui.region") 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.""" out, errors = ui_to_options_partial(model) if errors: raise errors[0] return out def ui_to_options_partial(model: dict): """(blob, [DeriveError]): every derivation that can be made, and the stored values (`keep`) for each one that cannot. For the checks only, so a step can be blocked by its own field and the rest still checked (UX-15 F1). Never saved: `ui_to_options()` raises instead.""" 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.")} errors = [] 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}) continue try: out.update(encode(model, keep)) except DeriveError as err: err.keys = keys errors.append(err) out.update({k: copy.deepcopy(keep[k]) for k in keys if k in keep}) return out, errors