goodwe-addon/goodwe_controller/tools/ux13_demo.py

"""UX-13: serve the real page and routes over a fixture controller, for a
headless browser check.  `python3 tools/ux13_demo.py STATE PORT [HOST]`

STATE is running, shadow, alarm, new (P4) or setup (UX-15's setup-only
mode, P5). The controller is the shipped `Controller`
with its readings set by hand (two batteries, solar, today's prices in the
mockup's shape, trading on). Nothing here talks to Home Assistant or to an
inverter: `/action` changes only this fixture. Used for the P1 render check
(no JS errors, tooltips, phone width, dark mode) described in the PR.

P2: the fixture Supervisor serves config.yaml's defaults with this site's
options, a schema read off config.yaml the way the Supervisor renders it
(`schema_ui()`) and translations/en.yaml, and keeps what the settings view
saves in memory, so the Installer view can be clicked through end to end.

P4: STATE `new` is a new install (config.yaml's defaults with one power
request entity, an empty Store made at this boot, control never started), so the page opens
the setup wizard by itself. A one-entity read of sensor.grid steps through
KETTLE_READS, so the kettle test sees the kettle go on (up) and then a
reversed meter (down).
"""

import asyncio
import io
import json
import os
import re
import sys
from datetime import datetime, timedelta, timezone

HERE = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, HERE)
sys.path.insert(0, os.path.join(HERE, "tools"))

from aiohttp import web as aioweb                                # noqa: E402

from app import homeview, web                                    # noqa: E402
from app.main import Controller                                  # noqa: E402
from app.setuponly import SetupOnly                              # noqa: E402
from app.peakguard import PeakDecision                           # noqa: E402
from app.price import PricePoint, PriceProduct, PriceRead        # noqa: E402
from app.static_tariff import TariffPlan                         # noqa: E402
import translations_lint as TL                                   # noqa: E402

PRICES = [.24, .22, .21, .20, .21, .22, .24, .27, .26, .22, .18, .16, .15,
          .14, .16, .19, .23, .28, .33, .36, .34, .29, .26, .24]


class Store(dict):
    fresh = False           # app/store.py: no state file at this boot

    def set(self, k, v):
        self[k] = v

    def set_time(self, k, v):
        self[k] = v.isoformat()

    def get_time(self, k):
        v = self.get(k)
        return None if v is None else datetime.fromisoformat(v)

    @property
    def data(self):
        return self


class Pub:
    enabled = False

    def publish(self, _v):
        pass

    def close(self):
        pass


class Hass:
    session = None
    token = ""

    async def limits(self, _e):
        return (-5000.0, 5000.0, 1.0)

    async def number(self, _e, invert=False):
        return None

    async def state(self, e):
        s = next((dict(x) for x in STATES if x["entity_id"] == e), None)
        if s and e == "sensor.grid":
            n = KETTLE_COUNT[0] = KETTLE_COUNT[0] + 1
            s["state"] = str(KETTLE_READS[(n - 1) % len(KETTLE_READS)])
        return s

    async def set_number(self, _e, _v):
        return True

    async def call(self, *_a):
        return True

    async def states(self):
        return STATES


KETTLE_READS = (-10, 1990, 1990, -10)
KETTLE_COUNT = [0]


def _state(eid, name, state, unit=None, cls=None):
    return {"entity_id": eid, "state": state,
            "attributes": {"friendly_name": name, "unit_of_measurement": unit,
                           "device_class": cls}}


STATES = [
    _state("sensor.grid", "Grid power", "-10", "W", "power"),
    _state("sensor.grid_kw", "Grid power in kW", "-0.01", "kW", "power"),
    _state("sensor.pv", "Solar power", "3900", "W", "power"),
    _state("sensor.soc_a", "Battery A level", "62", "%", "battery"),
    _state("sensor.soc_b", "Battery B level", "35", "%", "battery"),
    _state("sensor.batt_a", "Battery A power", "-2500", "W", "power"),
    _state("sensor.batt_b", "Battery B power", "-300", "W", "power"),
    _state("number.a", "Battery A power request", "0", "W"),
    _state("number.b", "Battery B power request", "0", "W"),
    _state("sensor.peak_forecast", "Quarter-hour peak forecast", "1240", "W",
           "power"),
]


# --- config.yaml as the Supervisor renders it (App.schema_ui) --------------- #
def _node(name, spec):
    spec = spec.strip()
    optional = spec.endswith("?")
    spec = spec.rstrip("?")
    out = {"name": name, "required": not optional, "optional": optional}
    m = re.match(r"(int|float)\(([^,]*),([^)]*)\)$", spec)
    if m:
        out.update(type="integer" if m.group(1) == "int" else "float",
                   lengthMin=float(m.group(2)), lengthMax=float(m.group(3)))
    elif spec.startswith("list("):
        out.update(type="select", options=spec[5:-1].split("|"))
    elif spec in ("bool",):
        out["type"] = "boolean"
    elif spec in ("int", "float"):
        out["type"] = "integer" if spec == "int" else "float"
    else:
        out["type"] = "string"
    return out


def schema_ui(text):
    """config.yaml's `schema:` block in the Supervisor's schema_ui shape: the
    fixture for tests and this demo, which cannot run the Supervisor."""
    block = text[text.index("\nschema:"):].splitlines()[1:]
    out, i = [], 0
    while i < len(block):
        line = block[i]
        m = re.match(r"^  ([a-z_0-9]+):\s*(\S.*?)?\s*$", line)
        i += 1
        if not m:
            continue
        name, spec = m.group(1), m.group(2)
        if spec:
            out.append(_node(name, spec))
            continue
        items = []
        while i < len(block) and (block[i].startswith("    ") or not block[i].strip()):
            if block[i].strip() and not block[i].strip().startswith("#"):
                items.append(block[i].strip())
            i += 1
        if items and items[0].startswith("- ") and ":" in items[0]:
            fields = [items[0][2:]] + items[1:]
            out.append({"name": name, "type": "schema", "optional": True,
                        "multiple": True,
                        "schema": [_node(*f.split(":", 1)) for f in fields]})
        else:
            node = _node(name, items[0][2:] if items else "str")
            node.update(multiple=True, required=False, optional=True)
            out.append(node)
    return out


def defaults(text):
    """config.yaml's `options:` block (as test_uimodel.py reads it)."""
    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


CONFIG = io.open(os.path.join(HERE, "config.yaml"), encoding="utf-8").read()


class Supervisor:
    """The stored options, kept in memory: a save here writes nothing else."""

    def __init__(self, options):
        self.options = options
        self.writes = 0

    async def info(self):
        return {"options": json.loads(json.dumps(self.options)),
                "schema": schema_ui(CONFIG),
                "translations": {lang: TL.parse_translation_file(
                    os.path.join(HERE, "translations", f"{lang}.yaml"))
                    for lang in ("en", "nl")}}

    async def write_options(self, options):
        self.options = json.loads(json.dumps(options))
        self.writes += 1
        print(f"options saved ({self.writes})", flush=True)

    async def restart(self, confirm):
        print(f"restart asked ({confirm!r})", flush=True)


OPTS = {"meter_entity": "sensor.grid", "meter_source": "off",
        "pv_entity": "sensor.pv", "maintenance_enabled": True,
        "maintenance_start_hour": 18, "arbitrage_enabled": True,
        "cycle_cost_threshold_eur": 0.05,
        "inverters": [
            {"name": "Battery A", "setpoint_entity": "number.a",
             "soc_entity": "sensor.soc_a", "batt_entity": "sensor.batt_a",
             "max_w": 2500, "capacity_kwh": 9.6},
            {"name": "Battery B", "setpoint_entity": "number.b",
             "soc_entity": "sensor.soc_b", "batt_entity": "sensor.batt_b",
             "max_w": 2500, "capacity_kwh": 5.4}]}


def controller(state):
    opts = dict(OPTS, shadow_mode="on" if state == "shadow" else "off")
    c = Controller(opts, Hass(), Store(), Pub(), stop=asyncio.Event())
    c.auto = True
    now = datetime.now(timezone.utc)
    day0, _day1 = homeview.local_day(now)
    pts = tuple(PricePoint(day0 + timedelta(days=d, hours=k), p, False)
                for d in range(2) for k, p in enumerate(PRICES))
    c.tariff_plan = TariffPlan(PriceRead.series(
        pts, PriceProduct.DAY_AHEAD, "generic_dayahead", now), None,
        "provider")
    c.tariff_at = now
    c.peak_guard = PeakDecision(engaged=False, floor_w=None,
                                ceiling_w=3500.0, ceiling_from="cap",
                                projected_w=1240.0)
    for inv, soc, batt in zip(c.inverters, (62.0, 35.0), (-2500.0, -300.0)):
        inv.soc, inv.batt = soc, (0.0 if state == "shadow" else batt)
        inv.dev_min, inv.dev_max = -5000.0, 5000.0
        if state != "running":          # 0 W accepted: Home says "Held"
            inv.last_write = 0.0
    c.grid = -10.0 if state != "shadow" else -2810.0
    c.pv_w = 3900.0
    if state == "alarm":
        c.inverters[0].sign_wd.tripped = True
    return c


class ReadOnlyHass(Hass):
    read_only = True        # app/setuponly.py serves only on a read-only HA


async def main(state, port, host="127.0.0.1"):
    if state == "setup":
        # UX-15's setup-only mode: shipped defaults, nothing controlled.
        stored = defaults(CONFIG)
        c = SetupOnly(dict(stored), ReadOnlyHass(), Store())
        runner = aioweb.AppRunner(web.build_app(c, Supervisor(stored)))
        await runner.setup()
        await aioweb.TCPSite(runner, host, port).start()
        print(f"serving setup on http://127.0.0.1:{port}/", flush=True)
        await asyncio.Event().wait()
    if state == "new":
        stored = dict(defaults(CONFIG), setpoint_entity="number.a")
        store = Store()
        store.fresh = True          # a new /data: the wizard opens by itself
        c = Controller(dict(stored), Hass(), store, Pub(),
                       stop=asyncio.Event())
        c.auto = False
        runner = aioweb.AppRunner(web.build_app(c, Supervisor(stored)))
        await runner.setup()
        await aioweb.TCPSite(runner, host, port).start()
        print(f"serving new on http://127.0.0.1:{port}/", flush=True)
        await asyncio.Event().wait()
    c = controller(state)
    stored = dict(defaults(CONFIG), **OPTS)
    stored.update(setpoint_entity="", meter_source="ha_signed",
                  p1_net_entity="sensor.grid", price_provider="generic_dayahead",
                  peak_forecast_entity="sensor.peak_forecast",
                  shadow_mode=c.o.get("shadow_mode", "off"),
                  inverters=[dict(i, bms_soc_floor=10)
                             for i in OPTS["inverters"]])
    runner = aioweb.AppRunner(web.build_app(c, Supervisor(stored)))
    await runner.setup()
    await aioweb.TCPSite(runner, host, port).start()
    print(f"serving {state} on http://127.0.0.1:{port}/", flush=True)
    await asyncio.Event().wait()


if __name__ == "__main__":
    asyncio.run(main(sys.argv[1], int(sys.argv[2]), *sys.argv[3:4]))