Claude
c9ab48d6fc
- Settings tab: the Installer view (app/static/settings.js), drawn from
app/uilayout.py through the new app/uipage.py: search, six collapsible
groups with "n to fix / n to check / n changed" badges and an "applies
after a restart" chip, radio cards, segmented choices, sliders with units
(the peak limit in kW), entity pickers with search and the live reading,
inverter cards, solar rows, locked measured values behind "Change", and
the add-on's refusals, warnings and info beside the field they name.
- Save path: the page posts its model to the existing config/validate and
config/save; uimodel turns it into options before the shipped validator,
so nothing it refuses is saved. No route added. A value equal to the
stored one is written as stored, so a save that changed nothing writes an
identical blob. Save bar, then the restart dialog that names 0 W.
- The region choice shows its preset year; Wallonia and Brussels are listed
as having no preset yet (TARIFF-22). "Classic settings" keeps /config.
- P1 test findings: (a) "Going to 0 W" until the last accepted power request
is 0 W; (b) the trade blocks and the price line name the feed's own points
("Lowest published price"), the hour bars say "hour average"; (c) the entity
list's WARNING gives a reason on a timeout.
- test_settings.py; test_home, test_haentities, test_uilayout extended.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LyR88rXETvwhTgx6yHE39p
1024 lines
40 KiB
Python
1024 lines
40 KiB
Python
"""UX-13 P1: the Home screen's part of `/status`, built from what the add-on
|
|
already knows and already plans. Reads only; commands nothing.
|
|
|
|
`build(controller)` returns `now`, `batteries`, `plan`, `attention`,
|
|
`month_peak`, `control` and `state`. The old fields (`banner`, `rows`,
|
|
`checks`, ...) stay beside them for the installer details and for anything
|
|
that reads the JSON. See DECISIONS/UX-13-one-layout-table.md and the design
|
|
note, section 3.3.
|
|
|
|
The plan shows only what the add-on does today: trading on price (the
|
|
arbitrage verdict, the same call the control cycle makes), the monthly battery
|
|
cycle, the peak limit and "follows the house". Battery modes (OPT-11) are not
|
|
built, so nothing here invents a mode plan. Every sentence is in the words
|
|
section at the bottom and is held to STYLE.md by test_messages.py.
|
|
"""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from . import arbitrage, messages, peakguard
|
|
from .maintenance import CHARGE as M_CHARGE, DRAIN as M_DRAIN, HOLD as M_HOLD
|
|
from .maintenance import IDLE as M_IDLE
|
|
from .price import Absence, Uncertainty, brussels
|
|
from .static_tariff import static_plan
|
|
from .tariff import BILLING_TZ
|
|
from .uimodel import REGION_PRESETS, options_to_ui
|
|
|
|
# A battery at or under this many watts is resting, not charging or giving.
|
|
IDLE_W = 50.0
|
|
# Display only: the lowest third of today's price range is "cheap", the top
|
|
# third "expensive". A price range under this is flat, and nothing is marked.
|
|
FLAT_EUR_KWH = 0.01
|
|
|
|
# Severity words the page shows; the same four as STYLE.md section 3.
|
|
ALARM, WARNING, INFO = messages.ALARM, messages.WARNING, messages.INFO
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Time: today's local day, on the billing calendar
|
|
# --------------------------------------------------------------------------- #
|
|
def local_day(now: datetime):
|
|
"""`(midnight_utc, next_midnight_utc)` of the local day holding `now`."""
|
|
here = now.astimezone(BILLING_TZ)
|
|
start = datetime(here.year, here.month, here.day, tzinfo=BILLING_TZ)
|
|
nxt = start.date() + timedelta(days=1)
|
|
end = datetime(nxt.year, nxt.month, nxt.day, tzinfo=BILLING_TZ)
|
|
return start.astimezone(timezone.utc), end.astimezone(timezone.utc)
|
|
|
|
|
|
def _hours(t: datetime, day0: datetime) -> float:
|
|
return (t - day0).total_seconds() / 3600.0
|
|
|
|
|
|
def hhmm(t: datetime) -> str:
|
|
return f"{brussels(t):%H:%M}"
|
|
|
|
|
|
def _kw(w) -> str:
|
|
return f"{abs(float(w)) / 1000:.1f} kW"
|
|
|
|
|
|
def _eur(v) -> str:
|
|
return f"€{float(v):.2f}"
|
|
|
|
|
|
def _kwh(v) -> str:
|
|
return f"{float(v):.1f} kWh"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Now
|
|
# --------------------------------------------------------------------------- #
|
|
def flow(w, *, shadow=False):
|
|
"""`charging` / `giving` / `resting` / None for one battery power."""
|
|
if w is None:
|
|
return None
|
|
if shadow:
|
|
return "resting"
|
|
if w > IDLE_W:
|
|
return "giving"
|
|
if w < -IDLE_W:
|
|
return "charging"
|
|
return "resting"
|
|
|
|
|
|
def now_block(*, grid_w, pv_w, batt_w, has_solar, shadow, settled=True):
|
|
"""The four tiles. `house_w` needs all three readings, so without a solar
|
|
reading it is None rather than a number that leaves the sun out."""
|
|
house, disagree = None, False
|
|
if has_solar and None not in (grid_w, pv_w, batt_w):
|
|
house = float(grid_w) + float(pv_w) + float(batt_w)
|
|
# The three are read at slightly different moments, so the sum can dip
|
|
# under zero. Inside the idle band that is 0 W; beyond it the readings
|
|
# disagree and no house figure is shown.
|
|
if house < -IDLE_W:
|
|
house, disagree = None, True
|
|
elif house < 0.0:
|
|
house = 0.0
|
|
return {
|
|
"solar_w": None if pv_w is None else float(pv_w),
|
|
"house_w": house,
|
|
"grid_w": None if grid_w is None else float(grid_w),
|
|
"batt_w": None if batt_w is None else float(batt_w),
|
|
"has_solar": bool(has_solar),
|
|
"lines": {
|
|
"solar": tile_solar(pv_w, has_solar),
|
|
"house": tile_house(house, has_solar, disagree),
|
|
"grid": tile_grid(grid_w),
|
|
"batt": tile_batt(batt_w, shadow, settled),
|
|
},
|
|
}
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Prices for today's chart
|
|
# --------------------------------------------------------------------------- #
|
|
def _points(plan):
|
|
if plan is None:
|
|
return ()
|
|
return plan.read.match(series=lambda r: tuple(r.points),
|
|
no_data=lambda _r: ())
|
|
|
|
|
|
def _spans(points):
|
|
"""`[(point, end)]`: each point runs until the next one starts, and the
|
|
last one for the series' own spacing (15 minutes or an hour)."""
|
|
pts = sorted(points, key=lambda p: p.start)
|
|
gaps = [b.start - a.start for a, b in zip(pts, pts[1:])]
|
|
step = min([g for g in gaps if g > timedelta(0)] or [timedelta(hours=1)])
|
|
step = min(step, timedelta(hours=1))
|
|
out = []
|
|
for i, p in enumerate(pts):
|
|
nxt = pts[i + 1].start if i + 1 < len(pts) else None
|
|
end = nxt if nxt is not None and nxt - p.start <= timedelta(hours=1) \
|
|
else p.start + step
|
|
out.append((p, end))
|
|
return out
|
|
|
|
|
|
def _day_prices(c):
|
|
cache = getattr(c, "home_prices", None)
|
|
if cache is None:
|
|
cache = c.home_prices = DayPrices()
|
|
return cache
|
|
|
|
|
|
class DayPrices:
|
|
"""Today's prices as seen so far. The refresh keeps only the hours ahead,
|
|
so the hours already past are remembered here for the chart. Display only:
|
|
nothing reads this back into a decision."""
|
|
|
|
def __init__(self):
|
|
self.day = None
|
|
self.points = {}
|
|
|
|
def see(self, points, day0, day1):
|
|
if self.day != day0:
|
|
self.day, self.points = day0, {}
|
|
for p in points:
|
|
if day0 <= p.start < day1:
|
|
self.points[p.start] = p
|
|
return sorted(self.points.values(), key=lambda p: p.start)
|
|
|
|
|
|
def hour_bars(points, day0, day1):
|
|
"""One bar per local hour: the mean of its prices, or None when unknown.
|
|
`kind` is cheap / dear / mid by thirds of today's range (display only).
|
|
The page calls these "low price" and "high price", not cheap hours: the
|
|
add-on trades in the one cheapest and one dearest hour, not in a third."""
|
|
n = int(round(_hours(day1, day0)))
|
|
bars = []
|
|
for k in range(n):
|
|
a, b = day0 + timedelta(hours=k), day0 + timedelta(hours=k + 1)
|
|
inside = [p for p in points if a <= p.start < b]
|
|
price = (sum(p.eur_per_kwh for p in inside) / len(inside)
|
|
if inside else None)
|
|
bars.append({"h": k, "start": a.isoformat(), "label": hhmm(a),
|
|
"end_label": hhmm(b), "points": len(inside),
|
|
"eur_kwh": None if price is None else round(price, 4),
|
|
"estimate": bool(inside) and any(p.provisional
|
|
for p in inside),
|
|
"kind": None})
|
|
known = [b["eur_kwh"] for b in bars if b["eur_kwh"] is not None]
|
|
if known and max(known) - min(known) >= FLAT_EUR_KWH:
|
|
lo, hi = min(known), max(known)
|
|
third = (hi - lo) / 3.0
|
|
for b in bars:
|
|
p = b["eur_kwh"]
|
|
if p is None:
|
|
continue
|
|
b["kind"] = ("cheap" if p <= lo + third else
|
|
"dear" if p >= hi - third else "mid")
|
|
elif known:
|
|
for b in bars:
|
|
if b["eur_kwh"] is not None:
|
|
b["kind"] = "mid"
|
|
return bars
|
|
|
|
|
|
def day_range(points):
|
|
"""(lowest, highest) of today's published points, as {eur_kwh, label},
|
|
or (None, None). The feed's own points, like the trade legs, not the
|
|
chart's hour averages."""
|
|
firm = [p for p in points if not p.provisional]
|
|
if not firm:
|
|
return None, None
|
|
lo = min(firm, key=lambda p: p.eur_per_kwh)
|
|
hi = max(firm, key=lambda p: p.eur_per_kwh)
|
|
return ({"eur_kwh": round(lo.eur_per_kwh, 4), "label": hhmm(lo.start)},
|
|
{"eur_kwh": round(hi.eur_per_kwh, 4), "label": hhmm(hi.start)})
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# The plan, per battery
|
|
# --------------------------------------------------------------------------- #
|
|
def _run(i, kind, start, end, day0, day1, what, why, amount, *, kwh=None,
|
|
eur=None, price=None):
|
|
a = max(0.0, _hours(start, day0))
|
|
z = min(_hours(day1, day0), _hours(end, day0))
|
|
return {"battery": i, "kind": kind, "start": start.isoformat(),
|
|
"end": end.isoformat(), "from": hhmm(start),
|
|
"to": "24:00" if end == day1 else hhmm(end),
|
|
"start_h": round(a, 3), "end_h": round(z, 3),
|
|
"what": what, "why": why, "amount": amount,
|
|
"kwh": None if kwh is None else round(kwh, 2),
|
|
"eur": None if eur is None else round(eur, 2),
|
|
"eur_kwh": None if price is None else round(price, 4)}
|
|
|
|
|
|
def trade_verdict(c, plan):
|
|
"""The arbitrage verdict the control cycle would reach on this plan, or a
|
|
reason there is none. The SAME call and inputs as `Controller.cycle()`:
|
|
`plan.arbitrage()` with the peak guard's state, the cycle cost and the
|
|
no-weather uncertainty. Returns `(verdict_or_None, why_key)`."""
|
|
if not arbitrage.enabled(c.o):
|
|
return None, "off"
|
|
guard = arbitrage.guard_state(c.peak_guard)
|
|
if guard is None:
|
|
return None, "not_covered"
|
|
if plan is None:
|
|
return None, "no_prices"
|
|
verdict = plan.arbitrage(peak_guard_active=guard,
|
|
min_spread_eur_kwh=arbitrage.threshold_eur_kwh(c.o),
|
|
uncertainty=frozenset({Uncertainty.NO_WEATHER}))
|
|
return verdict.match(
|
|
allowed=lambda v: (v, "allowed"),
|
|
refused=lambda _v: (None, refusal_key(plan, guard)))
|
|
|
|
|
|
def refusal_key(plan, guard):
|
|
"""Which of `TariffPlan.arbitrage()`'s refusals fired, in its own order:
|
|
the peak guard, no prices, under two published points, then the gap."""
|
|
if guard:
|
|
return "peak"
|
|
if not plan.has_prices:
|
|
return "no_prices"
|
|
if sum(1 for p in _points(plan) if not p.provisional) < 2:
|
|
return "not_firm"
|
|
return "not_worth"
|
|
|
|
|
|
def _point_end(plan, point):
|
|
for p, end in _spans(_points(plan)):
|
|
if p.start == point.start:
|
|
return end
|
|
return point.start + timedelta(hours=1)
|
|
|
|
|
|
# Under this many kWh a block is not drawn: the cycle would not move it.
|
|
MIN_KWH = 0.05
|
|
|
|
|
|
def trade_runs(c, plan, verdict, day0, day1, now, shadow):
|
|
"""A charge block and a give block on each battery, where the verdict put
|
|
them. The site trades as one; each battery's share is capped by its own
|
|
rating and by the room its level leaves (arbitrage's own floor/ceiling).
|
|
Only what `arbitrage.decide()` would still do is drawn: no block on a
|
|
battery with no room, no sale at a negative price, and no sale while the
|
|
emptiest battery is at the floor (the cycle's `soc` is the emptiest).
|
|
With a level unknown the cycle does not trade, so nothing is drawn."""
|
|
floor, ceiling = arbitrage.soc_bounds(c.o)
|
|
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:
|
|
end = _point_end(plan, pt)
|
|
if end <= now:
|
|
continue
|
|
if not (day0 <= pt.start < day1):
|
|
later.append((kind, pt))
|
|
continue
|
|
if kind == "give" and (pt.eur_per_kwh < 0 or min(level) <= floor):
|
|
continue
|
|
hrs = (end - pt.start).total_seconds() / 3600.0
|
|
for i, inv in enumerate(c.inverters):
|
|
kwh = inv.max_w / 1000.0 * hrs
|
|
cap = inv.capacity_kwh
|
|
if cap:
|
|
room = (ceiling - level[i] if kind == "grid"
|
|
else level[i] - floor)
|
|
kwh = min(kwh, max(0.0, room) / 100.0 * cap)
|
|
# A later give may use what this charge put in: an upper
|
|
# bound, like every amount here ("up to").
|
|
level[i] += (kwh if kind == "grid" else -kwh) / cap * 100.0
|
|
if kwh < MIN_KWH:
|
|
continue
|
|
eur = kwh * pt.eur_per_kwh
|
|
if kind == "grid":
|
|
what, why, amount = (run_grid_what(kwh, shadow),
|
|
run_grid_why(pt.eur_per_kwh),
|
|
run_grid_amount(eur))
|
|
else:
|
|
what, why, amount = (run_give_what(kwh, shadow),
|
|
run_give_why(pt.eur_per_kwh),
|
|
run_give_amount(eur))
|
|
runs.append(_run(i, kind, pt.start, end, day0, day1, what, why,
|
|
amount, kwh=kwh, eur=eur, price=pt.eur_per_kwh))
|
|
return runs, later, None
|
|
|
|
|
|
def maintenance_runs(c, day0, day1, now, shadow):
|
|
"""The monthly battery cycle: under way now, or starting later today.
|
|
The cycle starts in the hour `maintenance_start_hour` names, read as the
|
|
control cycle reads it (UTC hour of `now`), shown in local time."""
|
|
runs, lines = [], []
|
|
waiting = []
|
|
for i, (inv, m) in enumerate(zip(c.inverters, c.maints)):
|
|
cfg = m.cfg
|
|
if not cfg.enabled:
|
|
continue
|
|
cap = inv.capacity_kwh
|
|
if m.phase != M_IDLE:
|
|
hours = _cycle_hours(m.phase, inv.soc, cap, cfg)
|
|
end = now + timedelta(hours=hours or 1.0)
|
|
runs.append(_run(i, "cycle", now, end, day0, day1,
|
|
run_cycle_what(m.phase, shadow),
|
|
run_cycle_why(shadow),
|
|
run_cycle_amount(hours)))
|
|
continue
|
|
if not m.due(now) and (m.next_due(now) is None
|
|
or m.next_due(now) >= day1):
|
|
continue
|
|
start = now.replace(hour=cfg.start_hour, minute=0, second=0,
|
|
microsecond=0)
|
|
if start + timedelta(hours=1) <= now:
|
|
start += timedelta(days=1)
|
|
due_at = m.next_due(now)
|
|
if due_at is not None and due_at > start:
|
|
continue
|
|
if not (day0 <= start < day1):
|
|
continue
|
|
waiting.append((m.waited_days(now), i, start))
|
|
# One battery cycles at a time (maintenance.arbitrate); the one waiting
|
|
# longest goes first, and none starts while another is under way.
|
|
busy = any(m.phase != M_IDLE for m in c.maints)
|
|
waiting.sort(key=lambda t: (-t[0], t[1]))
|
|
for n, (_w, i, start) in enumerate(waiting):
|
|
inv, m = c.inverters[i], c.maints[i]
|
|
hours = _cycle_hours(M_DRAIN, inv.soc, inv.capacity_kwh, m.cfg)
|
|
if n == 0 and not busy:
|
|
end = start + timedelta(hours=hours or 1.0)
|
|
runs.append(_run(i, "cycle", start, end, day0, day1,
|
|
run_cycle_what(M_IDLE, shadow),
|
|
run_cycle_why(shadow),
|
|
run_cycle_amount(hours)))
|
|
else:
|
|
lines.append(line_cycle_waits(inv.name))
|
|
return runs, lines
|
|
|
|
|
|
def _cycle_hours(phase, soc, cap_kwh, cfg):
|
|
"""Hours left in a cycle, or None when the capacity or level is unknown."""
|
|
if phase == M_HOLD:
|
|
return cfg.hold_min / 60.0
|
|
if not cap_kwh or soc is None:
|
|
return None
|
|
hours = cfg.hold_min / 60.0
|
|
fill_from = soc if phase == M_CHARGE else cfg.soc_floor
|
|
hours += (max(0.0, cfg.soc_target - fill_from) / 100.0 * cap_kwh
|
|
/ max(cfg.charge_w / 1000.0, 0.1))
|
|
if phase not in (M_CHARGE,):
|
|
hours += (max(0.0, soc - cfg.soc_floor) / 100.0 * cap_kwh
|
|
/ max(cfg.discharge_w / 1000.0, 0.1))
|
|
return hours
|
|
|
|
|
|
def peak_run(c, day0, day1, now, shadow, limit_w):
|
|
"""While the batteries hold the grid under the peak limit, one block for
|
|
the current 15-minute period on every battery."""
|
|
pg = c.peak_guard
|
|
if pg is None or not pg.engaged:
|
|
return []
|
|
start = now - timedelta(minutes=now.minute % 15, seconds=now.second,
|
|
microseconds=now.microsecond)
|
|
end = start + timedelta(minutes=15)
|
|
return [_run(i, "peak", start, end, day0, day1,
|
|
run_peak_what(pg.ceiling_w or limit_w, shadow),
|
|
run_peak_why(), None)
|
|
for i in range(len(c.inverters))]
|
|
|
|
|
|
def house_runs(n, day0, day1, now, shadow):
|
|
""""Follows the house" for the rest of the day, drawn under the others."""
|
|
return [_run(i, "house", max(now, day0), day1, day0, day1,
|
|
run_house_what(shadow), run_house_why(shadow),
|
|
run_house_amount())
|
|
for i in range(n)]
|
|
|
|
|
|
def region_of(opts):
|
|
"""The levies preset and its year, as the page model has it (UX-13 P0)."""
|
|
try:
|
|
ui = options_to_ui(opts)
|
|
except Exception: # noqa: BLE001
|
|
return {"id": None, "year": None}
|
|
rid = ui.get("ui.region")
|
|
return {"id": rid, "year": ui.get("region_year")
|
|
if rid in REGION_PRESETS else None}
|
|
|
|
|
|
def build_plan(c, now, shadow, running, held=False, settled=True):
|
|
day0, day1 = local_day(now)
|
|
fresh = c.fresh_tariff_plan(now)
|
|
plan = fresh
|
|
if plan is None or plan.origin == "static":
|
|
# A static profile is pure arithmetic: price the whole day from it.
|
|
try:
|
|
static = static_plan(c.o, day0, day1, now=now)
|
|
if static.has_prices:
|
|
plan = static
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
seen = _day_prices(c).see(_points(plan), day0, day1)
|
|
bars = hour_bars(seen, day0, day1)
|
|
|
|
limit_w = c.peak.cap_w # the limit the controller parsed at start
|
|
runs, lines = [], []
|
|
# The control cycle trades on the fresh plan only, so this does too.
|
|
verdict, trade_why = trade_verdict(c, fresh)
|
|
if running:
|
|
runs += house_runs(len(c.inverters), day0, day1, now, shadow)
|
|
if verdict is not None:
|
|
trade, later, no_trade = trade_runs(c, fresh, verdict, day0, day1,
|
|
now, shadow)
|
|
runs += trade
|
|
if no_trade:
|
|
lines.append(line_no_trade(no_trade))
|
|
for kind, pt in later:
|
|
lines.append(line_trade_tomorrow(kind, hhmm(pt.start),
|
|
pt.eur_per_kwh, shadow))
|
|
else:
|
|
lines.append(line_no_trade(trade_why))
|
|
cyc, waits = maintenance_runs(c, day0, day1, now, shadow)
|
|
runs += cyc
|
|
lines += waits
|
|
runs += peak_run(c, day0, day1, now, shadow, limit_w)
|
|
if c.p1_enabled:
|
|
# The guard's ceiling: the cap, or this month's peak once above it.
|
|
pg = c.peak_guard
|
|
lines.append(line_peak_limit(
|
|
(pg.ceiling_w if pg is not None else None) or limit_w, shadow))
|
|
lines = [x for x in lines if x]
|
|
|
|
kinds = {r["kind"] for r in runs}
|
|
if held:
|
|
verdict_key = "held"
|
|
elif not running:
|
|
verdict_key = "stopped"
|
|
elif "cycle" in kinds:
|
|
verdict_key = "cycle"
|
|
elif "grid" in kinds or "give" in kinds:
|
|
verdict_key = "trade"
|
|
else:
|
|
verdict_key = "house"
|
|
|
|
known = [b for b in bars if b["eur_kwh"] is not None]
|
|
price_now = None
|
|
try:
|
|
if plan is not None:
|
|
price_now = plan.price_at(now)
|
|
except Absence:
|
|
price_now = None
|
|
feed = fresh is not None and fresh.from_feed
|
|
tomorrow = any(not p.provisional and p.start >= day1
|
|
for p in _points(fresh))
|
|
region = region_of(c.o)
|
|
return {
|
|
"day": brussels(now).date().isoformat(),
|
|
"now_h": round(_hours(now, day0), 3),
|
|
"hours": int(round(_hours(day1, day0))),
|
|
"verdict": (VERDICT[verdict_key] if settled or verdict_key != "held"
|
|
else ZERO_GOING),
|
|
"verdict_key": verdict_key,
|
|
"head": plan_head(verdict_key, shadow, settled),
|
|
"prices": bars,
|
|
"runs": runs,
|
|
"lines": lines,
|
|
"why": c.why_words,
|
|
"price_line": line_prices(price_now, *day_range(seen)),
|
|
"tomorrow_line": line_tomorrow(tomorrow) if feed else "",
|
|
"region": region,
|
|
"region_line": (line_region(region["id"], region["year"])
|
|
if feed else ""),
|
|
"has_prices": bool(known),
|
|
}
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Batteries, attention, month peak, control
|
|
# --------------------------------------------------------------------------- #
|
|
def at_zero(inv) -> bool:
|
|
"""Has this inverter's last accepted power request reached 0 W? A stop or
|
|
an alarm asks for 0 W at once, but the inverter gets it with the next
|
|
write, so until then the page says "Going to 0 W" (UX-13 P1 test, a)."""
|
|
return abs(inv.standing_w()) < 1.0
|
|
|
|
|
|
def battery_rows(c, shadow, running, held=False):
|
|
shares = {}
|
|
alloc = getattr(c, "allocation", None)
|
|
if shadow and alloc is not None and getattr(alloc, "shares", None):
|
|
shares = dict(alloc.shares)
|
|
rows = []
|
|
for i, inv in enumerate(c.inverters):
|
|
m = c.maints[i] if i < len(c.maints) else None
|
|
phase = m.phase if m is not None else M_IDLE
|
|
cap = inv.capacity_kwh
|
|
rows.append({
|
|
"name": inv.name,
|
|
"level_pct": None if inv.soc is None else float(inv.soc),
|
|
"power_w": None if inv.batt is None else float(inv.batt),
|
|
"capacity_kwh": cap,
|
|
"stored_kwh": (None if cap is None or inv.soc is None
|
|
else round(cap * float(inv.soc) / 100.0, 2)),
|
|
"state": flow(inv.batt, shadow=shadow),
|
|
"mode": MODE_HOUSE,
|
|
"maintenance": phase,
|
|
"now": battery_now(inv.soc, inv.batt, phase=phase, shadow=shadow,
|
|
running=running, held=held,
|
|
would_w=shares.get(inv.name),
|
|
settled=at_zero(inv)),
|
|
# Per-battery reasons arrive with battery modes (OPT-11).
|
|
"why": None,
|
|
})
|
|
return rows
|
|
|
|
|
|
# Row id -> the title on Home. The row's own sentence is the body.
|
|
ATTN_TITLES = {
|
|
"controller": "Control is not running",
|
|
"meter": "Meter reading",
|
|
"meter_frozen": messages.T_METER_FROZEN,
|
|
"freeze_check": "Stuck meter check",
|
|
"month_peak": "This month's peak",
|
|
"peak_guard": "Peak limit",
|
|
"peak_forecast": "Peak limit forecast",
|
|
"sensor": "Battery readings",
|
|
"control": "Inverter control",
|
|
"power_too_high": messages.T_OUTPUT_CLAMP,
|
|
"not_keeping_up": messages.T_LOOP_BOUND,
|
|
"write": "Power request not accepted",
|
|
"direction": "Power went the wrong way",
|
|
"site_direction": "Power went the wrong way",
|
|
"circulating": messages.T_CIRCULATING,
|
|
"delivery": messages.S_NOT_DELIVERING,
|
|
"estop": "Emergency stop",
|
|
"pv": "Solar reading",
|
|
None: "Inverter connection",
|
|
}
|
|
# Row id -> the DOCS.md heading under Notifications, where one exists.
|
|
ATTN_DOCS = {
|
|
"meter_frozen": messages.T_METER_FROZEN,
|
|
"power_too_high": messages.T_OUTPUT_CLAMP,
|
|
"not_keeping_up": messages.T_LOOP_BOUND,
|
|
"circulating": messages.T_CIRCULATING,
|
|
"delivery": messages.S_NOT_DELIVERING,
|
|
}
|
|
|
|
|
|
def attention(checks):
|
|
"""Every row that is not plain green: red rows as alarms, amber rows as
|
|
warnings, and green-but-amber rows (a state still settling) as info.
|
|
Most urgent first; within a level, the page's own row order."""
|
|
rank = {ALARM: 0, WARNING: 1, INFO: 2}
|
|
out = []
|
|
for row in checks:
|
|
ok, warn = bool(row.get("ok")), bool(row.get("warn"))
|
|
if ok and not warn:
|
|
continue
|
|
sev = INFO if ok else (WARNING if warn else ALARM)
|
|
rid = row.get("id")
|
|
out.append({"severity": sev, "id": rid,
|
|
"title": ATTN_TITLES.get(rid, ATTN_TITLES[None]),
|
|
"body": row.get("text", ""),
|
|
"docs": ATTN_DOCS.get(rid)})
|
|
out.sort(key=lambda a: rank[a["severity"]])
|
|
return out
|
|
|
|
|
|
def month_peak(c, now):
|
|
"""This month's peak against the limit, and the 15-minute period so far."""
|
|
limit_w = c.peak.cap_w # the limit the controller parsed at start
|
|
local = brussels(now)
|
|
out = {"month": f"{local:%B}", "limit_w": limit_w, "peak_w": None,
|
|
"pct": None, "period_w": None, "state": "not_covered",
|
|
"line": ""}
|
|
if not c.p1_enabled:
|
|
out["line"] = line_peak_not_covered()
|
|
return out
|
|
reading = c.peak.peak(now)
|
|
if reading.is_unreadable:
|
|
out["state"] = "unreadable"
|
|
elif reading.is_observed:
|
|
out["state"] = "observed"
|
|
out["peak_w"] = float(reading.peak_w)
|
|
out["pct"] = round(100.0 * float(reading.peak_w) / limit_w, 1) \
|
|
if limit_w > 0 else None
|
|
else:
|
|
out["state"] = "none"
|
|
pg = c.peak_guard
|
|
ceiling = limit_w
|
|
if pg is not None:
|
|
out["period_w"] = pg.projected_w
|
|
ceiling = pg.ceiling_w or limit_w
|
|
out["line"] = line_period(out["period_w"], ceiling, c.shadow)
|
|
return out
|
|
|
|
|
|
def state_of(c, level, shadow):
|
|
"""One word for the pill and the banner's colour: running, shadow,
|
|
stopped, alarm or warning."""
|
|
if shadow:
|
|
return "shadow"
|
|
if level == "bad":
|
|
return "alarm"
|
|
if not c.auto:
|
|
return "stopped"
|
|
if level == "warn":
|
|
return "warning"
|
|
return "running"
|
|
|
|
|
|
def build(c, *, level: str, checks: list, now: datetime | None = None) -> dict:
|
|
"""The structured half of `/status`. Pure reads of the controller."""
|
|
now = now or datetime.now(timezone.utc)
|
|
shadow = bool(c.shadow)
|
|
kind = state_of(c, level, shadow)
|
|
# A red banner means every inverter is at 0 W (the status cascade), so
|
|
# nothing is planned until it clears. Shadow mode still plans: it is
|
|
# what the add-on would do.
|
|
held = level == "bad" and not shadow
|
|
running = (bool(c.auto) or shadow) and not held
|
|
settled = all(at_zero(inv) for inv in c.inverters)
|
|
ctl = next((r for r in checks if r.get("id") == "controller"), None)
|
|
return {
|
|
"state": {"kind": kind, "pill": pill(kind),
|
|
"ran": ctl.get("text", "") if ctl else "",
|
|
"sub": banner_sub(kind)},
|
|
"now": now_block(grid_w=c.grid, pv_w=c.pv_w, batt_w=c.site_batt_w,
|
|
has_solar=bool(c.pv_roster), shadow=shadow,
|
|
settled=settled),
|
|
"batteries": battery_rows(c, shadow, running, held),
|
|
"plan": build_plan(c, now, shadow, running, held=held,
|
|
settled=settled),
|
|
"attention": ([{"severity": INFO, "id": "shadow",
|
|
"title": ATTN_SHADOW, "body": attn_shadow_body(),
|
|
"docs": None}] if shadow else []) + attention(checks),
|
|
"month_peak": month_peak(c, now),
|
|
"control": {"running": bool(c.auto), "shadow": shadow,
|
|
"text": control_text(bool(c.auto), shadow),
|
|
"maintenance": c.maint.phase,
|
|
"docs_url": _docs_url()},
|
|
}
|
|
|
|
|
|
def _docs_url():
|
|
slug = messages.addon_slug()
|
|
return None if slug is None else messages.docs_url(slug,
|
|
messages._site["ha"])
|
|
|
|
|
|
# =========================================================================== #
|
|
# The words. STYLE.md: plain, short, present tense. A plan line is at most
|
|
# two sentences of at most 12 words each (UX-13 design, section 4); a title at
|
|
# most 5 words; everything else one sentence of at most 25 words. In shadow
|
|
# mode a text says what the add-on WOULD do. test_messages.py renders every
|
|
# function below through `text_samples()` and checks it.
|
|
# =========================================================================== #
|
|
MODE_HOUSE = "Follows the house"
|
|
ATTN_SHADOW = "Shadow mode is on"
|
|
|
|
|
|
def attn_shadow_body() -> str:
|
|
return "The add-on decides everything, but every inverter stays at 0 W."
|
|
|
|
ZERO_HELD, ZERO_GOING = "Held at 0 W", "Going to 0 W"
|
|
VERDICT = {
|
|
"held": ZERO_HELD,
|
|
"stopped": "Stopped",
|
|
"cycle": "Monthly cycle today",
|
|
"trade": "Trading on price",
|
|
"house": "Follows the house",
|
|
}
|
|
|
|
|
|
def _w(shadow, live, would):
|
|
return would if shadow else live
|
|
|
|
|
|
def pill(kind) -> str:
|
|
return {"running": "Running", "shadow": "Shadow mode",
|
|
"stopped": "Stopped", "alarm": "Needs you now",
|
|
"warning": "Running, with a warning"}[kind]
|
|
|
|
|
|
def banner_sub(kind) -> str:
|
|
return {
|
|
"running": "The batteries follow the plan below.",
|
|
"shadow": "The plans below say what the add-on would do.",
|
|
"stopped": "Start control below when you are ready.",
|
|
"alarm": "See Needs attention below for what to do.",
|
|
"warning": "See Needs attention below.",
|
|
}[kind]
|
|
|
|
|
|
def tile_solar(pv_w, has_solar) -> str:
|
|
if not has_solar:
|
|
return "Not set up"
|
|
if pv_w is None:
|
|
return "No reading"
|
|
return "Producing" if pv_w > IDLE_W else "Not producing"
|
|
|
|
|
|
def tile_house(house_w, has_solar, disagree=False) -> str:
|
|
if not has_solar:
|
|
return "Needs a solar reading"
|
|
if disagree:
|
|
return "Readings do not add up"
|
|
return "No reading" if house_w is None else "Using"
|
|
|
|
|
|
def tile_grid(grid_w) -> str:
|
|
if grid_w is None:
|
|
return "No reading"
|
|
if grid_w > IDLE_W:
|
|
return "Taking from the grid"
|
|
if grid_w < -IDLE_W:
|
|
return "Feeding in"
|
|
return "Balanced"
|
|
|
|
|
|
def _zero(settled) -> str:
|
|
return ZERO_HELD if settled else ZERO_GOING
|
|
|
|
|
|
def tile_batt(batt_w, shadow, settled=True) -> str:
|
|
if shadow:
|
|
return _zero(settled)
|
|
return {"giving": "Giving power", "charging": "Charging",
|
|
"resting": "Resting", None: "No reading"}[flow(batt_w)]
|
|
|
|
|
|
def battery_now(soc, batt_w, *, phase, shadow, running, held=False,
|
|
would_w=None, settled=True) -> str:
|
|
if soc is None or batt_w is None:
|
|
return "No reading from this battery."
|
|
zero = _zero(settled)
|
|
if shadow:
|
|
if would_w is None or abs(would_w) <= IDLE_W:
|
|
return f"{zero}. Would rest now."
|
|
verb = "give" if would_w > 0 else "charge"
|
|
return f"{zero}. Would {verb} {_kw(would_w)} now."
|
|
if held:
|
|
return f"{zero} until the alarm clears."
|
|
if not running:
|
|
return f"{zero} while control is stopped."
|
|
if phase != M_IDLE:
|
|
return {M_DRAIN: "Monthly cycle: emptying.",
|
|
M_CHARGE: "Monthly cycle: refilling.",
|
|
M_HOLD: "Monthly cycle: holding full."}.get(
|
|
phase, "Monthly cycle under way.")
|
|
st = flow(batt_w)
|
|
if st == "giving":
|
|
return f"Giving {_kw(batt_w)} now."
|
|
if st == "charging":
|
|
return f"Charging {_kw(batt_w)} now."
|
|
return "Resting now."
|
|
|
|
|
|
def plan_head(key, shadow, settled=True) -> str:
|
|
if key == "held":
|
|
return ("Every inverter is at 0 W until the alarm clears." if settled
|
|
else "Every inverter goes to 0 W until the alarm clears.")
|
|
if key == "stopped":
|
|
return "Control is stopped, so nothing is planned."
|
|
if not shadow:
|
|
return {
|
|
"cycle": "A battery runs its monthly cycle today.",
|
|
"trade": "Charges in the cheapest hour, gives power in the dearest.",
|
|
"house": "Charges from spare solar and covers the house.",
|
|
}[key]
|
|
return "Shadow mode: this is what the add-on would do. " + {
|
|
"cycle": "A battery would run its monthly cycle today.",
|
|
"trade": "Would charge when cheapest and give power when dearest.",
|
|
"house": "Would charge from spare solar and cover the house.",
|
|
}[key]
|
|
|
|
|
|
def run_grid_what(kwh, shadow) -> str:
|
|
return _w(shadow, "Charges", "Would charge") + \
|
|
f" up to {_kwh(kwh)} from the grid."
|
|
|
|
|
|
# The trade legs and the price line use the feed's own points (a 15-minute
|
|
# period on a quarter-hourly feed), so both name the same price and time.
|
|
def run_grid_why(price) -> str:
|
|
return f"Lowest published price, {_eur(price)} per kWh."
|
|
|
|
|
|
def run_grid_amount(eur) -> str:
|
|
return f"Costs up to {_eur(eur)}."
|
|
|
|
|
|
def run_give_what(kwh, shadow) -> str:
|
|
return _w(shadow, "Gives", "Would give") + \
|
|
f" up to {_kwh(kwh)}, to the house first."
|
|
|
|
|
|
def run_give_why(price) -> str:
|
|
return f"Highest published price, {_eur(price)} per kWh."
|
|
|
|
|
|
def run_give_amount(eur) -> str:
|
|
return f"Worth up to {_eur(eur)} at that price."
|
|
|
|
|
|
def run_cycle_what(phase, shadow) -> str:
|
|
if phase == M_IDLE:
|
|
return _w(shadow, "Starts", "Would start") + \
|
|
" the monthly battery cycle."
|
|
return _w(shadow, "Runs", "Would run") + " the monthly battery cycle now."
|
|
|
|
|
|
def run_cycle_why(shadow=False) -> str:
|
|
return _w(shadow, "Empties, refills and holds full",
|
|
"Would empty, refill and hold full") + \
|
|
", so the level reads right."
|
|
|
|
|
|
def run_cycle_amount(hours) -> str:
|
|
if hours is None:
|
|
return "How long depends on the battery."
|
|
return f"Takes about {max(1, round(hours))} hours."
|
|
|
|
|
|
def run_peak_what(limit_w, shadow) -> str:
|
|
return _w(shadow, "Gives", "Would give") + \
|
|
f" power to stay under {_kw(limit_w)}."
|
|
|
|
|
|
def run_peak_why() -> str:
|
|
return "The peak limit comes before everything else."
|
|
|
|
|
|
def run_house_what(shadow) -> str:
|
|
return _w(shadow, "Follows", "Would follow") + " the house."
|
|
|
|
|
|
def run_house_why(shadow=False) -> str:
|
|
return _w(shadow, "Charges from spare solar, gives power",
|
|
"Would charge from spare solar, give power") + \
|
|
" when the house needs it."
|
|
|
|
|
|
def run_house_amount() -> str:
|
|
return "Amounts depend on the sun and the house."
|
|
|
|
|
|
def line_no_trade(key) -> str:
|
|
return {
|
|
"off": "",
|
|
"not_covered": "No trading on price: nothing measures the 15-minute "
|
|
"peak.",
|
|
"no_prices": "No trading on price yet: no prices have arrived.",
|
|
"peak": "No trading on price now: the peak limit comes first.",
|
|
"not_firm": "No trading on price yet: the prices are not published.",
|
|
"not_worth": "No trading on price today. The price gap is too small.",
|
|
"no_level": "No trading on price now: a battery level is unknown.",
|
|
}[key]
|
|
|
|
|
|
def line_trade_tomorrow(kind, at, price, shadow) -> str:
|
|
verb = {"grid": _w(shadow, "Charges", "Would charge"),
|
|
"give": _w(shadow, "Gives power", "Would give power")}[kind]
|
|
return f"Tomorrow: {verb.lower()} at {at}, {_eur(price)} per kWh."
|
|
|
|
|
|
def line_cycle_waits(name) -> str:
|
|
return f"{name} waits its turn for the monthly cycle."
|
|
|
|
|
|
def line_peak_limit(limit_w, shadow) -> str:
|
|
return _w(shadow, "The batteries step in", "The batteries would step in") \
|
|
+ f" above {_kw(limit_w)}."
|
|
|
|
|
|
def line_prices(now_eur, lo, hi) -> str:
|
|
parts = []
|
|
if now_eur is not None:
|
|
parts.append(f"Price now {_eur(now_eur)} per kWh, all in.")
|
|
if lo is not None and hi is not None:
|
|
parts.append(f"Today: lowest {_eur(lo['eur_kwh'])} at "
|
|
f"{lo['label']}, highest {_eur(hi['eur_kwh'])} at "
|
|
f"{hi['label']}.")
|
|
return " ".join(parts) or "No prices for today."
|
|
|
|
|
|
def line_tomorrow(known) -> str:
|
|
return ("Tomorrow's prices are in." if known else
|
|
"Tomorrow's prices usually arrive after 13:00.")
|
|
|
|
|
|
def line_region(rid, year) -> str:
|
|
if rid in REGION_PRESETS and year:
|
|
return f"Levies and VAT: {REGION_WORDS.get(rid, rid)}, {year} figures."
|
|
return "Levies and VAT: your own figure."
|
|
|
|
|
|
REGION_WORDS = {"flanders": "Flanders, Fluvius Antwerpen"}
|
|
|
|
|
|
def line_peak_not_covered() -> str:
|
|
return "No peak limit: this meter does not measure 15-minute periods."
|
|
|
|
|
|
def line_period(period_w, limit_w, shadow) -> str:
|
|
step = _w(shadow, "The batteries step in", "The batteries would step in")
|
|
if period_w is None:
|
|
return f"{step} above {_kw(limit_w)}."
|
|
return (f"This 15-minute period is heading for {_kw(period_w)}. "
|
|
f"{step} above {_kw(limit_w)}.")
|
|
|
|
|
|
def control_text(running, shadow) -> str:
|
|
if shadow:
|
|
return "Shadow mode: every inverter stays at 0 W."
|
|
if running:
|
|
return "Control is running."
|
|
return "Control is stopped. Every inverter is at 0 W."
|
|
|
|
|
|
def text_samples() -> list:
|
|
"""`(kind, text)` for every sentence above, rendered with real values, so
|
|
test_messages.py can hold each to its STYLE limit. Kinds: title, plan
|
|
(2 sentences of 12 words), line (1 sentence of 25 words)."""
|
|
out = [("title", t) for t in VERDICT.values()] + [("title", ZERO_GOING)]
|
|
out += [("title", t) for t in ATTN_TITLES.values()]
|
|
out += [("title", MODE_HOUSE), ("title", ATTN_SHADOW),
|
|
("line", attn_shadow_body())]
|
|
for k in ("running", "shadow", "stopped", "alarm", "warning"):
|
|
out += [("title", pill(k)), ("line", banner_sub(k))]
|
|
for sh in (False, True):
|
|
out += [("title", tile_solar(v, s)) for v, s in
|
|
((None, False), (None, True), (3900, True), (0, True))]
|
|
out += [("title", tile_house(v, s)) for v, s in
|
|
((None, False), (None, True), (1100, True))]
|
|
out.append(("title", tile_house(None, True, True)))
|
|
out += [("title", tile_grid(v)) for v in (None, 1200, -800, 5)]
|
|
out += [("title", tile_batt(v, sh)) for v in (None, 2500, -2500, 0)]
|
|
out.append(("title", tile_batt(0, sh, settled=False)))
|
|
for phase in (M_IDLE, M_DRAIN, M_CHARGE, M_HOLD):
|
|
for w in (2500, -2500, 0):
|
|
out.append(("plan", battery_now(62, w, phase=phase,
|
|
shadow=sh, running=True,
|
|
would_w=-2400)))
|
|
out.append(("plan", battery_now(62, 0, phase=M_IDLE, shadow=sh,
|
|
running=False)))
|
|
for settled in (True, False):
|
|
out.append(("plan", battery_now(62, 0, phase=M_IDLE, shadow=False,
|
|
running=False, held=True,
|
|
settled=settled)))
|
|
out.append(("plan", battery_now(62, 0, phase=M_IDLE, shadow=sh,
|
|
running=False, would_w=-2400,
|
|
settled=settled)))
|
|
out.append(("plan", plan_head("held", sh, settled=False)))
|
|
out.append(("plan", battery_now(None, None, phase=M_IDLE, shadow=sh,
|
|
running=True)))
|
|
for k in VERDICT:
|
|
out.append(("plan", plan_head(k, sh)))
|
|
out += [("plan", run_grid_what(12.34, sh)),
|
|
("plan", run_give_what(9.6, sh)),
|
|
("plan", run_cycle_what(M_IDLE, sh)),
|
|
("plan", run_cycle_what(M_DRAIN, sh)),
|
|
("plan", run_peak_what(3500, sh)),
|
|
("plan", run_house_what(sh)),
|
|
("plan", line_trade_tomorrow("grid", "02:00", 0.1234, sh)),
|
|
("plan", line_trade_tomorrow("give", "19:00", 0.3612, sh)),
|
|
("plan", line_peak_limit(3500, sh)),
|
|
("plan", line_period(1240, 3500, sh)),
|
|
("plan", line_period(None, 3500, sh)),
|
|
("plan", control_text(True, sh)),
|
|
("plan", control_text(False, sh))]
|
|
out += [("plan", run_grid_why(0.1412)), ("plan", run_grid_amount(1.23)),
|
|
("plan", run_give_why(0.3612)), ("plan", run_give_amount(3.47)),
|
|
("plan", run_cycle_why()), ("plan", run_cycle_why(True)),
|
|
("plan", run_cycle_amount(None)),
|
|
("plan", run_cycle_amount(5.4)), ("plan", run_peak_why()),
|
|
("plan", run_house_why()), ("plan", run_house_why(True)),
|
|
("plan", run_house_amount()),
|
|
("plan", line_cycle_waits("Battery B")),
|
|
("plan", line_prices(0.16, {"eur_kwh": 0.14, "label": "13:00"},
|
|
{"eur_kwh": 0.36, "label": "19:00"})),
|
|
("plan", line_prices(None, None, None)),
|
|
("plan", line_tomorrow(True)), ("plan", line_tomorrow(False)),
|
|
("plan", line_region("flanders", 2026)),
|
|
("plan", line_region("own", None)),
|
|
("plan", line_peak_not_covered())]
|
|
out += [("plan", line_no_trade(k)) for k in
|
|
("not_covered", "no_prices", "peak", "not_firm", "not_worth",
|
|
"no_level")]
|
|
return out
|