glenn schrooyenandClaude
995028bc5d
OPT-23: a state-only forecast is read from now for every span (Forecast.between since=), so the plan's per-hour sun adds up to the day's figure, today and tomorrow. Hourly splits are unchanged. The Forecast tab reads spans the same way. OPT-28: a Leftover solar battery buys, from the floor, the sum over the expensive hours of max(0, house - trust x sun), / EFF. Several Leftover solar batteries hold it together by capacity, the ceiling passes the rest on, and what none can hold is said (Plan.uncovered_kwh). Home's grid block, the Batteries line and the mode help (EN + NL) say the level, what it covers, and that Leftover solar buys only what the sun will not give. Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Bdqr7PiaHcBnzPituu2yTC
449 lines
20 KiB
Python
449 lines
20 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, priceruns
|
|
from . import solarfc
|
|
from .daycurve import Batt, curve
|
|
from .packmode import SOLAR_FIRST
|
|
from .price import Uncertainty, brussels, required_spread, uncertainty_of
|
|
|
|
BLOCKS = ("grid", "give", "cycle") # the kinds a battery's blocks carry
|
|
LEVEL_DP = 1 # the tab shows a level to 0.1 %
|
|
|
|
|
|
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.
|
|
Every span is measured from `now` (`Forecast.between(since=)`, OPT-23),
|
|
as `dayplan.plan()` measures them: a state-only "left today" figure is
|
|
the sun from the moment it was taken, so a span that starts later must
|
|
not be spread over what is left of the bell from its own start
|
|
(DECISIONS/UX-54-forecast-tab.md)."""
|
|
return fc.between(a, b, since=now)
|
|
|
|
|
|
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 each block at the level it starts at (`ahead`), with
|
|
# their own room and energy filter. With modes on that level is the
|
|
# curve's (`_levels()`); with modes off there is no curve, and it is
|
|
# Home's own count (`_counted()`).
|
|
runs = [plan["runs"], []]
|
|
if running and not modes_on:
|
|
runs[1] = _off_tomorrow(c, runs[0], now, shadow, n0, n1)
|
|
elif running:
|
|
runs[1] = hv.runs_for(c, n0, n1, now, shadow, ahead=True)
|
|
|
|
# The level curve of every battery that is on and known. It also makes
|
|
# tomorrow's blocks again at the level each has when its run starts, and
|
|
# draws a grid block as long as control keeps charging (`_levels()`).
|
|
notes_b = {}
|
|
levels = {}
|
|
if running and missing is None:
|
|
levels, runs = _levels(c, est, fc, trust, now, n1, runs, floor, ceil,
|
|
notes_b, shadow, n0, n1)
|
|
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, shadow,
|
|
n0, n1):
|
|
"""`({name: [(t, pct), ...]}, runs)`: `daycurve.curve()` for the batteries
|
|
that are on, have a level and are not in their monthly cycle, and `runs`
|
|
with tomorrow's blocks made as control will meet them.
|
|
|
|
Tomorrow's blocks are `homeview.mode_runs()`'s, read at the level the
|
|
curve has when each block starts (its `ahead` function), so its own room
|
|
and energy filter and `MIN_KWH` decide: no give block where the battery
|
|
is at the floor when the run starts, no grid block where it is at its
|
|
planned level. The curve in turn follows the blocks, so the two are taken
|
|
to a fixed point.
|
|
|
|
A cheap run is sized for all the batteries charging together, so a slower
|
|
one is still below `modetrade.charge_to()` when it ends, and control then
|
|
plans the next cheap slots (`Controller._price_runs()`). That is asked of
|
|
the same functions, `modetrade.energy()` and `priceruns.plan()`, with the
|
|
packs at the curve's level, and the block is drawn to where control
|
|
stops charging. Today's blocks are Home's, as they are."""
|
|
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 {}
|
|
|
|
def settle(ext, lv):
|
|
"""Tomorrow's blocks read at the levels of `lv`, and the curve they
|
|
make, again until the blocks stop changing."""
|
|
seen = None
|
|
for _ in range(4):
|
|
def at(name, _kind, start, lv=lv):
|
|
return _level_at(lv.get(name, ()), start)
|
|
rt = hv.runs_for(c, n0, n1, now, shadow, ahead=at, rp=ext)
|
|
lv = run([runs[0], rt])
|
|
sig = [(r["battery"], r["kind"], r["start"], r["end"])
|
|
for r in rt]
|
|
if sig == seen:
|
|
break
|
|
seen = sig
|
|
return rt, lv
|
|
|
|
lv0 = run([runs[0], []])
|
|
ext = rp
|
|
plan = c.fresh_tariff_plan(now)
|
|
cheap = None if rp is None else rp.cheap
|
|
end_t = None
|
|
if plan_on and cheap is not None and rp.charge_useful() \
|
|
and n0 <= cheap.start < n1 and plan is not None:
|
|
required = required_spread(
|
|
arbitrage.threshold_eur_kwh(c.o),
|
|
uncertainty_of(plan.read.product) | {Uncertainty.NO_WEATHER})
|
|
end_t = cheap.end
|
|
for _ in range(24):
|
|
ext = dataclasses.replace(rp, cheap=priceruns.Run(
|
|
cheap.start, end_t, cheap.mean_eur_kwh))
|
|
rt, lv = settle(ext, lv0)
|
|
at_end = [dataclasses.replace(packs[n], soc=v)
|
|
for n, v in ((n, _level_at(lv.get(n, ()), end_t))
|
|
for n in lv) if v is not None]
|
|
e = modetrade.energy(at_end, floor, ceil, targets=targets)
|
|
if e[0] <= 1e-6:
|
|
break
|
|
new = priceruns.plan(
|
|
plan.read.points, end_t, required_eur_kwh=required,
|
|
charge_kwh=e[0], charge_kw=e[1], sell_kwh=0.0, sell_kw=0.0)
|
|
nxt = new.cheap
|
|
if nxt is None or not new.charge_useful() or nxt.start > end_t \
|
|
or nxt.end <= end_t:
|
|
break
|
|
end_t = nxt.end
|
|
ext = dataclasses.replace(rp, cheap=priceruns.Run(
|
|
cheap.start, end_t, cheap.mean_eur_kwh))
|
|
rt, lv = settle(ext, lv0)
|
|
# A run is for all the batteries charging together; each stops when it is
|
|
# at its own level (control's `chargers()`), so its block ends there.
|
|
names = [inv.name for inv in c.inverters]
|
|
rt = [_stops(r, names, lv, packs, targets, floor, ceil, n0) for r in rt]
|
|
return lv, [runs[0], rt]
|
|
|
|
|
|
def _off_tomorrow(c, today, now, shadow, n0, n1):
|
|
"""Modes off: tomorrow's blocks by `homeview.trade_runs()`, each read at
|
|
the level `_counted()` gives when it starts. A charge may come before a
|
|
give, so this repeats until the blocks stop changing."""
|
|
sig, rt = None, []
|
|
for _ in range(4):
|
|
rt = hv.runs_for(c, n0, n1, now, shadow,
|
|
ahead=_counted(c, [today, rt]))
|
|
new = [(r["battery"], r["kind"], r["start"], r["end"]) for r in rt]
|
|
if new == sig:
|
|
break
|
|
sig = new
|
|
return rt
|
|
|
|
|
|
def _counted(c, rs):
|
|
"""Modes off reads no forecast and keeps no house profile, so there is no
|
|
curve. A give is read at Home's own count (`trade_runs()`): the level now,
|
|
plus what the blocks in `rs` that start before it put in, less what they
|
|
take out. The house can only take from that and no sun is read, so a
|
|
battery at the floor by the count is at the floor then. A charge is read
|
|
from the floor (None): the house uses some of a full battery before the
|
|
cheap hour, and control charges it back then."""
|
|
_floor, ceil = arbitrage.soc_bounds(c.o)
|
|
|
|
def at(name, kind, start):
|
|
i = next(k for k, inv in enumerate(c.inverters) if inv.name == name)
|
|
inv = c.inverters[i]
|
|
if kind == "grid" or inv.soc is None or not inv.capacity_kwh:
|
|
return None
|
|
v = float(inv.soc)
|
|
mine = sorted((r for day in rs for r in day if r["battery"] == i
|
|
and r["kind"] in ("grid", "give")
|
|
and r.get("kwh") is not None
|
|
and _times(r)[0] < start),
|
|
key=lambda r: _times(r)[0])
|
|
for r in mine:
|
|
d = float(r["kwh"]) / float(inv.capacity_kwh) * 100.0
|
|
v = min(max(v, ceil), v + d) if r["kind"] == "grid" else v - d
|
|
return v
|
|
return at
|
|
|
|
|
|
def _stops(r, names, lv, packs, targets, floor, ceil, day0):
|
|
"""`r` ended where its battery stops, if that is before the run ends: a
|
|
grid block where the curve has it at the level `modetrade.charge_to()`
|
|
allows (the curve has a point there), a give block at the first point
|
|
where the level the tab shows (`LEVEL_DP`) is at the floor. A run is for
|
|
all the batteries together; each leaves it on its own."""
|
|
if r["kind"] not in ("grid", "give"):
|
|
return r
|
|
name = names[r["battery"]]
|
|
start, end = _times(r)
|
|
if r["kind"] == "grid":
|
|
to = modetrade.charge_to(packs[name], ceil, targets)
|
|
if to is None:
|
|
return r
|
|
hit = lambda v: v >= to - 1e-6 # noqa: E731
|
|
else:
|
|
hit = lambda v: round(v, LEVEL_DP) <= floor # noqa: E731
|
|
done = next((t for t, v in lv.get(name, ()) if start < t <= end
|
|
and hit(v)), None)
|
|
if done is None or done >= end:
|
|
return r
|
|
return dict(r, end=done.isoformat(), to=hv.hhmm(done),
|
|
end_h=round(_hours_between(day0, done), 3))
|
|
|
|
|
|
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, LEVEL_DP)]
|
|
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], LEVEL_DP))
|
|
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", at=_sun_from()))
|
|
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 _sun_from() -> str:
|
|
"""When the sun of a day starts for `solarfc` (its bell's first hour):
|
|
with no forecast for tomorrow the sun is unknown from then, and so the
|
|
curve stops there (`Forecast.between()` answers None past it)."""
|
|
return f"{int(solarfc.BELL_FROM_H):02d}:00"
|
|
|
|
|
|
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 add-on plan a give in `[d0, d1)` not yet over? Modes on: a
|
|
give run of the run plan. Modes off: a give leg of the verdict that the
|
|
gate takes (`homeview.trade_legs()`), at a price that is not negative."""
|
|
if not getattr(c, "modes_on", False):
|
|
plan = c.fresh_tariff_plan(now)
|
|
verdict, _why = hv.trade_verdict(c, plan, now)
|
|
return verdict is not None and any(
|
|
k == "give" and pt.eur_per_kwh >= 0 and pt.start < d1
|
|
and hv._point_end(plan, pt) > max(now, d0)
|
|
for k, pt in hv.trade_legs(c, plan, verdict, now))
|
|
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, so a give that
|
|
the add-on plans 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")
|
|
modes_on = getattr(c, "modes_on", False)
|
|
gives = not modes_on or row.get("mode_choice") in hv.PRICE_MODES
|
|
if gives and not any(r["kind"] == "give" for r in mine) \
|
|
and _give_planned(c, d0, d1, now):
|
|
return hv.day_note("give_later")
|
|
if mine:
|
|
return ""
|
|
if modes_on and gives:
|
|
return hv.day_note("no_blocks")
|
|
return hv.run_house_what(shadow) + " " + hv.run_house_why(shadow)
|