glenn schrooyenandClaude
5ab8622265
- Tomorrow's grid block is drawn to where control stops charging. A cheap run is sized for all batteries together; a slower one is still below `charge_to()` when it ends and control plans the next cheap slots. The tab asks `modetrade.energy()` and `priceruns.plan()` with the packs at the curve's level and extends the run, so the curve charges as long. - Each battery's block ends where it is at its own level (grid) or at the floor (give). The curve puts a point where a battery reaches its charge level inside a step. - `homeview.mode_runs(ahead=f, rp=...)` reads each block at the level the curve has when it starts, with Home's own filter and MIN_KWH: no give block at the floor, no grid block at the planned level. Home never passes either. - `no_tomorrow` names the hour from `solarfc.BELL_FROM_H`, where the curve stops. - test_ux54.py part 7 compares control (`_mode_roles()` every 15 minutes) with the tab with no tolerance, and adds the cloudy / floor case. No decision, command, timing or option key changes. Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Bdqr7PiaHcBnzPituu2yTC
388 lines
17 KiB
Python
388 lines
17 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
|
|
|
|
|
|
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 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 _stops(r, names, lv, packs, targets, floor, ceil, day0):
|
|
"""`r` ended where its battery stops: a grid block where the curve has it
|
|
at the level `modetrade.charge_to()` allows (the curve has a point there),
|
|
a give block where the curve has it at the floor, within the 0.1 % the
|
|
tab shows (a step end), if that is
|
|
before the run ends. 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: v <= floor + 0.05 # 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, 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", 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 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)
|