glenn schrooyenandClaude
75195f7073
F1: tomorrow's blocks no longer depend on the battery's level now. Home sizes a block from the level now (`room()`, `plan_need()` in mode_runs, the leg sizing in trade_runs) and draws none when there is no room or energy. For a day that has not started the builders take `ahead=True` (Home never passes it): a charge is read from the floor, a give from the ceiling. A grid block is then kept only where `modetrade.chargers()` would charge at the curve's level when the run starts. The UX-53 gate is unchanged. Today still equals Home; a give run in the plan with no block (empty battery) has its own sentence instead of "No charge or give blocks planned". F2: test_ux54.py part 7 walks the real `_mode_roles()` every 15 minutes through tomorrow and compares the roles with the tab's blocks (modes on, the peak gate, part-full batteries), and the control cycle's verdict at 07:00 (modes off). F3: the curve counts spare solar as trust x max(0, sun - house), stored at EFF, as `dayplan.plan()` does (10 kWh sun, 2 kWh house, 70 %: 5.32 kWh). Strings: cycle, no_tomorrow, ahead, no_level, measured; new give_later. No decision, command, timing or option key changes. Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Bdqr7PiaHcBnzPituu2yTC
300 lines
13 KiB
Python
300 lines
13 KiB
Python
"""UX-54: the Forecast tab, built on request (`GET forecast`). Reads only.
|
|
|
|
Two days, today and tomorrow, per local hour: the solar forecast, the solar
|
|
measured so far, the house's expected use, and each battery's expected level
|
|
with its blocks.
|
|
|
|
It follows the add-on, as Home does (UX-53), and decides nothing:
|
|
* the blocks are Home's own: today's are the `runs` of `/status`'s plan, and
|
|
tomorrow's come from the same builders (`homeview.runs_for()`), so a block
|
|
is on the tab exactly where the plan has one, with the plan's words;
|
|
* the sun is `solarfc.Forecast.between()` and the house's use is
|
|
`dayplan.HouseProfile.estimator()` read as `dayplan.plan()` reads it;
|
|
* the level is `daycurve.curve()`, fed the blocks, the give runs, the packs
|
|
and the plan's levels (`modetrade.charge_to()`), floor and ceiling.
|
|
A part that is not known yet is a plain note (`home_words.day`), never an
|
|
empty chart. DECISIONS/UX-54-forecast-tab.md has the record.
|
|
"""
|
|
|
|
import dataclasses
|
|
from datetime import datetime, timedelta
|
|
|
|
from . import arbitrage, dayplan, homeview as hv, modetrade
|
|
from .daycurve import Batt, curve
|
|
from .packmode import SOLAR_FIRST
|
|
from .price import brussels
|
|
|
|
BLOCKS = ("grid", "give", "cycle") # the kinds a battery's blocks carry
|
|
|
|
|
|
def _hours_between(a, b) -> float:
|
|
return (b - a).total_seconds() / 3600.0
|
|
|
|
|
|
def build(c, status: dict, now: datetime, lang: str = "en") -> dict:
|
|
"""The tab's JSON. `status` is the `/status` payload just built, whose
|
|
plan holds today's blocks."""
|
|
with hv.in_lang(lang):
|
|
return _build(c, status, now)
|
|
|
|
|
|
def sun_kwh(fc, now, a, b):
|
|
"""kWh the forecast gives in `[a, b)`, `a` at or after `now`, or None.
|
|
Both ends are measured from `now`: `Forecast.between()` reads its start
|
|
as the moment the "left today" figure was taken, so a span that starts
|
|
later would be spread over what is left of the bell, not over its own
|
|
hours (state-only forecasts, DECISIONS/UX-54-forecast-tab.md)."""
|
|
hi, lo = fc.between(now, b), fc.between(now, a)
|
|
return None if hi is None or lo is None else max(0.0, hi - lo)
|
|
|
|
|
|
def _times(run):
|
|
return (datetime.fromisoformat(run["start"]),
|
|
datetime.fromisoformat(run["end"]))
|
|
|
|
|
|
def _build(c, status, now):
|
|
plan, rows = status.get("plan"), status.get("batteries")
|
|
if not plan or rows is None:
|
|
return {"ok": False}
|
|
d0, d1 = hv.local_day(now)
|
|
n0, n1 = hv.local_day(d1 + timedelta(hours=2))
|
|
shadow = bool(c.shadow)
|
|
key = plan.get("verdict_key")
|
|
running = key not in ("held", "stopped")
|
|
modes_on = bool(getattr(c, "modes_on", False))
|
|
floor, ceil = arbitrage.soc_bounds(c.o)
|
|
fc = getattr(c, "solar_fc", None)
|
|
hp = getattr(c, "house_profile", None)
|
|
est = hp.estimator() if hp is not None and fc is not None else None
|
|
trust = max(0.0, min(1.0, float(getattr(c, "solar_trust_pct", 70))
|
|
/ 100.0))
|
|
|
|
# Why a part is not known (None: nothing is missing).
|
|
missing = ("modes_off" if not modes_on
|
|
else "not_set" if c.fc_entity is None
|
|
else "no_forecast" if fc is None
|
|
else "learning" if est is None else None)
|
|
|
|
# The blocks of each day: today's from Home, tomorrow's by its builders.
|
|
# Tomorrow has not started, so the level now says nothing about it: the
|
|
# builders read it with `ahead` and draw a block wherever the plan and the
|
|
# gate have a run (`homeview.mode_runs()` has the room and energy filter).
|
|
runs = [plan["runs"], hv.runs_for(c, n0, n1, now, shadow, ahead=True)
|
|
if running else []]
|
|
|
|
# The level curve of every battery that is on and known. It also drops a
|
|
# grid block of tomorrow that control will not take: the battery is then
|
|
# at or above its planned level when the run starts.
|
|
notes_b = {}
|
|
levels = {}
|
|
if running and missing is None:
|
|
levels, runs = _levels(c, est, fc, trust, now, n1, runs, floor, ceil,
|
|
notes_b)
|
|
blocks = [[[r for r in rs if r["battery"] == i and r["kind"] in BLOCKS]
|
|
for i in range(len(c.inverters))] for rs in runs]
|
|
# Prices are planned about a day ahead (`priceruns.HORIZON`), so tomorrow's
|
|
# blocks may be missing now and appear later.
|
|
priced = bool(plan.get("has_prices")) and (
|
|
any(c.pack_mode(i).use and c.pack_mode(i).mode in hv.PRICE_MODES
|
|
for i in c.inverters) if modes_on else arbitrage.enabled(c.o))
|
|
days = []
|
|
for k, (a, z) in enumerate(((d0, d1), (n0, n1))):
|
|
days.append(_day(c, k, a, z, now, fc, est, trust, levels, blocks[k],
|
|
rows, shadow, running, key, missing, notes_b, status,
|
|
priced))
|
|
return {"ok": True, "running": running, "shadow": shadow,
|
|
"now_h": round(_hours_between(d0, now), 3),
|
|
"updated": None if fc is None else hv.hhmm(fc.at),
|
|
"days": days}
|
|
|
|
|
|
def _levels(c, est, fc, trust, now, end, runs, floor, ceil, notes_b):
|
|
"""`({name: [(t, pct), ...]}, runs)`: `daycurve.curve()` for the batteries
|
|
that are on, have a level and are not in their monthly cycle, and `runs`
|
|
less the grid blocks of tomorrow that control will not take. Those are
|
|
asked of `modetrade.chargers()`, the cycle's own test, at the level the
|
|
curve has when the block starts (a block is drawn from the floor, see
|
|
`homeview.mode_runs(ahead=True)`, so that level is the filter)."""
|
|
packs = {p.name: p for p in c._mode_packs()}
|
|
targets = c.plan_targets()
|
|
rp = getattr(c, "mode_runs", None)
|
|
plan_on = rp is not None and rp.on
|
|
house_w, base_w = est
|
|
|
|
def make(rs):
|
|
batts = []
|
|
for i, inv in enumerate(c.inverters):
|
|
pm = c.pack_mode(inv)
|
|
if not pm.use:
|
|
continue
|
|
if inv.soc is None or not inv.capacity_kwh:
|
|
notes_b[inv.name] = hv.day_note("no_level", name=inv.name)
|
|
continue
|
|
mine = [r for day in rs for r in day
|
|
if r["battery"] == i and r["kind"] in BLOCKS]
|
|
cyc = [_times(r)[0] for r in mine if r["kind"] == "cycle"]
|
|
if cyc and min(cyc) <= now:
|
|
notes_b[inv.name] = hv.day_note("cycle", name=inv.name)
|
|
continue
|
|
if cyc:
|
|
notes_b[inv.name] = hv.day_note("cycle", name=inv.name)
|
|
to = modetrade.charge_to(packs[inv.name], ceil, targets)
|
|
grid = tuple((*_times(r), to) for r in mine
|
|
if r["kind"] == "grid" and to is not None)
|
|
priced = plan_on and pm.mode != SOLAR_FIRST
|
|
give = tuple((g.start, g.end) for g in rp.give) if priced else ()
|
|
batts.append(Batt(inv.name, float(inv.capacity_kwh),
|
|
inv.max_w / 1000.0, float(inv.soc),
|
|
pm.mode == SOLAR_FIRST, priced, grid, give,
|
|
min(cyc) if cyc else None))
|
|
return batts
|
|
|
|
def sun(a, z):
|
|
return sun_kwh(fc, now, a, z)
|
|
|
|
def use(a, z):
|
|
return sum(max(house_w(h), base_w) * _hours_between(s, e) / 1000.0
|
|
for s, e, h in dayplan._hour_slices(a, z))
|
|
|
|
def run(rs):
|
|
batts = make(rs)
|
|
return curve(batts, now, end, sun, use, floor_pct=floor,
|
|
ceiling_pct=ceil, trust=trust) if batts else {}
|
|
first = run([runs[0], [r for r in runs[1] if r["kind"] != "grid"]])
|
|
keep = []
|
|
for r in runs[1]:
|
|
name = c.inverters[r["battery"]].name
|
|
lvl = _level_at(first.get(name, ()), _times(r)[0])
|
|
if r["kind"] == "grid" and lvl is not None and not modetrade.chargers(
|
|
[dataclasses.replace(packs[name], soc=lvl)], ceil,
|
|
targets=targets):
|
|
continue
|
|
keep.append(r)
|
|
runs = [runs[0], keep]
|
|
return run(runs), runs
|
|
|
|
|
|
def _level_at(pts, t):
|
|
"""The curve's level at `t` (between two points, straight), or None when
|
|
the curve does not reach `t`."""
|
|
for (t0, v0), (t1, v1) in zip(pts, pts[1:]):
|
|
if t0 <= t <= t1:
|
|
span = (t1 - t0).total_seconds()
|
|
return v0 if span <= 0 else v0 + (v1 - v0) * (
|
|
(t - t0).total_seconds() / span)
|
|
return None
|
|
|
|
|
|
def _day(c, k, d0, d1, now, fc, house, trust, levels, blocks, rows, shadow,
|
|
running, key, missing, notes_b, status, priced):
|
|
day = (fc.today, fc.tomorrow)[k] if fc is not None else None
|
|
solar = getattr(c, "solar_today", None)
|
|
names = [inv.name for inv in c.inverters]
|
|
hours = []
|
|
for s, e, h in dayplan._hour_slices(d0, d1):
|
|
f = None
|
|
if fc is not None and e > now and (k == 0 or fc.tomorrow is not None):
|
|
f = sun_kwh(fc, now, max(s, now), e)
|
|
got = solar.kwh(s, e, now) if k == 0 and solar is not None \
|
|
and s < now else None
|
|
use = None if house is None else max(house[0](h), house[1]) \
|
|
* _hours_between(s, e) / 1000.0
|
|
row = {"start_h": round(_hours_between(d0, s), 3),
|
|
"end_h": round(_hours_between(d0, e), 3), "label": hv.hhmm(s),
|
|
"forecast_kwh": None if f is None else round(f, 3),
|
|
"measured_kwh": None if got is None else round(got, 3),
|
|
"house_kwh": None if use is None else round(use, 3),
|
|
"levels": []}
|
|
hours.append((s, row))
|
|
batts = []
|
|
for i, inv in enumerate(c.inverters):
|
|
row = rows[i] if i < len(rows) else {}
|
|
pts = [(t, v) for t, v in levels.get(inv.name, ()) if d0 <= t <= d1]
|
|
mine = blocks[i]
|
|
batts.append({
|
|
"name": inv.name, "capacity_kwh": inv.capacity_kwh,
|
|
"off": row.get("use") is False,
|
|
"levels": [[round(_hours_between(d0, t), 3), round(v, 1)]
|
|
for t, v in pts],
|
|
"blocks": [{x: r[x] for x in ("kind", "from", "to", "start_h",
|
|
"end_h", "what", "why", "amount")}
|
|
for r in mine],
|
|
"line": _line(c, inv, row, mine, running, key, shadow, d0, d1,
|
|
now)})
|
|
at = {t: v for t, v in pts}
|
|
for s, hr in hours:
|
|
hr["levels"].append(None if s not in at else round(at[s], 1))
|
|
has_levels = any(b["levels"] for b in batts)
|
|
fcs = [h["forecast_kwh"] for _s, h in hours]
|
|
notes = []
|
|
if missing is not None:
|
|
notes.append(hv.day_note(missing))
|
|
if not running:
|
|
notes.append(hv.why_words(key))
|
|
if shadow:
|
|
notes.append(hv.why_words("shadow"))
|
|
line = ""
|
|
if fc is not None:
|
|
if k == 0:
|
|
line = hv.line_solar(fc.today.kwh, fc.today.kwh * trust,
|
|
hv.hhmm(fc.at))
|
|
elif fc.tomorrow is not None:
|
|
line = hv.day_note("tomorrow_total",
|
|
kwh=hv._kwh(fc.tomorrow.kwh),
|
|
count=hv._kwh(fc.tomorrow.kwh * trust))
|
|
if day is not None and not day.periods:
|
|
notes.append(hv.day_note("estimate"))
|
|
if k == 1 and fc.tomorrow is None:
|
|
notes.append(hv.day_note("no_tomorrow"))
|
|
if k == 0 and solar is not None and solar.first is not None \
|
|
and solar.day == brussels(now).date() \
|
|
and _hours_between(d0, solar.first) > 0.5:
|
|
notes.append(hv.day_note("measured_from", at=hv.hhmm(solar.first)))
|
|
if has_levels:
|
|
notes.append(hv.day_note("trust", pct=round(trust * 100)))
|
|
if k == 1 and _tomorrow_line(status):
|
|
notes.append(_tomorrow_line(status))
|
|
if k == 1 and running and priced:
|
|
notes.append(hv.day_note("ahead"))
|
|
notes += [notes_b[n] for n in names if n in notes_b]
|
|
return {"key": ("today", "tomorrow")[k], "date": brussels(d0).date()
|
|
.isoformat(),
|
|
"length_h": int(round(_hours_between(d0, d1))),
|
|
"line": line, "notes": list(dict.fromkeys(notes)),
|
|
"sun": {"forecast": any(x is not None for x in fcs),
|
|
"estimate": day is not None and not day.periods,
|
|
"measured": any(h["measured_kwh"] is not None
|
|
for _s, h in hours),
|
|
"house": house is not None},
|
|
"hours": [h for _s, h in hours], "batteries": batts}
|
|
|
|
|
|
def _tomorrow_line(status) -> str:
|
|
"""Home's own line about tomorrow's prices, for the tomorrow tab."""
|
|
return (status.get("plan") or {}).get("tomorrow_line") or ""
|
|
|
|
|
|
def _give_planned(c, d0, d1, now) -> bool:
|
|
"""Does the run plan have a give run in `[d0, d1)` not yet over?"""
|
|
rp = getattr(c, "mode_runs", None)
|
|
return rp is not None and rp.on and any(
|
|
g.end > max(now, d0) and g.start < d1 for g in rp.give)
|
|
|
|
|
|
def _line(c, inv, row, mine, running, key, shadow, d0, d1, now) -> str:
|
|
"""What to say for a battery with no block: the plan's own words. Home
|
|
draws no give block for a battery with no energy to give now, so a give run
|
|
in the plan with no block gets its own sentence."""
|
|
if not running:
|
|
return hv.why_words(key)
|
|
if row.get("use") is False:
|
|
return hv.why_words("off")
|
|
if mine:
|
|
return ""
|
|
if getattr(c, "modes_on", False) and row.get("mode_choice") in \
|
|
hv.PRICE_MODES:
|
|
return hv.day_note("give_later" if _give_planned(c, d0, d1, now)
|
|
else "no_blocks")
|
|
return hv.run_house_what(shadow) + " " + hv.run_house_why(shadow)
|