glenn schrooyenandClaude
17c6048417
With 15-minute prices the chart now has one bar per quarter (92/96/100), not hour averages; hourly and static tariffs still draw one bar per hour. Cheap and dear marks come from the kept priceruns result (battery modes on) or the arbitrage verdict's two points (off), not a thirds split. A missing quarter is an empty bar. Removes plan.average_tip (EN/NL). Co-Authored-By: Claude <noreply@anthropic.com>
802 lines
38 KiB
Python
802 lines
38 KiB
Python
"""UX-13 P1: the Home screen's `/status` and the page shell. `python3 -B test_home.py`
|
|
|
|
1. `/status` keeps every old field (`banner`, `level`, `rows`, `checks`,
|
|
`hint`, `events`) AND carries the structured half the Home screen reads:
|
|
`state`, `now`, `batteries`, `plan`, `attention`, `month_peak`, `control`.
|
|
Both are pinned here, by name and shape.
|
|
2. The plan shows only what the add-on does today. Trading on price is the
|
|
verdict the control cycle would reach (the same `plan.arbitrage()` call),
|
|
drawn where the cheapest and dearest prices are, with kWh and € per block.
|
|
The monthly cycle, the peak limit and "follows the house" are the rest.
|
|
3. Shadow mode says "would", a stopped site plans nothing, and a red row is an
|
|
alarm under "Needs attention".
|
|
4. A fault in the Home view never costs the page its old rows.
|
|
5. The page: served at `/` from app/static, no external file, relative URLs
|
|
only, every label resolves in app/ui_text, and the region year reaches
|
|
`config/model` for the P2 Settings screens.
|
|
|
|
The STYLE scan of every Home sentence is test_messages.py part 8.
|
|
Mutating a module this file grades? Read MUTATION.md first (purge __pycache__).
|
|
"""
|
|
|
|
import asyncio
|
|
import io
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from aiohttp import ClientSession, web as aioweb
|
|
|
|
from app import homeview, web
|
|
from app.main import Controller
|
|
from app.peakguard import PeakDecision
|
|
from app.price import PricePoint, PriceProduct, PriceRead
|
|
from app.static_tariff import TariffPlan
|
|
from app.uimodel import REGION_PRESETS
|
|
from named_absence import field, nth, present
|
|
|
|
FLANDERS = REGION_PRESETS["flanders"]["taxes_eur_kwh"]
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
fails = []
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Fixtures: a real Controller, with its readings set directly.
|
|
# --------------------------------------------------------------------------- #
|
|
class _Store:
|
|
def __init__(self):
|
|
self.data = {}
|
|
|
|
def set(self, k, v):
|
|
self.data[k] = v
|
|
|
|
def get(self, k, d=None):
|
|
return self.data.get(k, d)
|
|
|
|
def set_time(self, k, v):
|
|
self.data[k] = v.isoformat()
|
|
|
|
def get_time(self, k):
|
|
v = self.data.get(k)
|
|
return None if v is None else datetime.fromisoformat(v)
|
|
|
|
|
|
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):
|
|
return None
|
|
|
|
async def set_number(self, _e, _v):
|
|
return True
|
|
|
|
async def call(self, *_a):
|
|
return True
|
|
|
|
async def states(self):
|
|
return []
|
|
|
|
|
|
ONE = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
|
|
"soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
|
|
"meter_source": "off", "max_w": 5000, "shadow_mode": "off",
|
|
"maintenance_enabled": False}
|
|
TWO = dict(ONE, 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}])
|
|
|
|
# Sunday 27 September 2026, 11:40 in Brussels (09:40 UTC): the mockup's day.
|
|
NOW = datetime(2026, 9, 27, 9, 40, tzinfo=timezone.utc)
|
|
DAY0 = datetime(2026, 9, 26, 22, 0, tzinfo=timezone.utc)
|
|
MOCK = [.24, .22, .21, .20, .21, .22, .24, .27, .26, .22, .18, .16, .15, .14,
|
|
.16, .19, .23, .28, .33, .36, .34, .29, .26, .24]
|
|
|
|
|
|
def make(opts, *, soc=(62.0, 35.0), batt=(-2500.0, -300.0), grid=10.0,
|
|
auto=True, wrote=0.0):
|
|
"""`wrote` is each inverter's last accepted power request (None: none yet)."""
|
|
c = Controller(dict(opts), _Hass(), _Store(), _Pub(), stop=asyncio.Event())
|
|
c.auto = auto
|
|
for i, inv in enumerate(c.inverters):
|
|
inv.soc, inv.batt = soc[i], batt[i]
|
|
inv.last_write = wrote
|
|
inv.dev_min, inv.dev_max = -5000.0, 5000.0
|
|
c.grid = grid
|
|
return c
|
|
|
|
|
|
def provider_plan(prices=MOCK, day0=DAY0, days=2, provisional_from=None):
|
|
pts = []
|
|
for d in range(days):
|
|
for k, p in enumerate(prices):
|
|
start = day0 + timedelta(days=d, hours=k)
|
|
pts.append(PricePoint(start, p, provisional_from is not None
|
|
and start >= provisional_from))
|
|
return TariffPlan(PriceRead.series(tuple(pts), PriceProduct.DAY_AHEAD,
|
|
"generic_dayahead", NOW), None,
|
|
"provider")
|
|
|
|
|
|
def quiet_guard(projected_w=1240.0):
|
|
return PeakDecision(engaged=False, floor_w=None, ceiling_w=3500.0,
|
|
ceiling_from="cap", projected_w=projected_w)
|
|
|
|
|
|
def status(c, now=NOW):
|
|
s = c.status_payload()
|
|
# The status route builds at wall-clock time; rebuild the structured half
|
|
# at the fixture's instant so the day and the hours are the mockup's.
|
|
s.update(homeview.build(c, level=s["level"], checks=s["checks"], now=now))
|
|
return s
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("1. /status: the old fields stay, the structured half is added")
|
|
# --------------------------------------------------------------------------- #
|
|
c1 = make(ONE, soc=(62.0,), batt=(960.0,), grid=800.0)
|
|
s1 = c1.status_payload()
|
|
OLD = {"banner", "level", "rows", "checks", "hint", "events"}
|
|
NEW = {"state", "now", "batteries", "plan", "attention", "month_peak",
|
|
"control"}
|
|
check(f"every old field is still there, with its old type ({sorted(OLD)})",
|
|
OLD <= set(s1) and isinstance(s1["rows"], list)
|
|
and all(len(r) == 2 for r in s1["rows"])
|
|
and all({"ok", "warn", "text"} <= set(r) for r in s1["checks"])
|
|
and isinstance(s1["banner"], str) and s1["level"] in ("ok", "warn", "bad")
|
|
and isinstance(s1["events"], list))
|
|
check(f"...and the structured half is added beside them ({sorted(NEW)})",
|
|
NEW <= set(s1) and "home_error" not in s1)
|
|
check("...and it is JSON, as the route serves it", json.loads(json.dumps(s1))
|
|
== json.loads(json.dumps(s1)))
|
|
check("`state` names the kind, the pill word, the control row and a sub-line",
|
|
set(s1["state"]) == {"kind", "pill", "ran", "sub"}
|
|
and s1["state"]["kind"] == "running")
|
|
check("`now` carries the four tiles in W and a plain line for each",
|
|
{"solar_w", "house_w", "grid_w", "batt_w", "has_solar", "lines"}
|
|
<= set(s1["now"])
|
|
and set(s1["now"]["lines"]) == {"solar", "house", "grid", "batt"}
|
|
and s1["now"]["grid_w"] == 800.0 and s1["now"]["batt_w"] == 960.0)
|
|
_nh = homeview.now_block(grid_w=-3000.0, pv_w=1000.0, batt_w=1500.0,
|
|
has_solar=True, shadow=False)
|
|
_nz = homeview.now_block(grid_w=-1030.0, pv_w=1000.0, batt_w=0.0,
|
|
has_solar=True, shadow=False)
|
|
_np = homeview.now_block(grid_w=800.0, pv_w=1000.0, batt_w=-300.0,
|
|
has_solar=True, shadow=False)
|
|
check("House = grid (+ import) + solar + battery (+ giving): 800 + 1000 - 300 "
|
|
"= 1500 W",
|
|
_np["house_w"] == 1500.0 and _np["lines"]["house"] == "Using")
|
|
check("...a sum below zero is not shown as a positive figure: 0 W inside "
|
|
"the idle band, and no figure beyond it",
|
|
_nz["house_w"] == 0.0 and _nh["house_w"] is None
|
|
and _nh["lines"]["house"] == homeview.tile_house(None, True, True))
|
|
check("...and with no solar reading the house figure is NOT invented: it is "
|
|
"None and the tile says why",
|
|
s1["now"]["house_w"] is None and s1["now"]["has_solar"] is False
|
|
and s1["now"]["lines"]["house"] == homeview.tile_house(None, False))
|
|
_b = s1["batteries"][0]
|
|
check("`batteries[]` has level, power, state, mode, the now line and why",
|
|
{"name", "level_pct", "power_w", "state", "mode", "now", "why",
|
|
"capacity_kwh", "stored_kwh", "maintenance"} <= set(_b)
|
|
and _b["level_pct"] == 62.0 and _b["state"] == "giving"
|
|
and _b["now"] == "Giving 1.0 kW now.")
|
|
_p = s1["plan"]
|
|
check("`plan` carries the verdict, the prices, the runs, the lines and "
|
|
"the region",
|
|
{"verdict", "head", "prices", "runs", "lines", "now_h", "hours",
|
|
"price_line", "tomorrow_line", "region", "region_line"} <= set(_p)
|
|
and len(_p["prices"]) in (23, 24, 25))
|
|
check("`attention[]` has severity, title, body and docs; `month_peak` and "
|
|
"`control` are there",
|
|
all({"severity", "title", "body", "docs"} <= set(a)
|
|
for a in s1["attention"])
|
|
and {"month", "limit_w", "peak_w", "pct", "period_w", "state", "line"}
|
|
<= set(s1["month_peak"])
|
|
and {"running", "shadow", "text", "maintenance", "docs_url"}
|
|
<= set(s1["control"]))
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n2. the plan is what the add-on actually does")
|
|
# --------------------------------------------------------------------------- #
|
|
s_house = status(make(TWO))
|
|
check("with trading off and no cycle due, each battery follows the house "
|
|
"from now to the end of the day, and nothing else is drawn",
|
|
[r["kind"] for r in s_house["plan"]["runs"]] == ["house", "house"]
|
|
and s_house["plan"]["verdict_key"] == "house"
|
|
and all(r["end_h"] == 24.0 and abs(r["start_h"] - 11 - 40 / 60) < 0.01
|
|
for r in s_house["plan"]["runs"])
|
|
and s_house["plan"]["runs"][0]["to"] == "24:00")
|
|
|
|
ARB = dict(TWO, arbitrage_enabled=True, cycle_cost_threshold_eur=0.05)
|
|
c_arb = make(ARB)
|
|
c_arb.tariff_plan, c_arb.tariff_at = provider_plan(), NOW
|
|
c_arb.peak_guard = quiet_guard()
|
|
s_arb = status(c_arb)
|
|
_runs = s_arb["plan"]["runs"]
|
|
_grid = [r for r in _runs if r["kind"] == "grid"]
|
|
_give = [r for r in _runs if r["kind"] == "give"]
|
|
check("trading on price: a charge block at the cheapest hour (13:00) and a "
|
|
"give block at the dearest (19:00), on each battery",
|
|
[(r["battery"], r["from"], r["to"]) for r in _grid]
|
|
== [(0, "13:00", "14:00"), (1, "13:00", "14:00")]
|
|
and [(r["battery"], r["from"], r["to"]) for r in _give]
|
|
== [(0, "19:00", "20:00"), (1, "19:00", "20:00")]
|
|
and s_arb["plan"]["verdict_key"] == "trade")
|
|
check("...placed where the control cycle's own verdict puts them: the same "
|
|
"plan.arbitrage() call on the same plan",
|
|
c_arb.fresh_tariff_plan(NOW).arbitrage(
|
|
peak_guard_active=False, min_spread_eur_kwh=0.05,
|
|
uncertainty=frozenset({__import__("app.price", fromlist=["x"])
|
|
.Uncertainty.NO_WEATHER}))
|
|
.match(allowed=lambda v: (v.cheapest.start, v.dearest.start),
|
|
refused=lambda v: None)
|
|
== (DAY0 + timedelta(hours=13), DAY0 + timedelta(hours=19)))
|
|
# Battery A: 2.5 kW for 1 h = 2.5 kWh; room to the 95 % ceiling is 33 % of
|
|
# 9.6 kWh = 3.17 kWh, so 2.5. Battery B: 60 % of 5.4 = 3.24, so 2.5.
|
|
check("each block carries kWh and €: charging is capped by the battery's own "
|
|
"rating and the room its level leaves, € = kWh x that hour's price",
|
|
[r["kwh"] for r in _grid] == [2.5, 2.5]
|
|
and [r["eur"] for r in _grid] == [0.35, 0.35]
|
|
and [r["eur_kwh"] for r in _grid] == [0.14, 0.14]
|
|
and "2.5 kWh" in _grid[0]["what"] and "€0.14" in _grid[0]["why"]
|
|
and "€0.35" in _grid[0]["amount"])
|
|
# Giving at 19:00 comes after the 13:00 charge, so each battery may give what
|
|
# it will hold then: B is at 35 % + 2.5 kWh of 5.4 = 81 %, and 81 - 20 = 61 %
|
|
# of 5.4 = 3.3 kWh, so 2.5 (an upper bound, like every amount here).
|
|
check("...and giving is capped by what each battery holds above arbitrage's "
|
|
"floor after the charge before it",
|
|
[r["kwh"] for r in _give] == [2.5, 2.5]
|
|
and [r["eur"] for r in _give] == [0.9, 0.9])
|
|
|
|
|
|
def trade(soc, prices=MOCK, provisional_from=None, now=NOW, opts=ARB):
|
|
c = make(opts, soc=soc)
|
|
c.tariff_plan = provider_plan(prices, provisional_from=provisional_from)
|
|
c.tariff_at, c.peak_guard = now, quiet_guard()
|
|
s = status(c, now=now)
|
|
return ([(r["kind"], r["battery"], r["kwh"]) for r in s["plan"]["runs"]
|
|
if r["kind"] in ("grid", "give")], s["plan"]["lines"])
|
|
|
|
|
|
# Review of #144: only what arbitrage.decide() would still do is drawn.
|
|
_t, _l = trade((96.0, 35.0))
|
|
check("a battery at arbitrage's ceiling gets no charge block, not "
|
|
"'Charges up to 0.0 kWh'", ("grid", 0, 0.0) not in _t
|
|
and [k for k in _t if k[0] == "grid"] == [("grid", 1, 2.5)])
|
|
# 20:00 local: the dearest (19:00) is over; tomorrow's charge is a line. Put
|
|
# the dearest later today instead, with the emptiest pack at the floor.
|
|
_late = MOCK[:22] + [.40, .24]
|
|
_t, _l = trade((62.0, 20.0), prices=_late, now=NOW + timedelta(hours=8))
|
|
_t2, _l2 = trade((62.0, 35.0), prices=_late, now=NOW + timedelta(hours=8))
|
|
check("no give block while the emptiest battery is at arbitrage's floor: "
|
|
"the cycle's sale is graded on the emptiest pack (at 35 % both give)",
|
|
not [k for k in _t if k[0] == "give"]
|
|
and [k[:2] for k in _t2] == [("give", 0), ("give", 1)])
|
|
_t, _l = trade((62.0, 35.0), prices=[x - 0.5 for x in MOCK])
|
|
check("no give block at a negative price (the cycle refuses that sale)",
|
|
not [k for k in _t if k[0] == "give"]
|
|
and [k for k in _t if k[0] == "grid"])
|
|
# Running, a missing level is a red row and the site is held. In shadow mode
|
|
# the plan is still drawn, so this is where the level gate shows.
|
|
_t, _l = trade((None, 35.0), opts=dict(ARB, shadow_mode="on"))
|
|
check("with a battery level unknown the cycle does not trade, so nothing is "
|
|
"drawn and a line says why (shadow mode, which still plans)",
|
|
_t == [] and homeview.line_no_trade("no_level") in _l)
|
|
_t, _l = trade((62.0, 35.0), provisional_from=DAY0)
|
|
check("a plan with no published prices yet says so, not 'the price gap is "
|
|
"too small'",
|
|
_t == [] and homeview.line_no_trade("not_firm") in _l
|
|
and homeview.line_no_trade("not_worth") not in _l)
|
|
check("the price bars are today's 24 hours, from the real series, with the "
|
|
"cheap and dear hour marked",
|
|
[b["eur_kwh"] for b in s_arb["plan"]["prices"]] == MOCK
|
|
and s_arb["plan"]["prices"][13]["kind"] == "cheap"
|
|
and s_arb["plan"]["prices"][19]["kind"] == "dear"
|
|
and s_arb["plan"]["prices"][0]["kind"] == "mid"
|
|
and s_arb["plan"]["prices"][13]["label"] == "13:00")
|
|
check("the price line names the price now, today's lowest and highest, and "
|
|
"that tomorrow's prices are in",
|
|
s_arb["plan"]["price_line"] == ("Price now €0.16 per kWh, all in. "
|
|
"Today: lowest €0.14 at 13:00, "
|
|
"highest €0.36 at 19:00.")
|
|
and s_arb["plan"]["tomorrow_line"] == "Tomorrow's prices are in.")
|
|
|
|
c_late = make(ARB)
|
|
c_late.tariff_plan, c_late.tariff_at = provider_plan(), NOW
|
|
c_late.peak_guard = quiet_guard()
|
|
s_late = status(c_late, now=NOW + timedelta(hours=9)) # 20:40 local
|
|
check("a block whose hour is over is not drawn: at 20:40 no trade is left "
|
|
"today",
|
|
not [r for r in s_late["plan"]["runs"] if r["kind"] in ("grid", "give")])
|
|
|
|
c_peak = make(ARB)
|
|
c_peak.tariff_plan, c_peak.tariff_at = provider_plan(), NOW
|
|
c_peak.peak_guard = PeakDecision(engaged=True, floor_w=800.0,
|
|
ceiling_w=3500.0, ceiling_from="cap",
|
|
projected_w=3900.0)
|
|
s_peak = status(c_peak)
|
|
check("while the peak limit holds, the verdict refuses as the cycle does: no "
|
|
"trade block, a plain line, and a peak block for this 15-minute period",
|
|
not [r for r in s_peak["plan"]["runs"] if r["kind"] in ("grid", "give")]
|
|
and homeview.line_no_trade("peak") in s_peak["plan"]["lines"]
|
|
and [(r["from"], r["to"]) for r in s_peak["plan"]["runs"]
|
|
if r["kind"] == "peak"] == [("11:30", "11:45"), ("11:30", "11:45")])
|
|
|
|
c_pk = make(ARB)
|
|
c_pk.p1_enabled = True # the peak line is shown where P1 runs
|
|
c_pk.tariff_plan, c_pk.tariff_at = provider_plan(), NOW
|
|
c_pk.peak_guard = PeakDecision(engaged=False, floor_w=None, ceiling_w=4200.0,
|
|
ceiling_from="month_peak", projected_w=1240.0)
|
|
s_pk = status(c_pk)
|
|
check("the plan's peak line names the guard's ceiling, which is this month's "
|
|
"peak once it is above the limit",
|
|
homeview.line_peak_limit(4200.0, False) in s_pk["plan"]["lines"]
|
|
and s_pk["month_peak"]["line"].endswith("above 4.2 kW."))
|
|
|
|
c_blind = make(ARB)
|
|
c_blind.tariff_plan, c_blind.tariff_at = provider_plan(), NOW
|
|
s_blind = status(c_blind) # meter_source off: the guard never runs
|
|
check("with nothing measuring the 15-minute peak, trading is NOT drawn, as "
|
|
"the cycle refuses it (guard state None)",
|
|
not [r for r in s_blind["plan"]["runs"] if r["kind"] in ("grid", "give")]
|
|
and homeview.line_no_trade("not_covered") in s_blind["plan"]["lines"])
|
|
|
|
c_flat = make(ARB)
|
|
c_flat.tariff_plan, c_flat.tariff_at = provider_plan(prices=[.2] * 24), NOW
|
|
c_flat.peak_guard = quiet_guard()
|
|
s_flat = status(c_flat)
|
|
check("a flat day: no trade, no cheap or expensive hour marked",
|
|
not [r for r in s_flat["plan"]["runs"] if r["kind"] in ("grid", "give")]
|
|
and {b["kind"] for b in s_flat["plan"]["prices"]} == {"mid"}
|
|
and homeview.line_no_trade("not_worth") in s_flat["plan"]["lines"])
|
|
|
|
c_old = make(ARB)
|
|
c_old.tariff_plan = provider_plan()
|
|
c_old.tariff_at = NOW - timedelta(hours=3) # past its max age
|
|
c_old.peak_guard = quiet_guard()
|
|
s_old = status(c_old)
|
|
check("a stale plan is not traded on by the cycle, so it is not drawn",
|
|
not [r for r in s_old["plan"]["runs"] if r["kind"] in ("grid", "give")])
|
|
|
|
MAINT = dict(TWO, maintenance_enabled=True, maintenance_start_hour=12)
|
|
c_m = make(MAINT)
|
|
s_m = status(c_m)
|
|
_cyc = [r for r in s_m["plan"]["runs"] if r["kind"] == "cycle"]
|
|
check("a monthly cycle due today is drawn on ONE battery, at the start hour "
|
|
"the cycle uses (12 UTC is 14:00 here), and the other one waits",
|
|
len(_cyc) == 1 and field(nth(_cyc), "from") == "14:00"
|
|
and s_m["plan"]["verdict_key"] == "cycle"
|
|
and any("waits its turn" in x for x in s_m["plan"]["lines"]))
|
|
check("...with its duration from the battery's capacity and the cycle's rates "
|
|
"and hold time",
|
|
present(nth(_cyc)) and _cyc[0]["amount"].startswith("Takes about ")
|
|
and _cyc[0]["end_h"] > _cyc[0]["start_h"] + 2)
|
|
# MAINT-10: with the site's zone set, the hour is read on the local clock.
|
|
from app.tariff import _EUTimezone # noqa: E402
|
|
c_ml = make(dict(MAINT, maintenance_start_hour=14))
|
|
for _mm in c_ml.maints:
|
|
_mm.tz = _EUTimezone()
|
|
_cyc_l = [r for r in status(c_ml)["plan"]["runs"] if r["kind"] == "cycle"]
|
|
check("MAINT-10: start hour 14 in the local zone is drawn at 14:00, the "
|
|
"instant the cycle starts", len(_cyc_l) == 1
|
|
and field(nth(_cyc_l), "from") == "14:00")
|
|
c_m.maints[0].store.set("phase", "drain")
|
|
s_md = status(c_m)
|
|
check("a cycle under way is drawn from now, and the battery's now line says "
|
|
"so",
|
|
[r["from"] for r in s_md["plan"]["runs"] if r["kind"] == "cycle"]
|
|
== ["11:40"]
|
|
and s_md["batteries"][0]["now"] == "Monthly cycle: emptying."
|
|
and any("Battery B waits its turn" in x for x in s_md["plan"]["lines"]))
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n3. shadow mode, a stopped site, an alarm")
|
|
# --------------------------------------------------------------------------- #
|
|
c_sh = make(dict(ARB, shadow_mode="on"))
|
|
c_sh.tariff_plan, c_sh.tariff_at = provider_plan(), NOW
|
|
c_sh.peak_guard = quiet_guard()
|
|
s_sh = status(c_sh)
|
|
_texts = [r["what"] for r in s_sh["plan"]["runs"]] + [s_sh["plan"]["head"]]
|
|
check("shadow: the state is shadow, the battery tile is held at 0 W, and "
|
|
"every block and the head say 'would'",
|
|
s_sh["state"]["kind"] == "shadow"
|
|
and s_sh["now"]["lines"]["batt"] == "Held at 0 W"
|
|
and all("ould" in x for x in _texts)
|
|
and s_sh["plan"]["head"].startswith("Shadow mode:"))
|
|
check("...and 'Shadow mode is on' is the first item under Needs attention",
|
|
s_sh["attention"][0]["title"] == homeview.ATTN_SHADOW
|
|
and s_sh["attention"][0]["severity"] == "info")
|
|
check("...and each battery's now line says it is held at 0 W",
|
|
all(b["now"].startswith("Held at 0 W.") for b in s_sh["batteries"]))
|
|
|
|
s_stop = status(make(TWO, auto=False))
|
|
check("stopped: nothing is planned and the head says so",
|
|
s_stop["plan"]["runs"] == [] and s_stop["state"]["kind"] == "stopped"
|
|
and s_stop["plan"]["verdict_key"] == "stopped"
|
|
and s_stop["control"]["running"] is False
|
|
and all(b["now"] == "Held at 0 W while control is stopped."
|
|
for b in s_stop["batteries"]))
|
|
|
|
c_al = make(ONE, soc=(62.0,), batt=(0.0,))
|
|
c_al.inverters[0].sign_wd.tripped = True
|
|
s_al = c_al.status_payload()
|
|
_red = [a for a in s_al["attention"] if a["severity"] == "alarm"]
|
|
check("an alarm: level bad, state alarm, and the red row is the first item "
|
|
"under Needs attention",
|
|
s_al["level"] == "bad" and s_al["state"]["kind"] == "alarm"
|
|
and _red and s_al["attention"][0] is _red[0]
|
|
and _red[0]["title"] == homeview.ATTN_TITLES["direction"])
|
|
check("...and nothing is planned while every inverter is held at 0 W",
|
|
s_al["plan"]["runs"] == [] and s_al["plan"]["verdict_key"] == "held")
|
|
_rows = [r for r in s_al["checks"] if not r["ok"] and not r["warn"]]
|
|
check("...and every red, amber and settling row reaches Needs attention with "
|
|
"its own sentence as the body, and no green row does",
|
|
[a["body"] for a in homeview.attention(s_al["checks"])]
|
|
== [r["text"] for r in s_al["checks"] if not r["ok"] and not r["warn"]]
|
|
+ [r["text"] for r in s_al["checks"] if not r["ok"] and r["warn"]]
|
|
+ [r["text"] for r in s_al["checks"] if r["ok"] and r["warn"]])
|
|
|
|
# UX-13 P1 test finding (a): the page said "Held at 0 W" about 15 s before
|
|
# the inverter's power request reached 0 W. Until the last accepted request
|
|
# is 0 W it says "Going to 0 W".
|
|
c_al.inverters[0].last_write = 20.0
|
|
s_go = homeview.build(c_al, level="bad", checks=s_al["checks"], now=NOW)
|
|
c_al.inverters[0].last_write = 0.0
|
|
s_at = homeview.build(c_al, level="bad", checks=s_al["checks"], now=NOW)
|
|
check("an alarm while the last power request is still 20 W says 'Going to "
|
|
"0 W', and 'Held at 0 W' once 0 W was accepted",
|
|
s_go["plan"]["verdict"] == "Going to 0 W"
|
|
and s_go["batteries"][0]["now"] == "Going to 0 W until the alarm clears."
|
|
and s_go["plan"]["head"].startswith("Every inverter goes to 0 W")
|
|
and s_at["plan"]["verdict"] == "Held at 0 W"
|
|
and s_at["batteries"][0]["now"] == "Held at 0 W until the alarm clears.")
|
|
c_sp = make(TWO, auto=False)
|
|
c_sp.inverters[1].last_write = -1500.0
|
|
s_sp = status(c_sp)
|
|
s_new = status(make(TWO, auto=False, wrote=None))
|
|
check("...and before any power request was accepted (just started, or every "
|
|
"write refused) it says 'Going to 0 W', not 'Held'",
|
|
[b["now"] for b in s_new["batteries"]]
|
|
== ["Going to 0 W while control is stopped."] * 2)
|
|
check("...and a stop does the same, per battery",
|
|
[b["now"] for b in s_sp["batteries"]]
|
|
== ["Held at 0 W while control is stopped.",
|
|
"Going to 0 W while control is stopped."])
|
|
|
|
# Finding (b): a quarter-hourly feed. The block's price and the price line
|
|
# name the same 15-minute point, and nothing says "the list".
|
|
_q = []
|
|
for _k, _p in enumerate(MOCK):
|
|
for _m in range(4):
|
|
_q.append(PricePoint(DAY0 + timedelta(hours=_k, minutes=15 * _m),
|
|
round(_p + (0.0 if _m != 3 else -0.01), 4), False))
|
|
_q += [PricePoint(p.start + timedelta(days=1), p.eur_per_kwh, False)
|
|
for p in list(_q)]
|
|
c_q = make(ARB)
|
|
c_q.tariff_plan = TariffPlan(PriceRead.series(tuple(_q), PriceProduct.DAY_AHEAD,
|
|
"generic_dayahead", NOW), None,
|
|
"provider")
|
|
c_q.tariff_at = NOW
|
|
c_q.peak_guard = quiet_guard()
|
|
s_q = status(c_q)
|
|
_grid = [r for r in s_q["plan"]["runs"] if r["kind"] == "grid"]
|
|
_lo = min(_q[:96], key=lambda p: p.eur_per_kwh)
|
|
check("quarter-hour prices: the charge block, its price and the price line's "
|
|
"lowest are the same 15-minute period",
|
|
_grid and _grid[0]["from"] == homeview.hhmm(_lo.start)
|
|
and _grid[0]["why"] == f"Lowest published price, €{_lo.eur_per_kwh:.2f} "
|
|
"per kWh."
|
|
and f"Today: lowest €{_lo.eur_per_kwh:.2f} at "
|
|
f"{homeview.hhmm(_lo.start)}," in s_q["plan"]["price_line"]
|
|
and "the list" not in json.dumps(s_q["plan"]))
|
|
|
|
# UX-45: the chart draws the steps the add-on decides on.
|
|
_pq = s_q["plan"]["prices"]
|
|
check("UX-45: 96 quarter prices -> 96 bars, each one quarter, labelled "
|
|
"from its own start, with its own price and no average",
|
|
len(_pq) == 96 and _pq[0]["label"] == "00:00"
|
|
and _pq[0]["end_label"] == "00:15" and _pq[57]["label"] == "14:15"
|
|
and _pq[57]["end_label"] == "14:30"
|
|
and [b["eur_kwh"] for b in _pq] == [round(p.eur_per_kwh, 4)
|
|
for p in _q[:96]]
|
|
and all("points" not in b for b in _pq)
|
|
and s_q["plan"]["hours"] == 24)
|
|
check("...the quarters of one hour differ (the chart shows the steps)",
|
|
len({b["eur_kwh"] for b in _pq[52:56]}) > 1)
|
|
check("...and the page has no 'hour average' tip any more",
|
|
"average_tip" not in io.open(os.path.join(
|
|
HERE, "app", "static", "app.js"), encoding="utf-8").read()
|
|
and "average_tip" not in io.open(os.path.join(
|
|
HERE, "app", "ui_text", "en.json"), encoding="utf-8").read()
|
|
and "average_tip" not in io.open(os.path.join(
|
|
HERE, "app", "ui_text", "nl.json"), encoding="utf-8").read())
|
|
_cheap_q = [b["start"] for b in _pq if b["kind"] == "cheap"]
|
|
_dear_q = [b["start"] for b in _pq if b["kind"] == "dear"]
|
|
_hi = max(_q[:96], key=lambda p: p.eur_per_kwh)
|
|
check("...without battery modes the marks are the two quarters the "
|
|
"arbitrage verdict trades in, and no others",
|
|
_cheap_q == [_lo.start.isoformat()] and _dear_q == [_hi.start.isoformat()]
|
|
and sum(1 for b in _pq if b["kind"] == "mid") == 94)
|
|
|
|
# Battery modes on: the marks ARE the kept priceruns result.
|
|
from app import priceruns
|
|
_rp = priceruns.plan(tuple(_q), NOW, required_eur_kwh=0.05, charge_kwh=2.0,
|
|
charge_kw=4.0, sell_kwh=0.0, sell_kw=0.0)
|
|
c_m = make(ARB)
|
|
c_m.tariff_plan, c_m.tariff_at = c_q.tariff_plan, NOW
|
|
c_m.peak_guard = quiet_guard()
|
|
c_m.modes_on, c_m.mode_runs = True, _rp
|
|
_pm = status(c_m)["plan"]["prices"]
|
|
_today = [(a, z) for a, z, _p in priceruns.firm_slots(_q, NOW)
|
|
if a < DAY0 + timedelta(hours=24)]
|
|
_want_c = {a.isoformat() for a, z in _today
|
|
if _rp.cheap.start <= a < _rp.cheap.end}
|
|
_want_d = {a.isoformat() for g in _rp.give for a, z in _today
|
|
if g.start <= a < g.end}
|
|
check("UX-45: with battery modes on, the cheap and dear quarters are the "
|
|
"priceruns cheap run and give runs for the same input",
|
|
_rp.on and _rp.charge_useful() and _want_c and _want_d
|
|
and {b["start"] for b in _pm if b["kind"] == "cheap"} == _want_c
|
|
and {b["start"] for b in _pm if b["kind"] == "dear"} == _want_d
|
|
and len(_pm) == 96)
|
|
c_m2 = make(ARB)
|
|
c_m2.tariff_plan, c_m2.tariff_at = c_q.tariff_plan, NOW
|
|
c_m2.peak_guard = quiet_guard()
|
|
c_m2.modes_on = True
|
|
c_m2.mode_runs = priceruns.plan(tuple(_q), NOW, required_eur_kwh=5.0,
|
|
charge_kwh=2.0, charge_kw=4.0, sell_kwh=0.0,
|
|
sell_kw=0.0)
|
|
check("...a run plan that says no trade marks nothing",
|
|
not c_m2.mode_runs.on
|
|
and {b["kind"] for b in status(c_m2)["plan"]["prices"]} == {"mid"})
|
|
|
|
# A quarter that is over keeps the mark it had while it was ahead.
|
|
# (`status()` also builds at wall-clock time, which starts the day's memory
|
|
# over, so the two builds below go straight to `homeview.build`.)
|
|
_later = NOW + timedelta(hours=4)
|
|
_s0 = c_q.status_payload()
|
|
homeview.build(c_q, level=_s0["level"], checks=_s0["checks"], now=NOW)
|
|
c_q.tariff_at = _later
|
|
_pl = homeview.build(c_q, level=_s0["level"], checks=_s0["checks"],
|
|
now=_later)["plan"]["prices"]
|
|
check("...and a quarter already over keeps the mark it had while ahead",
|
|
[b["kind"] for b in _pl if b["start"] == _lo.start.isoformat()]
|
|
== ["cheap"]
|
|
and [b["kind"] for b in _pl if b["label"] == "19:00"] == ["dear"])
|
|
c_q.tariff_at = NOW
|
|
|
|
# A missing quarter is an empty bar in its place, nothing shifts.
|
|
_hole = [p for p in _q if p.start != DAY0 + timedelta(hours=10, minutes=30)]
|
|
c_h = make(ARB)
|
|
c_h.tariff_plan = TariffPlan(PriceRead.series(tuple(_hole), PriceProduct.DAY_AHEAD,
|
|
"generic_dayahead", NOW), None,
|
|
"provider")
|
|
c_h.tariff_at = NOW
|
|
c_h.peak_guard = quiet_guard()
|
|
_ph = status(c_h)["plan"]["prices"]
|
|
check("UX-45: a missing quarter is a 'none' bar, and the others stay where "
|
|
"they were",
|
|
len(_ph) == 96 and _ph[42]["eur_kwh"] is None
|
|
and _ph[42]["kind"] is None and _ph[42]["label"] == "10:30"
|
|
and _ph[41]["eur_kwh"] == _pq[41]["eur_kwh"]
|
|
and _ph[43]["eur_kwh"] == _pq[43]["eur_kwh"] and _ph[43]["label"] == "10:45"
|
|
and sum(1 for b in _ph if b["eur_kwh"] is None) == 1)
|
|
|
|
# A cheap run that is not worth charging in is not marked; an estimate is.
|
|
import types
|
|
_pts = [PricePoint(DAY0 + timedelta(minutes=15 * i),
|
|
.30 if i < 4 else .10, False) for i in range(8)]
|
|
_rp2 = priceruns.plan(tuple(_pts), DAY0, required_eur_kwh=0.05,
|
|
charge_kwh=0.5, charge_kw=4.0, sell_kwh=0.0, sell_kw=0.0)
|
|
_w2 = homeview.price_windows(types.SimpleNamespace(modes_on=True,
|
|
mode_runs=_rp2), None, None)
|
|
check("UX-45: a cheap run that comes after every dear slot is not worth a "
|
|
"charge, so it is not marked (the dear run is)",
|
|
_rp2.on and _rp2.cheap is not None and not _rp2.charge_useful()
|
|
and [w[0] for w in _w2] == ["dear"])
|
|
_est = [PricePoint(DAY0 + timedelta(minutes=15 * i), .2, False)
|
|
for i in range(80)] + [PricePoint(DAY0 + timedelta(hours=20), .25,
|
|
True)]
|
|
_be = homeview.price_bars(_est, DAY0, DAY0 + timedelta(hours=24))
|
|
check("UX-45: a carried (estimate) hourly price after quarter prices covers "
|
|
"its own hour as four estimate bars, and the rest stays empty",
|
|
[b["estimate"] for b in _be[78:85]]
|
|
== [False, False, True, True, True, True, False]
|
|
and [b["eur_kwh"] for b in _be[80:84]] == [.25] * 4
|
|
and _be[84]["eur_kwh"] is None)
|
|
|
|
# Hourly prices -> 24 bars (above); a day/night tariff -> 24 bars.
|
|
_ph1 = s_arb["plan"]["prices"]
|
|
check("UX-45: hourly prices -> 24 hour bars with the hour's own price",
|
|
len(_ph1) == 24 and _ph1[13]["end_label"] == "14:00"
|
|
and "points" not in _ph1[13])
|
|
c_dn = make(dict(ARB, static_profile=(
|
|
'{"periods": [{"name":"day","from":"07:00","to":"22:00","days":"all",'
|
|
'"eur_per_kwh":0.34},{"name":"night","from":"22:00","to":"07:00",'
|
|
'"days":"all","eur_per_kwh":0.26}]}')))
|
|
c_dn.peak_guard = quiet_guard()
|
|
_pd = status(c_dn)["plan"]["prices"]
|
|
check("UX-45: a day/night tariff -> 24 bars, 0.26 at night and 0.34 by day",
|
|
len(_pd) == 24
|
|
and [b["eur_kwh"] for b in _pd]
|
|
== [.26] * 7 + [.34] * 15 + [.26] * 2
|
|
and all(b["kind"] == "mid" for b in _pd))
|
|
|
|
# Clock-change days: 92 and 100 quarters.
|
|
_d0 = datetime(2026, 3, 28, 23, 0, tzinfo=timezone.utc)
|
|
_d1 = datetime(2026, 3, 29, 22, 0, tzinfo=timezone.utc)
|
|
_sp = [PricePoint(_d0 + timedelta(minutes=15 * i), .1, False)
|
|
for i in range(92)]
|
|
check("UX-45: a 23-hour day has 92 quarter bars, a 25-hour day 100",
|
|
len(homeview.price_bars(_sp, _d0, _d1)) == 92
|
|
and len(homeview.price_bars(
|
|
_sp, _d0, _d0 + timedelta(hours=25))) == 100)
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n4. a fault in the Home view never costs the page its rows")
|
|
# --------------------------------------------------------------------------- #
|
|
_real = homeview.build
|
|
try:
|
|
homeview.build = lambda *a, **k: 1 / 0
|
|
s_err = make(ONE, soc=(50.0,), batt=(0.0,)).status_payload()
|
|
finally:
|
|
homeview.build = _real
|
|
check("the old fields are served and `home_error` says the rest is missing",
|
|
OLD <= set(s_err) and s_err.get("home_error") is True
|
|
and not (NEW & set(s_err)))
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n5. the page")
|
|
# --------------------------------------------------------------------------- #
|
|
PAGE = web.page()
|
|
JS = io.open(os.path.join(HERE, "app", "static", "app.js"),
|
|
encoding="utf-8").read()
|
|
UI = json.load(io.open(os.path.join(HERE, "app", "ui_text", "en.json"),
|
|
encoding="utf-8"))["home"]
|
|
|
|
|
|
def resolve(path):
|
|
node = UI
|
|
for k in path.split("."):
|
|
if not isinstance(node, dict) or k not in node:
|
|
return None
|
|
node = node[k]
|
|
return node
|
|
|
|
|
|
_WORDS = json.loads(PAGE.split('id="ui-text">', 1)[1]
|
|
.split("</script>", 1)[0].replace("<\\/", "</"))
|
|
check("the page is assembled: no placeholder is left and the words parse, "
|
|
"English and Dutch (UX-13 P5)",
|
|
"/*APP_CSS*/" not in PAGE and "/*APP_JS*/" not in PAGE
|
|
and "/*UI_TEXT*/" not in PAGE
|
|
and _WORDS["en"]["home"] == UI and set(_WORDS) == {"en", "nl"})
|
|
check("four tabs with the four hash routes, in a keyboard tablist",
|
|
re.findall(r'data-hash="(\w+)"', PAGE) == ["home", "batteries",
|
|
"settings", "setup"]
|
|
and 'role="tablist"' in PAGE and PAGE.count('role="tabpanel"') == 4
|
|
and "ArrowRight" in JS and "ArrowLeft" in JS)
|
|
_ext = re.findall(r"""(?:https?:)?//[A-Za-z0-9.-]+\.[a-z]{2,}""", PAGE)
|
|
check(f"no external file and no web font: no URL leaves the add-on "
|
|
f"({_ext})", not _ext and "@import" not in PAGE
|
|
and "<link" not in PAGE and "src=" not in PAGE)
|
|
_abs = re.findall(r"""(?:fetch\(|href=)["'](/[^"']*)["']""", PAGE)
|
|
check(f"every fetch and link is relative, for ingress ({_abs})",
|
|
not _abs and 'fetch("status?lang=' in JS and 'fetch("action"' in JS)
|
|
_keys = set(re.findall(r'data-t(?:-label)?="([\w.]+)"', PAGE)) | set(
|
|
re.findall(r'''\bt\("([\w.]+)"\)''', JS))
|
|
_missing = sorted(k for k in _keys if not isinstance(resolve(k), str))
|
|
check(f"every label the page asks for is in app/ui_text ({len(_keys)} "
|
|
f"keys; missing: {_missing})", len(_keys) > 30 and not _missing)
|
|
check("data is set as text, never as HTML, so no status sentence can inject "
|
|
"markup", "innerHTML" not in JS and "insertAdjacentHTML" not in JS)
|
|
check("every function of the old page is still on it: start/stop, the "
|
|
"maintenance start and abort, the status rows, the checks and the "
|
|
"event line",
|
|
"auto_toggle" in JS and '"maint_start"' in JS and '"maint_abort"' in JS
|
|
and 'id="rows"' in PAGE and 'id="checks"' in PAGE
|
|
and 'id="hint"' in PAGE and 'href="config"' in PAGE)
|
|
check("stopping, starting and both maintenance actions ask first",
|
|
all(f'ask("{k}"' in JS for k in ("stop", "start", "maint_start",
|
|
"maint_abort")))
|
|
check("tooltips open on hover, on a tap and on keyboard focus",
|
|
"pointermove" in JS and "pointerdown" in JS and "focusin" in JS)
|
|
check("the chart is not colour alone: expensive hours and blocks are "
|
|
"patterned, and every item has a text label",
|
|
"repeating-linear-gradient" in PAGE and '"aria-label": tipText' in JS)
|
|
check("the phone layout puts the tabs at the bottom under 640 px, and dark "
|
|
"mode follows the system",
|
|
"@media (max-width:640px)" in PAGE and "position:fixed;bottom:0" in PAGE
|
|
and "prefers-color-scheme:dark" in PAGE)
|
|
|
|
|
|
async def _serve():
|
|
class _Sup:
|
|
async def info(self):
|
|
return {"options": {"price_taxes_eur_kwh": FLANDERS,
|
|
"price_taxes_profile": ""},
|
|
"schema": [], "translations": {}}
|
|
c = make(ONE, soc=(50.0,), batt=(0.0,))
|
|
runner = aioweb.AppRunner(web.build_app(c, _Sup()))
|
|
await runner.setup()
|
|
site = aioweb.TCPSite(runner, "127.0.0.1", 0)
|
|
await site.start()
|
|
port = site._server.sockets[0].getsockname()[1]
|
|
out = {}
|
|
try:
|
|
async with ClientSession() as s:
|
|
async with s.get(f"http://127.0.0.1:{port}/") as r:
|
|
out["page"] = (r.status, r.content_type, await r.text())
|
|
async with s.get(f"http://127.0.0.1:{port}/status") as r:
|
|
out["status"] = await r.json()
|
|
async with s.get(f"http://127.0.0.1:{port}/status?lang=nl-BE") as r:
|
|
out["status_nl"] = await r.json()
|
|
async with s.get(f"http://127.0.0.1:{port}/config/model") as r:
|
|
out["model"] = await r.json()
|
|
finally:
|
|
await runner.cleanup()
|
|
return out
|
|
|
|
|
|
E = asyncio.run(_serve())
|
|
_nl_words = json.load(io.open(os.path.join(HERE, "app", "ui_text", "nl.json"),
|
|
encoding="utf-8"))["home_words"]
|
|
check("UX-13 P5: /status?lang=nl-BE builds Home in Dutch, and plain /status "
|
|
"stays English",
|
|
E["status_nl"]["state"]["pill"] in _nl_words["pill"].values()
|
|
and E["status_nl"]["control"]["text"] in _nl_words["control"].values()
|
|
and E["status"]["state"]["pill"] not in _nl_words["pill"].values()
|
|
and set(E["status_nl"]) == set(E["status"]))
|
|
check("GET / serves the assembled page as HTML", E["page"][0] == 200
|
|
and E["page"][1] == "text/html" and E["page"][2] == PAGE)
|
|
check("GET status over HTTP carries both halves",
|
|
OLD <= set(E["status"]) and NEW <= set(E["status"]))
|
|
check("the levies preset and its year reach config/model for the P2 "
|
|
"Settings screens (P0 review follow-up)",
|
|
E["model"].get("region") == {"id": "flanders", "year": 2026})
|
|
c_reg = make(dict(ARB, price_taxes_eur_kwh=FLANDERS))
|
|
c_reg.tariff_plan, c_reg.tariff_at = provider_plan(), NOW
|
|
s_reg = status(c_reg)
|
|
check("...and Home names the preset with its year wherever prices come from "
|
|
"a feed, and says 'your own figure' otherwise",
|
|
s_reg["plan"]["region"] == {"id": "flanders", "year": 2026}
|
|
and s_reg["plan"]["region_line"]
|
|
== "Levies and VAT: Flanders, Fluvius Antwerpen, 2026 figures."
|
|
and s_arb["plan"]["region_line"] == "Levies and VAT: your own figure."
|
|
and s_house["plan"]["region_line"] == "")
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED")
|
|
sys.exit(1)
|
|
print("all passed")
|