"""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)