Claude
d0db5b8c39
At N>=2 a battery at or under its own peak reserve is held: it gives nothing for the peak (weight 0 while the site gives) and the guard's need is placed on the live, non-refusing batteries above theirs. The guard is floor-limited, and alarms, only when those cannot hold the limit at full power for ALARM_HOLD_S. The peak alarms no longer read "Not ready" or hold the Home page at 0 W, because control keeps running. N=1 is byte-identical. DECISIONS/SAFETY-87-guard-uses-able-batteries.md. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019DBUfvndgHwaV339Fvd2gr
1403 lines
55 KiB
Python
1403 lines
55 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; P3 adds each battery's price runs and its role this
|
|
cycle (`Controller.mode_runs`, `mode_tick`), read as the control cycle left
|
|
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.
|
|
"""
|
|
|
|
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 .packmode import CHEAP_HOURS, GRID_CHARGE_MODES, LEFTOVER_SOLAR, SOLAR_FIRST
|
|
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):
|
|
if _off(c, inv):
|
|
continue # OPT-11: an Off battery does not trade
|
|
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
|
|
|
|
|
|
PRICE_MODES = (LEFTOVER_SOLAR, CHEAP_HOURS)
|
|
|
|
|
|
def mode_runs(c, day0, day1, now, shadow):
|
|
"""OPT-11 P3: each battery's price runs today, as the control cycle keeps
|
|
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)`."""
|
|
rp = getattr(c, "mode_runs", None)
|
|
if getattr(c, "mode_tick", None) is None:
|
|
return [], None # no cycle yet: nothing is known
|
|
if rp is None:
|
|
return [], "no_prices"
|
|
if not rp.on:
|
|
return [], "not_worth"
|
|
floor, ceiling = arbitrage.soc_bounds(c.o)
|
|
out = []
|
|
for i, inv in enumerate(c.inverters):
|
|
pm = c.pack_mode(inv)
|
|
if not pm.use or pm.mode not in PRICE_MODES:
|
|
continue
|
|
soc, cap = inv.soc, inv.capacity_kwh
|
|
|
|
def room(kind, hrs):
|
|
kwh = inv.max_w / 1000.0 * hrs
|
|
if cap and soc is not None:
|
|
kwh = min(kwh, max(0.0, (ceiling - soc) if kind == "grid"
|
|
else (soc - floor)) / 100.0 * cap)
|
|
return kwh
|
|
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):
|
|
blocks.append(("grid", rp.cheap))
|
|
sell = rp.sell if pm.sell and pm.can_sell else None
|
|
for g in rp.give:
|
|
if sell is not None and g.start <= sell.start < g.end:
|
|
blocks.append(("sell", sell))
|
|
else:
|
|
blocks.append(("give", g))
|
|
for kind, run in blocks:
|
|
if run.end <= now or not (run.start < day1 and run.end > day0):
|
|
continue
|
|
hrs = (run.end - run.start).total_seconds() / 3600.0
|
|
kwh = room(kind, hrs)
|
|
if kind == "grid" and need is not None:
|
|
kwh = min(kwh, need)
|
|
if kwh < MIN_KWH:
|
|
continue
|
|
price = run.mean_eur_kwh
|
|
if kind == "grid":
|
|
what, why, amount = (run_grid_what(kwh, shadow),
|
|
run_cheap_why(price),
|
|
run_grid_amount(kwh * price))
|
|
elif kind == "sell":
|
|
# No money line: the price is what the house pays, and a sale
|
|
# earns the lower injection price (OPT-11 P3 review).
|
|
what, why, amount = (run_sell_what(kwh, shadow),
|
|
run_dear_why(price), None)
|
|
else:
|
|
what, why, amount = (run_give_what(kwh, shadow),
|
|
run_dear_why(price),
|
|
run_give_amount(kwh * price))
|
|
out.append(_run(i, "give" if kind == "sell" else kind,
|
|
max(run.start, day0), min(run.end, day1), day0,
|
|
day1, what, why, amount, kwh=kwh,
|
|
eur=None if kind == "sell" else kwh * price,
|
|
price=price))
|
|
return out, None
|
|
|
|
|
|
def plan_need(c, name):
|
|
"""P4: kWh this battery buys from the grid by the day plan, or None when
|
|
there is no plan for it (then P3 option A applies)."""
|
|
targets = c.plan_targets() if hasattr(c, "plan_targets") else None
|
|
if not targets or name not in targets:
|
|
return None
|
|
inv = next((i for i in c.inverters if i.name == name), None)
|
|
if inv is None or inv.soc is None or not inv.capacity_kwh:
|
|
return None
|
|
return max(0.0, targets[name] - float(inv.soc)) / 100.0 * float(
|
|
inv.capacity_kwh)
|
|
|
|
|
|
def plan_state(c):
|
|
"""P4: `"plan"` (a day plan is made), `"learning"` (no house profile
|
|
yet), `"no_forecast"` (set but missing or stale), or None (not set, or
|
|
battery modes off)."""
|
|
if not getattr(c, "modes_on", False) or \
|
|
getattr(c, "fc_entity", None) is None:
|
|
return None
|
|
if getattr(c, "solar_fc", None) is None:
|
|
return "no_forecast"
|
|
hp = getattr(c, "house_profile", None)
|
|
if hp is not None and not hp.ready():
|
|
return "learning"
|
|
return "plan" if getattr(c, "day_plan", None) is not None else None
|
|
|
|
|
|
def solar_block(c, now):
|
|
"""P4: the solar forecast line for the Batteries screen and the Simple
|
|
settings view: `{state, fc_kwh, count_kwh, updated, line}`."""
|
|
st = plan_state(c)
|
|
out = {"state": st, "fc_kwh": None, "count_kwh": None, "updated": None,
|
|
"line": ""}
|
|
if st is None and getattr(c, "solar_fc", None) is None:
|
|
return out
|
|
fc = getattr(c, "solar_fc", None)
|
|
if fc is not None:
|
|
trust = float(getattr(c, "solar_trust_pct", 70)) / 100.0
|
|
out.update(fc_kwh=round(fc.today.kwh, 1),
|
|
count_kwh=round(fc.today.kwh * trust, 1),
|
|
updated=hhmm(fc.at))
|
|
if st in ("no_forecast", "learning"):
|
|
out["line"] = _t("solar." + st)
|
|
elif fc is not None:
|
|
out["line"] = line_solar(fc.today.kwh, out["count_kwh"],
|
|
out["updated"])
|
|
return out
|
|
|
|
|
|
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 or _off(c, inv):
|
|
continue # OPT-11: no monthly cycle while it is off
|
|
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. An Off battery gets it
|
|
only while it gives power (OPT-11: Off covers only the shortfall)."""
|
|
pg = c.peak_guard
|
|
if not peak_engaged(c):
|
|
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, inv in enumerate(c.inverters)
|
|
if not _off(c, inv) or flow(inv.batt) == "giving"]
|
|
|
|
|
|
def house_runs(n, day0, day1, now, shadow, skip=()):
|
|
""""Follows the house" for the rest of the day, drawn under the others.
|
|
`skip`: the batteries that are switched off (OPT-11)."""
|
|
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) if i not in skip]
|
|
|
|
|
|
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)
|
|
modes_on = bool(getattr(c, "modes_on", False))
|
|
if running:
|
|
off = {i for i, inv in enumerate(c.inverters) if _off(c, inv)}
|
|
runs += house_runs(len(c.inverters), day0, day1, now, shadow, off)
|
|
if modes_on:
|
|
# OPT-11 P3: with battery modes on, each battery's price runs.
|
|
mine, no_trade = mode_runs(c, day0, day1, now, shadow)
|
|
runs += mine
|
|
if no_trade:
|
|
lines.append(line_no_trade(no_trade))
|
|
elif 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 _off(c, inv) -> bool:
|
|
"""OPT-11: is this battery switched off? Only with battery modes on."""
|
|
return bool(getattr(c, "modes_on", False)) and not c.pack_mode(inv).use
|
|
|
|
|
|
def refusing(c, inv) -> int:
|
|
"""SAFETY-52/73: the direction this battery refuses (+1 give, -1 charge,
|
|
0 none). Per inverter at two or more, the site's own at one."""
|
|
if len(c.inverters) > 1:
|
|
return int(getattr(inv, "refused", 0) or 0)
|
|
return int(getattr(c, "batt_refused", 0) or 0)
|
|
|
|
|
|
def peak_engaged(c) -> bool:
|
|
"""The peak guard asks the batteries for power this cycle. Not when it is
|
|
blind or floor-limited: those are engaged with a 0 W floor (no charging
|
|
only), so no battery gives power for it."""
|
|
pg = getattr(c, "peak_guard", None)
|
|
return bool(pg is not None and pg.engaged and (pg.floor_w or 0.0) > 0.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)
|
|
modes_on = bool(getattr(c, "modes_on", False))
|
|
pms = {inv.name: c.pack_mode(inv) for inv in c.inverters} \
|
|
if modes_on else {}
|
|
peak = peak_engaged(c)
|
|
# OPT-11 P3: each battery's role this cycle and the price runs, as the
|
|
# control cycle left them.
|
|
tick = getattr(c, "mode_tick", None) if modes_on else None
|
|
rp = getattr(c, "mode_runs", None)
|
|
now = datetime.now(timezone.utc)
|
|
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
|
|
pm = pms.get(inv.name)
|
|
off = pm is not None and not pm.use
|
|
# UX-13 P3: does a Solar first battery that is on come before this one?
|
|
sf_ahead = any(o.use and o.mode == SOLAR_FIRST
|
|
for n, o in pms.items() if n != inv.name)
|
|
# ...and does another battery refuse to give power, so this one
|
|
# carries its share (SAFETY-73)?
|
|
other = next((o.name for o in c.inverters if o is not inv
|
|
and refusing(c, o) > 0 and not _off(c, o)), None)
|
|
# SAFETY-78: another battery drains for its monthly cycle, so this
|
|
# one never charges now. Not while that drain waits: it files no
|
|
# floor then, so this battery may charge from solar.
|
|
waits = getattr(c, "_drain_waits", ())
|
|
cycling = next((o.name for j, o in enumerate(c.inverters)
|
|
if o is not inv and j < len(c.maints)
|
|
and c.maints[j].phase == M_DRAIN
|
|
and o.name not in waits), None)
|
|
role = None if tick is None else tick.roles.get(inv.name)
|
|
# Only once a cycle has run with modes on (`tick`): before that
|
|
# nothing is known, so nothing is claimed about the prices.
|
|
plan = (None if tick is None or pm.mode not in PRICE_MODES
|
|
else "no_prices" if rp is None
|
|
else "no_spread" if not rp.on else None)
|
|
nxt = rp.next_give(now) if rp is not None and rp.on else None
|
|
at = "" if nxt is None else (
|
|
f"{hhmm(nxt.start)} - {hhmm(nxt.end)}")
|
|
why = battery_why(
|
|
modes_on=modes_on, mode=None if pm is None else pm.mode, off=off,
|
|
reading=inv.soc is not None and inv.batt is not None,
|
|
phase=phase, shadow=shadow, running=running, held=held,
|
|
refused=refusing(c, inv), peak=peak,
|
|
giving=flow(inv.batt) == "giving", other_refuses=other,
|
|
sf_ahead=sf_ahead, role=role, plan=plan, has_at=bool(at),
|
|
other_cycle=cycling,
|
|
day=None if pm is None or pm.mode not in PRICE_MODES
|
|
else day_key(c, inv.name))
|
|
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), off=off, peak=peak),
|
|
# UX-13 P3: why it does what "now" says, from what the add-on
|
|
# knows this cycle. Price and forecast reasons are OPT-11 P3/P4.
|
|
"why_key": why,
|
|
"why": why_words(why, name=cycling if why == "other_cycle"
|
|
else other, at=at),
|
|
})
|
|
if modes_on:
|
|
# 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.
|
|
rows[-1].update({"use": pm.use, "mode_key": pm.key,
|
|
"mode_choice": pm.mode,
|
|
"mode": mode_word(pm.key),
|
|
# P3: selling, for Leftover solar and Cheap
|
|
# hours only (Solar first never sells).
|
|
"sell": pm.sell, "can_sell": pm.can_sell,
|
|
"role": role})
|
|
return rows
|
|
|
|
|
|
def day_key(c, name):
|
|
"""P4: the day plan's word for one Leftover solar or Cheap hours
|
|
battery: `sun_fills`, `tops_up`, `no_forecast`, or None."""
|
|
st = plan_state(c)
|
|
if st == "no_forecast":
|
|
mode = c.pack_mode(next(i for i in c.inverters if i.name == name))
|
|
return "no_forecast" if mode.mode == LEFTOVER_SOLAR else None
|
|
if st != "plan":
|
|
return None
|
|
need = plan_need(c, name)
|
|
if need is None:
|
|
return None
|
|
return "tops_up" if need >= MIN_KWH else "sun_fills"
|
|
|
|
|
|
def battery_why(*, modes_on, mode, off, reading, phase, shadow, running,
|
|
held, refused, peak, sf_ahead, giving=False,
|
|
other_refuses=None, role=None, plan=None,
|
|
has_at=False, other_cycle=None, day=None) -> str:
|
|
"""The key of one battery's "why" line (`home_words.why`), most urgent
|
|
first. Pure, so every branch is tested without a controller."""
|
|
if not reading:
|
|
return "no_reading"
|
|
if held:
|
|
return "held"
|
|
if not running:
|
|
return "stopped"
|
|
if shadow:
|
|
return "shadow"
|
|
if phase != M_IDLE:
|
|
return "cycle_mode" if modes_on else "cycle"
|
|
if off:
|
|
# Off gives way to the peak limit only (OPT-11 design section 2).
|
|
return "off_peak" if peak and giving else "off"
|
|
if refused:
|
|
return "refuses_give" if refused > 0 else "refuses_charge"
|
|
if peak:
|
|
return "peak"
|
|
if other_cycle:
|
|
return "other_cycle"
|
|
if giving and other_refuses:
|
|
return "other_refuses"
|
|
if not modes_on:
|
|
return "modes_off"
|
|
# OPT-11 P3: the battery's role in the price runs this cycle.
|
|
if role == "hold" and day is not None:
|
|
return day # P4: what the day plan decided
|
|
if role in ROLE_WHY:
|
|
if role == "hold" and not has_at:
|
|
return "saves_later"
|
|
return ROLE_WHY[role]
|
|
if plan in ("no_prices", "no_spread"):
|
|
return plan
|
|
if giving:
|
|
return "covers_house"
|
|
if mode in PRICE_MODES:
|
|
return "leftover" if sf_ahead else "leftover_alone"
|
|
return "solar_first"
|
|
|
|
|
|
# OPT-11 P3: a role (modetrade) -> its Why key.
|
|
ROLE_WHY = {"charge": "grid_charge", "rest": "rests_charge",
|
|
"sell": "selling", "give": "expensive", "hold": "saves"}
|
|
|
|
|
|
def attach_today(rows, runs, *, modes_on, running=True, held=False,
|
|
shadow=False) -> None:
|
|
"""UX-13 P3: each battery's runs today from the plan (the same blocks as
|
|
Home's chart), or one line saying there are none. With battery modes on,
|
|
a Leftover solar or Cheap hours battery with no price run left today
|
|
says so; the others follow the house."""
|
|
for i, row in enumerate(rows):
|
|
mine = [r for r in runs if r["battery"] == i and r["kind"] != "house"]
|
|
row["today"] = [{"from": r["from"], "to": r["to"], "kind": r["kind"],
|
|
"what": r["what"]} for r in mine]
|
|
if mine:
|
|
row["today_line"] = ""
|
|
elif held:
|
|
row["today_line"] = today_words("held")
|
|
elif not running:
|
|
row["today_line"] = today_words("stopped")
|
|
elif row.get("use") is False:
|
|
row["today_line"] = today_words("off")
|
|
elif modes_on and row.get("mode_choice") in PRICE_MODES:
|
|
# OPT-11 P3: a price mode with no run left today, or no plan
|
|
# made yet (no cycle has run with modes on).
|
|
row["today_line"] = today_words("no_runs" if row.get("role")
|
|
else "not_planned")
|
|
else:
|
|
row["today_line"] = today_words("house_would" if shadow
|
|
else "house")
|
|
|
|
|
|
# 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", held: bool | None = None) -> 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, held=held)
|
|
finally:
|
|
_LANG[0] = "en"
|
|
|
|
|
|
def _build(c, *, level, checks, now, held=None):
|
|
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` False: an alarm while control runs
|
|
# (SAFETY-87, the peak alarms).
|
|
held = level == "bad" and not shadow and held is not False
|
|
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)
|
|
rows = battery_rows(c, shadow, running, held)
|
|
plan = build_plan(c, now, shadow, running, held=held, settled=settled)
|
|
attach_today(rows, plan["runs"],
|
|
modes_on=bool(getattr(c, "modes_on", False)),
|
|
running=running, held=held, shadow=shadow)
|
|
return {
|
|
"solar": solar_block(c, now),
|
|
"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": rows,
|
|
"plan": plan,
|
|
"modes_on": bool(getattr(c, "modes_on", False)),
|
|
"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, off=False, peak=False) -> 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 off:
|
|
# OPT-11: an Off battery is asked for 0 W, but the peak limit may
|
|
# still need it. Say which, and never "0 W" beside a power figure.
|
|
if st == "resting":
|
|
return _t("now.off_resting")
|
|
if st == "giving" and peak:
|
|
return _t("now.off_peak", kw=_kw(batt_w))
|
|
return _t("now.off_going", kw=_kw(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 why_words(key, name=None, at="") -> str:
|
|
"""One battery's "why" line (UX-13 P3), from `battery_why()`'s key.
|
|
`name`: the battery that refuses, for `other_refuses`; `at`: the
|
|
give run's time, for `saves`."""
|
|
return _t("why." + key, name=name or "", at=at)
|
|
|
|
|
|
def today_words(key) -> str:
|
|
"""One battery's "today" line when the plan has no run for it."""
|
|
return _t("today." + key)
|
|
|
|
|
|
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_sell_what(kwh, shadow) -> str:
|
|
return _w(shadow, "run.sell_what", kwh=_kwh(kwh))
|
|
|
|
|
|
def run_cheap_why(price) -> str:
|
|
return _t("run.cheap_why", eur=_eur(price))
|
|
|
|
|
|
def run_dear_why(price) -> str:
|
|
return _t("run.dear_why", eur=_eur(price))
|
|
|
|
|
|
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_solar(fc_kwh, count_kwh, at) -> str:
|
|
return _t("solar.line", kwh=_kwh(fc_kwh), count=_kwh(count_kwh), at=at)
|
|
|
|
|
|
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())]
|
|
for off in (False, True):
|
|
for peak in (False, True):
|
|
for w in (2500, -2500, 0, 500):
|
|
out.append(("plan", battery_now(62, w, phase=M_IDLE,
|
|
shadow=False, running=True,
|
|
off=off, peak=peak)))
|
|
out += [("plan", why_words(k, name="Battery A")) for k in words["why"]]
|
|
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", line_no_trade(k)) for k in
|
|
("not_covered", "no_prices", "peak", "not_firm", "not_worth",
|
|
"no_level")]
|
|
return out
|