glenn schrooyenandClaude
d2820e8b9b
dayplan._hour_slices() added an hour to the local wall clock and read it back. Without tzdata (tariff._EUTimezone) 02:00 on the spring change reads as 00:00 UTC again, so the loop never advanced and the list grew until the process died. Each step now adds the time left in the local hour to the UTC instant. Ordinary days give the old slices; the spring day gives 23 slices and the autumn day 25. No option, no text change. test_safety123.py: every quarter of 27-29 March and 30 Oct-1 Nov 2027 (and the US change days) for the slices, dayplan.plan() and Controller._mode_roles(), each call under a hard time limit. Co-Authored-By: Claude <noreply@anthropic.com> Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Bdqr7PiaHcBnzPituu2yTC
265 lines
10 KiB
Python
265 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 batteries by their room.
|
|
* Cheap hours only ignores the plan and fills to the ceiling, as in P3 and
|
|
the design (section 1). 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 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 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
|
|
|
|
|
|
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 _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 == 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())
|
|
|
|
# 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
|
|
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
|