Claude dev agent
d69fd8fbfd
A battery's mode is a soft unit limit (source `mode`, P_MODE 35) kept
apart from the hard limits in pre_resolve(). The allocator applies it
only while the site target is decided at or below 35 (new
Resolution.deciding_priority), so the peak floor releases it and a mode
`hi` never meets the site `lo`. At N=1 resolve() applies the same rule
to the site target after stage 2. A mode range without 0 W is dropped
as a software fault, so REFUSE_SIGN stays unreachable from a mode.
compute() gets the loop bound (sum over live units of hard limits with
modes applied), so the loop never asks for watts no battery may give
and the saturation freeze never parks on them. While charging, Leftover
solar weighs 0 so Solar first fills first. Store keys pack_use and
pack_mode per battery, set by two new /action values; Off stops and
skips maintenance with a log line. Minimal Batteries cards (EN/NL).
Off by default (battery_modes_enabled); off, commands are byte-identical
to a5dddad at N=1 and N=2 (test_modes.py part 1).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1046 lines
39 KiB
Python
1046 lines
39 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". OPT-11 P1 adds each battery's
|
|
switch and mode to its row; the mode plan arrives with P3/P4, so nothing here
|
|
invents one. 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, uitext
|
|
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}"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 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_words(verdict_key, settled),
|
|
"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 accepted a 0 W power request? 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). Before the first
|
|
accepted write nothing is known to be held; the heartbeat writes every
|
|
`heartbeat_s`, stopped or not, so this does not stick."""
|
|
return inv.last_write is not None and abs(inv.last_write) < 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 the plan (OPT-11 P3/P4).
|
|
"why": None,
|
|
})
|
|
if getattr(c, "modes_on", False):
|
|
# OPT-11 P1: the battery's switch and mode, for the Batteries
|
|
# screen. Only with battery modes on, so the page is unchanged
|
|
# without them. UX-13 P3 builds the real screen on these keys.
|
|
pm = c.pack_mode(inv)
|
|
rows[-1].update({"use": pm.use, "mode_key": pm.key,
|
|
"mode_choice": pm.mode,
|
|
"mode": mode_word(pm.key)})
|
|
return rows
|
|
|
|
|
|
# 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_title(rid),
|
|
"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": month_name(local), "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,
|
|
lang: str = "en") -> dict:
|
|
"""The structured half of `/status`, in `lang` (UX-13 P5). Pure reads of
|
|
the controller."""
|
|
_LANG[0] = uitext.lang_of(lang)
|
|
try:
|
|
return _build(c, level=level, checks=checks, now=now)
|
|
finally:
|
|
_LANG[0] = "en"
|
|
|
|
|
|
def _build(c, *, level, checks, now):
|
|
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. The sentences themselves are in
|
|
# app/ui_text/<lang>.json under `home_words` (UX-13 P5: English and Dutch).
|
|
# test_messages.py and test_ui_text.py render every function below through
|
|
# `text_samples()` in each language and check it.
|
|
# =========================================================================== #
|
|
_LANG = ["en"] # the language of the build in progress (build())
|
|
_WORDS: dict = {}
|
|
|
|
|
|
def _hw() -> dict:
|
|
lang = _LANG[0]
|
|
if lang not in _WORDS:
|
|
_WORDS[lang] = uitext.words(lang)
|
|
return _WORDS[lang]
|
|
|
|
|
|
def _t(path, **values) -> str:
|
|
node = _hw()["home_words"]
|
|
for k in path.split("."):
|
|
node = node[k]
|
|
return uitext.say(node, **values)
|
|
|
|
|
|
def _dec(text) -> str:
|
|
"""A decimal comma in Dutch."""
|
|
return text.replace(".", ",") if _LANG[0] == "nl" else text
|
|
|
|
|
|
def _kw(w) -> str:
|
|
return _dec(f"{abs(float(w)) / 1000:.1f}") + " kW"
|
|
|
|
|
|
def _eur(v) -> str:
|
|
return "€" + _dec(f"{float(v):.2f}")
|
|
|
|
|
|
def _kwh(v) -> str:
|
|
return _dec(f"{float(v):.1f}") + " kWh"
|
|
|
|
|
|
# Row id -> the title on Home. The row's own sentence is the body, and it stays
|
|
# in English (app/messages.py). Five titles are the notification's own title,
|
|
# which test_ui_text.py checks.
|
|
_EN = uitext.words("en")["home_words"]
|
|
ATTN_TITLES = {(None if k == "none" else k): v for k, v in _EN["attn"].items()}
|
|
ATTN_FROM_MESSAGES = {
|
|
"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,
|
|
}
|
|
VERDICT = dict(_EN["verdict"])
|
|
MODE_HOUSE, ATTN_SHADOW = _EN["mode_house"], _EN["attn_shadow"]
|
|
ZERO_HELD, ZERO_GOING = _EN["zero_held"], _EN["zero_going"]
|
|
|
|
|
|
def mode_house() -> str:
|
|
return _t("mode_house")
|
|
|
|
|
|
def mode_word(key) -> str:
|
|
"""OPT-11: a battery's mode, or Off, in the page's language."""
|
|
return _t("modes." + key)
|
|
|
|
|
|
def attn_shadow() -> str:
|
|
return _t("attn_shadow")
|
|
|
|
|
|
def attn_shadow_body() -> str:
|
|
return _t("attn_shadow_body")
|
|
|
|
|
|
def verdict_words(key, settled=True) -> str:
|
|
return _t("verdict." + key) if settled or key != "held" \
|
|
else _t("zero_going")
|
|
|
|
|
|
def attn_title(rid) -> str:
|
|
table = _hw()["home_words"]["attn"]
|
|
return table.get(rid or "none", table["none"])
|
|
|
|
|
|
def month_name(local) -> str:
|
|
return _hw()["home_words"]["months"][local.month - 1]
|
|
|
|
|
|
def _w(shadow, key, **values) -> str:
|
|
"""`key`, or its `key_would` form in shadow mode."""
|
|
return _t(key + "_would" if shadow else key, **values)
|
|
|
|
|
|
def pill(kind) -> str:
|
|
return _t("pill." + kind)
|
|
|
|
|
|
def banner_sub(kind) -> str:
|
|
return _t("banner_sub." + kind)
|
|
|
|
|
|
def tile_solar(pv_w, has_solar) -> str:
|
|
if not has_solar:
|
|
return _t("tile.not_set_up")
|
|
if pv_w is None:
|
|
return _t("tile.no_reading")
|
|
return _t("tile.producing") if pv_w > IDLE_W else _t("tile.not_producing")
|
|
|
|
|
|
def tile_house(house_w, has_solar, disagree=False) -> str:
|
|
if not has_solar:
|
|
return _t("tile.needs_solar")
|
|
if disagree:
|
|
return _t("tile.not_adding_up")
|
|
return _t("tile.no_reading") if house_w is None else _t("tile.using")
|
|
|
|
|
|
def tile_grid(grid_w) -> str:
|
|
if grid_w is None:
|
|
return _t("tile.no_reading")
|
|
if grid_w > IDLE_W:
|
|
return _t("tile.taking")
|
|
if grid_w < -IDLE_W:
|
|
return _t("tile.feeding")
|
|
return _t("tile.balanced")
|
|
|
|
|
|
def _zero(settled) -> str:
|
|
return _t("zero_held") if settled else _t("zero_going")
|
|
|
|
|
|
def tile_batt(batt_w, shadow, settled=True) -> str:
|
|
if shadow:
|
|
return _zero(settled)
|
|
return _t("tile." + {"giving": "giving", "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 _t("now.no_reading")
|
|
zero = _zero(settled)
|
|
if shadow:
|
|
if would_w is None or abs(would_w) <= IDLE_W:
|
|
return _t("now.would_rest", zero=zero)
|
|
verb = "would_give" if would_w > 0 else "would_charge"
|
|
return _t("now." + verb, zero=zero, kw=_kw(would_w))
|
|
if held:
|
|
return _t("now.held", zero=zero)
|
|
if not running:
|
|
return _t("now.stopped", zero=zero)
|
|
if phase != M_IDLE:
|
|
return _t("now." + {M_DRAIN: "cycle_drain", M_CHARGE: "cycle_charge",
|
|
M_HOLD: "cycle_hold"}.get(phase, "cycle"))
|
|
st = flow(batt_w)
|
|
if st == "giving":
|
|
return _t("now.giving", kw=_kw(batt_w))
|
|
if st == "charging":
|
|
return _t("now.charging", kw=_kw(batt_w))
|
|
return _t("now.resting")
|
|
|
|
|
|
def plan_head(key, shadow, settled=True) -> str:
|
|
if key == "held":
|
|
return _t("head.held" if settled else "head.held_going")
|
|
if key == "stopped":
|
|
return _t("head.stopped")
|
|
if not shadow:
|
|
return _t("head." + key)
|
|
return _t("head.shadow") + " " + _t("head." + key + "_would")
|
|
|
|
|
|
def run_grid_what(kwh, shadow) -> str:
|
|
return _w(shadow, "run.grid_what", kwh=_kwh(kwh))
|
|
|
|
|
|
# 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 _t("run.grid_why", eur=_eur(price))
|
|
|
|
|
|
def run_grid_amount(eur) -> str:
|
|
return _t("run.grid_amount", eur=_eur(eur))
|
|
|
|
|
|
def run_give_what(kwh, shadow) -> str:
|
|
return _w(shadow, "run.give_what", kwh=_kwh(kwh))
|
|
|
|
|
|
def run_give_why(price) -> str:
|
|
return _t("run.give_why", eur=_eur(price))
|
|
|
|
|
|
def run_give_amount(eur) -> str:
|
|
return _t("run.give_amount", eur=_eur(eur))
|
|
|
|
|
|
def run_cycle_what(phase, shadow) -> str:
|
|
return _w(shadow, "run.cycle_start" if phase == M_IDLE else "run.cycle_now")
|
|
|
|
|
|
def run_cycle_why(shadow=False) -> str:
|
|
return _w(shadow, "run.cycle_why")
|
|
|
|
|
|
def run_cycle_amount(hours) -> str:
|
|
if hours is None:
|
|
return _t("run.cycle_amount_unknown")
|
|
return _t("run.cycle_amount", hours=max(1, round(hours)))
|
|
|
|
|
|
def run_peak_what(limit_w, shadow) -> str:
|
|
return _w(shadow, "run.peak_what", kw=_kw(limit_w))
|
|
|
|
|
|
def run_peak_why() -> str:
|
|
return _t("run.peak_why")
|
|
|
|
|
|
def run_house_what(shadow) -> str:
|
|
return _w(shadow, "run.house_what")
|
|
|
|
|
|
def run_house_why(shadow=False) -> str:
|
|
return _w(shadow, "run.house_why")
|
|
|
|
|
|
def run_house_amount() -> str:
|
|
return _t("run.house_amount")
|
|
|
|
|
|
def line_no_trade(key) -> str:
|
|
return "" if key == "off" else _t("no_trade." + key)
|
|
|
|
|
|
def line_trade_tomorrow(kind, at, price, shadow) -> str:
|
|
return _w(shadow, "tomorrow_trade." + kind, at=at, eur=_eur(price))
|
|
|
|
|
|
def line_cycle_waits(name) -> str:
|
|
return _t("cycle_waits", name=name)
|
|
|
|
|
|
def line_peak_limit(limit_w, shadow) -> str:
|
|
return _w(shadow, "step_in", kw=_kw(limit_w))
|
|
|
|
|
|
def line_prices(now_eur, lo, hi) -> str:
|
|
parts = []
|
|
if now_eur is not None:
|
|
parts.append(_t("prices.now", eur=_eur(now_eur)))
|
|
if lo is not None and hi is not None:
|
|
parts.append(_t("prices.today", lo=_eur(lo["eur_kwh"]),
|
|
lo_at=lo["label"], hi=_eur(hi["eur_kwh"]),
|
|
hi_at=hi["label"]))
|
|
return " ".join(parts) or _t("prices.none")
|
|
|
|
|
|
def line_tomorrow(known) -> str:
|
|
return _t("tomorrow.known" if known else "tomorrow.unknown")
|
|
|
|
|
|
def line_region(rid, year) -> str:
|
|
if rid in REGION_PRESETS and year:
|
|
name = _hw()["choices"]["region"].get(rid, rid)
|
|
return _t("region.preset", region=name, year=year)
|
|
return _t("region.own")
|
|
|
|
|
|
def line_peak_not_covered() -> str:
|
|
return _t("peak_not_covered")
|
|
|
|
|
|
def line_period(period_w, limit_w, shadow) -> str:
|
|
step = _w(shadow, "step_in", kw=_kw(limit_w))
|
|
if period_w is None:
|
|
return step
|
|
return _t("period", kw=_kw(period_w)) + " " + step
|
|
|
|
|
|
def control_text(running, shadow) -> str:
|
|
if shadow:
|
|
return _t("control.shadow")
|
|
if running:
|
|
return _t("control.running")
|
|
return _t("control.stopped")
|
|
|
|
|
|
def text_samples(lang="en") -> list:
|
|
"""`(kind, text)` for every sentence above in `lang`, 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)."""
|
|
_LANG[0] = uitext.lang_of(lang)
|
|
try:
|
|
return _samples()
|
|
finally:
|
|
_LANG[0] = "en"
|
|
|
|
|
|
def _samples() -> list:
|
|
words = _hw()["home_words"]
|
|
out = [("title", t) for t in words["verdict"].values()]
|
|
out += [("title", _t("zero_going")), ("title", _t("zero_held"))]
|
|
out += [("title", t) for t in words["attn"].values()]
|
|
out += [("title", mode_word(k)) for k in words["modes"]]
|
|
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
|