adminandClaude Opus 5.5
92a80c09b2
The new status row's id `solar_forecast` had no Home title, so Home showed it under the fallback "Inverter connection". Add "Solar forecast" / "Zonneprognose" to ui_text, and checks for the title, the tomorrow-gap hint and blank list rows (review mutations R7, R9, R14 now killed). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Bdqr7PiaHcBnzPituu2yTC
499 lines
23 KiB
Python
499 lines
23 KiB
Python
"""Runnable check for OPT-17, the solar forecast summed from several entities.
|
|
`python3 test_opt17.py`
|
|
|
|
1. `solarfc.read_sum()`: sums in mixed units, refusals, a member down or
|
|
stale makes that day unusable (never a partial sum), hourly splits, today
|
|
and tomorrow, an empty list, and one member equal to `read()`.
|
|
2. `solarfc.entities()`: which key a day uses.
|
|
3. Options: the validator, the settings notes and config.yaml.
|
|
4. The REAL controller (tools/opt11p3_replay.py): a one-member list runs
|
|
exactly like the single key, two halves like one whole, a member down
|
|
like a forecast down; the status row and the action log name the member
|
|
once; every member is read in one gather; modes off reads nothing.
|
|
|
|
DECISIONS/OPT-17-several-forecast-entities.md has the rules. Mutations
|
|
(MUTATION.md, purged copies) are listed in the pull request.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
import re
|
|
import sys
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, HERE)
|
|
sys.path.insert(0, os.path.join(HERE, "tools"))
|
|
logging.disable(logging.WARNING)
|
|
|
|
import opt11p3_replay as rp # noqa: E402
|
|
from app import (dayplan, homeview, messages, solarfc, uitext, # noqa: E402
|
|
validate)
|
|
|
|
fails = []
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
L = timezone(timedelta(hours=2))
|
|
UTC0 = datetime(2026, 9, 21, tzinfo=L).astimezone(timezone.utc)
|
|
TOM = UTC0 + timedelta(days=1)
|
|
NOW = datetime(2026, 9, 21, 9, 0, tzinfo=L).astimezone(timezone.utc)
|
|
END = UTC0 + timedelta(days=1)
|
|
OLD = NOW - timedelta(seconds=solarfc.FRESH_S + 1)
|
|
|
|
|
|
def st(state, unit="kWh", at=NOW, **attrs):
|
|
a = dict(attrs)
|
|
if unit is not None:
|
|
a["unit_of_measurement"] = unit
|
|
return {"state": state, "attributes": a,
|
|
"last_updated": at.isoformat(), "last_reported": at.isoformat()}
|
|
|
|
|
|
def members(*states):
|
|
return [(f"sensor.fc_{i}", s) for i, s in enumerate(states, 1)]
|
|
|
|
|
|
def near(a, b, eps=1e-6):
|
|
return a is not None and b is not None and abs(a - b) < eps
|
|
|
|
|
|
# ============================================================================ #
|
|
print("1. read_sum")
|
|
fc, why, gaps = solarfc.read_sum(members(st("5"), st("7500", "Wh")), [], NOW)
|
|
check("two members, kWh + Wh: 5 + 7.5 = 12.5 kWh",
|
|
fc is not None and near(fc.today.kwh, 12.5) and why is None
|
|
and gaps == ())
|
|
check("...and the rest of the day adds up to the sum",
|
|
near(fc.between(NOW, END), 12.5))
|
|
fc, why, gaps = solarfc.read_sum(
|
|
members(st("5"), st("2500", "Wh"), st("0.005", "MWh")), [], NOW)
|
|
check("three members, kWh + Wh + MWh: 5 + 2.5 + 5 = 12.5 kWh",
|
|
fc is not None and near(fc.today.kwh, 12.5))
|
|
for unit in ("W", None, "kW", "%"):
|
|
fc, why, gaps = solarfc.read_sum(members(st("5"), st("5", unit)), [],
|
|
NOW)
|
|
check(f"a member in {unit or 'no unit'} makes today unusable (unit), "
|
|
f"and is named",
|
|
fc is None and why == solarfc.UNIT
|
|
and gaps == (("today", 2, "sensor.fc_2", solarfc.UNIT),))
|
|
for bad, w in (("unavailable", solarfc.MISSING), ("unknown", solarfc.MISSING),
|
|
(None, solarfc.MISSING), ("-1", solarfc.BAD),
|
|
("lots", solarfc.BAD)):
|
|
s2 = None if bad is None else st(bad)
|
|
fc, why, gaps = solarfc.read_sum(members(st("5"), s2, st("4")), [], NOW)
|
|
check(f"a member that reads {bad!r}: no forecast at all ({w}), not "
|
|
f"the 9 kWh of the others",
|
|
fc is None and why == w
|
|
and gaps == (("today", 2, "sensor.fc_2", w),))
|
|
fc, why, gaps = solarfc.read_sum(members(st("5"), st("4", at=OLD)), [], NOW)
|
|
check("a stale member (older than 3 h): no forecast (stale), named",
|
|
fc is None and why == solarfc.STALE
|
|
and gaps == (("today", 2, "sensor.fc_2", solarfc.STALE),))
|
|
EVE = datetime(2026, 9, 20, 23, 30, tzinfo=L).astimezone(timezone.utc)
|
|
MID = datetime(2026, 9, 21, 0, 10, tzinfo=L).astimezone(timezone.utc)
|
|
fc, why, gaps = solarfc.read_sum(members(st("5", at=MID), st("0", at=EVE)),
|
|
[], MID)
|
|
check("a member from before local midnight is stale, not a dark member",
|
|
fc is None and why == solarfc.STALE)
|
|
fc, why, gaps = solarfc.read_sum(members(st("unavailable"), st("4", at=OLD)),
|
|
[], NOW)
|
|
check("two members down: both named, the first reason is the day's",
|
|
fc is None and why == solarfc.MISSING and [g[1] for g in gaps] == [1, 2])
|
|
check("an empty list is not set", solarfc.read_sum([], [], NOW)
|
|
== (None, solarfc.NOT_SET, ())
|
|
and solarfc.read_sum([], members(st("5")), NOW)[1] == solarfc.NOT_SET)
|
|
fc, _w, _g = solarfc.read_sum(members(st("5", at=NOW - timedelta(hours=2)),
|
|
st("4")), [], NOW)
|
|
check("the forecast's time is the oldest member's report",
|
|
fc.at == NOW - timedelta(hours=2))
|
|
|
|
# Tomorrow.
|
|
fc, why, gaps = solarfc.read_sum(members(st("5"), st("4")),
|
|
members(st("12"), st("8000", "Wh")), NOW)
|
|
check("tomorrow: 12 + 8 = 20 kWh, and today is 9",
|
|
fc is not None and near(fc.tomorrow.kwh, 20.0) and near(fc.today.kwh,
|
|
9.0)
|
|
and near(fc.between(TOM, TOM + timedelta(days=1)), 20.0))
|
|
for name, t2, w in (("unavailable", st("unavailable"), solarfc.MISSING),
|
|
("stale", st("8", at=OLD), solarfc.STALE),
|
|
("in W", st("8", "W"), solarfc.UNIT)):
|
|
fc, why, gaps = solarfc.read_sum(members(st("5"), st("4")),
|
|
members(st("12"), t2), NOW)
|
|
check(f"a tomorrow member {name}: tomorrow is not used (not 12 kWh), "
|
|
f"today is kept, the member is named",
|
|
fc is not None and fc.tomorrow is None and near(fc.today.kwh, 9.0)
|
|
and why is None
|
|
and gaps == (("tomorrow", 2, "sensor.fc_2", w),)
|
|
and fc.between(NOW, TOM + timedelta(hours=9)) is None)
|
|
fc, _w, gaps = solarfc.read_sum(members(st("5"), st("4")), [], NOW)
|
|
check("no tomorrow list: today only, nothing named",
|
|
fc.tomorrow is None and gaps == ())
|
|
|
|
# Hourly splits.
|
|
sol = [{"period_start": (UTC0 + timedelta(minutes=30 * i)).isoformat(),
|
|
"pv_estimate": 2.0 if 20 <= i < 36 else 0.0} for i in range(48)]
|
|
wh = {(UTC0 + timedelta(hours=h)).isoformat(): (1000.0 if 8 <= h < 16
|
|
else 0.0)
|
|
for h in range(24)}
|
|
fc, _w, _g = solarfc.read_sum(members(st("8.0", detailedForecast=sol),
|
|
st("7.0", wh_period=wh)), [], NOW)
|
|
one_s = solarfc.read(st("8.0", detailedForecast=sol), None, NOW)[0]
|
|
one_w = solarfc.read(st("7.0", wh_period=wh), None, NOW)[0]
|
|
spans = [(NOW, END)] + [
|
|
(datetime(2026, 9, 21, h, tzinfo=L), datetime(2026, 9, 21, h + 1,
|
|
tzinfo=L))
|
|
for h in range(9, 20)]
|
|
check("every member has a split: the splits are merged (hour by hour, the "
|
|
"sum of each member's own split)",
|
|
fc.today.periods and near(fc.today.kwh, 15.0)
|
|
and all(near(fc.between(a, b), one_s.between(a, b)
|
|
+ one_w.between(a, b)) for a, b in spans))
|
|
check("...so 10:00-12:00 is 2 (Solcast) + 2 (Open-Meteo) = 4 kWh",
|
|
near(fc.between(datetime(2026, 9, 21, 10, tzinfo=L),
|
|
datetime(2026, 9, 21, 12, tzinfo=L)), 4.0))
|
|
fc, _w, _g = solarfc.read_sum(members(st("8.0", detailedForecast=sol),
|
|
st("7.0")), [], NOW)
|
|
bell = solarfc.read(st("15.0"), None, NOW)[0]
|
|
check("one member without a split: the sum goes on the bell, as one "
|
|
"state-only sensor of 15 kWh would",
|
|
fc.today.periods == () and near(fc.today.kwh, 15.0)
|
|
and all(near(fc.between(a, b), bell.between(a, b)) for a, b in spans))
|
|
wht = {(TOM + timedelta(hours=h)).isoformat(): (500.0 if 8 <= h < 16 else 0)
|
|
for h in range(24)}
|
|
fc, _w, _g = solarfc.read_sum(members(st("5")),
|
|
members(st("4", wh_period=wht),
|
|
st("4", wh_period=wht)), NOW)
|
|
check("tomorrow's splits are merged too (8 x 0.5 kWh each, 8 in all)",
|
|
fc.tomorrow.periods and near(fc.between(TOM, TOM + timedelta(days=1)),
|
|
8.0)
|
|
and near(fc.between(TOM + timedelta(hours=8),
|
|
TOM + timedelta(hours=10)), 2.0))
|
|
fc, _w, _g = solarfc.read_sum(members(st("5")),
|
|
members(st("4", wh_period=wht), st("4")), NOW)
|
|
check("...and fall back to the bell when one has none",
|
|
fc.tomorrow.periods == () and near(fc.tomorrow.kwh, 8.0))
|
|
|
|
# One member is read().
|
|
TODAYS = [st("12.5"), st("12500", "Wh"), st("0.0125", "MWh"), st("12.5", "W"),
|
|
st("12.5", None), st("unavailable"), st("unknown"), None, st("-1"),
|
|
st("lots"), st("12.5", at=OLD), st("12.5", at=NOW + timedelta(
|
|
hours=1)), st("8.0", detailedForecast=sol),
|
|
st("7.0", wh_period=wh), st("0")]
|
|
TOMS = [None, st("20"), st("20", at=OLD), st("unavailable"), st("20", "W"),
|
|
st("4", wh_period=wht)]
|
|
WHENS = [NOW, MID, datetime(2026, 9, 21, 19, 30, tzinfo=L)]
|
|
same, n = True, 0
|
|
for now in WHENS:
|
|
for a in TODAYS:
|
|
for b in TOMS:
|
|
n += 1
|
|
one = solarfc.read(a, b, now)
|
|
got = solarfc.read_sum([("sensor.f", a)],
|
|
[("sensor.t", b)] if b is not None
|
|
else [], now)
|
|
same = same and got[:2] == one
|
|
check(f"one member per day gives exactly what read() gives ({n} cases: "
|
|
f"units, refusals, stale, splits, tomorrow)", same and n == 270)
|
|
|
|
# ============================================================================ #
|
|
print("2. entities")
|
|
E = solarfc.entities
|
|
check("the single key alone", E({"pv_forecast_entity": "sensor.a"})
|
|
== ("sensor.a",))
|
|
check("the list alone, in order", E({"pv_forecast_entities": [
|
|
"sensor.b", "sensor.a"]}) == ("sensor.b", "sensor.a"))
|
|
check("a list with an entry replaces the single key",
|
|
E({"pv_forecast_entity": "sensor.a",
|
|
"pv_forecast_entities": ["sensor.b"]}) == ("sensor.b",))
|
|
check("an empty list or blank rows use the single key",
|
|
E({"pv_forecast_entity": "sensor.a", "pv_forecast_entities": []})
|
|
== ("sensor.a",)
|
|
and E({"pv_forecast_entity": "sensor.a",
|
|
"pv_forecast_entities": ["", " "]}) == ("sensor.a",))
|
|
check("a repeated id counts once (it would double that forecast)",
|
|
E({"pv_forecast_entities": ["sensor.a", "sensor.b", "sensor.a"]})
|
|
== ("sensor.a", "sensor.b"))
|
|
check("tomorrow reads its own keys",
|
|
E({"pv_forecast_entities": ["sensor.a"],
|
|
"pv_forecast_tomorrow_entities": ["sensor.t1", "sensor.t2"]},
|
|
tomorrow=True) == ("sensor.t1", "sensor.t2")
|
|
and E({"pv_forecast_tomorrow_entity": "sensor.t"}, tomorrow=True)
|
|
== ("sensor.t",))
|
|
check("nothing set: ()", E({}) == () and E({}, tomorrow=True) == ())
|
|
|
|
# ============================================================================ #
|
|
print("3. options")
|
|
F = validate._forecast_problems
|
|
for one, many in (solarfc.TODAY_KEYS, solarfc.TOMORROW_KEYS):
|
|
p = F({one: "sensor.a", many: ["sensor.b", "sensor.c"]})
|
|
check(f"{one} and {many} both set: refused, one plain message that "
|
|
f"says which to keep",
|
|
len(p) == 1 and "not both" in p[0] and "Keep the list" in p[0]
|
|
and one in p[0] and many in p[0] and len(p[0].split()) <= 30)
|
|
check(f"...{many} alone is fine, and so is {one} alone",
|
|
not F({many: ["sensor.b", "sensor.c"]}) and not F({one: "sensor.a"})
|
|
and not F({one: "sensor.a", many: []})
|
|
and not F({one: "", many: ["sensor.b"]}))
|
|
check(f"...blank rows in {many} are ignored, alone or beside {one}",
|
|
not F({many: ["", None, "sensor.b"]})
|
|
and not F({one: "sensor.a", many: ["", None]}))
|
|
p = F({many: ["sensor.b", "number.x"]})
|
|
check(f"...an entry in {many} that is not a sensor is refused",
|
|
len(p) == 1 and "only sensors" in p[0] and "number.x" in p[0])
|
|
check(f"...and {many} that is not a list is refused",
|
|
F({many: "sensor.b"}) and "list of sensors" in F({many: "x"})[0])
|
|
with open(os.path.join(HERE, "translations", "en.yaml"),
|
|
encoding="utf-8") as fh:
|
|
EN = fh.read()
|
|
for k in solarfc.TODAY_KEYS + solarfc.TOMORROW_KEYS:
|
|
name = re.search(rf"^ {k}:\n name: \"Hardware · ([^\"(]+?)"
|
|
rf"(?: \(kWh\))?\"$", EN, re.M)
|
|
check(f"the refusal names {k} by its label on the tab",
|
|
name is not None
|
|
and f"“{name.group(1)}” ({k})" in validate.S(k))
|
|
print(" " + F({"pv_forecast_entity": "sensor.a",
|
|
"pv_forecast_entities": ["sensor.b"]})[0])
|
|
VBASE = {"setpoint_entity": "number.goodwe_setpoint", "max_w": 2000,
|
|
"soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
|
|
"meter_source": "off", "maintenance_charge_w": 2000,
|
|
"maintenance_discharge_w": 2000, "battery_modes_enabled": True}
|
|
check("the start-up check refuses both keys set for a day",
|
|
any("not both" in p for p in validate.validate_startup_config(
|
|
dict(VBASE, pv_forecast_tomorrow_entity="sensor.a",
|
|
pv_forecast_tomorrow_entities=["sensor.b"])))
|
|
and validate.validate_startup_config(
|
|
dict(VBASE, pv_forecast_entities=["sensor.a", "sensor.b"],
|
|
pv_forecast_tomorrow_entities=["sensor.c", "sensor.d"]))
|
|
== [])
|
|
notes = []
|
|
validate._entity_notes(notes, dict(VBASE, pv_forecast_entities=[
|
|
"sensor.a", "sensor.b", "sensor.gone"]),
|
|
{"sensor.a": {"unit": "kWh"}, "sensor.b": {"unit": "W"}})
|
|
check("settings notes: a listed sensor not in kWh, and one Home Assistant "
|
|
"does not have, are each named",
|
|
any("sensor.b" in n.text and "kWh" in n.text for n in notes)
|
|
and any("sensor.gone" in n.text for n in notes)
|
|
and not any("sensor.a " in n.text for n in notes))
|
|
notes = []
|
|
validate._entity_notes(notes, dict(VBASE, pv_forecast_tomorrow_entities=[
|
|
"sensor.t"]), {"sensor.t": {"unit": "kWh"}})
|
|
check("a tomorrow list without today's forecast gets the same warning as "
|
|
"the single key", any("used only with today's" in n.text
|
|
for n in notes))
|
|
notes = []
|
|
validate._entity_notes(notes, {"pv_forecast_entities": ["sensor.a"]},
|
|
{"sensor.a": {"unit": "kWh"}})
|
|
check("a list with battery modes off gets the 'unused' note",
|
|
any("battery modes on" in n.text for n in notes))
|
|
with open(os.path.join(HERE, "config.yaml"), encoding="utf-8") as fh:
|
|
CFG = fh.read()
|
|
for k in ("pv_forecast_entities", "pv_forecast_tomorrow_entities"):
|
|
check(f"config.yaml: {k} defaults to [] (load-bearing) and is a list "
|
|
f"of str",
|
|
re.search(rf"^ {k}: \[\]$", CFG, re.M) is not None
|
|
and re.search(rf"^ {k}:\n - str$", CFG, re.M) is not None)
|
|
for k in solarfc.TODAY_KEYS[:1] + solarfc.TOMORROW_KEYS[:1]:
|
|
check(f"config.yaml: {k} is unchanged (str?, default \"\")",
|
|
re.search(rf"^ {k}: \"\"$", CFG, re.M) is not None
|
|
and re.search(rf"^ {k}: str\?$", CFG, re.M) is not None)
|
|
|
|
# ============================================================================ #
|
|
print("4. the real controller")
|
|
A, B = "inverter-a", "inverter-b"
|
|
# Prices as in test_modes_p4: cheap 01:00-03:00, dear 19:00-20:00 local.
|
|
DAYW = [0.20] * 96 * 2
|
|
for i in list(range(4, 12)) + list(range(100, 108)):
|
|
DAYW[i] = 0.02
|
|
for i in list(range(76, 80)) + list(range(172, 176)):
|
|
DAYW[i] = 0.40
|
|
WPTS = [(UTC0 + timedelta(minutes=15 * i), 900, p)
|
|
for i, p in enumerate(DAYW)]
|
|
SF_LS = {"a": (True, "solar_first", False),
|
|
"b": (True, "leftover_solar", False)}
|
|
# Each fake sensor: (share of the day's sun, has a split, status).
|
|
SENSORS = {}
|
|
FC_KWH = [1.0]
|
|
START = [UTC0]
|
|
CALLS = []
|
|
|
|
|
|
async def _sensor(self, e):
|
|
CALLS.append(("start", e))
|
|
await asyncio.sleep(0)
|
|
CALLS.append(("end", e))
|
|
now = START[0] + timedelta(seconds=self.t)
|
|
share, split, status = SENSORS[e]
|
|
if status == "down":
|
|
return {"state": "unavailable", "attributes": {},
|
|
"last_updated": now.isoformat()}
|
|
at = now - timedelta(hours=4) if status == "stale" else now
|
|
loc = now.astimezone(L)
|
|
d0 = loc.replace(hour=0, minute=0, second=0, microsecond=0)
|
|
if e.startswith("sensor.tom"):
|
|
d0 += timedelta(days=1)
|
|
wh = {(d0 + timedelta(hours=h)).astimezone(timezone.utc).isoformat():
|
|
(FC_KWH[0] * 100.0 * share if 8 <= h < 18 else 0.0)
|
|
for h in range(24)}
|
|
left = sum(v for k, v in wh.items()
|
|
if datetime.fromisoformat(k) >= now) / 1000.0
|
|
attrs = {"unit_of_measurement": "kWh"}
|
|
if split:
|
|
attrs["wh_period"] = wh
|
|
return {"state": f"{left:.6f}", "attributes": attrs,
|
|
"last_updated": at.isoformat(), "last_reported": at.isoformat()}
|
|
|
|
|
|
rp.Hass.state = _sensor
|
|
|
|
|
|
def seed(c):
|
|
if c.house_profile is not None:
|
|
hp = dayplan.HouseProfile()
|
|
t = UTC0 - timedelta(days=1)
|
|
while t < UTC0:
|
|
h = t.astimezone(L).hour
|
|
hp.add(t, 3000.0 if 17 <= h < 22 else 600.0)
|
|
t += timedelta(seconds=5)
|
|
hp.add(t, None)
|
|
c.house_profile = hp
|
|
|
|
|
|
def run(opts, hours=4.0, modes=True):
|
|
start = UTC0 + timedelta(hours=0.75)
|
|
START[0] = start
|
|
rows = []
|
|
|
|
def cb(c, h, now, pr):
|
|
sh = dict(c.allocation.shares) if c.allocation is not None and \
|
|
c.allocation.shares else {}
|
|
rows.append((now, round(h.grid(), 6), tuple(sorted(sh.items())),
|
|
getattr(c.arb, "action", None),
|
|
tuple(sorted(c.mode_tick.roles.items()))
|
|
if c.mode_tick else (),
|
|
tuple(sorted((u, round(p.soc, 6))
|
|
for u, p in h.plants.items())),
|
|
tuple(sorted((c.plan_targets() or {}).items())),
|
|
None if c.day_plan is None else c.day_plan.verdict))
|
|
_st, c = rp.run("p3_ls" if modes else "modes_off", [(UTC0.timestamp()
|
|
- 10, 800.0)],
|
|
WPTS, start, 3600 * hours, check=cb,
|
|
soc0={"a": 60.0, "b": 30.0}, choice=SF_LS,
|
|
extra=dict({"cycle_cost_threshold_eur": 0.20}, **opts),
|
|
patch=seed)
|
|
return rows, c
|
|
|
|
|
|
def fc_rows(c):
|
|
return [r for r in c.checks() if r["id"] == "solar_forecast"]
|
|
|
|
|
|
def fc_events(c):
|
|
return [e for e in c.events if "solar forecast" in e]
|
|
|
|
|
|
SENSORS.update({
|
|
"sensor.all": (1.0, True, "ok"), "sensor.h1": (0.5, True, "ok"),
|
|
"sensor.h2": (0.5, True, "ok"), "sensor.h2_flat": (0.5, False, "ok"),
|
|
"sensor.down": (0.5, True, "down"), "sensor.stale": (0.5, True, "stale"),
|
|
"sensor.tom_all": (1.0, True, "ok"), "sensor.tom_h1": (0.5, True, "ok"),
|
|
"sensor.tom_down": (0.5, True, "down"), "sensor.none": (1.0, True, "down"),
|
|
})
|
|
single, c_single = run({"pv_forecast_entity": "sensor.all"})
|
|
listed, c_listed = run({"pv_forecast_entities": ["sensor.all"]})
|
|
plans = [r for r in single if r[6]]
|
|
check(f"the run makes a day plan with a level ({len(plans)} of "
|
|
f"{len(single)} cycles), so the forecast steers something",
|
|
plans and any(r[7] is not None for r in single))
|
|
check("a one-member list runs exactly like the single key: every grid "
|
|
"figure, share, action, role, level and plan, cycle by cycle",
|
|
single == listed and len(single) > 2000)
|
|
check("...and says the same on the status page and in the action log",
|
|
fc_rows(c_single) == fc_rows(c_listed) == []
|
|
and fc_events(c_single) == fc_events(c_listed))
|
|
halves, c_halves = run({"pv_forecast_entities": ["sensor.h1", "sensor.h2"]})
|
|
check("two members of half the sun each run like one sensor of all of it",
|
|
[r[:6] + (r[7],) for r in halves] == [r[:6] + (r[7],) for r in single]
|
|
and all(len(a[6]) == len(b[6]) and all(
|
|
k1 == k2 and abs(v1 - v2) < 1e-6 for (k1, v1), (k2, v2)
|
|
in zip(a[6], b[6])) for a, b in zip(halves, single)))
|
|
none_, c_none = run({"pv_forecast_entity": "sensor.none"})
|
|
check("(the comparison bites: no forecast runs differently)",
|
|
[r[:6] for r in none_] != [r[:6] for r in single])
|
|
for name, mem in (("unavailable", "sensor.down"), ("stale", "sensor.stale")):
|
|
rows, c = run({"pv_forecast_entities": ["sensor.h1", mem]})
|
|
check(f"a member {name}: the controller runs exactly as with the single "
|
|
f"forecast unavailable (no partial sum)", rows == none_)
|
|
check(f"...the status page names it by place, in one short line",
|
|
[r["text"] for r in fc_rows(c)] == [
|
|
"Solar forecast for today: not used, source 2 of 2 has no "
|
|
"reading"] and fc_rows(c)[0]["warn"])
|
|
ev = fc_events(c)
|
|
check(f"...and the action log names it once in {len(single)} cycles",
|
|
len(ev) == 1 and mem in ev[0] and "today" in ev[0])
|
|
one_down, c_one_down = run({"pv_forecast_entities": ["sensor.none"]})
|
|
check("a one-member list that is down runs, shows and logs exactly like the "
|
|
"single key down", one_down == none_
|
|
and fc_rows(c_one_down) == fc_rows(c_none)
|
|
and fc_events(c_one_down) == fc_events(c_none))
|
|
# A forecast missing from the start logs nothing (the start state is
|
|
# already "missing"); that is release/1.0's behaviour, kept.
|
|
check("the single key down: no new row and no new log line",
|
|
fc_rows(c_none) == [] and fc_events(c_none) == [])
|
|
CALLS.clear()
|
|
rows, c = run({"pv_forecast_entities": ["sensor.h1", "sensor.h2"],
|
|
"pv_forecast_tomorrow_entities": ["sensor.tom_h1",
|
|
"sensor.tom_down"]},
|
|
hours=0.2)
|
|
check("a tomorrow member down: today is used, tomorrow is not, and the row "
|
|
"says so",
|
|
c.solar_fc is not None and c.solar_fc.tomorrow is None
|
|
and [r["text"] for r in fc_rows(c)] == [
|
|
"Solar forecast for tomorrow: not used, source 2 of 2 has no "
|
|
"reading"]
|
|
and len(fc_events(c)) == 1 and "sensor.tom_down" in fc_events(c)[0])
|
|
check("...and the page hint says tomorrow's forecast is not used",
|
|
(c.hint or "").endswith(messages.hint("forecast_no_tomorrow")))
|
|
check("...and Home titles the row as the solar forecast, in EN and NL",
|
|
[a["title"] for a in homeview.attention(fc_rows(c))]
|
|
== ["Solar forecast"]
|
|
and uitext.words("nl")["home_words"]["attn"]["solar_forecast"]
|
|
== "Zonneprognose")
|
|
first = CALLS[:8]
|
|
check("every member of both days is read in one gather (all four start "
|
|
"before any ends)",
|
|
[k for k, _e in first] == ["start"] * 4 + ["end"] * 4
|
|
and sorted(e for _k, e in first[:4]) == sorted(
|
|
["sensor.h1", "sensor.h2", "sensor.tom_h1", "sensor.tom_down"]))
|
|
rows, c = run({"pv_forecast_entities": ["sensor.h1"],
|
|
"pv_forecast_tomorrow_entities": ["sensor.tom_all"]},
|
|
hours=0.2)
|
|
check("tomorrow: a one-member list is read and counted",
|
|
c.solar_fc is not None and c.solar_fc.tomorrow is not None)
|
|
rows, c = run({"pv_forecast_entities": ["sensor.h1", "sensor.h2_flat"]},
|
|
hours=0.2)
|
|
check("a member without a split: the controller's forecast is on the bell",
|
|
c.solar_fc is not None and c.solar_fc.today.periods == ())
|
|
CALLS.clear()
|
|
run({"pv_forecast_entities": ["sensor.h1", "sensor.h2"]}, hours=0.2,
|
|
modes=False)
|
|
check("battery modes off: no forecast member is read", CALLS == [])
|
|
check("the status row is a STYLE status row (<= 15 words, no entity id)",
|
|
all(len(t.split()) <= 15 and "sensor." not in t for t in (
|
|
messages.row_forecast_gap("today", [2, 3], 3),
|
|
messages.row_forecast_gap("tomorrow", [1], 2))))
|
|
|
|
print()
|
|
if fails:
|
|
print(f"FAILED: {len(fails)}")
|
|
for f in fails:
|
|
print(" -", f)
|
|
sys.exit(1)
|
|
print("all checks passed")
|