Compare commits

..
4 Commits
Author SHA1 Message Date
glenn schrooyenandClaude b6b2d1b184 UX-54: a Forecast tab with the sun, the house and each battery's level, per hour
checks / unit (pull_request) Successful in 48m29s
infra-pin / fresh (pull_request) Successful in 1s
A new tab beside Home and Batteries for today and tomorrow: the solar
forecast, the solar measured so far, the house's expected use and each
battery's expected level, with the plan's own charge and give blocks under it.

- Blocks are Home's: today's are the plan's `runs`, tomorrow's come from the
  same builders (`homeview.runs_for()`), in the plan lines' words.
- The sun is what solarfc reads; a daily total is spread over the bell and the
  tab says so. The house is HouseProfile.estimator(), as the plan reads it.
- The level is new: `daycurve.curve()` gives each battery the role
  `modetrade.decide()` gives it and moves it hour by hour inside the plan's
  floor and ceiling. It is an estimate and counts on `solar_trust_pct`.
- Built on request (`GET forecast`), no Home Assistant call. Measured solar is
  kept per hour in memory from the reading control already takes.
- A part not known yet is a note: modes off, no forecast, learning, no
  forecast for tomorrow, a monthly cycle, no level, stopped, shadow.

No decision, command, timing or option key changes. Found, not changed:
`dayplan.plan()` over-counts a state-only forecast (DECISIONS/UX-54).

Co-Authored-By: Claude <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Bdqr7PiaHcBnzPituu2yTC
2026-10-26 19:38:59 +01:00
admin a1d07030f8 Merge pull request 'UX-53: Home draws a grid charge only where the add-on will take it' (#209) from UX-53 into release/1.0
infra-pin / fresh (push) Successful in 2s
2026-10-26 13:59:22 +01:00
glenn schrooyenandClaude Opus 5.5 5dfe9720c5 UX-53 review: DECISIONS text, day/night weekend case and peak guard limit
checks / unit (pull_request) Successful in 48m2s
infra-pin / fresh (pull_request) Successful in 1s
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-26 13:10:49 +01:00
glenn schrooyenandClaude 242cf434f4 UX-53: Home draws a grid charge only where the add-on will take it
checks / unit (pull_request) Successful in 48m7s
infra-pin / fresh (pull_request) Successful in 2s
Modes on: Controller._mode_roles() files the trade gate it just asked as
mode_gate; while it refuses, Home draws no grid charge and marks no cheap
quarter. Modes off: a charge after the last dear period is asked of the same
gate at its own start (homeview.leg_taken) and dropped when refused there.
Words: "Lowest/Highest price left" (EN, NL); a flat remainder on a dynamic or
day/night site is no longer called a fully fixed product.

No change to control decisions, commands or options.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-10-26 11:15:29 +01:00
19 changed files with 2597 additions and 35 deletions
@@ -0,0 +1,40 @@
# UX-53 - Home draws a grid charge only where the add-on will take it
**Status: DONE 2026-10-26 (owner go 2026-10-25, from the PRICE-02 rig test F-1 and F-2).**
## Decision
Home asks the add-on's own gate, and does not decide for itself.
- **Modes on.** `Controller._mode_roles()` files the gate it just asked
(`TariffPlan.arbitrage()` with the measured peak guard) as `mode_gate`: None
when it allowed a charge, else its reason (`peak`, `not_worth`,
`not_covered`). While it is set, `homeview.mode_runs()` draws no grid charge
and `price_windows()` marks no cheap quarter. Control never reads it. A
refusal that is not the gate (no measured headroom, a battery cycling) is
not filed, so those pages are as before.
- **Modes off.** A charge the verdict puts after its last dear period is
refused when its time comes: the quarters left are all equal. `leg_taken()`
asks `trade_verdict()` (the control cycle's own call) at the leg's start and
keeps the leg only if the gate allows it there. A leg in progress is the
verdict's own answer now. The sale, "Tomorrow:" lines and chart marks use
the same answer.
- **Words.** "Lowest/Highest published price" becomes "Lowest/Highest price
left" (the verdict reads only the periods not yet over). The flat-tariff
refusal calls a site "a fully fixed product" only when its product is fixed.
## What does not change
No control decision, command or option. Day/night and hourly sites with a
second day of prices keep both legs at every half hour of a day
(`test_ux53.py` part 4), except where every price left is equal: with an
all-night weekend, Friday's 22:00 charge is no longer drawn, as the add-on
does not take it. Price-off sites draw nothing, as before.
## Limits
The leg's gate is asked on today's plan, as the control cycle will find it at
that time. A plan that gets new prices in between is drawn again at the next
refresh. While the peak guard holds, a modes-on page shows no grid charge at
all, also not the rest of a run the add-on charges in once the guard clears,
as the modes-off page already did.
@@ -0,0 +1,68 @@
# UX-54 - a Forecast tab: the sun, the house and each battery's level, per hour
**Status: DONE 2026-10-26 (owner go 2026-10-26: "Day curve", "New tab").**
## Decision
A tab beside Home and Batteries shows today and tomorrow, per hour: the solar
forecast, the solar measured so far, the house's expected use and each
battery's expected level with its blocks. It follows the add-on, as Home does
(UX-53), and decides nothing.
- **Blocks are Home's.** Today's are the `runs` of the same plan Home draws
(grid, give, monthly cycle). Tomorrow's come from the same builders
(`homeview.runs_for()`). The words are the plan lines'. A block is on the tab
exactly where Home has one, so the UX-53 gate is followed too.
- **The sun** is what `solarfc` reads. An hourly split (Solcast
`detailedForecast`, Open-Meteo `wh_period`) is used as given. A daily total
(Forecast.Solar, and any other state-only sensor, the owner's three "today
remaining" and three "tomorrow") is spread over the 06:00-20:00 bell, and the
tab says so.
- **The house** is `HouseProfile.estimator()`, never below the night median, as
the plan reads it.
- **The level did not exist** (the plan holds targets and run blocks only), so
`daycurve.curve()` derives it: each battery has the role `modetrade.decide()`
gives (follow the house, hold, give, charge from the grid, rest), spare solar
goes to Solar first batteries first, a shortfall is met by those that may
give, both at `dayplan.EFF`, inside the plan's floor and ceiling, from the
level now. It counts on `solar_trust_pct` of the sun, as the plan does, and
says so. Grid charging goes to `modetrade.charge_to()`. It is an estimate.
- **Measured solar** is kept per local hour in memory (`SolarToday`), from the
reading the control cycle already takes; nothing is read back.
- **On request.** `GET forecast` builds the tab (`Controller.forecast_payload()`
reads `/status`' plan first, so today's blocks are the very ones Home has). No
Home Assistant call, nothing cached, nothing in the cycle but one in-memory
add. The page asks every 30 s while the tab shows.
- **Not known yet is a note, not an empty chart:** modes off, no forecast set,
forecast missing or old, learning, no tomorrow forecast, a monthly cycle, no
level reading, control stopped, shadow.
## What does not change
No decision, command, timing or option key. The one write in the control path
is `SolarToday.add()` after the solar read, in a `try`, in memory.
## Limits
- Prices are planned about a day ahead (`priceruns.HORIZON`), so tomorrow's
blocks appear as the day comes near. The tab says so.
- A give run drawn on a battery that is empty now has no block on Home (no
energy to give), but the curve gives in it if the sun fills it first.
- Tomorrow shows price blocks only. A monthly cycle is drawn today, where the
curve then stops for that battery.
- The curve uses the arbitrage floor and ceiling, as the plan does, not the
battery's own cut-out level. The house profile is a median per hour, net of
the solar the house used.
- Measured solar starts when the add-on starts; the tab says from when.
## Found, not changed
`dayplan.plan()` reads the forecast hour by hour with `Forecast.between(s, e)`.
For a state-only forecast `between()` takes its start as the moment the "left
today" figure was read, so each later hour is spread over what is left of the
bell from that hour: on the owner's shape 14.4 kWh left gives 57 kWh over the
hours to midnight (`test_ux54.py` part 1). The plan's sun is overcounted for
those sites. It does not show on a site whose batteries are all Solar first
(no level is planned for them). The tab measures both ends from now
(`dayview.sun_kwh()`), which keeps the total and gives the same hours for an
hourly split. A card should fix the plan.
+15
View File
@@ -3530,6 +3530,21 @@ the peak limit, the batteries' switches and modes, and shadow mode, each
marked "Applies right away" or "Applies after a restart". **Installer** shows
every setting.
### The Forecast screen
The **Forecast** tab shows today and tomorrow, hour by hour: the solar forecast,
the solar measured so far today, the house's expected use, and each battery's
expected level. Under each battery's chart are its planned blocks, in the same
words as Home. A block is there only where the add-on plans one.
The level is an estimate. It counts on `solar_trust_pct` of the forecast, as the
plan does, and stays between the battery's low and high levels. A forecast that
gives one total for the day (Forecast.Solar) is spread over the hours, and the
screen says so. Where a part is not known yet (battery modes off, no forecast,
still learning your house, no forecast for tomorrow) the screen says that
instead of drawing an empty chart. It changes nothing in control, and it asks
Home Assistant for nothing: it is built when you open it.
### Setup, for a new install
The **Setup** tab walks an installer through six steps: the meter, the
+216
View File
@@ -0,0 +1,216 @@
"""UX-54: the expected battery level through the day, for the Forecast tab.
Pure: batteries, the sun and the house's use in; a level per battery over
time out. Display only: nothing here is read by control, and it never
decides a block. The blocks are the plan's (`homeview.mode_runs()`); this
only says where the level goes, hour by hour, if the add-on does what the
plan says.
Each step gives every battery a role, as `modetrade.decide()` does (the
names are modetrade's): in a grid block it charges (CHARGE) and the others
rest (REST, no charge); else it follows the house (HOUSE) if it is Solar
first or no price plan is on; inside a give run it covers the house (GIVE);
else it holds (HOLD, may charge from spare solar, gives nothing). Spare
solar (sun less the house's use) goes to Solar first batteries first, then
the rest, by room. A shortfall comes from the batteries that may give,
those in a give run first, by what they hold above the floor. Charging and
giving lose `dayplan.EFF`, as the plan counts. A level stays inside the
floor and ceiling the plan uses, and a battery already outside them does
not move further out.
`SolarToday` keeps the site's measured solar per hour, in memory, from the
reading the control cycle already has. It is shown and never read back
(TEL-11 keeps site solar out of control).
"""
import math
from datetime import datetime, timedelta, timezone
from dataclasses import dataclass
from .dayplan import EFF, _hour_slices, _local
from .modetrade import CHARGE, GIVE, HOLD, HOUSE, REST
EPS = 1e-9
GAP_S = 60.0 # a longer gap between readings is not counted
MIN_COVER = 0.5 # share of the elapsed hour a reading must cover
@dataclass(frozen=True)
class Batt:
"""One battery that is on, with a level and a capacity."""
name: str
capacity_kwh: float
rating_kw: float
soc: float # % now
solar_first: bool # Solar first: follows the house all day
priced: bool # a price plan is on and its mode is a price mode
grid: tuple = () # ((start, end, to_pct), ...) grid blocks
give: tuple = () # ((start, end), ...) give runs
stop_at: datetime | None = None # a monthly cycle starts: no level after
def _role(b, t, charging, own):
"""The battery's role at instant `t`; `own`: it charges from the grid;
`charging`: some battery does. Returns `(role, may_charge, may_give)`."""
in_give = any(s <= t < e for s, e in b.give)
if own:
return CHARGE, False, False
if charging:
return REST, False, b.solar_first or not b.priced or in_give
if b.solar_first or not b.priced:
return HOUSE, True, True
if in_give:
return GIVE, True, True
return HOLD, True, False
def _spread(amount, caps, weights):
"""Split `amount` over the keys of `caps` by `weights`, none over its
cap. Returns `(taken, left)`: what each took and what no one could."""
got = {k: 0.0 for k in caps}
left = float(amount)
live = {k for k, c in caps.items() if c > EPS and weights.get(k, 0.0) > 0}
while left > EPS and live:
total = sum(weights[k] for k in live)
used, full = 0.0, set()
for k in live:
part = left * weights[k] / total
room = caps[k] - got[k]
if part >= room - EPS:
part = room
full.add(k)
got[k] += part
used += part
left -= used
if not full:
break
live -= full
return got, max(0.0, left)
def curve(batts, start, end, sun_kwh, use_kwh, *, floor_pct, ceiling_pct):
"""`{name: [(t, pct), ...]}` from `start` (each battery's level now) to
`end`, a point at every hour edge and every block edge. A battery's list
stops where `sun_kwh(a, b)` is None (the sun is not known) or its
`stop_at` comes. `sun_kwh` and `use_kwh` give kWh for the span `[a, b)`."""
floor, ceil = float(floor_pct), float(ceiling_pct)
cuts = {end}
for a, _b, _h in _hour_slices(start, end):
cuts.add(a)
for b in batts:
for s, e, *_ in b.grid + b.give:
cuts.update(x for x in (s, e) if start < x < end)
cuts.add(start)
ts = sorted(cuts)
level = {b.name: float(b.soc) for b in batts}
out = {b.name: [(start, level[b.name])] for b in batts}
alive = {b.name for b in batts}
for a, z in zip(ts, ts[1:]):
if not alive:
break
sun = sun_kwh(a, z)
if sun is None:
break
for b in batts:
if b.name in alive and b.stop_at is not None and a >= b.stop_at:
alive.discard(b.name)
net = sun - use_kwh(a, z)
hrs = (z - a).total_seconds() / 3600.0
want = {} # name -> grid target %
for b in batts:
if b.name in alive:
for s, e, to in b.grid:
if s <= a < e and level[b.name] < to - EPS:
want[b.name] = float(to)
roles = {b.name: _role(b, a, bool(want), b.name in want)
for b in batts if b.name in alive}
by = {b.name: b for b in batts}
for n, to in want.items(): # charging from the grid
b = by[n]
room = (to - level[n]) / 100.0 * b.capacity_kwh
level[n] += min(b.rating_kw * hrs * EFF, room) \
/ b.capacity_kwh * 100.0
if net > EPS:
caps = {n: min(by[n].rating_kw * hrs,
max(0.0, ceil - level[n]) / 100.0
* by[n].capacity_kwh / EFF)
for n, r in roles.items() if r[1]}
first = {n: c for n, c in caps.items() if by[n].solar_first}
rest = {n: c for n, c in caps.items() if n not in first}
left = net
for grp in (first, rest):
got, left = _spread(left, grp, dict(grp))
for n, v in got.items():
level[n] += v * EFF / by[n].capacity_kwh * 100.0
elif net < -EPS:
caps = {n: min(by[n].rating_kw * hrs,
max(0.0, level[n] - floor) / 100.0
* by[n].capacity_kwh * EFF)
for n, r in roles.items() if r[2]}
first = {n: c for n, c in caps.items() if roles[n][0] == GIVE}
rest = {n: c for n, c in caps.items() if n not in first}
left = -net
for grp in (first, rest):
got, left = _spread(left, grp, dict(grp))
for n, v in got.items():
level[n] -= v / EFF / by[n].capacity_kwh * 100.0
for b in batts:
if b.name in alive:
out[b.name].append((z, level[b.name]))
return out
# --------------------------------------------------------------------------- #
# Measured solar, per local hour
# --------------------------------------------------------------------------- #
class SolarToday:
"""The site's solar power, averaged per local hour of one local day. In
memory only: a restart starts it again, and `first` says from when it
has counted. Display only."""
def __init__(self):
self.day = None
self.first = None
self._hours = {} # hour start (UTC) -> [W x s, seconds]
self._last = None # (time, W)
def add(self, now: datetime, w) -> None:
"""One reading (`None`: no reading, which ends the interval)."""
day = _local(now).date()
if day != self.day:
self.day, self.first, self._hours, self._last = day, None, {}, None
try:
ok = w is not None and math.isfinite(float(w))
except (TypeError, ValueError):
ok = False
if not ok:
self._last = None
return
w = max(0.0, float(w))
if self.first is None:
self.first = now
if self._last is not None and now > self._last[0]:
dt = min(GAP_S, (now - self._last[0]).total_seconds())
key = _hour_start(self._last[0])
row = self._hours.setdefault(key, [0.0, 0.0])
row[0] += (self._last[1] + w) / 2.0 * dt
row[1] += dt
self._last = (now, w)
def kwh(self, start: datetime, end: datetime, now: datetime):
"""kWh for the hour `[start, end)` up to `now`: the mean power it was
read at over the part of the hour that has passed, or None when the
readings cover under half of that part (or none has passed)."""
if self.day != _local(now).date():
return None
over = (min(end, now) - start).total_seconds()
row = self._hours.get(start)
if over <= 0 or row is None or row[1] < MIN_COVER * over:
return None
return row[0] / row[1] * over / 3.6e6
def _hour_start(t: datetime) -> datetime:
lt = _local(t)
return lt.replace(minute=0, second=0, microsecond=0).astimezone(
timezone.utc)
+251
View File
@@ -0,0 +1,251 @@
"""UX-54: the Forecast tab, built on request (`GET forecast`). Reads only.
Two days, today and tomorrow, per local hour: the solar forecast, the solar
measured so far, the house's expected use, and each battery's expected level
with its blocks.
It follows the add-on, as Home does (UX-53), and decides nothing:
* the blocks are Home's own: today's are the `runs` of `/status`'s plan, and
tomorrow's come from the same builders (`homeview.runs_for()`), so a block
is on the tab exactly where the plan has one, with the plan's words;
* the sun is `solarfc.Forecast.between()` and the house's use is
`dayplan.HouseProfile.estimator()` read as `dayplan.plan()` reads it;
* the level is `daycurve.curve()`, fed the blocks, the give runs, the packs
and the plan's levels (`modetrade.charge_to()`), floor and ceiling.
A part that is not known yet is a plain note (`home_words.day`), never an
empty chart. DECISIONS/UX-54-forecast-tab.md has the record.
"""
from datetime import datetime, timedelta
from . import arbitrage, dayplan, homeview as hv, modetrade
from .daycurve import Batt, curve
from .packmode import SOLAR_FIRST
from .price import brussels
BLOCKS = ("grid", "give", "cycle") # the kinds a battery's blocks carry
def _hours_between(a, b) -> float:
return (b - a).total_seconds() / 3600.0
def build(c, status: dict, now: datetime, lang: str = "en") -> dict:
"""The tab's JSON. `status` is the `/status` payload just built, whose
plan holds today's blocks."""
with hv.in_lang(lang):
return _build(c, status, now)
def sun_kwh(fc, now, a, b):
"""kWh the forecast gives in `[a, b)`, `a` at or after `now`, or None.
Both ends are measured from `now`: `Forecast.between()` reads its start
as the moment the "left today" figure was taken, so a span that starts
later would be spread over what is left of the bell, not over its own
hours (state-only forecasts, DECISIONS/UX-54-forecast-tab.md)."""
hi, lo = fc.between(now, b), fc.between(now, a)
return None if hi is None or lo is None else max(0.0, hi - lo)
def _times(run):
return (datetime.fromisoformat(run["start"]),
datetime.fromisoformat(run["end"]))
def _build(c, status, now):
plan, rows = status.get("plan"), status.get("batteries")
if not plan or rows is None:
return {"ok": False}
d0, d1 = hv.local_day(now)
n0, n1 = hv.local_day(d1 + timedelta(hours=2))
shadow = bool(c.shadow)
key = plan.get("verdict_key")
running = key not in ("held", "stopped")
modes_on = bool(getattr(c, "modes_on", False))
floor, ceil = arbitrage.soc_bounds(c.o)
fc = getattr(c, "solar_fc", None)
hp = getattr(c, "house_profile", None)
est = hp.estimator() if hp is not None and fc is not None else None
trust = max(0.0, min(1.0, float(getattr(c, "solar_trust_pct", 70))
/ 100.0))
# Why a part is not known (None: nothing is missing).
missing = ("modes_off" if not modes_on
else "not_set" if c.fc_entity is None
else "no_forecast" if fc is None
else "learning" if est is None else None)
# The blocks of each day: today's from Home, tomorrow's by its builders.
runs = [plan["runs"], hv.runs_for(c, n0, n1, now, shadow) if running
else []]
blocks = [[[r for r in rs if r["battery"] == i and r["kind"] in BLOCKS]
for i in range(len(c.inverters))] for rs in runs]
# The level curve of every battery that is on and known.
notes_b = {}
levels = {}
if running and missing is None:
levels = _levels(c, est, fc, trust, now, n1, runs, floor, ceil,
notes_b)
# Prices are planned about a day ahead (`priceruns.HORIZON`), so tomorrow's
# blocks may be missing now and appear later.
priced = bool(plan.get("has_prices")) and (
any(c.pack_mode(i).use and c.pack_mode(i).mode in hv.PRICE_MODES
for i in c.inverters) if modes_on else arbitrage.enabled(c.o))
days = []
for k, (a, z) in enumerate(((d0, d1), (n0, n1))):
days.append(_day(c, k, a, z, now, fc, est, trust, levels, blocks[k],
rows, shadow, running, key, missing, notes_b, status,
priced))
return {"ok": True, "running": running, "shadow": shadow,
"now_h": round(_hours_between(d0, now), 3),
"updated": None if fc is None else hv.hhmm(fc.at),
"days": days}
def _levels(c, est, fc, trust, now, end, runs, floor, ceil, notes_b):
"""`{name: [(t, pct), ...]}` from `daycurve.curve()` for the batteries
that are on, have a level and are not in their monthly cycle."""
packs = {p.name: p for p in c._mode_packs()}
targets = c.plan_targets()
rp = getattr(c, "mode_runs", None)
plan_on = rp is not None and rp.on
house_w, base_w = est
batts = []
for i, inv in enumerate(c.inverters):
pm = c.pack_mode(inv)
if not pm.use:
continue
if inv.soc is None or not inv.capacity_kwh:
notes_b[inv.name] = hv.day_note("no_level", name=inv.name)
continue
mine = [r for rs in runs for r in rs
if r["battery"] == i and r["kind"] in BLOCKS]
cyc = [_times(r)[0] for r in mine if r["kind"] == "cycle"]
if cyc and min(cyc) <= now:
notes_b[inv.name] = hv.day_note("cycle", name=inv.name)
continue
if cyc:
notes_b[inv.name] = hv.day_note("cycle", name=inv.name)
to = modetrade.charge_to(packs[inv.name], ceil, targets)
grid = tuple((*_times(r), to) for r in mine
if r["kind"] == "grid" and to is not None)
priced = plan_on and pm.mode != SOLAR_FIRST
give = tuple((g.start, g.end) for g in rp.give) if priced else ()
batts.append(Batt(inv.name, float(inv.capacity_kwh),
inv.max_w / 1000.0, float(inv.soc),
pm.mode == SOLAR_FIRST, priced, grid, give,
min(cyc) if cyc else None))
def sun(a, z):
kwh = sun_kwh(fc, now, a, z)
return None if kwh is None else kwh * trust
def use(a, z):
return sum(max(house_w(h), base_w) * _hours_between(s, e) / 1000.0
for s, e, h in dayplan._hour_slices(a, z))
if not batts:
return {}
return curve(batts, now, end, sun, use, floor_pct=floor, ceiling_pct=ceil)
def _day(c, k, d0, d1, now, fc, house, trust, levels, blocks, rows, shadow,
running, key, missing, notes_b, status, priced):
day = (fc.today, fc.tomorrow)[k] if fc is not None else None
solar = getattr(c, "solar_today", None)
names = [inv.name for inv in c.inverters]
hours = []
for s, e, h in dayplan._hour_slices(d0, d1):
f = None
if fc is not None and e > now and (k == 0 or fc.tomorrow is not None):
f = sun_kwh(fc, now, max(s, now), e)
got = solar.kwh(s, e, now) if k == 0 and solar is not None \
and s < now else None
use = None if house is None else max(house[0](h), house[1]) \
* _hours_between(s, e) / 1000.0
row = {"start_h": round(_hours_between(d0, s), 3),
"end_h": round(_hours_between(d0, e), 3), "label": hv.hhmm(s),
"forecast_kwh": None if f is None else round(f, 3),
"measured_kwh": None if got is None else round(got, 3),
"house_kwh": None if use is None else round(use, 3),
"levels": []}
hours.append((s, row))
batts = []
for i, inv in enumerate(c.inverters):
row = rows[i] if i < len(rows) else {}
pts = [(t, v) for t, v in levels.get(inv.name, ()) if d0 <= t <= d1]
mine = blocks[i]
batts.append({
"name": inv.name, "capacity_kwh": inv.capacity_kwh,
"off": row.get("use") is False,
"levels": [[round(_hours_between(d0, t), 3), round(v, 1)]
for t, v in pts],
"blocks": [{x: r[x] for x in ("kind", "from", "to", "start_h",
"end_h", "what", "why", "amount")}
for r in mine],
"line": _line(c, inv, row, mine, running, key, shadow)})
at = {t: v for t, v in pts}
for s, hr in hours:
hr["levels"].append(None if s not in at else round(at[s], 1))
has_levels = any(b["levels"] for b in batts)
fcs = [h["forecast_kwh"] for _s, h in hours]
notes = []
if missing is not None:
notes.append(hv.day_note(missing))
if not running:
notes.append(hv.why_words(key))
if shadow:
notes.append(hv.why_words("shadow"))
line = ""
if fc is not None:
if k == 0:
line = hv.line_solar(fc.today.kwh, fc.today.kwh * trust,
hv.hhmm(fc.at))
elif fc.tomorrow is not None:
line = hv.day_note("tomorrow_total",
kwh=hv._kwh(fc.tomorrow.kwh),
count=hv._kwh(fc.tomorrow.kwh * trust))
if day is not None and not day.periods:
notes.append(hv.day_note("estimate"))
if k == 1 and fc.tomorrow is None:
notes.append(hv.day_note("no_tomorrow"))
if k == 0 and solar is not None and solar.first is not None \
and solar.day == brussels(now).date() \
and _hours_between(d0, solar.first) > 0.5:
notes.append(hv.day_note("measured_from", at=hv.hhmm(solar.first)))
if has_levels:
notes.append(hv.day_note("trust", pct=round(trust * 100)))
if k == 1 and _tomorrow_line(status):
notes.append(_tomorrow_line(status))
if k == 1 and running and priced:
notes.append(hv.day_note("ahead"))
notes += [notes_b[n] for n in names if n in notes_b]
return {"key": ("today", "tomorrow")[k], "date": brussels(d0).date()
.isoformat(),
"length_h": int(round(_hours_between(d0, d1))),
"line": line, "notes": list(dict.fromkeys(notes)),
"sun": {"forecast": any(x is not None for x in fcs),
"estimate": day is not None and not day.periods,
"measured": any(h["measured_kwh"] is not None
for _s, h in hours),
"house": house is not None},
"hours": [h for _s, h in hours], "batteries": batts}
def _tomorrow_line(status) -> str:
"""Home's own line about tomorrow's prices, for the tomorrow tab."""
return (status.get("plan") or {}).get("tomorrow_line") or ""
def _line(c, inv, row, mine, running, key, shadow) -> str:
"""What to say for a battery with no block: the plan's own words."""
if not running:
return hv.why_words(key)
if row.get("use") is False:
return hv.why_words("off")
if mine:
return ""
if getattr(c, "modes_on", False) and row.get("mode_choice") in \
hv.PRICE_MODES:
return hv.day_note("no_blocks")
return hv.run_house_what(shadow) + " " + hv.run_house_why(shadow)
+72 -17
View File
@@ -16,6 +16,7 @@ them, so nothing here invents a plan. Every sentence is in the words
section at the bottom and is held to STYLE.md by test_messages.py.
"""
import contextlib
from datetime import datetime, timedelta, timezone
from . import arbitrage, messages, peakguard, uitext
@@ -212,25 +213,26 @@ def _price_mode_in_use(c):
return False
def price_windows(c, plan, verdict):
def price_windows(c, plan, verdict, now=None):
"""`[(kind, start, end)]`: where the add-on itself buys ("cheap") and
gives ("dear"). With battery modes on that is the kept `priceruns.RunPlan`
(`c.mode_runs`): the cheap run when charging in it pays, and every give
run. With them off it is the arbitrage verdict's two points, the ones
`trade_runs()` draws. Nothing is classified here."""
(`c.mode_runs`): the cheap run when charging in it pays and the trade gate
allows it (UX-53), and every give run. With them off it is the arbitrage
verdict's legs the add-on will take, the ones `trade_runs()` draws
(`trade_legs()`, given `now`). Nothing is classified here."""
if getattr(c, "modes_on", False):
rp = getattr(c, "mode_runs", None)
if rp is None or not rp.on or not _price_mode_in_use(c):
return []
out = [("cheap", rp.cheap.start, rp.cheap.end)] \
if rp.charge_useful() else []
if rp.charge_useful() and getattr(c, "mode_gate", None) is None \
else []
return out + [("dear", g.start, g.end) for g in rp.give]
if verdict is None:
return []
return [("cheap", verdict.cheapest.start,
_point_end(plan, verdict.cheapest)),
("dear", verdict.dearest.start,
_point_end(plan, verdict.dearest))]
return [("cheap" if kind == "grid" else "dear", pt.start,
_point_end(plan, pt))
for kind, pt in trade_legs(c, plan, verdict, now)]
def day_range(points):
@@ -306,6 +308,27 @@ def _point_end(plan, point):
return point.start + timedelta(hours=1)
def leg_taken(c, plan, pt, now):
"""UX-53: will the add-on act on this point of the verdict? One in
progress is the verdict's own answer now. One still ahead is asked of the
same gate (`trade_verdict()`, the call the control cycle makes) at the
point's own start. A charge that comes after the last dear period meets a
flat remainder there and is refused, so it is not announced. When the gate
allows it, the point is still the cheapest (dearest) one: the points
before it are gone and it was the first extreme of the points from now."""
if now is None or pt.start <= now:
return True
return trade_verdict(c, plan, pt.start)[0] is not None
def trade_legs(c, plan, verdict, now):
"""`[(kind, point)]` by start: the verdict's charge ("grid") and give
("give") legs that the add-on will take (`leg_taken()`)."""
legs = sorted((("grid", verdict.cheapest), ("give", verdict.dearest)),
key=lambda kp: kp[1].start)
return [(k, pt) for k, pt in legs if leg_taken(c, plan, pt, now)]
# Under this many kWh a block is not drawn: the cycle would not move it.
MIN_KWH = 0.05
@@ -322,10 +345,8 @@ def trade_runs(c, plan, verdict, day0, day1, now, shadow):
if any(inv.soc is None for inv in c.inverters):
return [], [], "no_level"
level = [float(inv.soc) for inv in c.inverters]
legs = sorted((("grid", verdict.cheapest), ("give", verdict.dearest)),
key=lambda kp: kp[1].start)
runs, later = [], []
for kind, pt in legs:
for kind, pt in trade_legs(c, plan, verdict, now):
end = _point_end(plan, pt)
if end <= now:
continue
@@ -371,8 +392,9 @@ def mode_runs(c, day0, day1, now, shadow):
them (`c.mode_runs`): the cheap run on the batteries that may charge
from the grid, the give runs on Leftover solar and Cheap hours, and the
sell run on those with selling on. Amounts are upper bounds ("up to"),
from the battery's own room, rating and the run's length. Returns
`(runs, no_trade_key)`."""
from the battery's own room, rating and the run's length. UX-53: no grid
charge is drawn while the trade gate refuses one (`c.mode_gate`, what the
control cycle found). Returns `(runs, no_trade_key)`."""
rp = getattr(c, "mode_runs", None)
if getattr(c, "mode_tick", None) is None:
return [], None # no cycle yet: nothing is known
@@ -381,6 +403,7 @@ def mode_runs(c, day0, day1, now, shadow):
if not rp.on:
return [], "not_worth"
floor, ceiling = arbitrage.soc_bounds(c.o)
gate = getattr(c, "mode_gate", None)
out = []
for i, inv in enumerate(c.inverters):
pm = c.pack_mode(inv)
@@ -397,8 +420,9 @@ def mode_runs(c, day0, day1, now, shadow):
blocks = []
# P4: with a day plan, the grid block is the planned shortfall.
need = plan_need(c, inv.name)
if rp.charge_useful() and (need > 0.0 if need is not None
else pm.mode in GRID_CHARGE_MODES):
if gate is None and rp.charge_useful() and (
need > 0.0 if need is not None
else pm.mode in GRID_CHARGE_MODES):
blocks.append(("grid", rp.cheap))
sell = rp.sell if pm.sell and pm.can_sell else None
for g in rp.give:
@@ -589,6 +613,19 @@ def region_of(opts):
if rid in REGION_PRESETS else None}
def runs_for(c, day0, day1, now, shadow):
"""UX-54: the price blocks of the window `[day0, day1)`, by the same
builders `build_plan()` uses for today: the batteries' price runs with
modes on, the verdict's legs with them off. Nothing is decided here."""
if getattr(c, "modes_on", False):
return mode_runs(c, day0, day1, now, shadow)[0]
fresh = c.fresh_tariff_plan(now)
verdict, _why = trade_verdict(c, fresh, now)
if verdict is None:
return []
return trade_runs(c, fresh, verdict, day0, day1, now, shadow)[0]
def build_plan(c, now, shadow, running, held=False, settled=True):
day0, day1 = local_day(now)
fresh = c.fresh_tariff_plan(now)
@@ -608,7 +645,7 @@ def build_plan(c, now, shadow, running, held=False, settled=True):
runs, lines = [], []
# The control cycle trades on the fresh plan only, so this does too.
verdict, trade_why = trade_verdict(c, fresh, now)
bars = price_bars(seen, day0, day1, price_windows(c, fresh, verdict),
bars = price_bars(seen, day0, day1, price_windows(c, fresh, verdict, now),
now, days.kinds)
modes_on = bool(getattr(c, "modes_on", False))
if running:
@@ -978,6 +1015,16 @@ def state_of(c, level, shadow, waiting=False):
return "running"
@contextlib.contextmanager
def in_lang(lang):
"""UX-54: build sentences in `lang` for the length of a `with` block."""
_LANG[0] = uitext.lang_of(lang)
try:
yield
finally:
_LANG[0] = "en"
def build(c, *, level: str, checks: list, now: datetime | None = None,
lang: str = "en", held: bool | None = None,
waiting: bool = False) -> dict:
@@ -1306,6 +1353,11 @@ def run_house_amount() -> str:
return _t("run.house_amount")
def day_note(key, **values) -> str:
"""UX-54: one sentence of the Forecast tab (`home_words.day`)."""
return _t("day." + key, **values)
def line_solar(fc_kwh, count_kwh, at) -> str:
return _t("solar.line", kwh=_kwh(fc_kwh), count=_kwh(count_kwh), at=at)
@@ -1479,6 +1531,9 @@ def _samples() -> list:
out += [("plan", today_words(k)) for k in words["today"]]
out += [("plan", line_solar(9.2, 6.4, "10:15")),
("plan", _t("solar.no_forecast")), ("plan", _t("solar.learning"))]
out += [("plan", day_note(k, pct=70, name="Battery A", at="08:15",
kwh="9.2 kWh", count="6.4 kWh"))
for k in words["day"]]
out += [("plan", line_no_trade(k)) for k in
("not_covered", "no_prices", "peak", "not_firm", "not_worth",
"no_level")]
+30 -1
View File
@@ -64,7 +64,8 @@ from .clamp import (
shortfall_note, write_clamp_note,
)
from . import freshness
from . import homeview
from . import dayview, homeview
from .daycurve import SolarToday
from . import messages
from .freeze import (
FreezeWatchdog, FreezeWatchdogConfig, looks_frozen, rail_unreachable, rail_w,
@@ -1001,6 +1002,10 @@ class Controller:
self.mode_runs = None
self._runs_key = None
self.mode_tick = None
# UX-53: why the trade gate refused a grid charge this tick (`peak`,
# `not_worth`, `not_covered`), or None when it allowed one. Home reads
# it; nothing in control does.
self.mode_gate = None
self._run_done = {}
self._mode_trade_was = False
self._charge_blocked_q = None
@@ -1257,6 +1262,9 @@ class Controller:
# UX-13 P1: today's prices as the Home chart has seen them. Display
# only; nothing reads it back into a decision.
self.home_prices = homeview.DayPrices()
# UX-54: the site's measured solar per hour, for the Forecast tab.
# In memory, display only; nothing reads it back into a decision.
self.solar_today = SolarToday()
# TARIFF-11. When the most recent cycle() STARTED. `arb_at < cycle_at`
# is the exact, derived statement of "the latest cycle did not get as
# far as the arbitrage step", which is the fact the *Arbitrage* row
@@ -2380,6 +2388,7 @@ class Controller:
charge_ok = sell_ok and allowed(guard)
trade_why = (None if charge_ok else "peak" if sell_ok
else "not_worth")
self.mode_gate = None if charge_ok else trade_why
price_now = None
if plan is not None:
try:
@@ -3376,6 +3385,11 @@ class Controller:
*(self.hass.state(entity) for entity in roster))
self.pv_w, self.pv_why, self.pv_parts = pv_read.total(
list(zip(roster, states)), pv_read.inverted(self.o))
# UX-54: kept for the Forecast tab only. It must never cost a cycle.
try:
self.solar_today.add(datetime.now(timezone.utc), self.pv_w)
except Exception: # noqa: BLE001
_LOG.debug("measured solar not kept", exc_info=True)
def _peak_fc_unit_log(self, fault) -> None:
"""SAFETY-118. One WARNING when the peak forecast's unit cannot be read
@@ -7630,6 +7644,21 @@ class Controller:
out["home_error"] = True
return out
def forecast_payload(self, lang: str = "en") -> dict:
"""UX-54: the Forecast tab's data. Built on request, from what the
control cycle already holds; no Home Assistant call. Today's blocks are
Home's own, so `status_payload()` is built first and read."""
try:
return dayview.build(self, self.status_payload(lang=lang),
datetime.now(timezone.utc), lang)
except Exception: # noqa: BLE001
if not getattr(self, "_forecast_failed", False):
self._forecast_failed = True
_LOG.exception("The Forecast tab could not be built")
else:
_LOG.debug("The Forecast tab failed again", exc_info=True)
return {"ok": False}
async def handle_action(self, what: str, *, unit=None, value=None):
now = datetime.now(timezone.utc)
if what in ("pack_use", "pack_mode", "pack_sell"):
+26
View File
@@ -246,3 +246,29 @@ select.sel:disabled{opacity:.6}
.simple .card .lbl{font-weight:600;font-size:15px}
.simple .inv .hd{border-bottom:0}
.simple .inv .f{border-top:0;padding-top:0}
/* UX-54: the Forecast screen */
.fcline{text-align:right}
.fcnote{background:var(--info-bg);border-radius:10px;padding:8px 10px;margin:0 0 8px}
.fcbars{display:flex;align-items:flex-end;gap:2px;height:110px}
.fcol{flex:1;position:relative;height:100%;min-width:0}
.fcol i{position:absolute;display:block}
.fcol .sun{bottom:0;left:0;right:0;background:var(--solar);border-radius:3px 3px 0 0;opacity:.85}
.fcol .sun.est{opacity:.5}
.fcol .got{bottom:0;left:30%;right:30%;background:repeating-linear-gradient(135deg,var(--ink) 0 3px,var(--card) 3px 5px);border:1px solid var(--ink);border-bottom:0}
.fcol .use{left:-1px;right:-1px;height:3px;background:var(--house)}
.fcols{position:absolute;inset:0;display:flex;gap:0}
.fcols>i{flex:1;min-width:0;display:block}
.lvlchart{position:relative;height:110px;margin:18px 0 4px}
.lvlchart svg{position:absolute;inset:0;width:100%;height:100%;display:block}
.lvlchart .yl{position:absolute;right:2px;font-size:10px;color:var(--faint);pointer-events:none}
.fcnow{position:absolute;top:-4px;bottom:0;width:2px;background:var(--ink);opacity:.7;pointer-events:none}
.fcnow span{position:absolute;top:-15px;left:-10px;font-size:10px}
.legend i.sun{background:var(--solar)}
.legend i.got{background:repeating-linear-gradient(135deg,var(--ink) 0 2px,var(--card) 2px 4px);border:1px solid var(--ink)}
.legend i.use{height:3px;background:var(--house)}
.legend i.lv{height:3px;background:var(--batt)}
table.fct{width:100%;border-collapse:collapse;font-size:12px;font-variant-numeric:tabular-nums}
table.fct th,table.fct td{padding:4px 3px;text-align:right;border-bottom:1px solid var(--line)}
table.fct th:first-child,table.fct td:first-child{text-align:left}
table.fct th{color:var(--muted);font-weight:600}
.ymax{position:absolute;top:-2px;left:2px;font-size:10px;color:var(--faint);pointer-events:none}
+173 -2
View File
@@ -448,6 +448,175 @@
});
}
/* ---------------- Forecast (UX-54) ---------------- */
// Today and tomorrow, per hour: the solar forecast and what was measured,
// the house's expected use, and each battery's expected level with its
// blocks. The data is built by app/dayview.py when the tab asks (GET
// forecast), the sentences in the page's language. A part that is not
// known yet comes as a note, so no chart is ever drawn empty.
// The SVG namespace, read off the page's own icons: the page names no URL.
const SVGNS = document.querySelector("svg").namespaceURI;
function sv(tag, attrs) {
const e = document.createElementNS(SVGNS, tag);
for (const k in attrs || {}) e.setAttribute(k, attrs[k]);
return e;
}
let FC = null;
let fcDay = 0;
let fcBusy = false;
let fcTimer = null;
const fmtKwh = (x) => dec(Number(x).toFixed(1));
const hoursOf = (d) => d.length_h || d.hours.length || 24;
function endLabel(d, i) { return i + 1 < d.hours.length ? d.hours[i + 1].label : "24:00"; }
function hourTip(d, i) {
const x = d.hours[i];
const lines = [];
if (x.forecast_kwh != null) lines.push(t("forecast.forecast") + ": " + fmtKwh(x.forecast_kwh) + " kWh");
if (x.measured_kwh != null) lines.push(t("forecast.measured") + ": " + fmtKwh(x.measured_kwh) + " kWh");
if (x.house_kwh != null) lines.push(t("forecast.house") + ": " + fmtKwh(x.house_kwh) + " kWh");
d.batteries.forEach((b, j) => { if (x.levels[j] != null) lines.push(b.name + ": " + dec(Math.round(x.levels[j]).toString()) + " %"); });
return [x.label + " " + t("plan.time_to") + " " + endLabel(d, i)].concat(lines);
}
function hourItem(tag, cls, d, i, kids) {
const tip = hourTip(d, i);
const e = h(tag, { class: cls, tabindex: "-1", role: "img", "aria-label": tip.join(". "), "data-tip": JSON.stringify(tip) }, kids);
return e;
}
function renderSun(d) {
const card = $("#fc-sun");
const any = d.sun.forecast || d.sun.measured || d.sun.house;
card.hidden = !any;
if (!any) return;
const mx = Math.max(0.5, ...d.hours.map((x) => Math.max(x.forecast_kwh || 0, x.measured_kwh || 0, x.house_kwh || 0)));
const pct = (v) => Math.max(0, Math.min(100, v / mx * 100)) + "%";
const bars = $("#fc-bars");
bars.setAttribute("aria-label", t("forecast.sun"));
bars.replaceChildren(...d.hours.map((x, i) => hourItem("div", "fcol", d, i, [
x.forecast_kwh != null ? h("i", { class: "sun" + (d.sun.estimate ? " est" : ""), style: "height:" + pct(x.forecast_kwh) }) : null,
x.measured_kwh != null ? h("i", { class: "got", style: "height:" + pct(x.measured_kwh) }) : null,
x.house_kwh != null ? h("i", { class: "use", style: "bottom:calc(" + pct(x.house_kwh) + " - 1px)" }) : null])));
roving(bars);
const chart = $("#fc-sunchart");
chart.querySelectorAll(".fcnow, .ymax").forEach((e) => e.remove());
chart.append(h("span", { class: "ymax", "aria-hidden": "true" }, fmtKwh(mx)));
if (fcDay === 0) chart.append(nowMark(d));
const used = [["sun", d.sun.forecast, "forecast"], ["got", d.sun.measured, "measured"], ["use", d.sun.house, "house"]];
$("#fc-legend").replaceChildren(...used.filter((u) => u[1]).map((u) => h("span", {}, h("i", { class: u[0], "aria-hidden": "true" }), t("forecast." + u[2]))));
}
function nowMark(d) {
return h("div", { class: "fcnow", "aria-hidden": "true", style: "left:" + Math.max(0, Math.min(100, FC.now_h / hoursOf(d) * 100)) + "%" }, h("span", {}, t("plan.now")));
}
function levelChart(d, b, bi) {
const L = hoursOf(d), W = L * 10;
const x = (hr) => hr * 10;
const box = h("div", { class: "chart" });
const area = h("div", { class: "lvlchart" });
const svg = sv("svg", { viewBox: "0 0 " + W + " 100", preserveAspectRatio: "none", "aria-hidden": "true", focusable: "false" });
[0, 50, 100].forEach((p) => svg.append(sv("line", { x1: 0, x2: W, y1: 100 - p, y2: 100 - p, stroke: "var(--line)", "stroke-width": 1, "vector-effect": "non-scaling-stroke" })));
const fill = { grid: "var(--accent)", give: "var(--house)", cycle: "var(--warn)" };
for (const r of b.blocks) {
if (r.end_h <= r.start_h) continue;
svg.append(sv("rect", { x: x(r.start_h), y: 0, width: x(r.end_h) - x(r.start_h), height: 100, fill: fill[r.kind] || "var(--mid)", opacity: 0.18 }));
}
const pts = b.levels.map((p) => x(p[0]) + "," + (100 - p[1]));
svg.append(sv("polygon", { points: x(b.levels[0][0]) + ",100 " + pts.join(" ") + " " + x(b.levels[b.levels.length - 1][0]) + ",100", fill: "var(--batt)", opacity: 0.12 }));
svg.append(sv("polyline", { points: pts.join(" "), fill: "none", stroke: "var(--batt)", "stroke-width": 2.5, "stroke-linejoin": "round", "vector-effect": "non-scaling-stroke" }));
area.append(svg);
[["100", 0], ["50", 50], ["0", 100]].forEach((y) => area.append(h("span", { class: "yl", "aria-hidden": "true", style: "top:" + y[1] + "%;transform:translateY(" + (y[1] === 100 ? "-100%" : y[1] === 0 ? "0" : "-50%") + ")" }, y[0] + " %")));
area.append(h("div", { class: "fcols", role: "group", "aria-label": b.name }, ...d.hours.map((hx, i) => hourItem("i", "", d, i, null))));
roving(area.querySelector(".fcols"));
if (fcDay === 0) area.append(nowMark(d));
box.append(area);
// The blocks under the chart, drawn as Home draws them.
if (b.blocks.length) {
const lane = h("div", { class: "lane", role: "group", "aria-label": b.name });
for (const r of b.blocks) {
if (r.end_h <= r.start_h) continue;
lane.append(chartItem("b", r.kind, "left:" + (r.start_h / L * 100) + "%;width:" + ((r.end_h - r.start_h) / L * 100) + "%",
b.name + ", " + r.from + " " + t("plan.time_to") + " " + r.to, [r.what, r.why, r.amount]));
}
box.append(lane);
roving(lane);
}
box.append(h("div", { class: "axis", "aria-hidden": "true" }, h("span", {}, "00:00"), h("span", {}, "06:00"), h("span", {}, "12:00"), h("span", {}, "18:00"), h("span", {}, "24:00")));
return box;
}
function batteryCardFc(d, b, bi) {
const drawn = b.levels.length > 1;
const list = b.blocks.map((r) => h("li", {}, h("strong", {}, r.from + " " + t("plan.time_to") + " " + r.to + ": "), tipText([r.what, r.why, r.amount])));
return h("div", { class: "card" },
h("div", { class: "row spread" }, h("h2", {}, b.name),
h("span", { class: "tiny muted" }, (b.capacity_kwh ? fmtKwh(b.capacity_kwh) + " kWh · " : "") + t("forecast.levels"))),
drawn ? levelChart(d, b, bi) : null,
list.length ? h("ul", { class: "plain small" }, ...list) : null,
b.line ? h("p", { class: "small muted", style: "margin:6px 0 0" }, b.line) : null);
}
function renderTable(d) {
const more = $("#fc-more");
const any = d.sun.forecast || d.sun.measured || d.sun.house;
more.hidden = !any;
if (!any) return;
const lv = d.batteries.filter((b) => b.levels.length > 1);
const head = h("tr", {}, h("th", { scope: "col" }, t("forecast.hour")),
d.sun.forecast ? h("th", { scope: "col" }, t("forecast.solar")) : null,
d.sun.measured ? h("th", { scope: "col" }, t("forecast.got")) : null,
d.sun.house ? h("th", { scope: "col" }, t("forecast.use")) : null,
...lv.map((b) => h("th", { scope: "col" }, b.name + " %")));
const cell = (v, f) => h("td", {}, v == null ? "–" : f(v));
const rows = d.hours.map((x) => h("tr", {},
h("td", {}, x.label),
d.sun.forecast ? cell(x.forecast_kwh, fmtKwh) : null,
d.sun.measured ? cell(x.measured_kwh, fmtKwh) : null,
d.sun.house ? cell(x.house_kwh, fmtKwh) : null,
...lv.map((b) => cell(x.levels[d.batteries.indexOf(b)], (v) => dec(Math.round(v).toString())))));
$("#fc-table").replaceChildren(h("thead", {}, head), h("tbody", {}, ...rows));
}
function renderForecast() {
const loading = $("#fc-loading");
if (!FC || FC.ok !== true) {
loading.hidden = false; loading.textContent = FC ? t("offline.name") : t("forecast.loading");
$("#fc-notes").replaceChildren(); $("#fc-sun").hidden = true; $("#fc-batts").replaceChildren(); $("#fc-more").hidden = true; $("#fc-line").textContent = "";
return;
}
loading.hidden = true;
const d = FC.days[fcDay];
$("#fc-line").textContent = d.line || "";
$("#fc-notes").replaceChildren(...d.notes.map((n) => h("p", { class: "fcnote small" }, n)));
renderSun(d);
$("#fc-batts").replaceChildren(...d.batteries.map((b, i) => batteryCardFc(d, b, i)));
renderTable(d);
document.querySelectorAll("#fc-day button").forEach((b) => {
const on = Number(b.dataset.d) === fcDay;
b.setAttribute("aria-checked", String(on)); b.tabIndex = on ? 0 : -1;
});
}
async function loadForecast() {
if (fcBusy) return;
fcBusy = true;
try {
const r = await fetch("forecast?lang=" + LANG, { cache: "no-store" });
if (!r.ok) throw new Error("forecast " + r.status);
FC = await r.json();
} catch (e) { FC = { ok: false }; }
finally { fcBusy = false; }
renderForecast();
}
document.querySelectorAll("#fc-day button").forEach((b, i, all) => {
b.addEventListener("click", () => { fcDay = Number(b.dataset.d); renderForecast(); });
b.addEventListener("keydown", (e) => {
const j = { ArrowRight: i + 1, ArrowLeft: i - 1 + all.length }[e.key];
if (j === undefined) return;
e.preventDefault(); fcDay = j % all.length; renderForecast(); all[fcDay].focus();
});
});
// Every 30 s while the screen shows: the data moves slowly, and each load
// is built on request.
function scheduleForecast() {
const on = !document.hidden && $("#t-fc").getAttribute("aria-selected") === "true";
if (on && !fcTimer) { loadForecast(); fcTimer = setInterval(loadForecast, 30000); }
if (!on && fcTimer) { clearInterval(fcTimer); fcTimer = null; }
}
/* ---------------- control buttons ---------------- */
// The action is a toggle, so the page reads the live state first and sends
// nothing when control is already where the person asked it to be.
@@ -511,11 +680,13 @@
}
addEventListener("hashchange", fromHash);
/* ---------------- polling: every 3 s while Home or Batteries shows ---------------- */
/* ---------------- polling: every 3 s while Home, Batteries or Forecast shows ---------------- */
let timer = null;
function schedule() {
scheduleForecast();
const live = !document.hidden && ($("#t-home").getAttribute("aria-selected") === "true" ||
$("#t-batt").getAttribute("aria-selected") === "true");
$("#t-batt").getAttribute("aria-selected") === "true" ||
$("#t-fc").getAttribute("aria-selected") === "true");
if (live && !timer) { refresh(); timer = setInterval(refresh, 3000); }
if (!live && timer) { clearInterval(timer); timer = null; }
}
+27
View File
@@ -18,6 +18,7 @@
<nav class="tabs" role="tablist" id="tabs" data-t-label="tabs.label">
<button class="tab" role="tab" id="t-home" aria-controls="s-home" aria-selected="true" data-hash="home"><svg viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" aria-hidden="true"><path d="M3 11l9-7 9 7v9a1 1 0 0 1-1 1h-5v-6H9v6H4a1 1 0 0 1-1-1z"/></svg><span data-t="tabs.home"></span><span class="badge" id="attn-badge" hidden></span></button>
<button class="tab" role="tab" id="t-batt" aria-controls="s-batt" aria-selected="false" data-hash="batteries" tabindex="-1"><svg viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" aria-hidden="true"><rect x="3" y="7" width="16" height="10" rx="2"/><path d="M21 11v2"/><rect x="5.5" y="9.5" width="7" height="5" fill="currentColor" stroke="none"/></svg><span data-t="tabs.batteries"></span></button>
<button class="tab" role="tab" id="t-fc" aria-controls="s-fc" aria-selected="false" data-hash="forecast" tabindex="-1"><svg viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" aria-hidden="true"><path d="M3 20h18M4 16l5-6 4 4 7-9"/></svg><span data-t="tabs.forecast"></span></button>
<button class="tab" role="tab" id="t-set" aria-controls="s-set" aria-selected="false" data-hash="settings" tabindex="-1"><svg viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" aria-hidden="true"><path d="M4 6h10M18 6h2M4 12h4M12 12h8M4 18h12M20 18h0"/><circle cx="16" cy="6" r="2"/><circle cx="10" cy="12" r="2"/><circle cx="18" cy="18" r="2"/></svg><span data-t="tabs.settings"></span></button>
<button class="tab" role="tab" id="t-wiz" aria-controls="s-wiz" aria-selected="false" data-hash="setup" tabindex="-1"><svg viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" aria-hidden="true"><path d="M14 7l3 3M5 19l9.5-9.5 3 3L8 22H5z"/><path d="M19 2v4M17 4h4"/></svg><span data-t="tabs.setup"></span></button>
</nav>
@@ -114,6 +115,32 @@
</details>
</section>
<section class="screen" id="s-fc" role="tabpanel" aria-labelledby="t-fc" hidden>
<div class="row spread" style="margin-bottom:8px">
<div class="seg" role="radiogroup" id="fc-day" data-t-label="forecast.day">
<button type="button" role="radio" aria-checked="true" data-d="0" data-t="forecast.today"></button>
<button type="button" role="radio" aria-checked="false" data-d="1" tabindex="-1" data-t="forecast.tomorrow"></button>
</div>
<span class="tiny muted grow fcline" id="fc-line"></span>
</div>
<div id="fc-notes" aria-live="polite"></div>
<p class="small muted" id="fc-loading" data-t="forecast.loading"></p>
<div class="card" id="fc-sun" hidden>
<div class="row spread"><h2 data-t="forecast.sun"></h2><span class="tiny muted" data-t="forecast.unit"></span></div>
<div class="chart" id="fc-sunchart">
<div class="fcbars" id="fc-bars" role="group"></div>
<div class="axis" aria-hidden="true"><span>00:00</span><span>06:00</span><span>12:00</span><span>18:00</span><span>24:00</span></div>
</div>
<div class="legend" id="fc-legend"></div>
<p class="tiny muted" style="margin:6px 0 0" data-t="plan.description"></p>
</div>
<div id="fc-batts"></div>
<details class="grp" id="fc-more" hidden>
<summary><span class="grow"><strong data-t="forecast.table"></strong></span><span class="chev" aria-hidden="true">&#8250;</span></summary>
<div class="body"><table class="fct" id="fc-table"></table></div>
</details>
</section>
<section class="screen" id="s-set" role="tabpanel" aria-labelledby="t-set" hidden>
<div class="row spread" style="margin-bottom:6px">
<h2 style="margin:0" data-s="name"></h2>
+6 -2
View File
@@ -1103,8 +1103,12 @@ class TariffPlan:
f"the tariff is flat - every one of the {len(firm)} firm "
f"points, spanning {_whole_hours(firm)} hours, costs "
f"{firm[0].eur_per_kwh:.4f} EUR/kWh, so buying low and "
f"selling high is arithmetically impossible. §9.2: a fully "
f"fixed product is self-consumption only")
f"selling high is arithmetically impossible"
# UX-53: only a real fixed product is "fully fixed". A
# dynamic or day/night site whose remaining periods happen to
# be equal trades again when they differ.
+ (". §9.2: a fully fixed product is self-consumption only"
if self.product is PriceProduct.FIXED else ""))
# REFUSAL 5, NEW IN FCST-03 (AC3). One widening path, driven by the set
# of active uncertainty sources: the product's own (§9.2 row 2, derived
+36 -3
View File
@@ -45,7 +45,7 @@
},
"home": {
"app": {"name": "Battery add-on"},
"tabs": {"home": "Home", "batteries": "Batteries", "settings": "Settings", "setup": "Setup", "label": "Screens"},
"tabs": {"home": "Home", "batteries": "Batteries", "forecast": "Forecast", "settings": "Settings", "setup": "Setup", "label": "Screens"},
"now": {"name": "Now", "description": "Updated every 3 seconds.", "solar": "Solar", "house": "House", "grid": "Grid", "batteries": "Batteries"},
"plan": {
"name": "Today's plan",
@@ -68,6 +68,24 @@
"no_prices": {"description": "No prices for today, so the chart shows only the plan."},
"list": "The plan in words"
},
"forecast": {
"name": "Forecast",
"day": "Day",
"today": "Today",
"tomorrow": "Tomorrow",
"sun": "Solar and house",
"unit": "kWh per hour",
"forecast": "Solar forecast",
"measured": "Solar measured",
"house": "House use",
"levels": "Battery level",
"table": "Hour by hour",
"hour": "Hour",
"solar": "Solar",
"got": "Measured",
"use": "House",
"loading": "Loading the forecast."
},
"peak": {"name": "Peak limit", "this_month": "This month's peak", "limit": "Limit", "none": "None recorded yet", "unreadable": "Cannot be read now"},
"attention": {"name": "Needs attention", "nothing": "Nothing to look at", "to_look": "to look at", "details": "Details", "alarm": "Alarm", "warning": "Warning", "info": "Info"},
"control": {"name": "Control", "stop": "Stop control", "start": "Start control", "maint_start": "Start a battery cycle now", "maint_abort": "Stop the battery cycle", "docs": "Add-on documentation", "more": "More actions"},
@@ -367,11 +385,11 @@
"run": {
"grid_what": "Charges up to {kwh} from the grid.",
"grid_what_would": "Would charge up to {kwh} from the grid.",
"grid_why": "Lowest published price, {eur} per kWh.",
"grid_why": "Lowest price left, {eur} per kWh.",
"grid_amount": "Costs up to {eur}.",
"give_what": "Gives up to {kwh}, to the house first.",
"give_what_would": "Would give up to {kwh}, to the house first.",
"give_why": "Highest published price, {eur} per kWh.",
"give_why": "Highest price left, {eur} per kWh.",
"sell_what": "Sells up to {kwh}, to the house first.",
"sell_what_would": "Would sell up to {kwh}, to the house first.",
"cheap_why": "Cheapest hours, about {eur} per kWh.",
@@ -425,6 +443,21 @@
"known": "Tomorrow's prices are in.",
"unknown": "Tomorrow's prices usually arrive after 13:00."
},
"day": {
"modes_off": "Battery modes are off. No forecast is read and no level is planned.",
"not_set": "No solar forecast is set up. Without one, no battery level is planned.",
"no_forecast": "The solar forecast is missing or too old. The levels return when it is back.",
"learning": "The add-on is learning your house's use. The levels start after one full day.",
"no_tomorrow": "There is no solar forecast for tomorrow yet. The levels stop at sunrise.",
"estimate": "The forecast gives one total for the day. The add-on spreads it over the hours.",
"trust": "The levels count on {pct} % of the forecast.",
"cycle": "{name} does its monthly cycle. Its level is not shown from the start.",
"no_level": "{name} has no level reading yet. Its levels are not shown.",
"measured_from": "Measured solar is counted from {at}, when the add-on started.",
"tomorrow_total": "Solar expected tomorrow: {kwh}, we count on {count}.",
"no_blocks": "No charge or give blocks planned.",
"ahead": "The add-on plans prices about a day ahead. Blocks for later hours appear as they come near."
},
"region": {
"preset": "Levies and VAT: {region}, {year} figures.",
"own": "Levies and VAT: your own figure."
+36 -3
View File
@@ -42,7 +42,7 @@
"applies": {"restart": {"name": "Geldt na een herstart"}, "now": {"name": "Geldt meteen"}},
"home": {
"app": {"name": "Batterij-add-on"},
"tabs": {"home": "Start", "batteries": "Batterijen", "settings": "Instellingen", "setup": "Installatie", "label": "Schermen"},
"tabs": {"home": "Start", "batteries": "Batterijen", "forecast": "Prognose", "settings": "Instellingen", "setup": "Installatie", "label": "Schermen"},
"now": {"name": "Nu", "description": "Elke 3 seconden bijgewerkt.", "solar": "Zonnepanelen", "house": "Huis", "grid": "Net", "batteries": "Batterijen"},
"plan": {
"name": "Plan voor vandaag",
@@ -65,6 +65,24 @@
"no_prices": {"description": "Geen prijzen voor vandaag, dus de grafiek toont alleen het plan."},
"list": "Het plan in woorden"
},
"forecast": {
"name": "Prognose",
"day": "Dag",
"today": "Vandaag",
"tomorrow": "Morgen",
"sun": "Zon en huis",
"unit": "kWh per uur",
"forecast": "Zonneprognose",
"measured": "Zon gemeten",
"house": "Huisverbruik",
"levels": "Laadniveau",
"table": "Uur per uur",
"hour": "Uur",
"solar": "Zon",
"got": "Gemeten",
"use": "Huis",
"loading": "De prognose wordt geladen."
},
"peak": {"name": "Piekgrens", "this_month": "Piek van deze maand", "limit": "Grens", "none": "Nog niets gemeten", "unreadable": "Nu niet leesbaar"},
"attention": {"name": "Vraagt aandacht", "nothing": "Niets om na te kijken", "to_look": "na te kijken", "details": "Details", "alarm": "Alarm", "warning": "Waarschuwing", "info": "Info"},
"control": {"name": "Regeling", "stop": "Regeling stoppen", "start": "Regeling starten", "maint_start": "Nu een batterijcyclus starten", "maint_abort": "Batterijcyclus stoppen", "docs": "Documentatie van de add-on", "more": "Meer acties"},
@@ -307,11 +325,11 @@
"run": {
"grid_what": "Laadt tot {kwh} van het net.",
"grid_what_would": "Zou tot {kwh} van het net laden.",
"grid_why": "Laagste gepubliceerde prijs, {eur} per kWh.",
"grid_why": "Laagste prijs die nog over is, {eur} per kWh.",
"grid_amount": "Kost tot {eur}.",
"give_what": "Levert tot {kwh}, eerst aan het huis.",
"give_what_would": "Zou tot {kwh} leveren, eerst aan het huis.",
"give_why": "Hoogste gepubliceerde prijs, {eur} per kWh.",
"give_why": "Hoogste prijs die nog over is, {eur} per kWh.",
"sell_what": "Verkoopt tot {kwh}, eerst aan het huis.",
"sell_what_would": "Zou tot {kwh} verkopen, eerst aan het huis.",
"cheap_why": "Goedkoopste uren, ongeveer {eur} per kWh.",
@@ -353,6 +371,21 @@
"period_unmeasured": "Een deel van dit kwartier is niet gemeten.",
"prices": {"now": "Prijs nu {eur} per kWh, alles inbegrepen.", "today": "Vandaag: laagste {lo} om {lo_at}, hoogste {hi} om {hi_at}.", "none": "Geen prijzen voor vandaag."},
"tomorrow": {"known": "De prijzen van morgen zijn binnen.", "unknown": "De prijzen van morgen komen meestal na 13:00."},
"day": {
"modes_off": "Batterijmodi staan uit. Er wordt geen prognose gelezen en geen niveau gepland.",
"not_set": "Er is geen zonneprognose ingesteld. Zonder prognose wordt geen laadniveau gepland.",
"no_forecast": "De zonneprognose ontbreekt of is te oud. De niveaus komen terug als ze er weer is.",
"learning": "De add-on leert het verbruik van uw huis. De niveaus starten na één volle dag.",
"no_tomorrow": "Er is nog geen zonneprognose voor morgen. De niveaus stoppen bij zonsopgang.",
"estimate": "De prognose geeft één totaal voor de dag. De add-on verdeelt het over de uren.",
"trust": "De niveaus rekenen op {pct} % van de prognose.",
"cycle": "{name} doet de maandelijkse cyclus. Het niveau wordt niet getoond vanaf de start.",
"no_level": "{name} geeft nog geen laadniveau door. De niveaus worden niet getoond.",
"measured_from": "Gemeten zon telt vanaf {at}, toen de add-on startte.",
"tomorrow_total": "Verwachte zon morgen: {kwh}, we rekenen op {count}.",
"no_blocks": "Geen laad- of leverblokken gepland.",
"ahead": "De add-on plant prijzen ongeveer een dag vooruit. Blokken voor latere uren verschijnen als ze naderen."
},
"region": {"preset": "Heffingen en btw: {region}, cijfers van {year}.", "own": "Heffingen en btw: uw eigen bedrag."},
"peak_not_covered": "Geen piekgrens: deze meter meet geen kwartieren.",
"control": {"shadow": "Schaduwmodus: elke omvormer blijft op 0 W.", "running": "De regeling is actief.", "stopped": "De regeling is gestopt. Elke omvormer staat op 0 W."},
+7
View File
@@ -150,6 +150,12 @@ def build_app(controller, selfcfg=None) -> web.Application:
return web.json_response(controller.status_payload())
return web.json_response(controller.status_payload(lang=lang))
async def forecast(request):
"""UX-54: the Forecast tab's data, built when the tab asks. Reads only."""
build = getattr(controller, "forecast_payload", None)
lang = uitext.lang_of(request.query.get("lang"))
return web.json_response(build(lang=lang) if build else {"ok": False})
async def action(request):
try:
body = await request.json()
@@ -471,6 +477,7 @@ def build_app(controller, selfcfg=None) -> web.Application:
app.add_routes([
web.get("/", index),
web.get("/status", status),
web.get("/forecast", forecast),
web.post("/action", action),
web.get("/config", config_index),
web.get("/config/model", config_model),
+6 -4
View File
@@ -523,7 +523,7 @@ _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} "
and _grid[0]["why"] == f"Lowest price left, €{_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"]
@@ -831,10 +831,12 @@ check("the page is assembled: no placeholder is left and the words parse, "
"/*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",
check("five tabs with the five hash routes, in a keyboard tablist (UX-54 "
"adds Forecast beside Home and Batteries)",
re.findall(r'data-hash="(\w+)"', PAGE) == ["home", "batteries",
"settings", "setup"]
and 'role="tablist"' in PAGE and PAGE.count('role="tabpanel"') == 4
"forecast", "settings",
"setup"]
and 'role="tablist"' in PAGE and PAGE.count('role="tabpanel"') == 5
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 "
+5 -3
View File
@@ -636,10 +636,12 @@ _app = web.build_app(FakeController(), FakeSupervisor())
_routes = sorted({(r.method, r.get_info().get("path"))
for r in _app.router.routes() if r.method != "HEAD"})
print(" " + " ".join(f"{m} {p}" for m, p in _routes))
check("the configuration surface adds five routes and UX-13 one GET, and "
"none touches the three that were there",
check("the configuration surface adds five routes, UX-13 one GET and "
"UX-54 one more GET (the Forecast tab), and none touches the three "
"that were there",
_routes == [("GET", "/"), ("GET", "/config"), ("GET", "/config/model"),
("GET", "/ha/entities"), ("GET", "/status"),
("GET", "/forecast"), ("GET", "/ha/entities"),
("GET", "/status"),
("POST", "/action"), ("POST", "/config/restart"),
("POST", "/config/save"), ("POST", "/config/validate")])
_ents_src = io.open(os.path.join(HERE, "app", "haentities.py"),
+589
View File
@@ -0,0 +1,589 @@
"""Runnable check for UX-53. `python3 test_ux53.py`
Home draws a grid charge only where the add-on will take it.
1. Modes on (F-1): inside a cheap run that is under way, the trade gate
(`TariffPlan.arbitrage()`, as `Controller._mode_roles()` asks it) refuses
with `not_worth`. The control cycle files that as `Controller.mode_gate`,
and Home draws no charge, and marks no cheap quarter, from then on. With
the gate open the charge is drawn as before. The real `_mode_roles()` runs
at each instant, so the page is graded against the decision it makes.
2. Modes off (F-2): the verdict's charge comes after its last dear period.
Home asks the same gate at the charge's own start (`homeview.leg_taken()`):
a flat remainder is refused there, so no charge is drawn and no quarter is
marked cheap. A charge the gate allows at its start is drawn as before, and
so is the sale.
3. Wording, English and Dutch: "lowest price left", and "a fully fixed
product" only for a real fixed product.
4. Day/night and price-off pages are the same as before.
Mutations (MUTATION.md, purged copies): listed in the pull request.
"""
import asyncio
import json
import logging
import os
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)
from app import arbitrage, homeview, modetrade, packmode # noqa: E402
from app.main import Controller # noqa: E402
from app.peakguard import PeakDecision # noqa: E402
from app.price import (PricePoint, PriceProduct, PriceRead, # noqa: E402
Uncertainty, brussels)
from app.static_tariff import TariffPlan, static_plan # noqa: E402
from app.store import unit_key # noqa: E402
fails = []
def check(name, cond):
print(f" {'ok ' if cond else 'FAIL'} {name}")
if not cond:
fails.append(name)
UTC = timezone.utc
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 []
BASE = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
"meter_source": "off", "max_w": 5000, "shadow_mode": "off",
"maintenance_enabled": False, "arbitrage_enabled": True,
"cycle_cost_threshold_eur": 0.10,
"inverters": [
{"name": "inverter-a", "setpoint_entity": "number.a",
"soc_entity": "sensor.soc_a", "batt_entity": "sensor.batt_a",
"max_w": 2500, "capacity_kwh": 2.0},
{"name": "inverter-b", "setpoint_entity": "number.b",
"soc_entity": "sensor.soc_b", "batt_entity": "sensor.batt_b",
"max_w": 2500, "capacity_kwh": 2.0}]}
# Tuesday 27 October 2026 in Brussels (UTC+1 since 25 October).
DAY0 = datetime(2026, 10, 26, 23, 0, tzinfo=UTC)
def at(h, m=0, s=0):
return DAY0 + timedelta(hours=h, minutes=m, seconds=s)
def quarters(price_of, days=2, product=PriceProduct.DAY_AHEAD):
"""48 hours of 15-minute prices; `price_of(local_hour_float)` -> EUR/kWh."""
pts = []
for i in range(96 * days):
t = DAY0 + timedelta(minutes=15 * i)
pts.append(PricePoint(brussels(t), price_of((i % 96) / 4.0), False))
return TariffPlan(PriceRead.series(tuple(pts), product,
"generic_dayahead", DAY0), None,
"provider")
def make(opts, plan, now, *, modes=None, soc=(50.0, 50.0)):
c = Controller(dict(opts), _Hass(), _Store(), _Pub(), stop=asyncio.Event())
c.auto = True
for i, inv in enumerate(c.inverters):
inv.soc, inv.batt, inv.last_write = soc[i], 0.0, 0.0
inv.dev_min, inv.dev_max = -5000.0, 5000.0
c.grid = 500.0
c.tariff_plan, c.tariff_at = plan, now
c.peak_guard = PeakDecision(engaged=False, floor_w=None, ceiling_w=3500.0,
ceiling_from="cap", projected_w=800.0)
c.arb_headroom_w = 3000.0
if modes:
for inv in c.inverters:
c.store.data[unit_key(inv.name, packmode.MODE_KEY)] = modes
return c
def build(c, now):
return homeview.build(c, level="ok", checks=[], now=now)["plan"]
def kinds_at(plan_json, label):
return [b["kind"] for b in plan_json["prices"] if b["label"] == label]
def grid_blocks(plan_json):
return [r for r in plan_json["runs"] if r["kind"] == "grid"]
# --------------------------------------------------------------------------- #
print("1. modes on: the gate refuses inside a cheap run (F-1)")
# All-in prices: 0.10 everywhere, 20:00 0.01, 20:15 0.09, 21:00-22:00 0.22. The
# cheap run is 20:00-20:30. At 20:00 the spread is 0.21. From 20:15 the quarters
# ahead are 0.09 to 0.22: 0.13 does not clear the 0.15 this cycle needs (0.10
# widened by 1.5 for the missing weather forecast).
def f1_price(h):
if h == 20.0:
return 0.01
if h == 20.25:
return 0.09
if 21.0 <= h < 22.0:
return 0.22
return 0.10
F1_PLAN = quarters(f1_price, days=1)
c1 = make(dict(BASE, battery_modes_enabled=True), F1_PLAN, at(19, 0, 5),
modes=packmode.CHEAP_HOURS)
def f1(now):
"""One real control tick of `_mode_roles()` at `now`, then Home."""
c1.tariff_at = now
tick = c1._mode_roles(now, None)
return tick, build(c1, now)
t_pre, p_pre = f1(at(19, 0, 5)) # builds the run plan
t_open, p_open = f1(at(20, 0, 5))
print(f" 20:00:05 roles {t_open.roles}, action {t_open.action}, "
f"gate {c1.mode_gate}, runs {c1.mode_runs.cheap}")
check("gate open (20:00): the control cycle charges, mode_gate is None, and "
"Home draws the charge on both batteries, 20:00 to 20:30",
t_open.action == modetrade.CHARGE and c1.mode_gate is None
and {r["battery"] for r in grid_blocks(p_open)} == {0, 1}
and all(r["from"] == "20:00" and r["to"] == "20:30"
for r in grid_blocks(p_open)))
check("...and marks the run's two quarters cheap on the chart",
kinds_at(p_open, "20:00") == ["cheap"]
and kinds_at(p_open, "20:15") == ["cheap"])
t_mid, p_mid = f1(at(20, 7, 30))
check("gate still open at 20:07: the charge goes on and is still drawn",
t_mid.action == modetrade.CHARGE and c1.mode_gate is None
and len(grid_blocks(p_mid)) == 2)
c1.inverters[0].soc = c1.inverters[1].soc = 52.0
t_ref, p_ref = f1(at(20, 17, 52))
print(f" 20:17:52 roles {t_ref.roles}, action {t_ref.action}, why "
f"{t_ref.why}, gate {c1.mode_gate}, run {c1.mode_runs.cheap}")
check("the run is still kept (20:00-20:30) and the gate refuses with "
"not_worth at 20:17: no charge, why not_worth",
c1.mode_runs.cheap.holds(at(20, 17, 52)) and c1.mode_gate == "not_worth"
and t_ref.action is None and t_ref.why == "not_worth"
and all(r != modetrade.CHARGE for r in t_ref.roles.values()))
check("Home draws no charge block for the refused part, on any battery",
grid_blocks(p_ref) == [] and not any(
r["what"].startswith("Charges") for r in p_ref["runs"]))
check("...marks no quarter from 20:15 cheap, keeps the 20:00 quarter it "
"did charge in, and keeps the give run marked dear",
kinds_at(p_ref, "20:00") == ["cheap"]
and kinds_at(p_ref, "20:15") == ["mid"]
and kinds_at(p_ref, "20:30") == ["mid"]
and kinds_at(p_ref, "21:00") == ["dear"])
saved = c1.mode_gate
c1.mode_gate = None # what Home did before UX-53
p_old = build(c1, at(20, 17, 52))
c1.mode_gate = saved
check("the same moment without the gate record draws the charge again (the "
"defect, shown): so the fixture is the F-1 case",
len(grid_blocks(p_old)) == 2 and kinds_at(p_old, "20:15") == ["cheap"])
row = homeview.battery_rows(c1, False, True)[0]
check("the battery card says what Home now says: it rests (role hold), and "
"does not say it charges",
t_ref.roles["inverter-a"] == modetrade.HOLD
and row["why_key"] in ("saves", "saves_later"))
t_n, p_n = f1(at(20, 31, 0))
check("after the run: nothing is drawn, as before",
grid_blocks(p_n) == [] and c1.mode_gate == "not_worth")
# Other refusals of the gate are followed the same way: no charge is drawn
# while the control cycle's gate says no.
c1b = make(dict(BASE, battery_modes_enabled=True), F1_PLAN, at(20, 0, 5),
modes=packmode.CHEAP_HOURS)
c1b.tariff_at = at(20, 0, 5)
c1b.peak_guard = PeakDecision(engaged=True, floor_w=1000.0, ceiling_w=3500.0,
ceiling_from="cap", projected_w=3600.0)
tb = c1b._mode_roles(at(20, 0, 5), None)
check("the peak guard holding refuses the charge (gate 'peak'): the cycle "
"does not charge and Home draws no charge",
c1b.mode_gate == "peak" and tb.action is None
and grid_blocks(build(c1b, at(20, 0, 5))) == [])
c1c = make(dict(BASE, battery_modes_enabled=True), F1_PLAN, at(20, 0, 5),
modes=packmode.CHEAP_HOURS)
c1c.peak_guard = None
tc = c1c._mode_roles(at(20, 0, 5), None)
check("a peak guard that measures nothing refuses it too (gate "
"'not_covered'): no charge, none drawn",
c1c.mode_gate == "not_covered" and tc.action is None
and grid_blocks(build(c1c, at(20, 0, 5))) == [])
c1d = make(dict(BASE, battery_modes_enabled=True), F1_PLAN, at(20, 0, 5),
modes=packmode.CHEAP_HOURS)
c1d.arb_headroom_w = None
td = c1d._mode_roles(at(20, 0, 5), None)
check("a refusal that is not the gate (no measured headroom yet) is not the "
"gate's: mode_gate stays None and the charge is still drawn",
c1d.mode_gate is None and td.why == "headroom"
and len(grid_blocks(build(c1d, at(20, 0, 5)))) == 2)
# --------------------------------------------------------------------------- #
print("\n2. modes off: a charge after the last dear period (F-2)")
NO_WEATHER = frozenset({Uncertainty.NO_WEATHER})
def decide_at(c, now, soc=50.0):
"""What the control cycle does at `now` on the plan (modes off)."""
return arbitrage.decide(
c.tariff_plan, now, peak_guard_active=False, charge_headroom_w=5000.0,
soc=soc, min_spread_eur_kwh=arbitrage.threshold_eur_kwh(c.o),
uncertainty=NO_WEATHER, charge_w=2500.0, discharge_w=2500.0,
soc_floor_pct=10.0, soc_ceiling_pct=90.0)
OFF = dict(BASE)
def f2_price(h):
return 0.54 if 21.5 <= h < 21.75 else 0.24
F2_PLAN = quarters(f2_price, days=1)
c2 = make(OFF, F2_PLAN, at(21, 36, 0))
p2 = build(c2, at(21, 36, 0))
g2 = [r for r in p2["runs"] if r["kind"] == "give"]
v2, _w = homeview.trade_verdict(c2, F2_PLAN, at(21, 36, 0))
print(f" 21:36 verdict: cheapest {homeview.hhmm(v2.cheapest.start)}, "
f"dearest {homeview.hhmm(v2.dearest.start)}")
check("the fixture: at 21:36 the verdict's cheapest quarter (21:45) comes "
"after its dearest (21:30, in progress), and the rest of the day is flat",
homeview.hhmm(v2.cheapest.start) == "21:45"
and homeview.hhmm(v2.dearest.start) == "21:30")
check("the add-on sells in 21:30, and at 21:45 it does nothing: idle, the "
"tariff is flat",
decide_at(c2, at(21, 30, 5)).action == arbitrage.DISCHARGE
and decide_at(c2, at(21, 45, 5)).action is None
and "the tariff is flat" in decide_at(c2, at(21, 45, 5)).reason)
check("Home draws the sale in progress and NO charge at 21:45",
len(g2) == 2 and all(r["from"] == "21:30" for r in g2)
and grid_blocks(p2) == [])
check("...and marks 21:30 dear and 21:45 not cheap",
kinds_at(p2, "21:30") == ["dear"] and kinds_at(p2, "21:45") == ["mid"])
real = homeview.leg_taken
homeview.leg_taken = lambda *a, **k: True # Home before UX-53
p2_old = build(c2, at(21, 36, 0))
homeview.leg_taken = real
check("without the question to the gate the page draws the 21:45 charge "
"(the defect, shown): so the fixture is the F-2 case",
[r["from"] for r in grid_blocks(p2_old)] == ["21:45", "21:45"]
and kinds_at(p2_old, "21:45") == ["cheap"])
# A charge the gate allows at its own start is still drawn, with the sale.
def q_price(h):
return {20.0: 0.10, 20.5: 0.12, 21.0: 0.59, 21.5: 0.54}.get(h, 0.24)
Q_PLAN = quarters(q_price, days=1)
c3 = make(OFF, Q_PLAN, at(20, 20, 0))
p3 = build(c3, at(20, 20, 0))
v3, _w = homeview.trade_verdict(c3, Q_PLAN, at(20, 20, 0))
check("gate allows (20:20, charge 20:30, sale 21:00): both are drawn, and "
"the add-on does charge at 20:30 and sell at 21:00",
[r["from"] for r in grid_blocks(p3)] == ["20:30", "20:30"]
and [r["from"] for r in p3["runs"] if r["kind"] == "give"] == ["21:00"] * 2
and decide_at(c3, at(20, 30, 5)).action == arbitrage.CHARGE
and decide_at(c3, at(21, 0, 5)).action == arbitrage.DISCHARGE
and kinds_at(p3, "20:30") == ["cheap"] and kinds_at(p3, "21:00") == ["dear"])
check("...and the charge in progress at 20:30 is drawn too",
len(grid_blocks(build(c3, at(20, 35, 0)))) == 2)
# Every probe on a day-ahead day: the charge is drawn exactly where the add-on
# will take it. "Takes it" is the control cycle's own `decide()` at the start
# of the charge's period, on the same plan.
def sweep(plan, opts, lo, hi, step=15):
rows = []
for m in range(lo, hi, step):
now = at(0, 0, 0) + timedelta(minutes=m, seconds=7)
c = make(opts, plan, now)
v, _w = homeview.trade_verdict(c, plan, now)
if v is None:
rows.append((now, None, None, None))
continue
legs = homeview.trade_legs(c, plan, v, now)
drawn = {k: homeview.hhmm(pt.start) for k, pt in legs}
rows.append((now, v, drawn, [k for k, _pt in legs]))
return rows
def takes(c, plan, kind, pt, now):
"""True when decide() acts in that leg's first moment (or now, when it
is already under way)."""
t = max(now, pt.start + timedelta(seconds=5))
d = decide_at(c, t)
return d.action == (arbitrage.CHARGE if kind == "grid"
else arbitrage.DISCHARGE)
bad, nlegs, nsuppressed = [], 0, 0
for now, v, drawn, kinds in sweep(F2_PLAN, OFF, 0, 24 * 60):
if v is None:
continue
c = make(OFF, F2_PLAN, now)
for kind, pt in (("grid", v.cheapest), ("give", v.dearest)):
nlegs += 1
shown = kind in kinds
if not shown:
nsuppressed += 1
if shown != takes(c, F2_PLAN, kind, pt, now):
bad.append((homeview.hhmm(now), kind, shown))
print(f" {nlegs} verdict legs over a day, {nsuppressed} not drawn, "
f"{len(bad)} where Home and decide() disagree {bad[:3]}")
check("over a whole day of 15-minute probes, a leg is drawn exactly when "
"decide() acts on it at its start; and some are not drawn",
not bad and nsuppressed > 0)
# The same, on 40 random days of 15-minute prices (rounded to a cent, so there
# are ties), today only, every quarter probed: Home and decide() agree on every
# leg, and the page does drop some.
import random # noqa: E402
rnd = random.Random(53)
n_leg = n_drop = 0
disagree = []
for day in range(40):
shape = [round(rnd.choice((0.10, 0.18, 0.24, 0.24, 0.30, 0.45, 0.60))
+ rnd.randrange(0, 3) * 0.01, 2) for _ in range(96)]
if day % 2:
shape = [shape[i // 4 * 4] for i in range(96)] # hour-constant
plan = quarters(lambda h, sh=shape: sh[int(h * 4)], days=1)
c = make(OFF, plan, at(0, 0, 0))
for q in range(96):
now = at(0, 0, 0) + timedelta(minutes=15 * q, seconds=7)
v, _w = homeview.trade_verdict(c, plan, now)
if v is None:
continue
kinds = [k for k, _pt in homeview.trade_legs(c, plan, v, now)]
for kind, pt in (("grid", v.cheapest), ("give", v.dearest)):
n_leg += 1
shown = kind in kinds
n_drop += not shown
if shown != takes(c, plan, kind, pt, now):
disagree.append((day, homeview.hhmm(now), kind, shown))
print(f" 40 random days: {n_leg} verdict legs, {n_drop} not drawn, "
f"{len(disagree)} disagreements with decide() {disagree[:3]}")
check("40 random 15-minute days, every quarter: a leg is drawn exactly "
"when decide() acts on it at its start",
not disagree and n_leg > 3000 and n_drop > 0)
# A charge tomorrow: announced only when the add-on will take it. Today's
# dear period is 21:00 (0.50, then 0.30). Tomorrow opens with a dear hour
# (0.45) and the cheapest quarter comes right after it, at 01:00 (0.24).
def two_days(tomorrow):
pts = []
for i in range(96 * 2):
t = DAY0 + timedelta(minutes=15 * i)
h = (i % 96) / 4.0
price = (0.50 if h == 21.0 else 0.30) if i < 96 else tomorrow(h)
pts.append(PricePoint(brussels(t), price, False))
return TariffPlan(PriceRead.series(tuple(pts), PriceProduct.DAY_AHEAD,
"generic_dayahead", DAY0), None,
"provider")
def tomorrow_lines(plan, now):
c = make(OFF, plan, now)
return [x for x in build(c, now)["lines"] if x.startswith("Tomorrow")]
T_FLAT = two_days(lambda h: 0.45 if h < 1.0 else 0.24)
T_WITH = two_days(lambda h: 0.45 if h < 1.0 else (0.50 if h == 18.0 else 0.24))
l_with = tomorrow_lines(T_WITH, at(21, 5, 0))
l_flat = tomorrow_lines(T_FLAT, at(21, 5, 0))
print(f" tomorrow lines: with a later dear period {l_with}; without {l_flat}")
check("tomorrow's charge (01:00) is announced when a dearer period follows "
"it, and not when the rest of tomorrow is flat",
len(l_with) == 1 and "01:00" in l_with[0] and l_flat == [])
check("...and the add-on charges at 01:00 on the first and does not on the "
"second",
decide_at(make(OFF, T_WITH, at(21, 5, 0)),
at(25, 0, 5)).action == arbitrage.CHARGE
and decide_at(make(OFF, T_FLAT, at(21, 5, 0)),
at(25, 0, 5)).action is None)
# --------------------------------------------------------------------------- #
print("\n3. wording, English and Dutch")
c6 = make(OFF, Q_PLAN, at(20, 20, 0))
en = homeview.build(c6, level="ok", checks=[], now=at(20, 20, 0),
lang="en")["plan"]
nl = homeview.build(c6, level="ok", checks=[], now=at(20, 20, 0),
lang="nl")["plan"]
gw_en = [r["why"] for r in en["runs"] if r["kind"] == "grid"]
gw_nl = [r["why"] for r in nl["runs"] if r["kind"] == "grid"]
dw_en = [r["why"] for r in en["runs"] if r["kind"] == "give"]
dw_nl = [r["why"] for r in nl["runs"] if r["kind"] == "give"]
print(f" {gw_en[:1]} {gw_nl[:1]} {dw_en[:1]} {dw_nl[:1]}")
check("the charge says 'lowest price left', English and Dutch",
gw_en and set(gw_en) == {"Lowest price left, \u20ac0.12 per kWh."}
and set(gw_nl) == {"Laagste prijs die nog over is, \u20ac0,12 per kWh."})
check("...and the sale 'highest price left'",
set(dw_en) == {"Highest price left, \u20ac0.59 per kWh."}
and set(dw_nl) == {"Hoogste prijs die nog over is, \u20ac0,59 per kWh."})
check("...and no 'published' is left in the run words of either language",
all("ublished" not in str(homeview.uitext.words(l)["home_words"]["run"])
and "gepubliceerd" not in str(
homeview.uitext.words(l)["home_words"]["run"])
for l in ("en", "nl")))
check("the short words fit the plan-line rule: one sentence of at most 12 "
"words",
all(len(x.split()) <= 12 for x in gw_en + gw_nl + dw_en + dw_nl))
NOW6 = at(21, 45, 5)
FLAT_DYN = quarters(lambda h: 0.24, days=1)
r_dyn = FLAT_DYN.arbitrage(peak_guard_active=False, min_spread_eur_kwh=0.10,
uncertainty=NO_WEATHER, now=NOW6).match(
allowed=lambda v: None, refused=lambda v: v.reason)
FIXED = {"periods": [{"name": "all", "from": "00:00", "to": "24:00",
"days": "all", "eur_per_kwh": 0.30}]}
FIXED_PLAN = static_plan({"static_profile": json.dumps(FIXED)}, at(0), at(36),
now=at(0))
r_fix = FIXED_PLAN.arbitrage(peak_guard_active=False,
min_spread_eur_kwh=0.10, uncertainty=NO_WEATHER,
now=NOW6).match(allowed=lambda v: None,
refused=lambda v: v.reason)
DN = {"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}]}
DN_FLAT = static_plan({"static_profile": json.dumps(DN)}, at(23, 0), at(30),
now=at(23, 0)) # 23:00 to 06:00: night only
r_dn = DN_FLAT.arbitrage(peak_guard_active=False, min_spread_eur_kwh=0.10,
uncertainty=NO_WEATHER, now=at(23, 10)).match(
allowed=lambda v: None, refused=lambda v: v.reason)
print(f" dynamic: {r_dyn}\n day/night: {r_dn}\n fixed: {r_fix}")
check("a dynamic site whose remaining quarters are equal is refused as flat, "
"and is not called a fully fixed product",
r_dyn and r_dyn.startswith("the tariff is flat - every one of the ")
and "fully fixed" not in r_dyn and "fixed product" not in r_dyn
and r_dyn.endswith("arithmetically impossible"))
check("...a day/night site whose remaining hours are all night: the same",
r_dn and r_dn.startswith("the tariff is flat")
and "fixed" not in r_dn)
check("...a site with a real fixed product is still told so",
r_fix and r_fix.endswith(
"§9.2: a fully fixed product is self-consumption only")
and FIXED_PLAN.product is PriceProduct.FIXED
and DN_FLAT.product is PriceProduct.STATIC
and FLAT_DYN.product is PriceProduct.DAY_AHEAD)
# --------------------------------------------------------------------------- #
print("\n4. day/night, hourly and price-off pages")
def old_vs_new(plan, opts, hours=24):
"""Verdict legs the page drew before (both, when there is a verdict) and
the ones it draws now, probed every 30 minutes."""
same = diff = none = 0
for m in range(0, hours * 60, 30):
now = at(0) + timedelta(minutes=m, seconds=7)
c = make(opts, plan, now)
v, _w = homeview.trade_verdict(c, plan, now)
if v is None:
none += 1
continue
if len(homeview.trade_legs(c, plan, v, now)) == 2:
same += 1
else:
diff += 1
return same, diff, none
DN_OPTS = dict(OFF, cycle_cost_threshold_eur=0.02,
static_profile=json.dumps(DN))
DN_48 = static_plan(DN_OPTS, at(0), at(48), now=at(0))
s_dn = old_vs_new(DN_48, DN_OPTS)
print(f" day/night, 48 h plan, 48 probes: same {s_dn[0]}, changed "
f"{s_dn[1]}, no verdict {s_dn[2]}")
check("day/night (a 48-hour plan, as the refresh builds it): at every probe "
"that has a verdict both legs are drawn, as before; and there are some",
s_dn[1] == 0 and s_dn[0] >= 20)
HOURLY = quarters(lambda h: [.24, .22, .21, .20, .21, .22, .24, .27, .26, .22,
.18, .16, .15, .14, .16, .19, .23, .28, .33, .36,
.34, .29, .26, .24][int(h)], days=2)
HR_OPTS = dict(OFF, cycle_cost_threshold_eur=0.05)
s_hr = old_vs_new(HOURLY, HR_OPTS)
print(f" hourly shape on a 48 h plan: same {s_hr[0]}, changed {s_hr[1]}, "
f"no verdict {s_hr[2]}")
check("an hourly-shaped day-ahead feed with tomorrow known: both legs are "
"drawn at every probe with a verdict, as before",
s_hr[1] == 0 and s_hr[0] >= 20)
s_one = old_vs_new(quarters(lambda h: [.24, .22, .21, .20, .21, .22, .24, .27,
.26, .22, .18, .16, .15, .14, .16, .19,
.23, .28, .33, .36, .34, .29, .26,
.24][int(h)], days=1), HR_OPTS)
print(f" the same with today only: same {s_one[0]}, changed {s_one[1]}, "
f"no verdict {s_one[2]}")
check("...and with today only, the probes that change are the ones where "
"the cheapest quarter left is after the last dear one (F-2)",
s_one[1] > 0 and s_one[0] > 0)
c7 = make(dict(OFF, arbitrage_enabled=False), DN_48, at(12, 0, 0))
p7 = build(c7, at(12, 0, 0))
check("price trading off (a day/night site, modes off): no charge or sale "
"is drawn and no bar is marked, as before",
[r for r in p7["runs"] if r["kind"] in ("grid", "give")] == []
and {b["kind"] for b in p7["prices"]} == {"mid"})
c8 = make(dict(OFF, arbitrage_enabled=False, battery_modes_enabled=True),
DN_48, at(12, 0, 0), modes=packmode.SOLAR_FIRST)
c8._mode_roles(at(12, 0, 0), None)
p8 = build(c8, at(12, 0, 0))
check("modes on, every battery Solar first (the owner's setting): no grid "
"charge is drawn, and the gate record changes nothing",
grid_blocks(p8) == [] and {b["kind"] for b in p8["prices"]} == {"mid"})
c9 = make(dict(OFF, battery_modes_enabled=True), DN_48, at(12, 0, 0),
modes=packmode.SOLAR_FIRST)
check("a controller that never ran a cycle has no gate record, and the "
"default is 'no refusal' (None)",
getattr(c9, "mode_gate", "missing") is None)
print(f"fails: {len(fails)}")
sys.exit(1 if fails else 0)
+719
View File
@@ -0,0 +1,719 @@
"""Runnable check for UX-54. `python3 test_ux54.py`
The Forecast tab: today's and tomorrow's solar, the house's expected use and
each battery's expected level, per hour. It reads the add-on and decides
nothing. Real `Controller`s in known situations (tools/ux54_scenarios.py).
1. The hourly shape: what each forecast integration gives, and that the tab
spreads a state-only total over the bell and says so.
2. The blocks are the plan's: a block is on the tab exactly where Home has
one, with its words; tomorrow's come from the same builders; the gate
(UX-53), modes off and every mode are followed.
3. The level curve (`daycurve.curve()`): inside floor and ceiling, the right
way in a charge and a give block, from spare solar and from a shortfall,
and a fuzz against the invariants.
4. A part that is not known yet is a note: modes off, no forecast, a lost
forecast, learning, no tomorrow, a monthly cycle, no level, stopped,
shadow.
5. Measured solar (`SolarToday`), and that keeping it can never cost a cycle.
6. The words, English and Dutch, and the page and its route. Reading it
asks Home Assistant for nothing and moves nothing in control.
Mutations (MUTATION.md, purged copies): listed in the pull request.
"""
import asyncio
import json
import logging
import os
import random
import re
import sys
import time
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 ux54_scenarios as S # noqa: E402
from app import arbitrage, dayplan, dayview, homeview, modetrade # noqa: E402
from app import packmode, solarfc, uitext # noqa: E402
from app.daycurve import Batt, SolarToday, curve # noqa: E402
fails = []
def check(name, cond):
print(f" {'ok ' if cond else 'FAIL'} {name}")
if not cond:
fails.append(name)
UTC = timezone.utc
DAY0 = datetime(2026, 10, 26, 23, 0, tzinfo=UTC) # Tue 27 Oct, local
KINDS = ("grid", "give", "cycle")
def at(h, m=0, s=5):
return DAY0 + timedelta(hours=h, minutes=m, seconds=s)
def home(c, now, lang="en"):
return homeview.build(c, level="ok", checks=[], now=now, lang=lang)
def tab(c, now, lang="en"):
return dayview.build(c, home(c, now, lang), now, lang)
def block_key(r):
return (r["kind"], r["from"], r["to"], r["start_h"], r["end_h"],
r["what"], r["why"], r["amount"])
def tab_blocks(out, day=0):
return [[block_key(r) for r in b["blocks"]]
for b in out["days"][day]["batteries"]]
def home_blocks(status, n):
return [[block_key(r) for r in status["plan"]["runs"]
if r["battery"] == i and r["kind"] in KINDS] for i in range(n)]
def notes_of(out, day=0):
return out["days"][day]["notes"]
def say(key, lang="en", **v):
with homeview.in_lang(lang):
return homeview.day_note(key, **v)
# --------------------------------------------------------------------------- #
print("1. the hourly shape")
# --------------------------------------------------------------------------- #
now = at(10, 7)
c = S.make("owner", now)
fc = c.solar_fc
out = tab(c, now)
d = out["days"][0]
fcs = [h["forecast_kwh"] for h in d["hours"]]
print(f" owner: {len(c.fc_entities)} + {len(c.fc_tomorrow_entities)} "
f"state-only sensors, today {fc.today.kwh:.2f} kWh left, tomorrow "
f"{fc.tomorrow.kwh:.2f}")
check("the owner's shape (three 'remaining' + three 'tomorrow' sensors) "
"carries no hourly split: the add-on sums them and uses its bell",
len(c.fc_entities) == 3 and len(c.fc_tomorrow_entities) == 3
and fc.today.periods == () and fc.tomorrow.periods == ()
and abs(fc.tomorrow.kwh - 16.0) < 0.01)
check("...so the tab says it is an estimate from a daily total, and fades "
"the bars", d["sun"]["estimate"] is True
and say("estimate") in notes_of(out)
and out["days"][1]["sun"]["estimate"] is True)
check("today the hours still to come add up to what the sensors say is left "
"(no hour before now)",
all(f is None for f in fcs[:10])
and abs(sum(f for f in fcs if f is not None) - fc.today.kwh) < 1e-3)
tm = [h["forecast_kwh"] for h in out["days"][1]["hours"]]
check("tomorrow's 24 hours add up to tomorrow's total, are 0 outside 06-20 "
"and peak at 13:00 (the bell)",
abs(sum(tm) - fc.tomorrow.kwh) < 0.01
and all(v == 0 for v in tm[:6] + tm[20:])
and tm.index(max(tm)) in (12, 13))
# The day plan reads the same forecast hour by hour with `between(s, e)`;
# for a state-only forecast that spreads what is left over each span's own
# tail, so its hours add up to more than the total (a finding in the pull
# request, not changed here: no control change).
plan_sum = sum(fc.between(s, e) for s, e, _h in dayplan._hour_slices(
now, solarfc._bounds(now)[1]))
print(f" dayplan's own per-hour sum: {plan_sum:.1f} kWh for {fc.today.kwh:.1f} kWh left")
check("the tab's hours use the span from now (`dayview.sun_kwh()`), not the "
"plan's per-hour call, and so keep the total",
abs(dayview.sun_kwh(fc, now, now, solarfc._bounds(now)[1])
- fc.today.kwh) < 1e-6)
cp = S.make("periods", now)
dp = tab(cp, now)["days"]
check("an hourly split (Open-Meteo's `wh_period`, Solcast's "
"`detailedForecast`) is used as it is, and the tab does not call it an "
"estimate",
cp.solar_fc.today.periods and cp.solar_fc.tomorrow.periods
and dp[0]["sun"]["estimate"] is False
and say("estimate") not in dp[0]["notes"]
and abs(sum(h["forecast_kwh"] for h in dp[1]["hours"])
- cp.solar_fc.tomorrow.kwh) < 1e-3)
# Solcast's shape: `detailedForecast` rows, kW per half hour.
d0 = S.day0_of(now)
rows = [{"period_start": (d0 + timedelta(minutes=30 * i)).isoformat(),
"pv_estimate": 4.0 if 20 <= i < 36 else 0.0} for i in range(48)]
st = {"state": "20.0", "attributes": {"unit_of_measurement": "kWh",
"detailedForecast": rows},
"last_updated": now.isoformat(), "last_reported": now.isoformat()}
sc = solarfc.read_sum([("sensor.solcast", st)], [], now)[0]
check("Solcast's `detailedForecast` gives a split too (kW per half hour)",
sc is not None and sc.today.periods and len(sc.today.periods) >= 16)
check("the forecast the tab reads is the one the controller holds "
"(`Controller.solar_fc`), read by `solarfc` as the plan reads it",
out["updated"] == homeview.hhmm(c.solar_fc.at))
# --------------------------------------------------------------------------- #
print("\n2. the blocks are the plan's")
# --------------------------------------------------------------------------- #
SCN = [("owner", 10.1), ("dynamic", 10.1), ("dynamic", 0.3),
("dynamic", 18.5), ("sunny", 0.3), ("sunny", 18.5), ("off", 10.1),
("off", 16.9), ("cycle", 10.1), ("nofc", 10.1), ("stale", 0.3),
("learning", 0.3), ("periods", 18.5)]
bad, seen = [], {"grid": 0, "give": 0, "cycle": 0}
for name, hh in SCN:
now = at(hh)
c = S.make(name, now)
st = home(c, now)
out = tab(c, now)
n = len(c.inverters)
if tab_blocks(out) != home_blocks(st, n):
bad.append((name, hh))
for rows in tab_blocks(out):
for r in rows:
seen[r[0]] += 1
print(f" {len(SCN)} scenarios, blocks seen on the tab: {seen}")
check("today, every battery's blocks on the tab are Home's `runs` of that "
"battery (grid, give, cycle): same times, same words, no more, no less "
f"({bad})", not bad and all(v > 0 for v in seen.values()))
now = at(10, 7)
c = S.make("dynamic", now)
d0, d1 = homeview.local_day(now)
n0, n1 = homeview.local_day(d1 + timedelta(hours=2))
check("Home's own price blocks for today are `homeview.runs_for()` of "
"today: one builder, two days",
[block_key(r) for r in homeview.runs_for(c, d0, d1, now, False)]
== [block_key(r) for r in home(c, now)["plan"]["runs"]
if r["kind"] in ("grid", "give")])
tmr = tab(c, now)["days"][1]["batteries"]
check("tomorrow's blocks are the same builder over tomorrow's window",
[[block_key(r) for r in b["blocks"]] for b in tmr]
== [[block_key(r) for r in homeview.runs_for(c, n0, n1, now, False)
if r["battery"] == i] for i in range(2)])
# Modes on: Solar first draws no price block; Leftover solar and Cheap hours do.
c_o = S.make("owner", at(10, 7))
o = tab(c_o, at(10, 7))
check("modes on, both batteries Solar first, a day/night tariff (the "
"owner's shape): no block on either battery, either day, and the "
"battery says it follows the house",
all(not b["blocks"] for dd in o["days"] for b in dd["batteries"])
and all(b["line"] == homeview.run_house_what(False) + " "
+ homeview.run_house_why(False)
for b in o["days"][0]["batteries"]))
c_d = S.make("dynamic", at(10, 7))
dd = tab(c_d, at(10, 7))["days"][0]["batteries"]
rp = c_d.mode_runs
check("Leftover solar and Cheap hours: a give block exactly on the give run "
"(17:00 to 21:00), on both",
all([(r["kind"], r["from"], r["to"]) for r in b["blocks"]]
== [("give", "17:00", "21:00")] for b in dd)
and rp.give[0].start == at(17, 0, 0) and rp.give[0].end == at(21, 0, 0))
c_n = S.make("dynamic", at(0, 18))
nb = tab(c_n, at(0, 18))["days"][0]["batteries"]
check("in the night before the cheap run both charge from the grid in it "
"(01:00 to 02:30), then give at 17:00",
all([r["kind"] for r in b["blocks"]] == ["grid", "give"] for b in nb)
and all(b["blocks"][0]["from"] == "01:00"
and b["blocks"][0]["to"] == "02:30" for b in nb))
# UX-53: no grid block while the gate refuses.
for why in ("peak", "not_worth"):
c_n.mode_gate = why
gb = tab(c_n, at(0, 18))["days"][0]["batteries"]
check(f"the trade gate refuses ({why}): no grid block on the tab, as on "
"Home, and the give block stays",
all([r["kind"] for r in b["blocks"]] == ["give"] for b in gb)
and tab_blocks(tab(c_n, at(0, 18))) == home_blocks(
home(c_n, at(0, 18)), 2))
c_n.mode_gate = None
def at_h(b, h):
"""A drawn level at hour `h` of the day, NaN where no point is drawn."""
return dict((round(x, 3), v) for x, v in b["levels"]).get(h, float("nan"))
check("the curve charges in a grid block: A (planned level 64 %) and B "
"(Cheap hours, to the end of its 01:30 run) are well above where they "
"were at 01:00, and stay put after the run",
all(at_h(b, 2.5) - at_h(b, 1.0) > 8.0
and abs(at_h(b, 3.0) - at_h(b, 2.5)) < 1e-6
for b in nb) and nb[0]["levels"][0][1] == 55.0)
# Modes off: the verdict's legs.
c_f = S.make("off", at(10, 7))
fb = tab(c_f, at(10, 7))
check("modes off, arbitrage on: the verdict's blocks are Home's, today, and "
"tomorrow's charge comes from the same builder; no level is drawn",
[b["kind"] for x in fb["days"][0]["batteries"] for b in x["blocks"]]
== ["give", "give"]
and [b["kind"] for x in fb["days"][1]["batteries"]
for b in x["blocks"]] == ["grid", "grid"]
and not any(x["levels"] for dd in fb["days"] for x in dd["batteries"]))
# A battery that is Off has no block and no curve.
c_d.store.data[S.unit_key("inverter-b", packmode.USE_KEY)] = False
off = tab(c_d, at(10, 7))["days"][0]["batteries"]
check("a battery switched Off: no block, no curve, and it says so in the "
"plan's words",
off[1]["off"] is True and not off[1]["blocks"] and not off[1]["levels"]
and off[1]["line"] == homeview.why_words("off")
and off[0]["levels"])
# The cycle is a block, and the curve stops where it starts.
c_c = S.make("cycle", at(10, 7))
cy = tab(c_c, at(10, 7))
a0 = cy["days"][0]["batteries"][0]
check("a monthly cycle waiting today is a block, the level is drawn only "
"until it starts, and a note says so",
[r["kind"] for r in a0["blocks"]] == ["give", "cycle"]
and a0["levels"][-1][0] <= 20.0 + 1e-6
and say("cycle", name="inverter-a") in notes_of(cy)
and not cy["days"][1]["batteries"][0]["levels"])
# --------------------------------------------------------------------------- #
print("\n3. the level curve")
# --------------------------------------------------------------------------- #
T0 = DAY0 + timedelta(hours=9)
T1 = T0 + timedelta(hours=24)
def flat(kwh_per_h):
return lambda a, z: kwh_per_h * (z - a).total_seconds() / 3600.0
def run(b, sun, use, **kw):
return curve([b] if isinstance(b, Batt) else b, T0, T1, sun, use,
floor_pct=20.0, ceiling_pct=95.0, **kw)
def lv(out, name, h):
"""The level at `h` hours after T0 (NaN when there is no such point)."""
return dict(out[name]).get(T0 + timedelta(hours=h), float("nan"))
B1 = Batt("a", 10.0, 5.0, 50.0, True, False)
o1 = run(B1, flat(3.0), flat(1.0))
check("Solar first: 2 kWh of spare solar an hour fills it at 95 % "
"efficiency, 19 % an hour",
abs(lv(o1, "a", 1) - 69.0) < 1e-6 and abs(lv(o1, "a", 2) - 88.0) < 1e-6)
check("...stops at the ceiling (95 %), and stays there", abs(lv(o1, "a", 3)
- 95.0) < 1e-6
and max(v for _t, v in o1["a"]) <= 95.0 + 1e-9)
o2 = run(B1, flat(0.0), flat(1.0))
check("a shortfall of 1 kWh an hour is given at 95 % efficiency, 10.53 % an "
"hour", abs(lv(o2, "a", 1) - (50.0 - 100.0 / 9.5)) < 1e-6)
check("...stops at the floor (20 %), and stays there",
abs(lv(o2, "a", 6) - 20.0) < 1e-6
and min(v for _t, v in o2["a"]) >= 20.0 - 1e-9)
check("a battery already above the ceiling does not charge and one below the "
"floor does not give",
lv(run(Batt("a", 10.0, 5.0, 98.0, True, False), flat(3), flat(1)),
"a", 3) == 98.0
and lv(run(Batt("a", 10.0, 5.0, 12.0, True, False), flat(0), flat(1)),
"a", 3) == 12.0)
slow = run(Batt("a", 10.0, 1.0, 50.0, True, False), flat(5.0), flat(0.0))
check("the rating caps the rate: 1 kW charges 9.5 % of 10 kWh an hour",
abs(lv(slow, "a", 1) - 59.5) < 1e-6)
# Roles.
GV = ((T0 + timedelta(hours=3), T0 + timedelta(hours=6)),)
hold = Batt("p", 10.0, 5.0, 60.0, False, True, (), GV)
oh = run(hold, flat(0.0), flat(1.0))
check("a price-mode battery holds outside a give run (no give), and gives "
"inside it",
lv(oh, "p", 3) == 60.0
and lv(oh, "p", 6) < 60.0 - 3 * 100.0 / 9.5 + 1e-6 + 0.01
and lv(oh, "p", 24) == lv(oh, "p", 6))
oh2 = run(hold, flat(3.0), flat(1.0))
check("...and charges from spare solar while it holds",
lv(oh2, "p", 3) > 60.0)
GR = ((T0 + timedelta(hours=2), T0 + timedelta(hours=4), 80.0),)
ch = Batt("c", 10.0, 5.0, 40.0, False, True, GR, ())
oc = run(ch, flat(0.0), flat(0.0))
check("a grid block charges to its level at the rating (5 kW x 95 %: 47.5 % "
"an hour on 10 kWh, so 80 % in the first hour), and stops there",
lv(oc, "c", 2) == 40.0 and abs(lv(oc, "c", 3) - 80.0) < 1e-6
and lv(oc, "c", 5) == 80.0)
two = [Batt("a", 10.0, 5.0, 50.0, True, False),
Batt("b", 10.0, 5.0, 50.0, False, True)]
ot = run(two, flat(3.0), flat(1.0))
check("two batteries: Solar first takes the spare solar first; the other "
"gets what is left",
lv(ot, "a", 1) > lv(ot, "b", 1) - 1e-9
and abs(lv(ot, "a", 1) - 69.0) < 1e-6 and lv(ot, "b", 1) == 50.0)
big = run(two, flat(8.0), flat(0.0))
check("...and the other charges once Solar first is at its rating or full",
lv(big, "b", 1) > 50.0)
GR2 = ((T0 + timedelta(hours=2), T0 + timedelta(hours=4), 80.0),)
mix = [Batt("a", 40.0, 5.0, 50.0, True, True, (), ()),
Batt("b", 10.0, 5.0, 40.0, False, True, GR2, ())]
om = run(mix, flat(0.0), flat(2.0))
check("while one battery charges from the grid the others do not charge "
"(they rest) and Solar first may still give",
lv(om, "b", 3) > lv(om, "b", 2) and lv(om, "a", 3) < lv(om, "a", 2))
ends = run(B1, lambda a, z: None if a >= T0 + timedelta(hours=5)
else flat(3.0)(a, z), flat(1.0))
check("an unknown sun ends the curve there", ends["a"][-1][0]
== T0 + timedelta(hours=5) and len(ends["a"]) == 6)
stop = run(Batt("a", 10.0, 5.0, 50.0, True, False, (), (),
T0 + timedelta(hours=2)), flat(3.0), flat(1.0))
check("a monthly cycle ends the battery's curve at its start",
stop["a"][-1][0] == T0 + timedelta(hours=2))
check("the curve starts at the battery's level now, at `start`",
o1["a"][0] == (T0, 50.0) and o2["a"][0] == (T0, 50.0))
# A fuzz against the invariants.
rnd = random.Random(54)
bad_f = []
for i in range(300):
n = rnd.choice((1, 2, 3))
bs = []
for k in range(n):
gs = ()
gw = ()
if rnd.random() < 0.5:
a = rnd.randint(0, 20)
gs = ((T0 + timedelta(hours=a),
T0 + timedelta(hours=a + rnd.randint(1, 4)),
rnd.uniform(40, 95)),)
if rnd.random() < 0.5:
a = rnd.randint(0, 20)
gw = ((T0 + timedelta(hours=a),
T0 + timedelta(hours=a + rnd.randint(1, 4))),)
bs.append(Batt(f"b{k}", rnd.uniform(5, 25), rnd.uniform(1, 6),
rnd.uniform(20, 95), rnd.random() < 0.5,
rnd.random() < 0.6, gs, gw))
sun = rnd.uniform(0, 5)
use = rnd.uniform(0, 3)
out = run(bs, lambda a, z, s=sun: s * (z - a).total_seconds() / 3600.0,
lambda a, z, u=use: u * (z - a).total_seconds() / 3600.0)
for b in bs:
pts = out[b.name]
vals = [v for _t, v in pts]
if min(vals) < 20.0 - 1e-6 or max(vals) > 95.0 + 1e-6:
bad_f.append(("range", i, b.name))
for (t1, v1), (t2, v2) in zip(pts, pts[1:]):
hrs = (t2 - t1).total_seconds() / 3600.0
if abs(v2 - v1) / 100.0 * b.capacity_kwh > b.rating_kw * hrs \
/ dayplan.EFF * 1.0001 + 1e-9:
bad_f.append(("rate", i, b.name))
# holds, and not in a grid block or a give run: never gives
in_give = any(s <= t1 < e for s, e in b.give)
in_grid = any(s <= t1 < e for s, e, _ in b.grid)
if b.priced and not b.solar_first and not in_give \
and not in_grid and v2 < v1 - 1e-9:
bad_f.append(("hold gave", i, b.name))
# Energy: what the batteries gain never exceeds the sun's spare (net of
# the grid blocks) at 95 %.
check(f"a fuzz of 300 days of 1-3 random batteries: every level stays inside "
f"the floor and ceiling, no step beats the rating, a holding battery "
f"never gives ({bad_f[:3]})", not bad_f)
# On the controllers: inside floor and ceiling, the right way in blocks.
rows_ok, moves_ok, n_curves = True, True, 0
for name, hh in SCN:
now = at(hh)
c = S.make(name, now)
out = tab(c, now)
fl, ce = arbitrage.soc_bounds(c.o)
for day in out["days"]:
for b in day["batteries"]:
n_curves += bool(b["levels"])
for h_, v in b["levels"]:
rows_ok &= fl - 1e-6 <= v <= ce + 1e-6
for r in b["blocks"]:
inside = [(h_, v) for h_, v in b["levels"]
if r["start_h"] - 1e-6 <= h_ <= r["end_h"] + 1e-6]
if len(inside) >= 2 and r["kind"] == "grid":
moves_ok &= inside[-1][1] >= inside[0][1] - 1e-6
if len(inside) >= 2 and r["kind"] == "give":
moves_ok &= inside[-1][1] <= inside[0][1] + 1e-6
print(f" {n_curves} battery curves checked on the controllers")
check("on the controllers: every drawn level is inside the floor and the "
"ceiling", rows_ok and n_curves >= 10)
check("...and the level rises in a grid block and does not rise in a give "
"block", moves_ok)
c_s = S.make("sunny", at(10, 7))
ls = tab(c_s, at(10, 7))["days"][0]["batteries"][0]["levels"]
check("a sunny day with Leftover solar: the level climbs by noon and "
"reaches the ceiling by evening",
ls[0][1] == c_s.inverters[0].soc and max(v for _h, v in ls) > 85.0)
# --------------------------------------------------------------------------- #
print("\n4. a part that is not known yet is a note")
# --------------------------------------------------------------------------- #
def no_charts(out):
return all(not b["levels"] for dd in out["days"]
for b in dd["batteries"])
c = S.make("off", at(10, 7))
o = tab(c, at(10, 7))
check("modes off: a note, no forecast, no house use, no curve",
say("modes_off") in notes_of(o) and no_charts(o)
and not o["days"][0]["sun"]["forecast"]
and not o["days"][0]["sun"]["house"])
o = tab(S.make("nofc", at(10, 7)), at(10, 7))
check("no forecast set up: a note, and the price blocks still show",
say("not_set") in notes_of(o) and no_charts(o)
and o["days"][0]["batteries"][0]["blocks"])
o = tab(S.make("stale", at(10, 7)), at(10, 7))
check("a forecast that is missing or too old: a note, no curve",
say("no_forecast") in notes_of(o) and no_charts(o))
o = tab(S.make("learning", at(10, 7)), at(10, 7))
check("learning (no house profile yet): a note, the sun is drawn, no house "
"use and no curve",
say("learning") in notes_of(o) and no_charts(o)
and o["days"][0]["sun"]["forecast"] and not o["days"][0]["sun"]["house"]
and all(h["house_kwh"] is None for h in o["days"][0]["hours"]))
o = tab(S.make("notomorrow", at(10, 7)), at(10, 7))
tmr = o["days"][1]
check("no forecast for tomorrow: a note on that day only; the curve stops at "
"sunrise; today is whole",
say("no_tomorrow") in tmr["notes"]
and say("no_tomorrow") not in notes_of(o, 0)
and not tmr["sun"]["forecast"]
and tmr["batteries"][0]["levels"][-1][0] <= 6.0 + 1e-6
and o["days"][0]["batteries"][0]["levels"][-1][0] > 23.0)
c = S.make("owner", at(10, 7))
c.inverters[1].soc = None
o = tab(c, at(10, 7))
check("a battery with no level reading: a note naming it, no curve for it, "
"and the other still has one",
say("no_level", name="inverter-b") in notes_of(o)
and not o["days"][0]["batteries"][1]["levels"]
and o["days"][0]["batteries"][0]["levels"])
c = S.make("owner", at(10, 7))
c.auto = False
o = tab(c, at(10, 7))
check("control stopped: the plan's words, no curve, no block",
homeview.why_words("stopped") in notes_of(o) and no_charts(o)
and not o["running"])
c = S.make("owner", at(10, 7))
c.o["shadow_mode"] = "on"
o = tab(c, at(10, 7))
check("shadow mode: the plan's own shadow sentence is on the tab",
homeview.why_words("shadow") in notes_of(o))
c = S.make("owner", at(10, 7))
check("the levels count on the trust the plan counts on, and the tab says "
"so", say("trust", pct=70) in notes_of(tab(c, at(10, 7))))
c.solar_trust_pct = 100
full = tab(c, at(10, 7))["days"][0]["batteries"][0]["levels"]
c.solar_trust_pct = 30
low = tab(c, at(10, 7))["days"][0]["batteries"][0]["levels"]
check("...a higher trust fills the battery more (the curve uses it)",
max(v for _h, v in full) > max(v for _h, v in low))
# --------------------------------------------------------------------------- #
print("\n5. measured solar")
# --------------------------------------------------------------------------- #
t0 = at(0, 0, 0)
sd = SolarToday()
for i in range(0, 3600, 5): # 2 kW for the hour 10:00-11:00
sd.add(t0 + timedelta(hours=10, seconds=i), 2000.0)
sd.add(t0 + timedelta(hours=11), 2000.0)
check("2 kW read all hour is 2.0 kWh for that hour",
abs(sd.kwh(t0 + timedelta(hours=10), t0 + timedelta(hours=11),
t0 + timedelta(hours=12)) - 2.0) < 0.02)
check("the hour in progress is what was made so far (half an hour: 1 kWh)",
abs(sd.kwh(t0 + timedelta(hours=10), t0 + timedelta(hours=11),
t0 + timedelta(hours=10, minutes=30)) - 1.0) < 0.05)
check("an hour with no reading, or under half covered, is not guessed",
sd.kwh(t0 + timedelta(hours=9), t0 + timedelta(hours=10),
t0 + timedelta(hours=12)) is None)
g = SolarToday()
g.add(t0 + timedelta(hours=10), 1000.0)
g.add(t0 + timedelta(hours=10, minutes=50), 1000.0) # a gap: 60 s counted
check("a gap in the readings counts for no more than a minute, so a "
"restart does not invent energy",
g.kwh(t0 + timedelta(hours=10), t0 + timedelta(hours=11),
t0 + timedelta(hours=11)) is None)
g.add(t0 + timedelta(hours=10, minutes=51), None)
g.add(t0 + timedelta(hours=10, minutes=52), 1000.0)
check("no reading ends an interval; a negative reading counts as 0",
g._last == (t0 + timedelta(hours=10, minutes=52), 1000.0))
nxt = t0 + timedelta(days=1, hours=1)
g.add(nxt, 500.0)
check("a new local day starts again", g.day != (t0).date()
and g.first == nxt and g._hours == {})
c = S.make("owner", at(10, 7))
o = tab(c, at(10, 7))
got = [h["measured_kwh"] for h in o["days"][0]["hours"]]
check("on the page: the hours already over have measured solar, the hours "
"to come do not, tomorrow has none",
got[9] is not None and got[11] is None and got[0] is not None
and all(h["measured_kwh"] is None for h in o["days"][1]["hours"]))
c = S.make("owner", at(10, 7))
c.solar_today = SolarToday()
c.solar_today.add(at(8, 0, 0), 1000.0)
o2 = tab(c, at(10, 7))
check("the add-on that started at 08:00 says measured solar counts from "
"then", say("measured_from", at="08:00") in notes_of(o2))
class _Hass:
def __init__(self):
self.reads = 0
async def state(self, e):
self.reads += 1
return {"entity_id": e, "state": "1500",
"attributes": {"unit_of_measurement": "W"}}
async def _read_pv_with(add):
c = S.make("owner", at(10, 7))
c.hass = _Hass()
c.solar_today.add = add
await c._read_pv()
return c.pv_w
def _boom(*_a):
raise RuntimeError("no")
def _survives():
try:
return asyncio.run(_read_pv_with(_boom)) == 1500.0
except Exception: # noqa: BLE001
return False
check("a fault in keeping the measured solar never costs the control "
"cycle its reading (`_read_pv` still returns the total)", _survives())
# --------------------------------------------------------------------------- #
print("\n6. words, page, route; reading asks for nothing and moves nothing")
# --------------------------------------------------------------------------- #
en = uitext.words("en")
nl = uitext.words("nl")
check("every sentence of the tab has an English and a Dutch form, with the "
"same placeholders",
set(en["home_words"]["day"]) == set(nl["home_words"]["day"])
and all(re.findall(r"\{\w+\}", en["home_words"]["day"][k])
== re.findall(r"\{\w+\}", nl["home_words"]["day"][k])
for k in en["home_words"]["day"]))
now = at(10, 7)
c = S.make("owner", now)
oe, on = tab(c, now, "en"), tab(c, now, "nl")
check("Dutch: the notes, the totals and the blocks are in Dutch, with a "
"decimal comma",
on["days"][0]["line"].startswith("Nog verwachte zon vandaag: 14,4 kWh")
and say("estimate", "nl") in on["days"][0]["notes"]
and on["days"][1]["line"].startswith("Verwachte zon morgen: 16,0 kWh")
and oe["days"][0]["line"].startswith("Solar still to come today: 14.4"))
c = S.make("dynamic", at(0, 18))
nl_b = tab(c, at(0, 18), "nl")["days"][0]["batteries"][0]["blocks"]
check("...and so are the plan's words on a block",
nl_b[0]["what"].startswith("Laadt tot") and nl_b[0]["why"].startswith(
"Goedkoopste uren") and nl_b[1]["what"].startswith("Levert tot"))
check("the language goes back to English after a Dutch build",
homeview.pill("running") == en["home_words"]["pill"]["running"])
c = S.make("dynamic", at(10, 7))
sent = []
orig_state = c.hass.state
c.hass.state = lambda *a, **k: sent.append(a) or orig_state(*a, **k)
snap = lambda: repr((c.mode_tick, c.mode_runs, c._plan_targets, c.arb,
[(i.soc, i.batt, i.last_write) for i in c.inverters],
c.mode_gate, c.day_plan))
before = snap()
t_a = time.perf_counter()
for _ in range(10):
c.forecast_payload("en")
per = (time.perf_counter() - t_a) / 10 * 1000.0
print(f" forecast_payload: {per:.1f} ms per build on this machine")
check("building the tab asks Home Assistant for nothing and changes nothing "
"control holds (modes, runs, levels, decision, readings)",
not sent and snap() == before)
def _broken(c):
real = dayview.build
dayview.build = lambda *a, **k: 1 / 0
logging.disable(logging.CRITICAL)
try:
return c.forecast_payload("en")
finally:
dayview.build = real
logging.disable(logging.WARNING)
pay = c.forecast_payload("en")
check("`Controller.forecast_payload()` gives the tab's JSON, and a fault in "
"building it gives `ok: false` rather than raising",
pay.get("ok") is True and len(pay["days"]) == 2
and _broken(c) == {"ok": False})
PAGE_SRC = None
async def _serve():
from aiohttp import web as aioweb, ClientSession
from app import web
class _Sup:
async def info(self):
return {"options": {}, "schema": [], "translations": {}}
c = S.make("owner", datetime.now(UTC))
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]
res = {}
try:
async with ClientSession() as s:
async with s.get(f"http://127.0.0.1:{port}/forecast") as r:
res["en"] = (r.status, await r.json())
async with s.get(f"http://127.0.0.1:{port}/forecast?lang=nl-BE") as r:
res["nl"] = await r.json()
finally:
await runner.cleanup()
return res
try:
res = asyncio.run(_serve())
except Exception: # noqa: BLE001
res = None
check("GET forecast answers 200 with the tab's JSON, in English by default "
"and in Dutch with ?lang=nl",
res is not None and res["en"][0] == 200 and res["en"][1]["ok"] is True
and res["en"][1]["days"][0]["line"].startswith("Solar still to come")
and res["nl"]["days"][0]["line"].startswith("Nog verwachte zon"))
PAGE = open(os.path.join(HERE, "app", "static", "index.html"),
encoding="utf-8").read()
JS = open(os.path.join(HERE, "app", "static", "app.js"),
encoding="utf-8").read()
check("the page has the Forecast tab between Batteries and Settings, with "
"its own panel", re.findall(r'data-hash="(\w+)"', PAGE)[:4]
== ["home", "batteries", "forecast", "settings"]
and 'id="s-fc"' in PAGE and 'aria-labelledby="t-fc"' in PAGE)
check("it asks for `forecast?lang=` (relative, for ingress), only while its "
"screen shows, and sets data as text",
'fetch("forecast?lang="' in JS and "scheduleForecast" in JS
and "innerHTML" not in JS)
labels = set(re.findall(r'''\bt\("(forecast\.[\w.]+)"\)''', JS)) | set(
re.findall(r'data-t(?:-label)?="(forecast\.[\w.]+)"', PAGE))
miss = sorted(k for k in labels if not isinstance(
en["home"]["forecast"].get(k.split(".", 1)[1]), str)
or not isinstance(nl["home"]["forecast"].get(k.split(".", 1)[1]), str))
check(f"every label the screen asks for is in both languages ({len(labels)} "
f"keys; missing {miss})", len(labels) >= 12 and not miss)
CSS = open(os.path.join(HERE, "app", "static", "app.css"),
encoding="utf-8").read()
check("it is readable on a phone: columns flex to the width, the chart has "
"no fixed pixel width, and the controls are 44 px tall",
"min-width:0" in CSS and "width:100%" in CSS
and not re.search(r"\.fc[a-z]*\{[^}]*width:\s*\d{3,}px", CSS)
and "min-height:44px" in CSS)
print()
if fails:
print(f"{len(fails)} FAILED")
sys.exit(1)
print("all passed")
+275
View File
@@ -0,0 +1,275 @@
"""UX-54: controllers in a known situation, for test_ux54.py and the browser
proof (`scratch/UX-54/dev/demo54.py`). Real `Controller`, readings set by hand,
no Home Assistant. `make(name, now)` returns one; the names are in `SCENARIOS`.
owner modes on, both batteries Solar first, a day/night tariff, a forecast
from three state-only sensors (Forecast.Solar's shape), a house
profile (the owner's site 2026-10-26)
dynamic modes on, A Leftover solar and B Cheap hours, 15-minute prices with a
cheap night and a dear evening, a dark day (a grid charge is due)
sunny as dynamic, a sunny day (the sun fills A, no grid charge)
periods as dynamic with a Solcast-shaped forecast (an hourly split)
off modes off, arbitrage on, the same prices
learning modes on, a forecast, and no house profile yet
nofc modes on, no forecast sensor set
stale modes on, a sensor set that gives no forecast now
notomorrow as owner, today's forecast only
cycle as dynamic, battery A waiting for its monthly cycle
"""
import asyncio
import json
import math
import os
import sys
from datetime import datetime, timedelta, timezone
HERE = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, HERE)
from app import dayplan, packmode, solarfc # noqa: E402
from app.main import Controller # noqa: E402
from app.peakguard import PeakDecision # noqa: E402
from app.price import PricePoint, PriceProduct, PriceRead, brussels # noqa
from app.static_tariff import TariffPlan, static_plan # noqa: E402
from app.store import unit_key # noqa: E402
UTC = timezone.utc
SCENARIOS = ("owner", "dynamic", "sunny", "periods", "off", "learning",
"nofc", "stale", "notomorrow", "cycle")
INV = [{"name": "inverter-a", "setpoint_entity": "number.a",
"soc_entity": "sensor.soc_a", "batt_entity": "sensor.batt_a",
"max_w": 5000, "capacity_kwh": 15.36},
{"name": "inverter-b", "setpoint_entity": "number.b",
"soc_entity": "sensor.soc_b", "batt_entity": "sensor.batt_b",
"max_w": 4600, "capacity_kwh": 20.48}]
BASE = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
"meter_source": "off", "max_w": 9600, "shadow_mode": "off",
"maintenance_enabled": False, "arbitrage_enabled": True,
"cycle_cost_threshold_eur": 0.10, "pv_entity": "sensor.pv",
"inverters": INV}
FC1 = "sensor.energy_production_today_remaining"
FCT1 = "sensor.energy_production_tomorrow"
DN = {"periods": [{"name": "day", "from": "07:00", "to": "22:00",
"days": "all", "eur_per_kwh": 0.2801},
{"name": "night", "from": "22:00", "to": "07:00",
"days": "all", "eur_per_kwh": 0.2333}]}
class Store:
fresh = False
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 []
def day0_of(now):
return solarfc._bounds(now)[0]
def quarters(now, price_of, days=2):
"""`days` days of 15-minute prices from local midnight; `price_of(h)`."""
d0 = day0_of(now)
pts = []
for i in range(96 * days):
t = d0 + timedelta(minutes=15 * i)
pts.append(PricePoint(brussels(t), price_of((i % 96) / 4.0), False))
return TariffPlan(PriceRead.series(tuple(pts), PriceProduct.DAY_AHEAD,
"generic_dayahead", now), None,
"provider")
def dyn_price(h):
if 1.0 <= h < 5.0:
return 0.08
if 11.0 <= h < 15.0:
return 0.14
if 17.0 <= h < 21.0:
return 0.34
return 0.20
def profile(now, *, day=600.0, evening=1800.0, night=350.0):
"""A house profile with three full days: the net use per local hour."""
shape = {h: (night if h < 6 or h >= 22 else 900.0 if h < 9
else day if h < 17 else evening) for h in range(24)}
d = brussels(now).date()
return dayplan.HouseProfile(
{(d - timedelta(days=k)).isoformat():
{str(h): w for h, w in shape.items()} for k in (1, 2, 3)})
def sensor(entity, kwh, now, *, periods=None, unit="kWh"):
"""A forecast sensor's state, as Home Assistant gives it."""
attrs = {"unit_of_measurement": unit}
if periods:
attrs["wh_period"] = periods
return {"entity_id": entity, "state": f"{kwh:.3f}", "attributes": attrs,
"last_updated": now.isoformat(),
"last_reported": now.isoformat()}
def bell_periods(now, day_kwh, *, tomorrow=False):
"""An hourly split (Open-Meteo's `wh_period`) of a bell, for `now`'s day."""
d0 = day0_of(now) + (timedelta(days=1) if tomorrow else timedelta(0))
out = {}
for h in range(24):
a = solarfc._bell(h + 1.0) - solarfc._bell(float(h))
out[(d0 + timedelta(hours=h)).isoformat()] = round(
a * day_kwh * 1000.0, 1)
return out
def left_today(now, day_kwh):
"""What the integration says is left today at `now` (state only)."""
d0 = day0_of(now)
h = (now - d0).total_seconds() / 3600.0
return day_kwh * (1.0 - solarfc._bell(h))
def forecast(now, today_kwh, tomorrow_kwh, *, split=False, members=1,
tomorrow=True):
"""A `solarfc.Forecast` through `read_sum()`, as the controller reads it.
`today_kwh` is the whole day's sun; the sensors say what is left."""
left = left_today(now, today_kwh)
ts = [(f"{FC1}_{i}", sensor(
f"{FC1}_{i}", left / members, now,
periods=(bell_periods(now, today_kwh / members) if split else None)))
for i in range(members)]
tm = [(f"{FCT1}_{i}", sensor(
f"{FCT1}_{i}", tomorrow_kwh / members, now,
periods=(bell_periods(now, tomorrow_kwh / members, tomorrow=True)
if split else None)))
for i in range(members)] if tomorrow else []
fc, why, _gaps = solarfc.read_sum(ts, tm, now)
return fc
def measure(c, now, peak_w):
"""The solar this site made today until `now`, one reading a minute."""
d0 = day0_of(now)
t = d0
while t <= now:
h = (t - d0).total_seconds() / 3600.0
c.solar_today.add(t, peak_w * math.sin(math.pi * (h - 6.0) / 14.0)
if 6.0 <= h < 20.0 else 0.0)
t += timedelta(minutes=1)
def make(name, now, *, soc=(55.0, 40.0), modes=("solar_first",
"solar_first")):
"""The controller of scenario `name` at `now` (UTC), its cycle run once."""
assert name in SCENARIOS, name
opts = dict(BASE)
plan = quarters(now, dyn_price)
sun_today, sun_tomorrow = 18.0, 16.0
members, split, tomorrow = 1, False, True
if name == "owner":
opts.update(battery_modes_enabled=True,
pv_forecast_entities=[f"{FC1}_{i}" for i in range(3)],
pv_forecast_tomorrow_entities=[f"{FCT1}_{i}"
for i in range(3)],
static_profile=json.dumps(DN), arbitrage_enabled=False)
plan = static_plan(opts, day0_of(now), day0_of(now) + timedelta(
hours=48), now=now)
members = 3
elif name in ("dynamic", "sunny", "periods", "learning", "stale",
"cycle"):
opts.update(battery_modes_enabled=True, pv_forecast_entity=FC1,
pv_forecast_tomorrow_entity=FCT1)
modes = ("leftover_solar", "cheap_hours")
if name == "dynamic":
sun_today, sun_tomorrow = 3.0, 3.0
if name == "periods":
split = True
sun_today, sun_tomorrow = 6.0, 6.0
if name == "cycle":
opts.update(maintenance_enabled=True)
elif name == "off":
opts.update(battery_modes_enabled=False, arbitrage_enabled=True)
elif name == "nofc":
opts.update(battery_modes_enabled=True)
modes = ("leftover_solar", "cheap_hours")
elif name == "notomorrow":
opts.update(battery_modes_enabled=True, pv_forecast_entity=FC1,
static_profile=json.dumps(DN), arbitrage_enabled=False)
plan = static_plan(opts, day0_of(now), day0_of(now) + timedelta(
hours=48), now=now)
tomorrow = False
if name == "sunny":
sun_today, sun_tomorrow = 40.0, 40.0
c = Controller(opts, Hass(), Store(), Pub(), stop=asyncio.Event())
c.auto = True
for i, inv in enumerate(c.inverters):
inv.soc, inv.batt, inv.last_write = soc[i], 0.0, 0.0
inv.dev_min, inv.dev_max = -5000.0, 5000.0
if opts.get("battery_modes_enabled"):
c.store.data[unit_key(inv.name, packmode.MODE_KEY)] = modes[i]
c.grid, c.pv_w = 300.0, 0.0
c.tariff_plan, c.tariff_at = plan, now
c.peak_guard = PeakDecision(engaged=False, floor_w=None, ceiling_w=7000.0,
ceiling_from="cap", projected_w=800.0)
c.arb_headroom_w = 6000.0
if name not in ("nofc", "off", "stale") and \
opts.get("battery_modes_enabled"):
c.solar_fc = forecast(now, sun_today, sun_tomorrow, split=split,
members=members, tomorrow=tomorrow)
c.solar_fc_why = None
if name != "learning":
c.house_profile = profile(now)
if name in ("owner", "dynamic", "sunny", "periods", "learning",
"notomorrow", "cycle"):
measure(c, now, 7000.0 if name != "dynamic" else 2500.0)
if opts.get("battery_modes_enabled"):
c.tariff_at = now
c._mode_roles(now, None)
return c