Claude
eb10537aee
A Home Assistant solar forecast (Forecast.Solar, Solcast or Open-Meteo
shapes) and the add-on's own hourly house use decide, before each cheap
run, whether the sun will fill each battery before the expensive hours.
Leftover solar then buys from the grid only the shortfall; Cheap hours only
skips the part the sun covers; Solar first never grid-charges. No, stale or
failed forecast: P3 option A exactly (12 digests pinned from 5a4cdc2).
Replay (16 recorded days, forecast errors from Elia's day-ahead BE solar):
year-earlier prices P4 19.15 vs P2 19.33 / option A 19.44 / old trade 18.14;
real 2026 prices P4 29.19 vs 32.26 / 32.01 / 31.40 EUR net.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
257 lines
10 KiB
Python
257 lines
10 KiB
Python
"""OPT-11 P4: the day plan. Will the sun fill each battery before the
|
|
expensive hours? If not, how much should it take from the grid?
|
|
|
|
Pure: the forecast (`solarfc`), the house profile, the price runs and the
|
|
batteries in; a level per battery (`Plan.targets`) out. `modetrade.charge_to()`
|
|
turns the level into permission to charge from the grid; main.py rebuilds the
|
|
plan with the price runs (at most every 15 minutes) and keeps a level unless
|
|
it moves by `HYST_KWH` or `HYST_FRAC` (`hold()`).
|
|
|
|
The sums (OPT-11-design section 1, "Excess solar", with the owner's answers):
|
|
* `sun`: trust x the solar the house will not use, between now and the
|
|
expensive run the cheap run is for, x the charge efficiency:
|
|
`trust x sum_h max(0, forecast_h - house_h) x EFF`.
|
|
* Solar first batteries take it first, up to the ceiling. What is left is
|
|
shared by the Leftover solar and Cheap hours batteries by their room.
|
|
* A Cheap hours battery charges from the grid up to `ceiling - its share`.
|
|
* A Leftover solar battery charges only up to what the house will draw from
|
|
it in the expensive runs before the next cheap run, less its share:
|
|
`min(ceiling, floor + house use / EFF) - share`. So it buys only the
|
|
shortfall, not up to the ceiling.
|
|
* Solar first never charges from the grid (owner, 2026-09-29).
|
|
|
|
The house profile is the add-on's own measurement, grid + battery power, per
|
|
local hour, over the last `PROFILE_DAYS` days; the estimate for an hour is the
|
|
median of that hour. It is net of the solar the house used itself, so for
|
|
the solar sum it is never taken below the night-time median (`base()`).
|
|
The site's own solar readings are not used: TEL-11 keeps them out of every
|
|
control decision.
|
|
"""
|
|
|
|
import statistics
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from .packmode import CHEAP_HOURS, LEFTOVER_SOLAR, SOLAR_FIRST
|
|
from .tariff import BILLING_TZ
|
|
|
|
EFF = 0.95 # charge (and discharge) efficiency
|
|
HYST_KWH = 1.0 # a kept level moves only by this much...
|
|
HYST_FRAC = 0.10 # ...or this share of the battery's shortfall
|
|
PROFILE_DAYS = 7
|
|
MIN_HOURS = 20 # hours of a day recorded before it counts
|
|
NIGHT_HOURS = (0, 1, 2, 3, 4, 5, 22, 23)
|
|
DEFAULT_TRUST_PCT = 70
|
|
|
|
# Site verdicts (the page's forecast line).
|
|
SUNNY, PARTLY, DARK = "sunny", "partly", "dark"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# The house profile
|
|
# --------------------------------------------------------------------------- #
|
|
def _local(t: datetime):
|
|
return t.astimezone(BILLING_TZ)
|
|
|
|
|
|
class HouseProfile:
|
|
"""Hourly means of the house's net use (grid + battery power), kept per
|
|
local day and hour: `{"YYYY-MM-DD": {hour: [mean_w, samples]}}`. The
|
|
store holds `dump()`; `load()` takes it back and drops what is bad."""
|
|
|
|
def __init__(self, data=None):
|
|
self.days = {}
|
|
self._cur = None # (day, hour, sum, n)
|
|
if isinstance(data, dict):
|
|
self.load(data)
|
|
|
|
def load(self, data):
|
|
for day, hours in data.items():
|
|
if not isinstance(hours, dict):
|
|
continue
|
|
row = {}
|
|
for h, v in hours.items():
|
|
try:
|
|
hi, w = int(h), float(v)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if 0 <= hi < 24 and abs(w) < 1e6:
|
|
row[hi] = w
|
|
if row:
|
|
self.days[str(day)] = row
|
|
|
|
def dump(self) -> dict:
|
|
return {d: {str(h): round(w, 1) for h, w in sorted(r.items())}
|
|
for d, r in sorted(self.days.items())}
|
|
|
|
def add(self, now: datetime, house_w) -> bool:
|
|
"""One reading. Returns True when an hour closed (then store it)."""
|
|
lt = _local(now)
|
|
key = (lt.strftime("%Y-%m-%d"), lt.hour)
|
|
closed = False
|
|
if self._cur is not None and self._cur[:2] != key:
|
|
day, hour, s, n = self._cur
|
|
if n >= 60: # about 5 minutes of readings
|
|
self.days.setdefault(day, {})[hour] = s / n
|
|
closed = True
|
|
self._cur = None
|
|
self._prune(lt)
|
|
if house_w is None:
|
|
return closed
|
|
if self._cur is None:
|
|
self._cur = (key[0], key[1], 0.0, 0)
|
|
d, h, s, n = self._cur
|
|
self._cur = (d, h, s + float(house_w), n + 1)
|
|
return closed
|
|
|
|
def _prune(self, lt):
|
|
keep = (lt.date() - timedelta(days=PROFILE_DAYS)).isoformat()
|
|
for d in [d for d in self.days if d < keep]:
|
|
del self.days[d]
|
|
|
|
def ready(self) -> bool:
|
|
"""At least one day with `MIN_HOURS` hours recorded."""
|
|
return any(len(r) >= MIN_HOURS for r in self.days.values())
|
|
|
|
def hour_w(self, hour: int):
|
|
vals = [r[hour] for r in self.days.values() if hour in r]
|
|
return statistics.median(vals) if vals else None
|
|
|
|
def base(self):
|
|
vals = [w for h in NIGHT_HOURS if (w := self.hour_w(h)) is not None]
|
|
return statistics.median(vals) if vals else None
|
|
|
|
def estimator(self):
|
|
"""`f(hour) -> W`, or None when not ready. Fixed at call time."""
|
|
if not self.ready():
|
|
return None
|
|
base = self.base() or 0.0
|
|
table = {}
|
|
for h in range(24):
|
|
w = self.hour_w(h)
|
|
table[h] = base if w is None else w
|
|
return lambda h: table[h], max(0.0, base)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# The plan
|
|
# --------------------------------------------------------------------------- #
|
|
@dataclass(frozen=True)
|
|
class Plan:
|
|
verdict: str # SUNNY / PARTLY / DARK
|
|
sun_kwh: float # solar the batteries can count on
|
|
fc_kwh: float # the forecast over the same span
|
|
until: datetime # the expensive run the plan is for
|
|
targets: dict # name -> level % (LS and CH only)
|
|
need_kwh: dict # name -> kWh from the grid (0: none)
|
|
share_kwh: dict # name -> kWh of solar counted for it
|
|
|
|
|
|
def _hour_slices(t0, t1):
|
|
"""Local-hour slices of [t0, t1): (start, end, local hour)."""
|
|
out, t = [], t0
|
|
while t < t1:
|
|
lt = _local(t)
|
|
nxt = (lt.replace(minute=0, second=0, microsecond=0)
|
|
+ timedelta(hours=1)).astimezone(timezone.utc)
|
|
e = min(nxt, t1)
|
|
out.append((t, e, lt.hour))
|
|
t = e
|
|
return out
|
|
|
|
|
|
def _room(p, to_pct):
|
|
if p.capacity_kwh is None or p.soc is None:
|
|
return 0.0
|
|
return max(0.0, float(to_pct) - float(p.soc)) / 100.0 * float(
|
|
p.capacity_kwh)
|
|
|
|
|
|
def plan(forecast, house, runs, now, packs, *, trust_pct, floor_pct,
|
|
ceiling_pct):
|
|
"""The day plan, or None when there is none to make: no forecast, no
|
|
house profile yet, no price plan, or no expensive run after the cheap
|
|
run. `house` is `HouseProfile.estimator()`'s answer. `packs` are
|
|
`modetrade.Pack`s."""
|
|
if forecast is None or house is None or runs is None or not runs.on \
|
|
or runs.cheap is None:
|
|
return None
|
|
house_w, base_w = house
|
|
until = next((g for g in runs.give if g.start >= runs.cheap.end), None)
|
|
if until is None:
|
|
return None
|
|
trust = max(0.0, min(1.0, float(trust_pct) / 100.0))
|
|
sun = fc_sum = 0.0
|
|
for s, e, h in _hour_slices(now, until.start):
|
|
pv = forecast.between(s, e)
|
|
if pv is None:
|
|
return None # the span reaches a day with no forecast
|
|
hrs = (e - s).total_seconds() / 3600.0
|
|
use = max(house_w(h), base_w) * hrs / 1000.0
|
|
sun += max(0.0, pv - use)
|
|
fc_sum += pv
|
|
sun *= trust * EFF
|
|
|
|
usable = [p for p in packs if p.use and p.live and p.idle
|
|
and p.soc is not None and p.capacity_kwh]
|
|
sf = [p for p in usable if p.mode == SOLAR_FIRST]
|
|
price = [p for p in usable if p.mode in (LEFTOVER_SOLAR, CHEAP_HOURS)]
|
|
sf_room = sum(_room(p, ceiling_pct) for p in sf)
|
|
excess = max(0.0, sun - sf_room)
|
|
rooms = {p.name: _room(p, ceiling_pct) for p in price}
|
|
total_room = sum(rooms.values())
|
|
|
|
# What the house draws in the expensive runs from the one this plan is
|
|
# for until the next cheap run (or a day), from the profile.
|
|
horizon = until.start + timedelta(hours=24)
|
|
use_kwh = 0.0
|
|
for g in runs.give:
|
|
if g.start < until.start or g.start >= horizon:
|
|
continue
|
|
for s, e, h in _hour_slices(g.start, g.end):
|
|
use_kwh += max(0.0, house_w(h)) * (
|
|
e - s).total_seconds() / 3600.0 / 1000.0
|
|
caps = {p.name: float(p.capacity_kwh) for p in price}
|
|
cap_sum = sum(caps.values()) or 1.0
|
|
|
|
targets, need, share = {}, {}, {}
|
|
for p in price:
|
|
sh = excess * rooms[p.name] / total_room if total_room > 0 else 0.0
|
|
sh_pct = sh / caps[p.name] * 100.0
|
|
if p.mode == CHEAP_HOURS:
|
|
top = float(ceiling_pct)
|
|
else:
|
|
want = use_kwh * caps[p.name] / cap_sum / EFF
|
|
top = min(float(ceiling_pct),
|
|
float(floor_pct) + want / caps[p.name] * 100.0)
|
|
level = max(0.0, top - sh_pct)
|
|
targets[p.name] = round(level, 2)
|
|
need[p.name] = round(_room(p, level), 3)
|
|
share[p.name] = round(sh, 3)
|
|
if sun < sf_room:
|
|
verdict = DARK
|
|
elif any(v > 0.05 for v in need.values()):
|
|
verdict = PARTLY
|
|
else:
|
|
verdict = SUNNY
|
|
return Plan(verdict, round(sun, 3), round(fc_sum, 3), until.start,
|
|
targets, need, share)
|
|
|
|
|
|
def hold(old: dict | None, new: dict, capacity: dict) -> dict:
|
|
"""Hysteresis: a battery keeps its old level unless the new one differs
|
|
by at least `HYST_KWH`, or by `HYST_FRAC` of the larger level's energy
|
|
(whichever is less). A battery new to the plan takes its level at once;
|
|
one that left the plan is dropped at once."""
|
|
out = {}
|
|
for name, level in new.items():
|
|
was = (old or {}).get(name)
|
|
cap = float(capacity.get(name) or 0.0)
|
|
if was is None or cap <= 0:
|
|
out[name] = level
|
|
continue
|
|
moved = abs(level - was) / 100.0 * cap
|
|
step = min(HYST_KWH, HYST_FRAC * max(was, level) / 100.0 * cap)
|
|
out[name] = level if moved >= step else was
|
|
return out
|