admin/goodwe-addon
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

UX-13 P1: page shell and Home screen on a structured /status
checks / unit (pull_request) Successful in 9m37s
Details
infra-pin / fresh (pull_request) Successful in 2s
Details

Browse Source 
- app/static/{index.html,app.css,app.js}: tabs (Home, Batteries, Settings,
  Setup), hash routes, phone bottom tabs, light/dark. Served inlined at "/";
  the route list is unchanged, and every URL is relative.
- app/homeview.py: /status gains state, now, batteries, plan, attention,
  month_peak and control beside the old fields. The plan is what the add-on
  does today: the arbitrage verdict (the same plan.arbitrage() call as the
  cycle), the monthly cycle, the peak limit, "follows the house".
- The Home plan chart has hover, tap and keyboard-focus tooltips: price bar =
  hour and price; block = what, when, why, kWh and EUR.
- Region preset and year: in config/model (for P2) and on the Home price line.
- test_home.py; test_messages part 8 scans every Home sentence; STYLE plan line.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LyR88rXETvwhTgx6yHE39p
This commit is contained in:
Claude
2026-09-28 00:45:01 +02:00
parent afd6af068d
commit f8d63cc614
12 changed files with 2388 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
goodwe_controller/app/homeview.py
+932
View File
@@ -0,0 +1,932 @@
"""UX-13 P1: the Home screen's part of `/status`, built from what the add-on
already knows and already plans. Reads only; commands nothing.
`build(controller)` returns `now`, `batteries`, `plan`, `attention`,
`month_peak`, `control` and `state`. The old fields (`banner`, `rows`,
`checks`, ...) stay beside them for the installer details and for anything
that reads the JSON. See DECISIONS/UX-13-one-layout-table.md and the design
note, section 3.3.
The plan shows only what the add-on does today: trading on price (the
arbitrage verdict, the same call the control cycle makes), the monthly battery
cycle, the peak limit and "follows the house". Battery modes (OPT-11) are not
built, so nothing here invents a mode plan. Every sentence is in the words
section at the bottom and is held to STYLE.md by test_messages.py.
"""
from datetime import datetime, timedelta, timezone
from . import arbitrage, messages, peakguard
from .maintenance import CHARGE as M_CHARGE, DRAIN as M_DRAIN, HOLD as M_HOLD
from .maintenance import IDLE as M_IDLE
from .price import Absence, Uncertainty, brussels
from .static_tariff import static_plan
from .tariff import BILLING_TZ
from .uimodel import REGION_PRESETS, options_to_ui
# A battery at or under this many watts is resting, not charging or giving.
IDLE_W = 50.0
# Display only: the lowest third of today's price range is "cheap", the top
# third "expensive". A price range under this is flat, and nothing is marked.
FLAT_EUR_KWH = 0.01
# Severity words the page shows; the same four as STYLE.md section 3.
ALARM, WARNING, INFO = messages.ALARM, messages.WARNING, messages.INFO
# --------------------------------------------------------------------------- #
# Time: today's local day, on the billing calendar
# --------------------------------------------------------------------------- #
def local_day(now: datetime):
"""`(midnight_utc, next_midnight_utc)` of the local day holding `now`."""
here = now.astimezone(BILLING_TZ)
start = datetime(here.year, here.month, here.day, tzinfo=BILLING_TZ)
nxt = start.date() + timedelta(days=1)
end = datetime(nxt.year, nxt.month, nxt.day, tzinfo=BILLING_TZ)
return start.astimezone(timezone.utc), end.astimezone(timezone.utc)
def _hours(t: datetime, day0: datetime) -> float:
return (t - day0).total_seconds() / 3600.0
def hhmm(t: datetime) -> str:
return f"{brussels(t):%H:%M}"
def _kw(w) -> str:
return f"{abs(float(w)) / 1000:.1f} kW"
def _eur(v) -> str:
return f"€{float(v):.2f}"
def _kwh(v) -> str:
return f"{float(v):.1f} kWh"
# --------------------------------------------------------------------------- #
# Now
# --------------------------------------------------------------------------- #
def flow(w, *, shadow=False):
"""`charging` / `giving` / `resting` / None for one battery power."""
if w is None:
return None
if shadow:
return "resting"
if w > IDLE_W:
return "giving"
if w < -IDLE_W:
return "charging"
return "resting"
def now_block(*, grid_w, pv_w, batt_w, has_solar, shadow):
"""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 = None
if has_solar and None not in (grid_w, pv_w, batt_w):
house = float(grid_w) + float(pv_w) + float(batt_w)
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),
"grid": tile_grid(grid_w),
"batt": tile_batt(batt_w, shadow),
},
}
# --------------------------------------------------------------------------- #
# 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)."""
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),
"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
# --------------------------------------------------------------------------- #
# 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, "peak" if guard else "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 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)."""
floor, ceiling = arbitrage.soc_bounds(c.o)
runs, later = [], []
for kind, pt in (("grid", verdict.cheapest), ("give", verdict.dearest)):
end = _point_end(plan, pt)
if end <= now:
continue
if not (day0 <= pt.start < day1):
later.append((kind, pt))
continue
hrs = (end - pt.start).total_seconds() / 3600.0
for i, inv in enumerate(c.inverters):
kwh = inv.max_w / 1000.0 * hrs
if inv.capacity_kwh and inv.soc is not None:
room = (ceiling - inv.soc if kind == "grid"
else inv.soc - floor)
kwh = min(kwh, max(0.0, room) / 100.0 * inv.capacity_kwh)
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
def maintenance_runs(c, day0, day1, now, shadow):
"""The monthly battery cycle: under way now, or starting later today.
The cycle starts in the hour `maintenance_start_hour` names, read as the
control cycle reads it (UTC hour of `now`), shown in local time."""
runs, lines = [], []
waiting = []
for i, (inv, m) in enumerate(zip(c.inverters, c.maints)):
cfg = m.cfg
if not cfg.enabled:
continue
cap = inv.capacity_kwh
if m.phase != M_IDLE:
hours = _cycle_hours(m.phase, inv.soc, cap, cfg)
end = now + timedelta(hours=hours or 1.0)
runs.append(_run(i, "cycle", now, end, day0, day1,
run_cycle_what(m.phase, shadow),
run_cycle_why(shadow),
run_cycle_amount(hours)))
continue
if not m.due(now) and (m.next_due(now) is None
or m.next_due(now) >= day1):
continue
start = now.replace(hour=cfg.start_hour, minute=0, second=0,
microsecond=0)
if start + timedelta(hours=1) <= now:
start += timedelta(days=1)
due_at = m.next_due(now)
if due_at is not None and due_at > start:
continue
if not (day0 <= start < day1):
continue
waiting.append((m.waited_days(now), i, start))
# One battery cycles at a time (maintenance.arbitrate); the one waiting
# longest goes first, and none starts while another is under way.
busy = any(m.phase != M_IDLE for m in c.maints)
waiting.sort(key=lambda t: (-t[0], t[1]))
for n, (_w, i, start) in enumerate(waiting):
inv, m = c.inverters[i], c.maints[i]
hours = _cycle_hours(M_DRAIN, inv.soc, inv.capacity_kwh, m.cfg)
if n == 0 and not busy:
end = start + timedelta(hours=hours or 1.0)
runs.append(_run(i, "cycle", start, end, day0, day1,
run_cycle_what(M_IDLE, shadow),
run_cycle_why(shadow),
run_cycle_amount(hours)))
else:
lines.append(line_cycle_waits(inv.name))
return runs, lines
def _cycle_hours(phase, soc, cap_kwh, cfg):
"""Hours left in a cycle, or None when the capacity or level is unknown."""
if phase == M_HOLD:
return cfg.hold_min / 60.0
if not cap_kwh or soc is None:
return None
hours = cfg.hold_min / 60.0
fill_from = soc if phase == M_CHARGE else cfg.soc_floor
hours += (max(0.0, cfg.soc_target - fill_from) / 100.0 * cap_kwh
/ max(cfg.charge_w / 1000.0, 0.1))
if phase not in (M_CHARGE,):
hours += (max(0.0, soc - cfg.soc_floor) / 100.0 * cap_kwh
/ max(cfg.discharge_w / 1000.0, 0.1))
return hours
def peak_run(c, day0, day1, now, shadow, limit_w):
"""While the batteries hold the grid under the peak limit, one block for
the current 15-minute period on every battery."""
pg = c.peak_guard
if pg is None or not pg.engaged:
return []
start = now - timedelta(minutes=now.minute % 15, seconds=now.second,
microseconds=now.microsecond)
end = start + timedelta(minutes=15)
return [_run(i, "peak", start, end, day0, day1,
run_peak_what(pg.ceiling_w or limit_w, shadow),
run_peak_why(), None)
for i in range(len(c.inverters))]
def house_runs(n, day0, day1, now, shadow):
""""Follows the house" for the rest of the day, drawn under the others."""
return [_run(i, "house", max(now, day0), day1, day0, day1,
run_house_what(shadow), run_house_why(shadow),
run_house_amount())
for i in range(n)]
def region_of(opts):
"""The levies preset and its year, as the page model has it (UX-13 P0)."""
try:
ui = options_to_ui(opts)
except Exception: # noqa: BLE001
return {"id": None, "year": None}
rid = ui.get("ui.region")
return {"id": rid, "year": ui.get("region_year")
if rid in REGION_PRESETS else None}
def build_plan(c, now, shadow, running, held=False):
day0, day1 = local_day(now)
fresh = c.fresh_tariff_plan(now)
plan = fresh
if plan is None or plan.origin == "static":
# A static profile is pure arithmetic: price the whole day from it.
try:
static = static_plan(c.o, day0, day1, now=now)
if static.has_prices:
plan = static
except Exception: # noqa: BLE001
pass
seen = _day_prices(c).see(_points(plan), day0, day1)
bars = hour_bars(seen, day0, day1)
limit_w = c.peak.cap_w # the limit the controller parsed at start
runs, lines = [], []
# The control cycle trades on the fresh plan only, so this does too.
verdict, trade_why = trade_verdict(c, fresh)
if running:
runs += house_runs(len(c.inverters), day0, day1, now, shadow)
if verdict is not None:
trade, later = trade_runs(c, fresh, verdict, day0, day1, now,
shadow)
runs += 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:
lines.append(line_peak_limit(limit_w, shadow))
lines = [x for x in lines if x]
kinds = {r["kind"] for r in runs}
if held:
verdict_key = "held"
elif not running:
verdict_key = "stopped"
elif "cycle" in kinds:
verdict_key = "cycle"
elif "grid" in kinds or "give" in kinds:
verdict_key = "trade"
else:
verdict_key = "house"
known = [b for b in bars if b["eur_kwh"] is not None]
price_now = None
try:
if plan is not None:
price_now = plan.price_at(now)
except Absence:
price_now = None
feed = fresh is not None and fresh.from_feed
tomorrow = any(not p.provisional and p.start >= day1
for p in _points(fresh))
region = region_of(c.o)
return {
"day": brussels(now).date().isoformat(),
"now_h": round(_hours(now, day0), 3),
"hours": int(round(_hours(day1, day0))),
"verdict": VERDICT[verdict_key],
"verdict_key": verdict_key,
"head": plan_head(verdict_key, shadow),
"prices": bars,
"runs": runs,
"lines": lines,
"why": c.why_words,
"price_line": line_prices(
price_now,
min(known, key=lambda b: b["eur_kwh"]) if known else None,
max(known, key=lambda b: b["eur_kwh"]) if known else None),
"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 battery_rows(c, shadow, running, held=False):
shares = {}
alloc = getattr(c, "allocation", None)
if shadow and alloc is not None and getattr(alloc, "shares", None):
shares = dict(alloc.shares)
rows = []
for i, inv in enumerate(c.inverters):
m = c.maints[i] if i < len(c.maints) else None
phase = m.phase if m is not None else M_IDLE
cap = inv.capacity_kwh
rows.append({
"name": inv.name,
"level_pct": None if inv.soc is None else float(inv.soc),
"power_w": None if inv.batt is None else float(inv.batt),
"capacity_kwh": cap,
"stored_kwh": (None if cap is None or inv.soc is None
else round(cap * float(inv.soc) / 100.0, 2)),
"state": flow(inv.batt, shadow=shadow),
"mode": MODE_HOUSE,
"maintenance": phase,
"now": battery_now(inv.soc, inv.batt, phase=phase, shadow=shadow,
running=running, held=held,
would_w=shares.get(inv.name)),
# Per-battery reasons arrive with battery modes (OPT-11).
"why": None,
})
return rows
# Row id -> the title on Home. The row's own sentence is the body.
ATTN_TITLES = {
"controller": "Control is not running",
"meter": "Meter reading",
"meter_frozen": messages.T_METER_FROZEN,
"freeze_check": "Stuck meter check",
"month_peak": "This month's peak",
"peak_guard": "Peak limit",
"peak_forecast": "Peak limit forecast",
"sensor": "Battery readings",
"control": "Inverter control",
"power_too_high": messages.T_OUTPUT_CLAMP,
"not_keeping_up": messages.T_LOOP_BOUND,
"write": "Power request not accepted",
"direction": "Power went the wrong way",
"site_direction": "Power went the wrong way",
"circulating": messages.T_CIRCULATING,
"delivery": messages.S_NOT_DELIVERING,
"estop": "Emergency stop",
"pv": "Solar reading",
None: "Inverter connection",
}
# Row id -> the DOCS.md heading under Notifications, where one exists.
ATTN_DOCS = {
"meter_frozen": messages.T_METER_FROZEN,
"power_too_high": messages.T_OUTPUT_CLAMP,
"not_keeping_up": messages.T_LOOP_BOUND,
"circulating": messages.T_CIRCULATING,
"delivery": messages.S_NOT_DELIVERING,
}
def attention(checks):
"""Every row that is not plain green: red rows as alarms, amber rows as
warnings, and green-but-amber rows (a state still settling) as info.
Most urgent first; within a level, the page's own row order."""
rank = {ALARM: 0, WARNING: 1, INFO: 2}
out = []
for row in checks:
ok, warn = bool(row.get("ok")), bool(row.get("warn"))
if ok and not warn:
continue
sev = INFO if ok else (WARNING if warn else ALARM)
rid = row.get("id")
out.append({"severity": sev, "id": rid,
"title": ATTN_TITLES.get(rid, ATTN_TITLES[None]),
"body": row.get("text", ""),
"docs": ATTN_DOCS.get(rid)})
out.sort(key=lambda a: rank[a["severity"]])
return out
def month_peak(c, now):
"""This month's peak against the limit, and the 15-minute period so far."""
limit_w = c.peak.cap_w # the limit the controller parsed at start
local = brussels(now)
out = {"month": f"{local:%B}", "limit_w": limit_w, "peak_w": None,
"pct": None, "period_w": None, "state": "not_covered",
"line": ""}
if not c.p1_enabled:
out["line"] = line_peak_not_covered()
return out
reading = c.peak.peak(now)
if reading.is_unreadable:
out["state"] = "unreadable"
elif reading.is_observed:
out["state"] = "observed"
out["peak_w"] = float(reading.peak_w)
out["pct"] = round(100.0 * float(reading.peak_w) / limit_w, 1) \
if limit_w > 0 else None
else:
out["state"] = "none"
pg = c.peak_guard
ceiling = limit_w
if pg is not None:
out["period_w"] = pg.projected_w
ceiling = pg.ceiling_w or limit_w
out["line"] = line_period(out["period_w"], ceiling, c.shadow)
return out
def state_of(c, level, shadow):
"""One word for the pill and the banner's colour: running, shadow,
stopped, alarm or warning."""
if shadow:
return "shadow"
if level == "bad":
return "alarm"
if not c.auto:
return "stopped"
if level == "warn":
return "warning"
return "running"
def build(c, *, level: str, checks: list, now: datetime | None = None) -> dict:
"""The structured half of `/status`. Pure reads of the controller."""
now = now or datetime.now(timezone.utc)
shadow = bool(c.shadow)
kind = state_of(c, level, shadow)
# A red banner means every inverter is at 0 W (the status cascade), so
# nothing is planned until it clears. Shadow mode still plans: it is
# what the add-on would do.
held = level == "bad" and not shadow
running = (bool(c.auto) or shadow) and not held
ctl = next((r for r in checks if r.get("id") == "controller"), None)
return {
"state": {"kind": kind, "pill": pill(kind),
"ran": ctl.get("text", "") if ctl else "",
"sub": banner_sub(kind)},
"now": now_block(grid_w=c.grid, pv_w=c.pv_w, batt_w=c.site_batt_w,
has_solar=bool(c.pv_roster), shadow=shadow),
"batteries": battery_rows(c, shadow, running, held),
"plan": build_plan(c, now, shadow, running, held=held),
"attention": ([{"severity": INFO, "id": "shadow",
"title": ATTN_SHADOW, "body": attn_shadow_body(),
"docs": None}] if shadow else []) + attention(checks),
"month_peak": month_peak(c, now),
"control": {"running": bool(c.auto), "shadow": shadow,
"text": control_text(bool(c.auto), shadow),
"maintenance": c.maint.phase,
"docs_url": _docs_url()},
}
def _docs_url():
slug = messages.addon_slug()
return None if slug is None else messages.docs_url(slug,
messages._site["ha"])
# =========================================================================== #
# The words. STYLE.md: plain, short, present tense. A plan line is at most
# two sentences of at most 12 words each (UX-13 design, section 4); a title at
# most 5 words; everything else one sentence of at most 25 words. In shadow
# mode a text says what the add-on WOULD do. test_messages.py renders every
# function below through `text_samples()` and checks it.
# =========================================================================== #
MODE_HOUSE = "Follows the house"
ATTN_SHADOW = "Shadow mode is on"
def attn_shadow_body() -> str:
return "The add-on decides everything, but every inverter stays at 0 W."
VERDICT = {
"held": "Held at 0 W",
"stopped": "Stopped",
"cycle": "Monthly cycle today",
"trade": "Trading on price",
"house": "Follows the house",
}
def _w(shadow, live, would):
return would if shadow else live
def pill(kind) -> str:
return {"running": "Running", "shadow": "Shadow mode",
"stopped": "Stopped", "alarm": "Needs you now",
"warning": "Running, with a warning"}[kind]
def banner_sub(kind) -> str:
return {
"running": "The batteries follow the plan below.",
"shadow": "The plans below say what the add-on would do.",
"stopped": "Start control below when you are ready.",
"alarm": "See Needs attention below for what to do.",
"warning": "See Needs attention below.",
}[kind]
def tile_solar(pv_w, has_solar) -> str:
if not has_solar:
return "Not set up"
if pv_w is None:
return "No reading"
return "Producing" if pv_w > IDLE_W else "Not producing"
def tile_house(house_w, has_solar) -> str:
if not has_solar:
return "Needs a solar reading"
return "No reading" if house_w is None else "Using"
def tile_grid(grid_w) -> str:
if grid_w is None:
return "No reading"
if grid_w > IDLE_W:
return "Taking from the grid"
if grid_w < -IDLE_W:
return "Feeding in"
return "Balanced"
def tile_batt(batt_w, shadow) -> str:
if shadow:
return "Held at 0 W"
return {"giving": "Giving power", "charging": "Charging",
"resting": "Resting", None: "No reading"}[flow(batt_w)]
def battery_now(soc, batt_w, *, phase, shadow, running, held=False,
would_w=None) -> str:
if soc is None or batt_w is None:
return "No reading from this battery."
if shadow:
if would_w is None or abs(would_w) <= IDLE_W:
return "Held at 0 W. Would rest now."
verb = "give" if would_w > 0 else "charge"
return f"Held at 0 W. Would {verb} {_kw(would_w)} now."
if held:
return "Held at 0 W until the alarm clears."
if not running:
return "Held at 0 W while control is stopped."
if phase != M_IDLE:
return {M_DRAIN: "Monthly cycle: emptying.",
M_CHARGE: "Monthly cycle: refilling.",
M_HOLD: "Monthly cycle: holding full."}.get(
phase, "Monthly cycle under way.")
st = flow(batt_w)
if st == "giving":
return f"Giving {_kw(batt_w)} now."
if st == "charging":
return f"Charging {_kw(batt_w)} now."
return "Resting now."
def plan_head(key, shadow) -> str:
if key == "held":
return "Every inverter is at 0 W until the alarm clears."
if key == "stopped":
return "Control is stopped, so nothing is planned."
if not shadow:
return {
"cycle": "A battery runs its monthly cycle today.",
"trade": "Charges in the cheapest hour, gives power in the dearest.",
"house": "Charges from spare solar and covers the house.",
}[key]
return "Shadow mode: this is what the add-on would do. " + {
"cycle": "A battery would run its monthly cycle today.",
"trade": "Would charge when cheapest and give power when dearest.",
"house": "Would charge from spare solar and cover the house.",
}[key]
def run_grid_what(kwh, shadow) -> str:
return _w(shadow, "Charges", "Would charge") + \
f" up to {_kwh(kwh)} from the grid."
def run_grid_why(price) -> str:
return f"Cheapest price in the list, {_eur(price)} per kWh."
def run_grid_amount(eur) -> str:
return f"Costs up to {_eur(eur)}."
def run_give_what(kwh, shadow) -> str:
return _w(shadow, "Gives", "Would give") + \
f" up to {_kwh(kwh)}, to the house first."
def run_give_why(price) -> str:
return f"Dearest price in the list, {_eur(price)} per kWh."
def run_give_amount(eur) -> str:
return f"Worth up to {_eur(eur)} at that price."
def run_cycle_what(phase, shadow) -> str:
if phase == M_IDLE:
return _w(shadow, "Starts", "Would start") + \
" the monthly battery cycle."
return _w(shadow, "Runs", "Would run") + " the monthly battery cycle now."
def run_cycle_why(shadow=False) -> str:
return _w(shadow, "Empties, refills and holds full",
"Would empty, refill and hold full") + \
", so the level reads right."
def run_cycle_amount(hours) -> str:
if hours is None:
return "How long depends on the battery."
return f"Takes about {max(1, round(hours))} hours."
def run_peak_what(limit_w, shadow) -> str:
return _w(shadow, "Gives", "Would give") + \
f" power to stay under {_kw(limit_w)}."
def run_peak_why() -> str:
return "The peak limit comes before everything else."
def run_house_what(shadow) -> str:
return _w(shadow, "Follows", "Would follow") + " the house."
def run_house_why(shadow=False) -> str:
return _w(shadow, "Charges from spare solar, gives power",
"Would charge from spare solar, give power") + \
" when the house needs it."
def run_house_amount() -> str:
return "Amounts depend on the sun and the house."
def line_no_trade(key) -> str:
return {
"off": "",
"not_covered": "No trading on price: nothing measures the 15-minute "
"peak.",
"no_prices": "No trading on price yet: no prices have arrived.",
"peak": "No trading on price now: the peak limit comes first.",
"not_worth": "No trading on price today. The price gap is too small.",
}[key]
def line_trade_tomorrow(kind, at, price, shadow) -> str:
verb = {"grid": _w(shadow, "Charges", "Would charge"),
"give": _w(shadow, "Gives power", "Would give power")}[kind]
return f"Tomorrow: {verb.lower()} at {at}, {_eur(price)} per kWh."
def line_cycle_waits(name) -> str:
return f"{name} waits its turn for the monthly cycle."
def line_peak_limit(limit_w, shadow) -> str:
return _w(shadow, "The batteries step in", "The batteries would step in") \
+ f" above {_kw(limit_w)}."
def line_prices(now_eur, lo, hi) -> str:
parts = []
if now_eur is not None:
parts.append(f"Price now {_eur(now_eur)} per kWh, all in.")
if lo is not None and hi is not None:
parts.append(f"Today: cheapest {_eur(lo['eur_kwh'])} at "
f"{lo['label']}, dearest {_eur(hi['eur_kwh'])} at "
f"{hi['label']}.")
return " ".join(parts) or "No prices for today."
def line_tomorrow(known) -> str:
return ("Tomorrow's prices are in." if known else
"Tomorrow's prices usually arrive after 13:00.")
def line_region(rid, year) -> str:
if rid in REGION_PRESETS and year:
return f"Levies and VAT: {REGION_WORDS.get(rid, rid)}, {year} figures."
return "Levies and VAT: your own figure."
REGION_WORDS = {"flanders": "Flanders, Fluvius Antwerpen"}
def line_peak_not_covered() -> str:
return "No peak limit: this meter does not measure 15-minute periods."
def line_period(period_w, limit_w, shadow) -> str:
step = _w(shadow, "The batteries step in", "The batteries would step in")
if period_w is None:
return f"{step} above {_kw(limit_w)}."
return (f"This 15-minute period is heading for {_kw(period_w)}. "
f"{step} above {_kw(limit_w)}.")
def control_text(running, shadow) -> str:
if shadow:
return "Shadow mode: every inverter stays at 0 W."
if running:
return "Control is running."
return "Control is stopped. Every inverter is at 0 W."
def text_samples() -> list:
"""`(kind, text)` for every sentence above, rendered with real values, so
test_messages.py can hold each to its STYLE limit. Kinds: title, plan
(2 sentences of 12 words), line (1 sentence of 25 words)."""
out = [("title", t) for t in VERDICT.values()]
out += [("title", t) for t in ATTN_TITLES.values()]
out += [("title", MODE_HOUSE), ("title", ATTN_SHADOW),
("line", attn_shadow_body())]
for k in ("running", "shadow", "stopped", "alarm", "warning"):
out += [("title", pill(k)), ("line", banner_sub(k))]
for sh in (False, True):
out += [("title", tile_solar(v, s)) for v, s in
((None, False), (None, True), (3900, True), (0, True))]
out += [("title", tile_house(v, s)) for v, s in
((None, False), (None, True), (1100, True))]
out += [("title", tile_grid(v)) for v in (None, 1200, -800, 5)]
out += [("title", tile_batt(v, sh)) for v in (None, 2500, -2500, 0)]
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)))
out.append(("plan", battery_now(62, 0, phase=M_IDLE, shadow=False,
running=False, held=True)))
out.append(("plan", battery_now(None, None, phase=M_IDLE, shadow=sh,
running=True)))
for k in VERDICT:
out.append(("plan", plan_head(k, sh)))
out += [("plan", run_grid_what(12.34, sh)),
("plan", run_give_what(9.6, sh)),
("plan", run_cycle_what(M_IDLE, sh)),
("plan", run_cycle_what(M_DRAIN, sh)),
("plan", run_peak_what(3500, sh)),
("plan", run_house_what(sh)),
("plan", line_trade_tomorrow("grid", "02:00", 0.1234, sh)),
("plan", line_trade_tomorrow("give", "19:00", 0.3612, sh)),
("plan", line_peak_limit(3500, sh)),
("plan", line_period(1240, 3500, sh)),
("plan", line_period(None, 3500, sh)),
("plan", control_text(True, sh)),
("plan", control_text(False, sh))]
out += [("plan", run_grid_why(0.1412)), ("plan", run_grid_amount(1.23)),
("plan", run_give_why(0.3612)), ("plan", run_give_amount(3.47)),
("plan", run_cycle_why()), ("plan", run_cycle_why(True)),
("plan", run_cycle_amount(None)),
("plan", run_cycle_amount(5.4)), ("plan", run_peak_why()),
("plan", run_house_why()), ("plan", run_house_why(True)),
("plan", run_house_amount()),
("plan", line_cycle_waits("Battery B")),
("plan", line_prices(0.16, {"eur_kwh": 0.14, "label": "13:00"},
{"eur_kwh": 0.36, "label": "19:00"})),
("plan", line_prices(None, None, None)),
("plan", line_tomorrow(True)), ("plan", line_tomorrow(False)),
("plan", line_region("flanders", 2026)),
("plan", line_region("own", None)),
("plan", line_peak_not_covered())]
out += [("plan", line_no_trade(k)) for k in
("not_covered", "no_prices", "peak", "not_worth")]
return out