goodwe-addon/goodwe_controller/app/uimodel.py

"""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", "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.<code>`), and `field` the
    page field it is shown beside."""

    def __init__(self, text, code="", field=""):
        super().__init__(text)
        self.code, self.field = code, field


# 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


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",
                              "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":
        if "static_profile" not in keep:
            raise DeriveError("odd tariff windows are edited in the "
                              "Configuration tab", "custom_profile",
                              "ui.contract")
        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."""
    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