Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6d2b9b5335 | ||
|
|
47c89a13d0 | ||
|
|
45535e19d7 | ||
|
|
af2d0d2f61 |
@@ -2,6 +2,7 @@
|
||||
|
||||
### Not in a release yet
|
||||
|
||||
- Home now draws today's charge from the grid for every hour the add-on will really charge, not only the run under way ([[UX-77]]).
|
||||
- When Home Assistant passes readings on late, control no longer swings on them: it waits until the battery shows its last power request, and the peak limit no longer asks for full power on them. The page says so, and if the readings stay too late every inverter goes to 0 W ([[SAFETY-128]]).
|
||||
|
||||
## 0.20.1
|
||||
|
||||
@@ -97,9 +97,14 @@ is `SolarToday.add()` after the solar read, in a `try`, in memory.
|
||||
- Prices are planned about a day ahead (`priceruns.HORIZON`), so a block of
|
||||
tomorrow later than 24 h from now, or one that needs prices not yet
|
||||
published, appears as the day comes near. The tab says so.
|
||||
- Today's blocks are Home's, and are not extended or shortened: control may
|
||||
plan more once they end, and Home draws it then. The curve stops charging
|
||||
at its level (`charge_to()`) and giving at the floor.
|
||||
- Today's blocks are Home's. UX-77: Home draws today's grid block to where
|
||||
control will stop charging, as the tab does for tomorrow's: the cheap run
|
||||
rolls on at its end while a battery is short (`cheaproll.roll_end()`, the
|
||||
same loop as `dayview._levels()`), the level read at the charge power the
|
||||
peak limit leaves (peak aim less the house, else the rating) and shared by
|
||||
room left to full as control shares it (`packshare.weights()`, re-read as
|
||||
the levels move), and each battery's block ends at its own stop. The curve stops charging at its level
|
||||
(`charge_to()`) and giving at the floor.
|
||||
- Today, a give run in the plan on a battery that is empty now has no block on
|
||||
Home (no energy to give). The tab equals Home, and says so: "The add-on plans
|
||||
to give power. No block shows while a battery has no energy to give." The
|
||||
|
||||
@@ -0,0 +1,201 @@
|
||||
"""UX-77: where the cheap run rolls on. Display only; control is unchanged.
|
||||
|
||||
A cheap run is sized for the batteries charging together, at their full
|
||||
rating, and starts at the first of the cheapest hours. The peak limit leaves
|
||||
less power than that, so a battery is still short when the run ends, and
|
||||
control plans the next run then (`Controller._price_runs()`). On a flat night
|
||||
that goes on until the batteries are at their level or the cheap hours end.
|
||||
|
||||
`roll_end()` asks what control asks at each run end, of the same functions
|
||||
(`modetrade.energy()`, `priceruns.plan()`), until nothing is short or the next
|
||||
run does not follow on. `for_today()` reads the batteries at the level they
|
||||
will have by then: the charge power is the peak limit's room (peak aim minus
|
||||
the house, as `dayplan` counts it), shared as control shares it: by room
|
||||
left to full (`packshare.weights()`), read again as the levels move, so a
|
||||
battery that fills drops out and its share goes to the others. Where that
|
||||
room is not known (no house profile or no aim), the rating is used.
|
||||
DECISIONS/UX-54-forecast-tab.md has the record.
|
||||
"""
|
||||
|
||||
import dataclasses
|
||||
import math
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from . import arbitrage, dayplan, modetrade, packshare, priceruns
|
||||
from .dayplan import EFF
|
||||
from .price import required_spread, uncertainty_of
|
||||
|
||||
SLOT_S = 900 # a price slot
|
||||
STEP = timedelta(minutes=5) # how finely the level is followed
|
||||
LIMIT = 24 # rolls at most (a day of one-hour runs)
|
||||
|
||||
|
||||
def roll_end(run, points, packs_at, *, required, floor_pct, ceiling_pct,
|
||||
targets=None, limit=LIMIT):
|
||||
"""Where the cheap run `run` ends once control has rolled it on.
|
||||
|
||||
`packs_at(end)` gives the `modetrade.Pack`s as they stand when a run
|
||||
ends at `end`. At each end the charge still short is asked of
|
||||
`modetrade.energy()` and the next run of `priceruns.plan()`, as control
|
||||
asks them. It stops when nothing is short or the next run does not follow
|
||||
on, and is `run.end` when control would not roll it."""
|
||||
end_t = run.end
|
||||
for _ in range(limit):
|
||||
e = modetrade.energy(packs_at(end_t), floor_pct, ceiling_pct,
|
||||
targets=targets)
|
||||
if e[0] <= 1e-6:
|
||||
break
|
||||
new = priceruns.plan(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
|
||||
return end_t
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Roll:
|
||||
"""`end`: where control stops charging, `stops`: name -> when that battery
|
||||
reaches its level (rounded up to a price slot), None while it still
|
||||
charges at `end`. A battery not in `stops` does not charge in the run."""
|
||||
end: datetime
|
||||
stops: dict
|
||||
|
||||
def end_for(self, name, base_end):
|
||||
"""The end of `name`'s block: its stop, no later than `end`."""
|
||||
if name not in self.stops:
|
||||
return base_end
|
||||
stop = self.stops[name]
|
||||
return self.end if stop is None else min(self.end, stop)
|
||||
|
||||
def last(self, base_end):
|
||||
"""The last end of any battery's block (the chart's cheap bar)."""
|
||||
return max([base_end] + [self.end_for(n, base_end)
|
||||
for n in self.stops])
|
||||
|
||||
|
||||
def _ceil_slot(t):
|
||||
return datetime.fromtimestamp(
|
||||
math.ceil(t.timestamp() / SLOT_S) * SLOT_S, tz=t.tzinfo
|
||||
or timezone.utc)
|
||||
|
||||
|
||||
class _Charge:
|
||||
"""The batteries' levels followed from `t0` while the chargers charge at
|
||||
the room the peak limit leaves: `min(sum of ratings, aim - house)`, shared
|
||||
by room left to full as control does (`_shares()`), each stopping at `modetrade.charge_to()`."""
|
||||
|
||||
def __init__(self, packs, t0, ceiling_pct, targets, aim_w, house):
|
||||
self.packs = {p.name: p for p in packs}
|
||||
self.t = t0
|
||||
self.ceil = ceiling_pct
|
||||
self.targets = targets
|
||||
self.aim = aim_w
|
||||
self.house = house
|
||||
self.stops = {}
|
||||
|
||||
def at(self, end):
|
||||
"""The packs when a run ends at `end`."""
|
||||
self._to(end)
|
||||
return list(self.packs.values())
|
||||
|
||||
def _room_w(self, hour, total):
|
||||
if self.aim is None or self.house is None:
|
||||
return total
|
||||
house_w, base_w = self.house
|
||||
return min(total, max(0.0, float(self.aim)
|
||||
- max(house_w(hour), base_w, 0.0)))
|
||||
|
||||
def _to(self, end):
|
||||
while self.t < end:
|
||||
nxt = min(end, self.t + STEP)
|
||||
for s, e, hour in dayplan._hour_slices(self.t, nxt):
|
||||
self._charge(s, e, hour)
|
||||
self.t = nxt
|
||||
|
||||
def _shares(self, ch, w):
|
||||
"""Watts per charger: `w` shared by `packshare.weights()` over all
|
||||
packs (room left to full, as control passes it), no charger above its
|
||||
rating, the surplus going to the others. By rating where the room left
|
||||
is not known (no capacity)."""
|
||||
wt = packshare.weights(list(self.packs.values()), -1.0)
|
||||
if wt is None:
|
||||
wt = {p.name: abs(p.rating_w) for p in ch}
|
||||
wt = {p.name: wt.get(p.name, 0.0) for p in ch}
|
||||
cap = {p.name: abs(p.rating_w) for p in ch}
|
||||
out, open_, left = {}, set(wt), float(w)
|
||||
while open_:
|
||||
tot = sum(wt[n] for n in open_)
|
||||
if tot <= 0.0:
|
||||
for n in open_:
|
||||
out[n] = 0.0
|
||||
break
|
||||
over = [n for n in open_ if left * wt[n] / tot > cap[n] + 1e-9]
|
||||
if not over:
|
||||
for n in open_:
|
||||
out[n] = left * wt[n] / tot
|
||||
break
|
||||
for n in over:
|
||||
out[n] = cap[n]
|
||||
left -= cap[n]
|
||||
open_.discard(n)
|
||||
return out
|
||||
|
||||
def _charge(self, s, e, hour):
|
||||
ch = modetrade.chargers(list(self.packs.values()), self.ceil,
|
||||
targets=self.targets)
|
||||
total = sum(abs(p.rating_w) for p in ch)
|
||||
if not ch or total <= 0.0:
|
||||
return
|
||||
w = self._room_w(hour, total)
|
||||
hrs = (e - s).total_seconds() / 3600.0
|
||||
share = self._shares(ch, w)
|
||||
for p in ch:
|
||||
self.stops.setdefault(p.name, None)
|
||||
if not p.capacity_kwh:
|
||||
continue
|
||||
to = modetrade.charge_to(p, self.ceil, self.targets)
|
||||
kwh = share[p.name] * hrs / 1000.0 * EFF
|
||||
room = (to - p.soc) / 100.0 * float(p.capacity_kwh)
|
||||
if kwh <= 0.0:
|
||||
continue
|
||||
if kwh >= room - 1e-9:
|
||||
self.stops[p.name] = _ceil_slot(
|
||||
s + (e - s) * (room / kwh))
|
||||
kwh = room
|
||||
self.packs[p.name] = dataclasses.replace(
|
||||
p, soc=p.soc + kwh / float(p.capacity_kwh) * 100.0)
|
||||
|
||||
|
||||
def for_today(c, rp, now):
|
||||
"""The `Roll` of the cheap run `rp.cheap` now under way or to come today,
|
||||
or None when control does not roll it on (no run left, charging not
|
||||
useful, the trade gate shut, nothing short at its end). Display only:
|
||||
nothing here is fed back to control."""
|
||||
if rp is None or not rp.on or rp.cheap is None or rp.cheap.end <= now \
|
||||
or not rp.charge_useful() \
|
||||
or getattr(c, "mode_gate", None) is not None:
|
||||
return None
|
||||
plan = c.fresh_tariff_plan(now)
|
||||
if plan is None:
|
||||
return None
|
||||
floor, ceil = arbitrage.soc_bounds(c.o)
|
||||
hp = getattr(c, "house_profile", None)
|
||||
sim = _Charge(c._mode_packs(), max(now, rp.cheap.start), ceil,
|
||||
c.plan_targets(), c.peak_aim_w(),
|
||||
None if hp is None else hp.estimator())
|
||||
required = required_spread(
|
||||
arbitrage.threshold_eur_kwh(c.o),
|
||||
uncertainty_of(plan.read.product)
|
||||
| arbitrage.weather_uncertainty(True, c.solar_fc,
|
||||
bool(c.weather_held(now))))
|
||||
end = roll_end(rp.cheap, plan.read.points, sim.at, required=required,
|
||||
floor_pct=floor, ceiling_pct=ceil,
|
||||
targets=c.plan_targets())
|
||||
if end <= rp.cheap.end:
|
||||
return None
|
||||
return Roll(end, dict(sim.stops))
|
||||
@@ -19,7 +19,8 @@ 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 arbitrage, cheaproll, dayplan, homeview as hv, modetrade
|
||||
from . import priceruns
|
||||
from . import solarfc
|
||||
from .daycurve import Batt, curve
|
||||
from .packmode import SOLAR_FIRST
|
||||
@@ -210,25 +211,18 @@ def _levels(c, est, fc, trust, now, end, runs, floor, ceil, notes_b, shadow,
|
||||
arbitrage.threshold_eur_kwh(c.o),
|
||||
uncertainty_of(plan.read.product)
|
||||
| arbitrage.weather_uncertainty(True, c.solar_fc))
|
||||
end_t = cheap.end
|
||||
for _ in range(24):
|
||||
ext = dataclasses.replace(rp, cheap=priceruns.Run(
|
||||
|
||||
def packs_at(end_t):
|
||||
"""The packs at the curve's level when a run ends at `end_t`."""
|
||||
at = 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
|
||||
_rt, lv = settle(at, lv0)
|
||||
return [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]
|
||||
end_t = cheaproll.roll_end(
|
||||
cheap, plan.read.points, packs_at, required=required,
|
||||
floor_pct=floor, ceiling_pct=ceil, targets=targets)
|
||||
ext = dataclasses.replace(rp, cheap=priceruns.Run(
|
||||
cheap.start, end_t, cheap.mean_eur_kwh))
|
||||
rt, lv = settle(ext, lv0)
|
||||
|
||||
@@ -19,7 +19,7 @@ section at the bottom and is held to STYLE.md by test_messages.py.
|
||||
import contextlib
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from . import arbitrage, messages, peakguard, uitext
|
||||
from . import arbitrage, cheaproll, messages, peakguard, priceruns, uitext
|
||||
from .dayplan import EFF
|
||||
from .maintenance import CHARGE as M_CHARGE, DRAIN as M_DRAIN, HOLD as M_HOLD
|
||||
from .maintenance import IDLE as M_IDLE, next_slot_start
|
||||
@@ -218,14 +218,20 @@ def price_windows(c, plan, verdict, now=None):
|
||||
"""`[(kind, start, end)]`: where the add-on itself buys ("cheap") and
|
||||
gives ("dear"). With battery modes on that is the kept `priceruns.RunPlan`
|
||||
(`c.mode_runs`): the cheap run when charging in it pays and the trade gate
|
||||
allows it (UX-53), and every give run. With them off it is the arbitrage
|
||||
verdict's legs the add-on will take, the ones `trade_runs()` draws
|
||||
(`trade_legs()`, given `now`). Nothing is classified here."""
|
||||
allows it (UX-53), and every give run. UX-77: the cheap run is drawn to
|
||||
where control stops charging as it rolls on (`cheaproll`). With them off
|
||||
it is the arbitrage verdict's legs the add-on will take, the ones
|
||||
`trade_runs()` draws (`trade_legs()`, given `now`). Nothing is classified
|
||||
here."""
|
||||
if getattr(c, "modes_on", False):
|
||||
rp = getattr(c, "mode_runs", None)
|
||||
if rp is None or not rp.on or not _price_mode_in_use(c):
|
||||
return []
|
||||
out = [("cheap", rp.cheap.start, rp.cheap.end)] \
|
||||
end = rp.cheap.end
|
||||
roll = None if now is None else cheaproll.for_today(c, rp, now)
|
||||
if roll is not None:
|
||||
end = roll.last(end) # UX-77: to where control stops
|
||||
out = [("cheap", rp.cheap.start, end)] \
|
||||
if rp.charge_useful() and getattr(c, "mode_gate", None) is None \
|
||||
else []
|
||||
return out + [("dear", g.start, g.end) for g in rp.give]
|
||||
@@ -422,6 +428,7 @@ def mode_runs(c, day0, day1, now, shadow, ahead=False, rp=None):
|
||||
`ahead=True`) a charge from the floor and a give from the ceiling. The
|
||||
same filter, the same `MIN_KWH`. `rp`: a run plan to read instead of
|
||||
`c.mode_runs`. Home never passes either."""
|
||||
today = rp is None and not ahead
|
||||
rp = rp if rp is not None else getattr(c, "mode_runs", None)
|
||||
if getattr(c, "mode_tick", None) is None:
|
||||
return [], None # no cycle yet: nothing is known
|
||||
@@ -431,6 +438,9 @@ def mode_runs(c, day0, day1, now, shadow, ahead=False, rp=None):
|
||||
return [], "not_worth"
|
||||
floor, ceiling = arbitrage.soc_bounds(c.o)
|
||||
gate = getattr(c, "mode_gate", None)
|
||||
# UX-77: today's grid block runs to where control stops charging (the run
|
||||
# rolls on when it ends), each battery's to its own stop.
|
||||
roll = cheaproll.for_today(c, rp, now) if today else None
|
||||
out = []
|
||||
for i, inv in enumerate(c.inverters):
|
||||
pm = c.pack_mode(inv)
|
||||
@@ -465,7 +475,10 @@ def mode_runs(c, day0, day1, now, shadow, ahead=False, rp=None):
|
||||
if gate is None and rp.charge_useful() and (
|
||||
need > 0.0 if need is not None
|
||||
else pm.mode in GRID_CHARGE_MODES):
|
||||
blocks.append(("grid", rp.cheap))
|
||||
cheap = rp.cheap if roll is None else priceruns.Run(
|
||||
rp.cheap.start, roll.end_for(inv.name, rp.cheap.end),
|
||||
rp.cheap.mean_eur_kwh)
|
||||
blocks.append(("grid", cheap))
|
||||
sell = rp.sell if pm.sell and pm.can_sell else None
|
||||
for g in rp.give:
|
||||
if sell is not None and g.start <= sell.start < g.end:
|
||||
|
||||
@@ -39,7 +39,8 @@ sys.path.insert(0, os.path.join(HERE, "tools"))
|
||||
logging.disable(logging.WARNING)
|
||||
|
||||
import ux54_scenarios as S # noqa: E402
|
||||
from app import arbitrage, dayplan, dayview, homeview, modetrade # noqa: E402
|
||||
from app import arbitrage, cheaproll, dayplan, dayview, homeview # noqa: E402
|
||||
from app import modetrade # noqa: E402
|
||||
from app import packmode, packshare, peakguard, solarfc, uitext # noqa: E402
|
||||
from app.main import pack_energy_of # noqa: E402
|
||||
from app.price import Uncertainty # noqa: E402
|
||||
@@ -232,11 +233,19 @@ END_S = S.brussels(END).strftime("%H:%M")
|
||||
check(f"the cheap run is 01:00 to {END_S} (02:30 for this house)",
|
||||
c_n.mode_runs.cheap.start == at(1, 0, 0) and END == at(2, 30, 0))
|
||||
nb = tab(c_n, at(0, 18))["days"][0]["batteries"]
|
||||
check("in the night before the cheap run both charge from the grid in it "
|
||||
f"(01:00 to {END_S}), then give at 17:00",
|
||||
all([r["kind"] for r in b["blocks"]] == ["grid", "give"] for b in nb)
|
||||
and all(b["blocks"][0]["from"] == "01:00"
|
||||
and b["blocks"][0]["to"] == END_S for b in nb))
|
||||
# UX-77: today's block runs to where control stops charging: the 1.5 h run is
|
||||
# sized at the full rating, the peak limit leaves 3.1 kW, so control plans the
|
||||
# next run when it ends (`cheaproll`), and each battery's block ends at its stop.
|
||||
RL = cheaproll.for_today(c_n, c_n.mode_runs, at(0, 18))
|
||||
ENDS = [S.brussels(RL.end_for(inv.name, END)).strftime("%H:%M")
|
||||
for inv in c_n.inverters]
|
||||
check("in the night before the cheap run both charge from the grid from "
|
||||
f"01:00 to where control stops ({ENDS}, the run rolls on after "
|
||||
f"{END_S}), then give at 17:00",
|
||||
RL is not None and RL.end > END and ENDS[0] < ENDS[1]
|
||||
and all([r["kind"] for r in b["blocks"]] == ["grid", "give"] for b in nb)
|
||||
and all(b["blocks"][0]["from"] == "01:00" for b in nb)
|
||||
and [b["blocks"][0]["to"] for b in nb] == ENDS)
|
||||
# UX-53: no grid block while the gate refuses.
|
||||
for why in ("peak", "not_worth"):
|
||||
c_n.mode_gate = why
|
||||
@@ -255,12 +264,16 @@ def at_h(b, h):
|
||||
|
||||
|
||||
LVL_A = c_n.plan_targets()[c_n.inverters[0].name]
|
||||
END_HS = [(S.brussels(RL.end_for(inv.name, END)).replace(tzinfo=None)
|
||||
- S.brussels(DAY0).replace(tzinfo=None)).total_seconds() / 3600.0
|
||||
for inv in c_n.inverters]
|
||||
check(f"the curve charges in a grid block: A (planned level {LVL_A:.0f} %) "
|
||||
"and B (Cheap hours, to the end of its run) are well above where they "
|
||||
"were at 01:00, and stay put after the run",
|
||||
all(at_h(b, END_H) - at_h(b, 1.0) > 8.0
|
||||
and abs(at_h(b, END_H + 0.5) - at_h(b, END_H)) < 1e-6
|
||||
for b in nb) and nb[0]["levels"][0][1] == 30.0)
|
||||
"and B (Cheap hours, to the end of its block) are well above where "
|
||||
"they were at 01:00, and stay put after the block",
|
||||
all(at_h(b, eh) - at_h(b, 1.0) > 8.0
|
||||
and all(abs(v - at_h(b, eh)) < 1e-6 for x, v in b["levels"]
|
||||
if eh + 1e-6 < x < eh + 3.0)
|
||||
for b, eh in zip(nb, END_HS)) and nb[0]["levels"][0][1] == 30.0)
|
||||
# Modes off: the verdict's legs.
|
||||
c_f = S.make("off", at(10, 7))
|
||||
fb = tab(c_f, at(10, 7))
|
||||
|
||||
@@ -0,0 +1,358 @@
|
||||
"""UX-77: Home draws today's grid charge for the hours control will charge.
|
||||
|
||||
The real `Controller._mode_roles()` and `homeview.build()`, stepped each
|
||||
minute over a flat night (the walk of the investigation): at five times, every
|
||||
minute control charges battery X in that local day lies inside X's drawn grid
|
||||
block, the block does not run past the hour control stops charging X by more
|
||||
than one slot, and the chart's cheap bars cover every hour with a charge.
|
||||
The peak-bound night is the one that needs the peak limit's room (a mutant that
|
||||
rolls at the full rating fails it, part 3); dynamic prices draw what they drew
|
||||
before (part 4). Mutations (MUTATION.md, purged copies): listed in the pull
|
||||
request. Run: python3 test_ux77.py
|
||||
"""
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
sys.path.insert(0, HERE)
|
||||
sys.path.insert(0, os.path.join(HERE, "tools"))
|
||||
logging.disable(logging.WARNING)
|
||||
|
||||
import ux54_scenarios as S # noqa: E402
|
||||
from app import dayplan, homeview, modetrade, packmode, packshare # noqa: E402
|
||||
from app.main import Controller # noqa: E402
|
||||
from app.peakguard import PeakDecision # noqa: E402
|
||||
from app.static_tariff import static_plan # noqa: E402
|
||||
from app.store import unit_key # noqa: E402
|
||||
from app.tariff import BILLING_TZ # noqa: E402
|
||||
|
||||
try: # not on the base, where it is red
|
||||
from app import cheaproll
|
||||
except ImportError:
|
||||
cheaproll = None
|
||||
|
||||
UTC = timezone.utc
|
||||
EFF = 0.95
|
||||
AIM = 3450.0
|
||||
SLOT = timedelta(hours=1) # the price slot of the day/night tariff
|
||||
QUARTER = timedelta(minutes=15) # the run's own slot
|
||||
SAMPLES = ("23:30", "00:30", "01:30", "02:30", "04:30")
|
||||
OLD = ("leftover_solar", "cheap_hours") # the owner's old shape
|
||||
NOW = ("leftover_solar", "leftover_solar") # the current one
|
||||
PASS = FAIL = 0
|
||||
|
||||
|
||||
def check(name, ok, detail=""):
|
||||
global PASS, FAIL
|
||||
if ok:
|
||||
PASS += 1
|
||||
else:
|
||||
FAIL += 1
|
||||
print(f"FAIL {name} {detail}")
|
||||
|
||||
|
||||
def no_ties(h):
|
||||
"""Two dips, no two slots at the same price: nothing rolls on."""
|
||||
return (0.05 if 1.0 <= h < 2.0 else 0.08 if 5.0 <= h < 6.0
|
||||
else 0.30) + 0.0001 * h
|
||||
|
||||
|
||||
def shape(night=None):
|
||||
d = {h: 0.973 * (600.0 if h < 6 or h >= 22 else 1200.0 if h < 10
|
||||
else 900.0 if h < 17 else 1800.0) for h in range(24)}
|
||||
if night is not None:
|
||||
for h in list(range(0, 7)) + [22, 23]:
|
||||
d[h] = night
|
||||
return d
|
||||
|
||||
|
||||
def profile(sh, now):
|
||||
d = now.astimezone(BILLING_TZ).date()
|
||||
return dayplan.HouseProfile({(d - timedelta(days=k)).isoformat():
|
||||
{str(h): w for h, w in sh.items()} for k in (1, 2, 3)})
|
||||
|
||||
|
||||
def loc(t):
|
||||
return t.astimezone(BILLING_TZ)
|
||||
|
||||
|
||||
def hm(s, day):
|
||||
h, m = map(int, s.split(":"))
|
||||
return day.replace(hour=h, minute=m)
|
||||
|
||||
|
||||
def split_by_room(w, ch, rating, wt, invs):
|
||||
"""Control's split of `w` over the chargers `ch`: by `packshare.weights()`,
|
||||
no battery over its rating, the surplus to the others."""
|
||||
p, open_, left = [0.0, 0.0], set(ch), w
|
||||
while open_:
|
||||
tot = sum(wt[invs[i].name] for i in open_)
|
||||
over = [i for i in open_
|
||||
if left * wt[invs[i].name] / tot > rating[i] + 1e-9]
|
||||
if not over:
|
||||
for i in open_:
|
||||
p[i] = left * wt[invs[i].name] / tot
|
||||
break
|
||||
for i in over:
|
||||
p[i] = rating[i]
|
||||
left -= rating[i]
|
||||
open_.discard(i)
|
||||
return p
|
||||
|
||||
|
||||
def walk(modes, soc0, house_w, night=None, dynamic=False, samples=SAMPLES,
|
||||
start_h=22, end_h=7, plain=None, by_room=True, fc=(10.2, 10.0)):
|
||||
"""Walk 22:00-07:15. Returns (charging, homes, commands):
|
||||
charging[minute] = set of batteries control charges (power > 0),
|
||||
homes[sample] = Home's `/status` plan at that minute, commands = the
|
||||
(minute, unit, role, size) control gave."""
|
||||
opts = dict(S.BASE)
|
||||
opts.update(battery_modes_enabled=True,
|
||||
pv_forecast_entities=[f"{S.FC1}_{i}" for i in range(3)],
|
||||
pv_forecast_tomorrow_entities=[f"{S.FCT1}_{i}"
|
||||
for i in range(3)],
|
||||
static_profile=json.dumps(S.DN), arbitrage_enabled=False,
|
||||
arbitrage_soc_floor=20, arbitrage_soc_ceiling=95,
|
||||
cycle_cost_threshold_eur=0.04)
|
||||
c = Controller(opts, S.Hass(), S.Store(), S.Pub(), stop=S.asyncio.Event())
|
||||
c.auto = True
|
||||
soc = list(soc0)
|
||||
caps = [inv.capacity_kwh for inv in c.inverters]
|
||||
rating = [inv.max_w for inv in c.inverters]
|
||||
for i, inv in enumerate(c.inverters):
|
||||
inv.batt, inv.last_write = 0.0, 0.0
|
||||
inv.dev_min, inv.dev_max = -5000.0, 5000.0
|
||||
c.store.data[unit_key(inv.name, packmode.MODE_KEY)] = modes[i]
|
||||
c.pv_w = 0.0
|
||||
c.peak_guard = PeakDecision(engaged=False, floor_w=None, ceiling_w=7000.0,
|
||||
ceiling_from="cap", projected_w=800.0)
|
||||
start = datetime(2026, 11, 1, start_h, 0, tzinfo=BILLING_TZ).astimezone(UTC)
|
||||
end = datetime(2026, 11, 2, end_h, 15, tzinfo=BILLING_TZ).astimezone(UTC)
|
||||
c.house_profile = profile(shape(night), start)
|
||||
now, plan_at = start, None
|
||||
charging, homes, cmds = {}, {}, []
|
||||
while now < end:
|
||||
lt = loc(now)
|
||||
today, tomorrow = fc if lt.hour < 12 else fc[::-1]
|
||||
c.solar_fc = S.forecast(now, today, tomorrow, members=3)
|
||||
c.solar_fc_why = None
|
||||
if plan_at is None or (now - plan_at).total_seconds() >= 1800:
|
||||
c.tariff_plan = (S.quarters(now, dynamic) if dynamic else
|
||||
static_plan(opts, now, now + timedelta(hours=48),
|
||||
now=now))
|
||||
plan_at = now
|
||||
c.tariff_at = plan_at
|
||||
for i, inv in enumerate(c.inverters):
|
||||
inv.soc = round(soc[i], 1)
|
||||
room = AIM - house_w
|
||||
c.arb_headroom_w = room
|
||||
c.grid = house_w
|
||||
tick = c._mode_roles(now, None)
|
||||
size = modetrade.size(tick, (-sum(rating), sum(rating)), room) \
|
||||
if tick.action else 0.0
|
||||
ch = [i for i, inv in enumerate(c.inverters)
|
||||
if tick.roles.get(inv.name) == modetrade.CHARGE]
|
||||
p = [0.0, 0.0]
|
||||
if size < 0 and ch:
|
||||
if by_room: # as control: room left to full, rating caps
|
||||
wt = packshare.weights(c.pack_energy, -1.0)
|
||||
p = split_by_room(-size, ch, rating, wt, c.inverters)
|
||||
else:
|
||||
tot = sum(rating[i] for i in ch)
|
||||
for i in ch:
|
||||
p[i] = -size * rating[i] / tot
|
||||
for i in range(2):
|
||||
soc[i] = min(100.0, soc[i] + p[i] / 1000.0 / 60.0 * EFF
|
||||
/ caps[i] * 100.0)
|
||||
charging[lt.replace(second=0, microsecond=0)] = {
|
||||
i for i in range(2) if p[i] > 0.0}
|
||||
cmds.append((lt.strftime("%H:%M"), tick.action,
|
||||
tuple(sorted(tick.roles.items())), round(size)))
|
||||
if lt.strftime("%H:%M") in samples:
|
||||
homes[lt.replace(second=0, microsecond=0)] = homeview.build(
|
||||
c, level="ok", checks=[], now=now, lang="en")
|
||||
if plain is not None: # the same, with no roll
|
||||
real = cheaproll.for_today
|
||||
cheaproll.for_today = lambda *_a: None
|
||||
try:
|
||||
plain[lt.replace(second=0, microsecond=0)] = \
|
||||
homeview.build(c, level="ok", checks=[], now=now,
|
||||
lang="en")
|
||||
finally:
|
||||
cheaproll.for_today = real
|
||||
now += timedelta(minutes=1)
|
||||
return charging, homes, cmds
|
||||
|
||||
|
||||
def blocks_of(st, i):
|
||||
"""Battery i's drawn grid blocks as (start, end) local datetimes."""
|
||||
return [(loc(datetime.fromisoformat(r["start"])),
|
||||
loc(datetime.fromisoformat(r["end"])))
|
||||
for r in st["plan"]["runs"]
|
||||
if r["kind"] == "grid" and r["battery"] == i]
|
||||
|
||||
|
||||
def today_of(st, i):
|
||||
return [(t["from"], t["to"]) for t in st["batteries"][i]["today"]]
|
||||
|
||||
|
||||
def violations(charging, homes, slot=SLOT):
|
||||
"""What is wrong with Home's drawing at each sample."""
|
||||
bad = []
|
||||
for t, st in sorted(homes.items()):
|
||||
day_end = (t.replace(hour=0, minute=0) + timedelta(days=1))
|
||||
mins = {m: who for m, who in charging.items() if t <= m < day_end}
|
||||
for i in range(2):
|
||||
mine = sorted(m for m, who in mins.items() if i in who)
|
||||
bl = blocks_of(st, i)
|
||||
for m in mine:
|
||||
if not any(a <= m < z for a, z in bl):
|
||||
bad.append(f"{t:%H:%M} battery {i}: charges at "
|
||||
f"{m:%H:%M}, not in a block {bl}")
|
||||
break
|
||||
if mine:
|
||||
stop = mine[-1] + timedelta(minutes=1)
|
||||
if slot == QUARTER: # the last charging quarter's end
|
||||
stop += timedelta(minutes=-stop.minute % 15)
|
||||
for a, z in bl:
|
||||
if z > stop + slot:
|
||||
bad.append(f"{t:%H:%M} battery {i}: block to "
|
||||
f"{z:%H:%M}, stops {stop:%H:%M}")
|
||||
# the Today line says the same as the block
|
||||
tl = today_of(st, i)
|
||||
for m in mine:
|
||||
if not any(f <= m.strftime("%H:%M") < (z if z != "24:00"
|
||||
else "99") for f, z in tl):
|
||||
bad.append(f"{t:%H:%M} battery {i}: Today line {tl} "
|
||||
f"has no {m:%H:%M}")
|
||||
break
|
||||
cheap = {b["label"][:2] for b in st["plan"]["prices"]
|
||||
if b["kind"] == "cheap"}
|
||||
for m in mins:
|
||||
if mins[m] and m.strftime("%H") not in cheap:
|
||||
bad.append(f"{t:%H:%M}: charge at {m:%H:%M}, hour not cheap")
|
||||
break
|
||||
return bad
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Run
|
||||
# --------------------------------------------------------------------------- #
|
||||
def show(tag, ch, homes, want_bad, slot=SLOT):
|
||||
bad = violations(ch, homes, slot)
|
||||
if want_bad:
|
||||
check(f"{tag}: red", bool(bad), "no violation found")
|
||||
else:
|
||||
check(f"{tag}: every charging minute is drawn, no block runs on",
|
||||
not bad, "\n " + "\n ".join(bad))
|
||||
return bad
|
||||
|
||||
|
||||
def charged(ch, i, after):
|
||||
return [m for m, who in ch.items() if i in who and m >= after]
|
||||
|
||||
|
||||
def main():
|
||||
global cheaproll
|
||||
if cheaproll is None:
|
||||
print("no app/cheaproll.py: the base; the walk is red there")
|
||||
cheaproll = type("Off", (), {"for_today": staticmethod(lambda *a: None)})
|
||||
|
||||
print("1. the walk: no-bind (0.6 kW house) and peak-bound (2.85 kW house)")
|
||||
CASES = (("old shape, no bind", OLD, (20.0, 20.0), 584.0, None),
|
||||
("current shape, no bind", NOW, (20.0, 20.0), 584.0, None),
|
||||
("old shape, peak-bound", OLD, (40.0, 90.0), 2850.0, 2850.0),
|
||||
("current shape, peak-bound", NOW, (40.0, 90.0), 2850.0, 2850.0))
|
||||
# the rig's checks 2 and 3 (UX-77 round 2): room left and rating disagree
|
||||
UNEQUAL = (("unequal levels, A+B Leftover solar, no bind", NOW,
|
||||
(30.0, 20.0), 584.0, None, (5.0, 5.0)),
|
||||
("unequal levels, A Leftover solar + B Cheap hours", OLD,
|
||||
(40.0, 88.0), 2850.0, 2850.0, (10.2, 10.0)))
|
||||
walked = {}
|
||||
for tag, modes, soc, hw, night in CASES:
|
||||
ch, homes, _cmds = walk(modes, soc, hw, night=night)
|
||||
walked[tag] = (ch, homes)
|
||||
show(tag, ch, homes, False)
|
||||
# the check has something to check: charging goes on past 00:30
|
||||
day = datetime(2026, 11, 2, 0, 30, tzinfo=BILLING_TZ)
|
||||
check(f"{tag}: control charges after 00:30", bool(
|
||||
charged(ch, 0, day) or charged(ch, 1, day)))
|
||||
|
||||
print("2. a battery with nothing to charge gets no block")
|
||||
ch, homes, _c = walk(OLD, (40.0, 94.0), 2850.0, night=2850.0)
|
||||
st = homes[datetime(2026, 11, 2, 2, 30, tzinfo=BILLING_TZ)]
|
||||
check("B is at its level at 02:30: no block", blocks_of(st, 1) == [],
|
||||
str(blocks_of(st, 1)))
|
||||
check("A still has one at 02:30", bool(blocks_of(st, 0)))
|
||||
|
||||
print("3. a roll at the full rating fails the peak-bound night")
|
||||
if hasattr(cheaproll, "_Charge"):
|
||||
real = cheaproll._Charge._room_w
|
||||
cheaproll._Charge._room_w = lambda self, hour, total: total
|
||||
try:
|
||||
for tag, modes, soc, hw, night in CASES[2:]:
|
||||
ch, homes, _c = walk(modes, soc, hw, night=night)
|
||||
show(f"mutant, {tag}", ch, homes, True)
|
||||
finally:
|
||||
cheaproll._Charge._room_w = real
|
||||
|
||||
print("4. dynamic prices: Home is what it was")
|
||||
for tag, modes, soc, hw, night in (CASES[2], CASES[3], CASES[1]):
|
||||
plain = {}
|
||||
ch, homes, _c = walk(modes, soc, hw, night=night, dynamic=no_ties,
|
||||
plain=plain)
|
||||
same = all(json.dumps(homes[t], sort_keys=True, default=str)
|
||||
== json.dumps(plain[t], sort_keys=True, default=str)
|
||||
for t in homes)
|
||||
check(f"dynamic, {tag}: same as with no roll", same)
|
||||
check(f"dynamic, {tag}: and there are grid blocks to compare",
|
||||
any(r["kind"] == "grid" for st in homes.values()
|
||||
for r in st["plan"]["runs"]))
|
||||
|
||||
print("5. unequal levels: the roll shares the room as control does")
|
||||
for tag, modes, soc, hw, night, fc in UNEQUAL:
|
||||
ch, homes, _c = walk(modes, soc, hw, night=night, fc=fc)
|
||||
show(tag, ch, homes, False, QUARTER)
|
||||
day = datetime(2026, 11, 2, 0, 30, tzinfo=BILLING_TZ)
|
||||
check(f"{tag}: control charges after 00:30", bool(
|
||||
charged(ch, 0, day) or charged(ch, 1, day)))
|
||||
if hasattr(cheaproll, "_Charge") \
|
||||
and hasattr(cheaproll._Charge, "_shares"):
|
||||
real = cheaproll._Charge._shares
|
||||
cheaproll._Charge._shares = lambda self, ch, w: {
|
||||
p.name: w * abs(p.rating_w) / sum(abs(q.rating_w) for q in ch)
|
||||
for p in ch}
|
||||
try:
|
||||
for tag, modes, soc, hw, night, fc in UNEQUAL:
|
||||
ch, homes, _c = walk(modes, soc, hw, night=night, fc=fc)
|
||||
show(f"mutant by rating, {tag}", ch, homes, True, QUARTER)
|
||||
finally:
|
||||
cheaproll._Charge._shares = real
|
||||
|
||||
print("6. the pieces")
|
||||
if hasattr(cheaproll, "Roll"):
|
||||
t = datetime(2026, 11, 2, 1, 46, tzinfo=UTC)
|
||||
check("a stop is rounded up to the slot", cheaproll._ceil_slot(t)
|
||||
== datetime(2026, 11, 2, 2, 0, tzinfo=UTC))
|
||||
t = datetime(2026, 11, 2, 1, 45, tzinfo=UTC)
|
||||
check("a stop on a slot stays", cheaproll._ceil_slot(t) == t)
|
||||
e0 = datetime(2026, 11, 2, 0, 0, tzinfo=UTC)
|
||||
e1 = datetime(2026, 11, 2, 6, 0, tzinfo=UTC)
|
||||
r = cheaproll.Roll(e1, {"a": None, "b": datetime(2026, 11, 2, 2, 0,
|
||||
tzinfo=UTC)})
|
||||
check("a battery still charging ends at the roll", r.end_for("a", e0) == e1)
|
||||
check("a battery at its level ends at its stop",
|
||||
r.end_for("b", e0) == datetime(2026, 11, 2, 2, 0, tzinfo=UTC))
|
||||
check("a battery that does not charge keeps the run's end",
|
||||
r.end_for("c", e0) == e0)
|
||||
check("the chart's bar runs to the last block", r.last(e0) == e1)
|
||||
|
||||
print(f"\n{PASS} passed, {FAIL} failed")
|
||||
return FAIL
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(1 if main() else 0)
|
||||
Reference in New Issue
Block a user