glenn schrooyenandClaude
45535e19d7
The cheap run is sized at the full rating, so under the peak limit a battery is still short when it ends, and control plans the next run then. Home drew only the run under way. `cheaproll` rolls the run on the way control does (`modetrade.energy()` + `priceruns.plan()` at each run end, lifted from `dayview._levels()`), reading the levels at the charge power the peak limit leaves (peak aim less the house, else the rating). `homeview.mode_runs()` and `price_windows()` use it for today: the grid block, the Today line and the chart's cheap bars run to where control stops, each battery's block to its own stop. Display only: no control change. test_ux54: today's grid block of the night-before scenario now follows the roll. Co-Authored-By: Claude <noreply@anthropic.com>
308 lines
13 KiB
Python
308 lines
13 KiB
Python
"""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 # 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
|
|
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 walk(modes, soc0, house_w, night=None, dynamic=False, samples=SAMPLES,
|
|
start_h=22, end_h=7, plain=None):
|
|
"""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 = (10.2, 10.0) if lt.hour < 12 else (10.0, 10.2)
|
|
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:
|
|
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):
|
|
"""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)
|
|
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):
|
|
bad = violations(ch, homes)
|
|
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))
|
|
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 = walked["old shape, peak-bound"]
|
|
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. 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)
|