"""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. """ import contextlib from datetime import datetime, timedelta, timezone from . import arbitrage, cheaproll, handback, messages, peakguard, priceruns, uitext from .dayplan import EFF from .maintenance import CHARGE as M_CHARGE, DRAIN as M_DRAIN, HOLD as M_HOLD from .maintenance import IDLE as M_IDLE, next_slot_start from .packmode import CHEAP_HOURS, GRID_CHARGE_MODES, LEFTOVER_SOLAR, SOLAR_FIRST from .price import Absence, brussels, _cover_table 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 # 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 = {} self.kinds = {} def see(self, points, day0, day1): if self.day != day0: self.day, self.points, self.kinds = 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 _bar_step(pts): """The series' own spacing: its smallest gap, never over an hour.""" gaps = [y.start - x.start for x, y in zip(pts, pts[1:]) if y.start > x.start] return min(gaps + [timedelta(hours=1)]) def _bar_cover(pts): """`[(start, end, point)]`: what each point is the price of on the chart. The add-on's own rule, not a copy: `price._cover_table()` (TARIFF-14), so a point covers until the next one starts and never over an hour, and the last one for the span of the one before it.""" return _cover_table(pts) def price_bars(points, day0, day1, windows=(), now=None, kept=None): """One bar per price step of the day: a quarter for a 15-minute series (96; 92 and 100 on the clock-change days), an hour for an hourly one or a static tariff. A bar is the price of the point that covers it, never an average, and a step with no price is a bar with `eur_kwh` None. `kind` is "cheap" or "dear" where `windows` (`[(kind, start, end)]`, the runs the add-on itself uses) cover the bar's start, else "mid". A bar that is over keeps the kind it had while it was still ahead (`kept`, a dict that lives a day). Display only.""" pts = sorted(points, key=lambda p: p.start) step = _bar_step(pts) cover = _bar_cover(pts) n = int((day1 - day0) / step) bars = [] for k in range(n): a = day0 + k * step z = a + step hit = next((p for s, e, p in cover if s <= a < e), None) kind = None if hit is not None: if kept is not None and now is not None and z <= now: kind = kept.get(a, "mid") else: kind = next((w[0] for w in windows if w[1] <= a < w[2]), "mid") if kept is not None: kept[a] = kind bars.append({"h": k, "start": a.isoformat(), "label": hhmm(a), "end_label": hhmm(z), "eur_kwh": None if hit is None else round(hit.eur_per_kwh, 4), "estimate": hit is not None and bool(hit.provisional), "kind": kind}) return bars def _price_mode_in_use(c): """True when at least one battery is on and uses a price mode. The runs belong to those batteries; with all on Solar first or off, nothing acts on a price.""" for inv in c.inverters: pm = c.pack_mode(inv) if pm.use and pm.mode in PRICE_MODES: return True return False def price_windows(c, plan, verdict, now=None): """`[(kind, start, end)]`: where the add-on itself buys ("cheap") and gives ("dear"). With battery modes on that is the kept `priceruns.RunPlan` (`c.mode_runs`): the cheap run when charging in it pays and the trade gate allows it (UX-53), and every give run. UX-77: the cheap run is drawn to where control stops charging as it rolls on (`cheaproll`). With them off it is the arbitrage verdict's legs the add-on will take, the ones `trade_runs()` draws (`trade_legs()`, given `now`). Nothing is classified here.""" if getattr(c, "modes_on", False): rp = getattr(c, "mode_runs", None) if rp is None or not rp.on or not _price_mode_in_use(c): return [] end = rp.cheap.end roll = None if now is None else cheaproll.for_today(c, rp, now) if roll is not None: end = roll.last(end) # UX-77: to where control stops out = [("cheap", rp.cheap.start, end)] \ if rp.charge_useful() and getattr(c, "mode_gate", None) is None \ else [] return out + [("dear", g.start, g.end) for g in rp.give] if verdict is None: return [] return [("cheap" if kind == "grid" else "dear", pt.start, _point_end(plan, pt)) for kind, pt in trade_legs(c, plan, verdict, now)] def day_range(points): """(lowest, highest) of today's published points, as {eur_kwh, label}, or (None, None). The feed's own points, like the trade legs.""" 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, now): """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, and (PRICE-02) the same `now`. 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=arbitrage.weather_uncertainty( bool(getattr(c, "modes_on", False)), getattr(c, "solar_fc", None), bool(getattr(c, "weather_held", lambda _n: False)(now))), now=now) return verdict.match( allowed=lambda v: (v, "allowed"), refused=lambda _v: (None, refusal_key(plan, guard, now))) def refusal_key(plan, guard, now): """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" pts = _points(plan) ahead = {id(p) for _s, end, p in _cover_table(pts) if end > now} pts = [p for p in pts if id(p) in ahead] if sum(1 for p in pts 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) def leg_taken(c, plan, pt, now): """UX-53: will the add-on act on this point of the verdict? One in progress is the verdict's own answer now. One still ahead is asked of the same gate (`trade_verdict()`, the call the control cycle makes) at the point's own start. A charge that comes after the last dear period meets a flat remainder there and is refused, so it is not announced. When the gate allows it, the point is still the cheapest (dearest) one: the points before it are gone and it was the first extreme of the points from now.""" if now is None or pt.start <= now: return True return trade_verdict(c, plan, pt.start)[0] is not None def trade_legs(c, plan, verdict, now): """`[(kind, point)]` by start: the verdict's charge ("grid") and give ("give") legs that the add-on will take (`leg_taken()`).""" legs = sorted((("grid", verdict.cheapest), ("give", verdict.dearest)), key=lambda kp: kp[1].start) return [(k, pt) for k, pt in legs if leg_taken(c, plan, pt, now)] # Under this many kWh a block is not drawn: the cycle would not move it. MIN_KWH = 0.05 def trade_runs(c, plan, verdict, day0, day1, now, shadow, ahead=False): """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. UX-54 `ahead`: the window has not started, so the level now says nothing about it. The same rules then read each battery at the level a leg starts at: `ahead(name, kind, start)`, a level in % or None, as in `mode_runs()`; for None (and for `ahead=True`) a charge is read from the floor and a give from the ceiling (the most it could be). The leg itself is still the gate's (`trade_legs()`). Home never passes it.""" floor, ceiling = arbitrage.soc_bounds(c.o) if not ahead and any(inv.soc is None for inv in c.inverters): return [], [], "no_level" level = [] if ahead else [float(inv.soc) for inv in c.inverters] def start_pct(kind, start): got = [ahead(inv.name, kind, start) if callable(ahead) else None for inv in c.inverters] return [(floor if kind == "grid" else ceiling) if v is None else float(v) for v in got] runs, later = [], [] for kind, pt in trade_legs(c, plan, verdict, now): end = _point_end(plan, pt) if end <= now: continue if not (day0 <= pt.start < day1): later.append((kind, pt)) continue if ahead: level = start_pct(kind, pt.start) 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"). if not ahead: 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, ahead=False, rp=None): """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. UX-53: no grid charge is drawn while the trade gate refuses one (`c.mode_gate`, what the control cycle found). Returns `(runs, no_trade_key)`. UX-54 `ahead`: the window has not started, so the level now says nothing about it. The room and energy filter (`room()` and `plan_need()` below) then reads the level a block starts at: `ahead(name, kind, start)`, a function giving the expected level in % or None, or for None (and for `ahead=True`) a charge from the floor and a give from the ceiling. The same filter, the same `MIN_KWH`. `rp`: a run plan to read instead of `c.mode_runs`. Home never passes either.""" today = rp is None and not ahead rp = rp if rp is not None else 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) gate = getattr(c, "mode_gate", None) # UX-77: today's grid block runs to where control stops charging (the run # rolls on when it ends), each battery's to its own stop. roll = cheaproll.for_today(c, rp, now) if today else None 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 start_pct(kind, start): """The level a block of `kind` is read at when `ahead`.""" got = ahead(inv.name, kind, start) if callable(ahead) else None return got if got is not None else ( floor if kind == "grid" else ceiling) grids = [] # (end, kWh) of the grid blocks drawn def room(kind, hrs, start): kwh = inv.max_w / 1000.0 * hrs lvl = start_pct(kind, start) if ahead else soc if kind != "grid" and not ahead and cap and lvl is not None: # OPT-36: a give block reads the level after the grid block # before it, not the level now. lvl = min(float(ceiling), lvl + sum( k for e, k in grids if e <= start) / cap * 100.0) if cap and lvl is not None: kwh = min(kwh, max(0.0, (ceiling - lvl) if kind == "grid" else (lvl - 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, start_pct("grid", rp.cheap.start) if ahead and rp.cheap is not None else None) if gate is None and rp.charge_useful() and ( need > 0.0 if need is not None else pm.mode in GRID_CHARGE_MODES): cheap = rp.cheap if roll is None else priceruns.Run( rp.cheap.start, roll.end_for(inv.name, rp.cheap.end), rp.cheap.mean_eur_kwh) blocks.append(("grid", 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, run.start) if kind == "grid" and need is not None: kwh = min(kwh, need) if kwh < MIN_KWH: continue if kind == "grid": grids.append((run.end, kwh)) price = run.mean_eur_kwh if kind == "grid": cov = plan_cover(c, inv.name) what, why, amount = (run_grid_what(kwh, shadow), run_cheap_why(price), run_grid_amount(kwh * price)) if cov is not None: # OPT-28: with a day plan the block says how far it # charges and what that covers. what = run_grid_what_to(kwh, cov[0], shadow) why += " " + run_covers(cov[1]) 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, soc=None): """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). `soc`: UX-54, a level to read it from instead of the one now (`mode_runs(ahead=True)`).""" 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) at = None if inv is None else (inv.soc if soc is None else soc) if inv is None or at is None or not inv.capacity_kwh: return None return max(0.0, targets[name] - float(at)) / 100.0 * float( inv.capacity_kwh) def plan_cover(c, name): """OPT-28: `(level %, kWh)` for a battery the day plan charges from the grid: the level it charges to, and the expensive-hour use of the house that charge covers and the sun will not. Read from the level control holds (`plan_targets()`), so Home says what control does. None when the plan buys nothing for it. OPT-36: the kWh is the plan's own cover figure (`Plan.cover_kwh`), not one rebuilt from the level.""" dp = getattr(c, "day_plan", None) targets = c.plan_targets() if hasattr(c, "plan_targets") else None inv = next((i for i in c.inverters if i.name == name), None) if dp is None or not targets or name not in targets or inv is None \ or not inv.capacity_kwh or targets[name] <= 0.0 \ or name not in dp.cover_kwh: return None return targets[name], dp.cover_kwh[name] def _latched_below(c) -> bool: """OPT-34: no forecast, but a cheap run is on and a battery is still below the level its night charge keeps (`Controller.plan_targets()`), so control goes on charging.""" rp = getattr(c, "mode_runs", None) if rp is None or rp.cheap is None or not rp.cheap.holds( datetime.now(timezone.utc)): return False targets = getattr(c, "plan_targets", lambda: None)() or {} return any(inv.soc is not None and inv.name in targets and inv.soc < targets[inv.name] for inv in c.inverters) def plan_state(c): """P4: `"plan"` (a day plan is made), `"learning"` (no house profile yet), `"no_forecast"` (set but missing or stale), `"latched"` (OPT-34: the same, but a night charge under way is still going on to its level), 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 "latched" if _latched_below(c) else "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", "latched"): out["line"] = _t("solar." + st) elif fc is not None: out["line"] = line_solar(fc.today.kwh, out["count_kwh"], out["updated"], kept=getattr(fc, "kept", False)) dp = getattr(c, "day_plan", None) if st == "plan" and dp is not None and dp.uncovered_kwh >= MIN_KWH: out["line"] += " " + line_uncovered(dp.uncovered_kwh) 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 in the site's zone exactly as the control cycle reads it (MAINT-10).""" 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 = next_slot_start(now, cfg.start_hour, m.tz) if start is None: continue 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 runs_for(c, day0, day1, now, shadow, ahead=False, rp=None): """UX-54: the price blocks of the window `[day0, day1)`, by the same builders `build_plan()` uses for today: the batteries' price runs with modes on, the verdict's legs with them off. Nothing is decided here. `ahead`: see `mode_runs()`; for a day that has not started.""" if getattr(c, "modes_on", False): return mode_runs(c, day0, day1, now, shadow, ahead, rp)[0] fresh = c.fresh_tariff_plan(now) verdict, _why = trade_verdict(c, fresh, now) if verdict is None: return [] return trade_runs(c, fresh, verdict, day0, day1, now, shadow, ahead)[0] 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 days = _day_prices(c) seen = days.see(_points(plan), 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, now) bars = price_bars(seen, day0, day1, price_windows(c, fresh, verdict, now), now, days.kinds) modes_on = bool(getattr(c, "modes_on", False)) if running: 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 if getattr(c, "price_gap", False): lines.append(line_no_prices(c.price_gap_since, now)) 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. Nor when the quarter ends on the aim anyway (SAFETY-126, `within_aim`).""" pg = getattr(c, "peak_guard", None) return bool(pg is not None and pg.engaged and (pg.floor_w or 0.0) > 0.0 and not pg.within_aim) 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, **plan_words(c, inv.name, why)), }) if getattr(inv, "released", False): # SMA-10. A unit handed back says so, in its driver's words, not # "Held at 0 W": nothing is written to it. The key exists only for # such a unit, so a GoodWe-only page is unchanged. rows[-1]["now"] = handback.unit_state(handback.unit_of(inv), _LANG[0]) rows[-1]["handed_back"] = True 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`, `others_cover`, `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 if need >= MIN_KWH: return "tops_up" # OPT-36: no grid charge because other batteries (Cheap hours only, # Solar first) cover the hours, not the sun. dp = getattr(c, "day_plan", None) if dp is not None and dp.saved_kwh.get(name, 0.0) >= MIN_KWH: return "others_cover" return "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, unmeasured=getattr(c, "peak_unmeasured_s", 0.0) > 0.0, cut=getattr(c, "room_charge", False)) return out def state_of(c, level, shadow, waiting=False): """One word for the pill and the banner's colour: running, shadow, stopped, alarm, waiting or warning. `waiting`: the red headline is Home Assistant still starting (SAFETY-112), which needs nobody.""" if shadow: return "shadow" if level == "bad": return "waiting" if waiting else "alarm" if not c.auto: return "stopped" if level == "warn": return "warning" return "running" @contextlib.contextmanager def in_lang(lang): """UX-54: build sentences in `lang` for the length of a `with` block.""" _LANG[0] = uitext.lang_of(lang) try: yield finally: _LANG[0] = "en" def build(c, *, level: str, checks: list, now: datetime | None = None, lang: str = "en", held: bool | None = None, waiting: bool = False) -> dict: """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, waiting=waiting) finally: _LANG[0] = "en" def _build(c, *, level, checks, now, held=None, waiting=False): now = now or datetime.now(timezone.utc) shadow = bool(c.shadow) kind = state_of(c, level, shadow, waiting) # 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/.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 _WORDS.get("") != uitext.generation(): # SMA-10: the roster's words moved _WORDS.clear() _WORDS[""] = uitext.generation() 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" def _pct(v) -> str: return _dec(f"{float(v):.0f}") + " %" # Row id -> the title on Home. The row's own sentence is the body, in the # page's language (UX-46, `messages.pick`). 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="", pct="", kwh="") -> 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`; `pct` and `kwh`: the planned level and what it covers, for `tops_up` (OPT-28).""" return _t("why." + key, name=name or "", at=at, pct=pct, kwh=kwh) def plan_words(c, name, why) -> dict: """OPT-28: the figures of a `tops_up` line, or nothing.""" cov = plan_cover(c, name) if why == "tops_up" else None return {} if cov is None else {"pct": _pct(cov[0]), "kwh": _kwh(cov[1])} 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)) def run_grid_what_to(kwh, pct, shadow) -> str: """OPT-28: a grid block of the day plan, with the level it charges to.""" return _w(shadow, "run.grid_what_to", kwh=_kwh(kwh), pct=_pct(pct)) def run_covers(kwh) -> str: """OPT-28: what a planned grid charge covers.""" return _t("run.covers", 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 day_note(key, **values) -> str: """UX-54: one sentence of the Forecast tab (`home_words.day`).""" return _t("day." + key, **values) def line_solar(fc_kwh, count_kwh, at, kept=False) -> str: """The forecast line. OPT-27: a reading kept past three hours (the evening's, used at night) says so instead of "Updated".""" if kept: return _t("solar.kept", at=at) return _t("solar.line", kwh=_kwh(fc_kwh), count=_kwh(count_kwh), at=at) def line_uncovered(kwh) -> str: """OPT-28: the Leftover solar batteries cannot hold the expensive hours' shortfall. The rest comes from the grid: OPT-36 counts the other batteries first.""" return _t("solar.uncovered", kwh=_kwh(kwh)) 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_no_prices(since, now) -> str: """PRICE-01: no price covers now. `since` is an instant or None.""" if since is None: return _t("no_prices_yet") local, today = brussels(since), brussels(now) if local.date() == today.date(): return _t("no_prices_since", at=f"{local:%H:%M}") return _t("no_prices_since_day", day=local.day, month=month_name(local), at=f"{local:%H:%M}") 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, unmeasured=False, cut=False) -> str: step = _w(shadow, "step_in", kw=_kw(limit_w)) if unmeasured: # SAFETY-90: the add-on started during this period. return _t("period_unmeasured") + " " + step if period_w is None: return step if period_w > limit_w: # SAFETY-125: the figure is the current rate, and a grid charge is # what gets cut back. return _t("period_over", kw=_kw(period_w)) + " " + ( _w(shadow, "charge_cut", kw=_kw(limit_w)) if cut else 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", "waiting", "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_grid_what_to(12.34, 82, 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", line_period(1240, 3500, sh, unmeasured=True)), ("plan", line_period(5814, 5413, sh)), ("plan", line_period(5814, 5413, sh, cut=True)), ("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_covers(9.4)), ("plan", line_uncovered(2.5)), ("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_no_prices(None, datetime(2026, 10, 14, 9, 0, tzinfo=timezone.utc))), ("plan", line_no_prices(datetime(2026, 10, 14, 0, 0, tzinfo=timezone.utc), datetime(2026, 10, 14, 9, 0, tzinfo=timezone.utc))), ("plan", line_no_prices(datetime(2026, 10, 13, 22, 0, tzinfo=timezone.utc), datetime(2026, 10, 14, 9, 0, tzinfo=timezone.utc))), ("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", pct="82 %", kwh="9.4 kWh")) 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")), ("plan", _t("solar.latched"))] out += [("plan", day_note(k, pct=70, name="Battery A", at="08:15", kwh="9.2 kWh", count="6.4 kWh")) for k in words["day"]] out += [("plan", line_no_trade(k)) for k in ("not_covered", "no_prices", "peak", "not_firm", "not_worth", "no_level")] return out def priced_hours(bars): """How many clock hours of the day have a price: distinct hours among the bars that have one, so 96 quarters count 24, not 96. The setup wizard's "N of 24 hours". A 23-hour day gives 23, a 25-hour day 24.""" return len({b["label"][:2] for b in bars if b.get("eur_kwh") is not None})