"""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). P3 adds the battery-mode states in `P3`: modes (on, B on Leftover solar), off (B switched off), off_peak (B off but giving for the peak limit), refusal (A refuses to give power) and maint (A in its monthly cycle). 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, packmode, web # noqa: E402 from app.main import Controller # noqa: E402 from app.maintenance import DRAIN # 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 == "password": out.update(type="string", format="password") 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}]} # UX-13 P3: the battery-mode states, each with battery modes on. P3 = ("modes", "off", "off_peak", "refusal", "maint") def controller(state): opts = dict(OPTS, shadow_mode="on" if state == "shadow" else "off") if state in P3: opts["battery_modes_enabled"] = True 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 if state in P3: a, b = c.inverters b.store_set(packmode.MODE_KEY, packmode.LEFTOVER_SOLAR) if state in ("off", "off_peak"): b.store_set(packmode.USE_KEY, False) b.batt = 0.0 if state == "off_peak": # A cloudy evening: the house takes 6 kW and the peak limit needs # both batteries, so B gives 0.5 kW although it is switched off. c.pv_w, c.grid = 0.0, 3490.0 a.batt, b.batt = 2000.0, 510.0 c.peak_guard = PeakDecision(engaged=True, floor_w=2500.0, ceiling_w=3500.0, ceiling_from="cap", projected_w=3600.0) if state == "refusal": # A is at its cut-out level and refuses: asked 0 W, B takes over. c.pv_w, c.grid = 0.0, 5.0 a.soc, a.batt, a.refused = 9.0, -4.0, 1 b.batt = 1210.0 if state == "maint": c.maints[0].store.set("phase", DRAIN) # MAINT-05: A's own key c.pv_w, c.grid = 0.0, 0.0 a.batt, b.batt = 2000.0, 0.0 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"), battery_modes_enabled=state in P3, 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]))