Claude
2d2b6afb07
- Batteries: one card per battery with On/Off, the mode list (cheap hours and selling greyed out until OPT-11 P3), level, and Now / Today / Why lines built in homeview.py. Replaces the OPT-11 P1 minimal cards and uses the existing pack_use / pack_mode actions and store keys. - An Off battery the peak limit still needs says so with its figure; no line shows 0 W beside moving power (OPT-11 P1 test F2). Today says "Not planned yet" until OPT-11 P3/P4. - Home's plan leaves Off batteries out, except for the peak limit. - Settings opens on Simple: the layout's simple fields with an applies chip and a note, plus the battery switches and modes. Installer is one tap. - English and Dutch words in app/ui_text; test_batteries_screen.py; browser check tools/ux13_browser_check.py (CHECK=1 ux13_screenshots.sh); new and changed screenshots only. No control, allocator, arbiter or command module changed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UBtqPrG1mDLXzTEBnhEkkV
337 lines
13 KiB
Python
337 lines
13 KiB
Python
"""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 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]))
|