glenn schrooyenandClaude
eb9ccc2730
Review F1: a slot after B's run counts only if no slot from the expensive
hours on is cheaper (control plans B's charge left in the cheapest window
ahead), and x 0.95 of B's charge is stored. Three-price night: 1ac7fe9 left
the expensive hours 2.61 kWh short; now 0.
Review F2: a Solar first battery's drain until the expensive hours reads each
hour at no less than the night base (the profile is net of the sun). Read net,
a dark morning drained the battery the plan still counted; A's level fell,
the cheap run moved and was lost before it started (+0.40 EUR per 3 days).
The two real-price replays now cost what 0.20.0 does.
F3: the ux54 sweep's peak limit is 15 kW (the option's top). F4: the rest
comes from the grid (EN/NL). F5: CHANGELOG and DECISIONS wording.
Co-Authored-By: Claude <noreply@anthropic.com>
432 lines
19 KiB
Python
432 lines
19 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;
|
|
OPT-28 for the second):
|
|
* `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 batteries by their room.
|
|
* `short` (OPT-28): what the house draws in the expensive run that the sun
|
|
will not give, `sum_h max(0, house_h - trust x forecast_h)` over every
|
|
hour of the run, early morning and evening. The Leftover solar batteries
|
|
hold it together: `short / EFF` above the floor, shared by capacity, and
|
|
a battery that reaches the ceiling passes the rest to the others. What
|
|
all of them cannot hold (`Plan.uncovered_kwh`) is not bought: another
|
|
battery (Cheap hours, Solar first) or the grid carries it.
|
|
* Cheap hours only ignores the plan and fills to the ceiling, as in P3 and
|
|
the design (section 1). OPT-36: what it, and a Solar first battery after
|
|
the night's use, will give in the expensive hours is taken off `short`
|
|
before the Leftover solar batteries share it (`_cheap_give_kwh()`,
|
|
`_solar_first_give_kwh()`). Skipping the sun's share cost money in the replay
|
|
(PR #156 review: 29.72 against 29.41 EUR on real 2026 prices).
|
|
* A Leftover solar battery charges from the grid up to the level it needs
|
|
for its part of `short`, less its share of the sun before the run:
|
|
`min(ceiling, floor + its part / EFF) - share`. So it buys only the
|
|
shortfall, not up to the ceiling.
|
|
* Solar first never charges from the grid (owner, 2026-09-29).
|
|
* A state-only forecast is read from now for every span (`Forecast.between`
|
|
`since=`, OPT-23), so the hours add up to the day's figure.
|
|
DECISIONS/OPT-23-28-the-night-charge-covers-the-dear-hours.md has the record.
|
|
|
|
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, field
|
|
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():
|
|
try:
|
|
datetime.strptime(str(day), "%Y-%m-%d")
|
|
except ValueError:
|
|
continue
|
|
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 the last `PROFILE_DAYS` days; a day after today (the clock
|
|
was wrong once) would never age out, so it goes too."""
|
|
keep = (lt.date() - timedelta(days=PROFILE_DAYS)).isoformat()
|
|
today = lt.date().isoformat()
|
|
for d in [d for d in self.days if d < keep or d > today]:
|
|
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 % (Leftover solar only)
|
|
need_kwh: dict # name -> kWh from the grid (0: none)
|
|
share_kwh: dict # name -> kWh of solar counted for it
|
|
# OPT-28: the expensive hours' shortfall, the part of it each battery
|
|
# covers, and what no battery can hold. All kWh on the house's side.
|
|
short_kwh: float = 0.0
|
|
cover_kwh: dict = field(default_factory=dict)
|
|
uncovered_kwh: float = 0.0
|
|
# OPT-36: what Cheap hours batteries will give in those hours (kWh on the
|
|
# house's side), and, per Leftover solar battery, the grid kWh it buys
|
|
# less because of it.
|
|
cheap_kwh: float = 0.0
|
|
saved_kwh: dict = field(default_factory=dict)
|
|
# ...and the same for Solar first batteries (owner, 2026-11-02): their
|
|
# level when the expensive hours start, after the night's use.
|
|
solar_first_kwh: float = 0.0
|
|
|
|
|
|
def _hour_slices(t0, t1):
|
|
"""Local-hour slices of [t0, t1): (start, end, local hour).
|
|
|
|
Steps in UTC, so `t` always advances (SAFETY-123, DECISIONS)."""
|
|
out, t = [], t0
|
|
while t < t1:
|
|
lt = _local(t)
|
|
left = timedelta(minutes=60 - lt.minute) - timedelta(
|
|
seconds=lt.second, microseconds=lt.microsecond)
|
|
e = min(t.astimezone(timezone.utc) + left, t1)
|
|
out.append((t, e, lt.hour))
|
|
t = e
|
|
return out
|
|
|
|
|
|
def _spread(amount, room, weight):
|
|
"""Share `amount` over the keys of `room` by `weight`, none past its
|
|
room, what one cannot take passed to the others. Returns `(taken,
|
|
left)`: what each took and what none could."""
|
|
got = {k: 0.0 for k in room}
|
|
left = float(amount)
|
|
live = {k for k, r in room.items() if r > 1e-9 and weight.get(k, 0) > 0}
|
|
while left > 1e-9 and live:
|
|
total = sum(weight[k] for k in live)
|
|
used, full = 0.0, set()
|
|
for k in sorted(live):
|
|
part = left * weight[k] / total
|
|
free = room[k] - got[k]
|
|
if part >= free - 1e-9:
|
|
part = free
|
|
full.add(k)
|
|
got[k] += part
|
|
used += part
|
|
left -= used
|
|
if not full:
|
|
break
|
|
live -= full
|
|
return got, max(0.0, left)
|
|
|
|
|
|
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 _cheap_give_kwh(runs, now, house, packs, *, floor_pct, ceiling_pct,
|
|
aim_w, until=None, slots=None, spread=None):
|
|
"""OPT-36: what the Cheap hours batteries in `packs` give in the
|
|
expensive hours, in kWh on the house's side: (the level at the start of
|
|
those hours - the floor) x capacity x `EFF`.
|
|
|
|
That level is the one its cheap charging reaches: the ceiling, or less
|
|
when the time left cannot get there. That time is the cheap run from now
|
|
(or the whole run ahead of it) and, given the firm price `slots`, each
|
|
later slot before the expensive hours (`until`, a give run) still worth
|
|
charging in (its price plus the spread control asks, `spread`, at most
|
|
the give run's dearest slot, `priceruns.plan()`'s test) and no dearer
|
|
than every slot from the start of those hours on: control plans the
|
|
charge left in the cheapest window ahead, so a slot with a cheaper one
|
|
after it may never be used (round 4, a three-price night). The run is
|
|
sized at the batteries' ratings (`priceruns.need_hours()`), not the peak
|
|
limit, so a run too short to fill them is planned again in that time;
|
|
counting the run alone let A's level fall in the night after A had
|
|
bought. In that time the batteries together charge at their ratings,
|
|
held to what the peak aim leaves above the house's use that hour
|
|
(`aim_w`, None: no limit), shared by rating; x `EFF` of it is stored.
|
|
No cheap run worth charging in: the level now.
|
|
A battery that refuses a direction counts 0."""
|
|
ps = [p for p in packs if p.mode == CHEAP_HOURS and p.refused == 0]
|
|
if not ps:
|
|
return 0.0
|
|
house_w, base_w = house
|
|
gain = {p.name: 0.0 for p in ps}
|
|
rate = {p.name: abs(float(p.rating_w)) for p in ps}
|
|
total = sum(rate.values())
|
|
run = runs.cheap
|
|
spans = [(max(now, run.start), run.end)]
|
|
top = max((eur for s, _e, eur in slots or ()
|
|
if until is not None and until.start <= s < until.end),
|
|
default=None)
|
|
low = min((eur for s, _e, eur in slots or ()
|
|
if until is not None and s >= until.start), default=None)
|
|
if top is not None and spread is not None:
|
|
spans += [(max(s, now, run.end), min(e, until.start))
|
|
for s, e, eur in slots
|
|
if eur + float(spread) <= top + 1e-9
|
|
and (low is None or eur <= low + 1e-9)
|
|
and e > run.end and s < until.start]
|
|
if runs.charge_useful() and total > 0.0:
|
|
for t0, t1 in spans:
|
|
for s, e, h in _hour_slices(t0, t1):
|
|
w = total
|
|
if aim_w is not None:
|
|
w = min(w, max(0.0, float(aim_w)
|
|
- max(house_w(h), base_w, 0.0)))
|
|
hrs = (e - s).total_seconds() / 3600.0
|
|
for p in ps:
|
|
gain[p.name] += w * rate[p.name] / total * hrs \
|
|
/ 1000.0 * EFF
|
|
out = 0.0
|
|
for p in ps:
|
|
cap = float(p.capacity_kwh)
|
|
lvl = max(float(p.soc), min(float(ceiling_pct),
|
|
float(p.soc) + gain[p.name] / cap * 100.0))
|
|
out += max(0.0, lvl - float(floor_pct)) / 100.0 * cap * EFF
|
|
return out
|
|
|
|
|
|
def _solar_first_give_kwh(house, now, start, packs, *, floor_pct):
|
|
"""OPT-36 (owner, 2026-11-02): what the Solar first batteries in `packs`
|
|
will still give when the expensive hours start at `start`, in kWh on the
|
|
house's side. They give at any hour, so the house's use from now to
|
|
`start` is taken off first, shared by the energy each holds above the
|
|
floor, and what is left x `EFF` counts. Each hour is read at no less than
|
|
the night base: the profile is net of the sun the house used, and read
|
|
net it let a dark day's morning drain a battery that the plan still
|
|
counted (round 4, real-price replays). A battery
|
|
that runs down to the floor before then counts 0; one that refuses a
|
|
direction counts 0."""
|
|
ps = [p for p in packs if p.mode == SOLAR_FIRST and p.refused == 0]
|
|
if not ps:
|
|
return 0.0
|
|
house_w, base_w = house
|
|
use = sum(max(0.0, house_w(h), base_w) * (e - s).total_seconds() / 3.6e6
|
|
for s, e, h in _hour_slices(now, start))
|
|
have = {p.name: max(0.0, float(p.soc) - float(floor_pct)) / 100.0
|
|
* float(p.capacity_kwh) for p in ps}
|
|
took, _ = _spread(use / EFF, have, have)
|
|
return sum(have[n] - took[n] for n in have) * EFF
|
|
|
|
|
|
def plan(forecast, house, runs, now, packs, *, trust_pct, floor_pct,
|
|
ceiling_pct, aim_w=None, slots=None, spread=None):
|
|
"""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. `aim_w`, `slots` (`priceruns.firm_slots()`) and
|
|
`spread` (the price gap control asks, EUR/kWh) size a Cheap hours
|
|
battery's charge (`_cheap_give_kwh()`)."""
|
|
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, since=now)
|
|
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 == LEFTOVER_SOLAR]
|
|
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())
|
|
|
|
# OPT-28: what the house draws in the expensive runs, from the one this
|
|
# plan is for until the next cheap run (or a day), that the sun will not
|
|
# give. An hour with sun is read at no less than the night base, as the
|
|
# sun sum above reads it (the profile is net of the sun the house used);
|
|
# an hour without is read as measured. A run hour with no forecast counts
|
|
# no sun.
|
|
horizon = until.start + timedelta(hours=24)
|
|
short = 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):
|
|
pv = forecast.between(s, e, since=now) or 0.0
|
|
w = max(0.0, house_w(h))
|
|
if pv > 0.0:
|
|
w = max(w, base_w)
|
|
short += max(0.0, w * (e - s).total_seconds() / 3.6e6
|
|
- trust * pv)
|
|
caps = {p.name: float(p.capacity_kwh) for p in price}
|
|
span = max(0.0, float(ceiling_pct) - float(floor_pct))
|
|
# OPT-36: the Cheap hours and Solar first batteries give first;
|
|
# Leftover solar holds the rest. A Leftover solar battery's own stored energy counts through its
|
|
# level: it buys from where it is to the level it needs.
|
|
cheap = _cheap_give_kwh(runs, now, house, usable, floor_pct=floor_pct,
|
|
ceiling_pct=ceiling_pct, aim_w=aim_w,
|
|
until=until, slots=slots, spread=spread) \
|
|
if price else 0.0
|
|
cheap = min(cheap, short)
|
|
sfirst = _solar_first_give_kwh(house, now, until.start, usable,
|
|
floor_pct=floor_pct) if price else 0.0
|
|
sfirst = min(sfirst, short - cheap)
|
|
others = cheap + sfirst
|
|
room_of = {n: span / 100.0 * c for n, c in caps.items()}
|
|
held, left = _spread((short - others) / EFF, room_of, caps)
|
|
held0 = _spread(short / EFF, room_of, caps)[0] if others > 0.0 else held
|
|
|
|
targets, need, share, cover, saved = {}, {}, {}, {}, {}
|
|
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
|
|
top = float(floor_pct) + held[p.name] / 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)
|
|
top0 = float(floor_pct) + held0[p.name] / caps[p.name] * 100.0
|
|
saved[p.name] = round(
|
|
_room(p, max(0.0, top0 - sh_pct)) - need[p.name], 3)
|
|
share[p.name] = round(sh, 3)
|
|
cover[p.name] = round(held[p.name] * EFF, 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, round(short, 3), cover,
|
|
round(left * EFF if price else 0.0, 3), round(cheap, 3),
|
|
saved, round(sfirst, 3))
|
|
|
|
|
|
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
|