"""Runnable check for MAINT-28. `python3 test_maint28.py` MAINT-24 follow-up. With A in maintenance CHARGE and a real peak (house over the month aim), MAINT-24's pause ended every time the guard idled: the forecast dipped under the aim, the floor charged A, the guard tripped again, and A flipped charge/give every few seconds (rig x10: 468 flips in 45 min, base 02eac3b 11). Now the pause is held (`peakguard.shaping_held()`) until the house is MAINT_RESUME_UNDER_W under the aim. All on the real Controller through `cycle()` (tools/opt20_scen.py): cheap window, cap 3500 / margin 50, a month peak of 6000 W banked (aim 5950 W), a peak forecast like the live sensor (quarter average so far, the rest at the present grid), house +-50 W every 5 s. 1. A real peak (house 3000 W, then 6500 W for 15 min, then 3000 W): once the guard has stepped in, A does not charge and does not flip; every quarter is at or under the aim; A's charge is back within 10 cycles after the load. 2. peakguard.shaping_held(): the pure rule, with its edges. 3. Unchanged: the MAINT-24 live shape (A 1475 W, no pause), N=1, the SAFETY-78 drain and battery modes off (digests of release/1.0 1b5f7c9, as test_maint24.py). Mutations: DECISIONS/MAINT-28-hold-the-pause-in-a-peak.md. Read MUTATION.md before you report a survivor. """ import logging import os import random import sys 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 opt11p3_replay as rp # noqa: E402 import opt20_scen as sc # noqa: E402 from app import ctl_cycle, peakguard, tariff # noqa: E402 from app.maintenance import DRAIN, IDLE, MaintResult # noqa: E402 fails = [] def check(name, cond): print(f" {'ok ' if cond else 'FAIL'} {name}") if not cond: fails.append(name) CH = sc.MAINT_CHARGE MONTH_PEAK = 6000.0 AIM = MONTH_PEAK - 50.0 DRAIN_R = MaintResult(DRAIN, True, 2000.0, False, []) T0 = sc.at(2.0) # a quarter boundary PRE, LOAD, POST = 120.0, 900.0, 120.0 # s (CI time: short runs) def stepped_house(base_w, hi_w): """[(t_utc, W)]: base_w, then hi_w from PRE for LOAD s, +-50 W per 5 s.""" rnd = random.Random(28) out = [(sc.UTC0.timestamp() - 10, base_w)] for t in range(0, int(PRE + LOAD + POST), 5): w = hi_w if PRE <= t < PRE + LOAD else base_w out.append((T0.timestamp() + t, w + rnd.uniform(-50.0, 50.0))) return out def scen(*, a=None, tick=None, soc=(49.0, 30.0), house=None, house_w=3000.0, peak=MONTH_PEAK, fc="grid", n=2, variant="p3_ch_sell", dur=600.0): """`a`: A's forced maintenance result. `fc`: "grid" (follows the grid), "quarter" (quarter average so far, the rest at the present grid, like the live sensor) or None (no forecast entity). Returns (rows, stats, c).""" rows = [] q = dict(q=None, s=0.0, last=None, start=None) def patch(c): if tick is not None: c.maints[0].tick = ( lambda now, soc_, blocked_by=None, gate_open=True: tick(now)) elif a is not None: c.maints[0].tick = ( lambda now, soc_, blocked_by=None, gate_open=True: a) if peak is not None: orig_peak = c.peak.peak def peak_(now): r = orig_peak(now) return tariff.PeakReading(tariff.OBSERVED, getattr(r, "month", ""), peak) c.peak.peak = peak_ orig = c.cycle susp = [] c._m28_susp = susp async def cyc(): c.cheap = True if fc and c.grid is not None: g = float(c.grid) if fc == "grid": c.peak_fc = g else: # the replay's clock; not main_files.live (it parses # every app file per access: ~0.1 s a cycle). now = ctl_cycle.datetime.now() qi = int(now.timestamp() // 900) if q["q"] != qi: q.update(q=qi, s=0.0, last=now, start=now) el = (now - q["start"]).total_seconds() q["s"] += g * (now - q["last"]).total_seconds() q["last"] = now c.peak_fc = (q["s"] + g * (900.0 - el)) / 900.0 await orig() susp.append(bool(c._maint_shape_suspended)) c.cycle = cyc def cb(c, h, now, pr): sh = dict(c.allocation.shares) if c.allocation is not None \ and c.allocation.shares else {} rows.append(dict( now=now, grid=h.grid(), shares=sh, action=getattr(c.arb, "action", None), reason=getattr(c.arb, "reason", None) or "", roles=dict(c.mode_tick.roles) if c.mode_tick else {}, why=c.mode_tick.why if c.mode_tick else None, p={u: p.p for u, p in h.plants.items()}, t=(now - T0).total_seconds(), a=h.plants["a"].p, susp=c._m28_susp[-1], hold=bool(getattr(c, "_maint_peak_hold", False)))) src = house if house is not None else [(sc.UTC0.timestamp() - 10, house_w)] stats, c = rp.run(variant, src, sc.WPTS, T0, dur, check=cb, lag=sc.LAG, n=n, soc0={"a": soc[0], "b": soc[1]}, choice=sc.CH_CH, refuse=(), patch=patch, extra={"cycle_cost_threshold_eur": 0.10}) return rows, stats, c def flips(rs): """Reversals of A between charging and giving (beyond 50 W).""" n, s = 0, 0 for r in rs: v = 1 if r["a"] > 50 else -1 if r["a"] < -50 else 0 if v and s and v != s: n += 1 if v: s = v return n # ============================================================================ # print("1. a real peak: house 3000 W, 6500 W for 15 min, 3000 W") rows, st, c = scen(a=CH, house=stepped_house(3000.0, 6500.0), fc="quarter", dur=PRE + LOAD + POST) end = PRE + LOAD trip = next((r["t"] for r in rows if r["t"] >= PRE and r["a"] > 50), None) load = [r for r in rows if trip is not None and trip <= r["t"] < end] after = [r for r in rows if r["t"] >= end] back = next((i for i, r in enumerate(after) if r["a"] < -50), None) # cycles print(f" first give at {trip}, flips in the load {flips(load)}, charge s in" f" the load {sum(1 for r in load if r['a'] < -50)}, charge back " f"{back} cycles after it, quarters " f"{[round(a[0] / a[1]) for a in st['quarters'].values()]}") check("the guard stepped in once the projection went over the aim", trip is not None) check("from the first give to the load's end A does not flip", flips(load) == 0) check("...and does not charge at all", not any(r["a"] < -50 for r in load)) check("...the pause is on in every one of those cycles", all(r["susp"] for r in load)) check("every quarter at or under the month aim", max(a[0] / a[1] for a in st["quarters"].values()) <= AIM + 1.0) check("after the load A's charge is back within 10 cycles (1 s each)", back is not None and back <= 10) tail = [r["a"] for r in after[10:]] check("...and stays on: at least 150 W (half the resume margin), no give", tail and sum(tail) / len(tail) <= -150.0 and max(tail) <= 50.0) # ============================================================================ # print("\n1b. the hold does not outlive the charge: a stale True must not " "pause the next charge") # The idle window opens on a quarter boundary, so the new charge meets a # fresh quarter whose projection (house 5700 W) is under the aim. HI_START, HI_END, IDLE_END, B_END = 300.0, 900.0, 1000.0, 1300.0 def _house_b(): rnd = random.Random(29) out = [(sc.UTC0.timestamp() - 10, 3000.0)] for t in range(0, int(B_END) + 100, 5): w = 3000.0 if t < HI_START else 6500.0 if t < HI_END else 5700.0 out.append((T0.timestamp() + t, w + rnd.uniform(-50.0, 50.0))) return out def _tick_b(now): t = (now - T0).total_seconds() return MaintResult(IDLE, False, None, False, []) \ if HI_END <= t < IDLE_END else CH rb, sb, cb_ = scen(tick=_tick_b, house=_house_b(), fc="quarter", dur=B_END) held_before = [r for r in rb if HI_START <= r["t"] < HI_END and r["hold"]] idle_rows = [r for r in rb if HI_END <= r["t"] < IDLE_END] later = [r["a"] for r in rb if IDLE_END + 10.0 <= r["t"] < B_END] print(f" hold on {len(held_before)} cycles before; in the idle window " f"{sum(r['hold'] for r in idle_rows)} of {len(idle_rows)}; A after " f"{sum(later) / len(later):+.0f} W (house 5700 W: 125 W)") check("the hold was on in the peak", len(held_before) > 100) check("it is off in every cycle the charge-only branch is not taken", not any(r["hold"] for r in idle_rows)) check("a new maintenance charge at house 5700 W (over the resume line) " "starts at once: (5950 - 5700) / 2 = 125 W, not paused", sum(later) / len(later) <= -60.0) # ============================================================================ # print("\n2. peakguard.shaping_held()") def held(was, decision, fc, aim, house): return peakguard.shaping_held( peakguard.shaping_suspended(decision, fc, aim), was, fc, aim, house) U = peakguard.MAINT_RESUME_UNDER_W eng = peakguard.PeakDecision(engaged=True, floor_w=100.0, ceiling_w=6000.0, ceiling_from="x", target_w=AIM) idle = peakguard.PeakDecision(engaged=False, floor_w=None, ceiling_w=6000.0, ceiling_from="x", target_w=AIM) check("the old conjunction still pauses: forecast over the aim; guard engaged", held(False, idle, AIM + 1, AIM, 3000.0) and held(False, eng, None, AIM, 3000.0)) check("not held and not suspended: not paused", not held(False, idle, AIM - 1, AIM, 6500.0)) check("held: stays paused while the house is over the aim minus the margin", held(True, idle, AIM - 1, AIM, 6500.0) and held(True, idle, AIM - 1, AIM, AIM - U + 1)) check("held: resumes when the house is the margin under the aim", not held(True, idle, AIM - 1, AIM, AIM - U) and not held(True, idle, AIM - 1, AIM, 3000.0)) check("held, house unread: the forecast stands in; neither read: resumes " "(SAFETY-63)", held(True, idle, AIM - 1, AIM, None) and not held(True, idle, AIM - U, AIM, None) and not held(True, idle, None, AIM, None)) check("a guard that is engaged pauses whatever the house reads", held(True, eng, AIM - 1, AIM, 3000.0)) # ============================================================================ # print("\n3. unchanged: the live shape, N=1, the drain, modes off") rl, sl, cl = scen(a=CH) body = rl[len(rl) // 6:] mean_a = sum(r["a"] for r in body) / len(body) print(f" A {mean_a:+.0f} W, paused {sum(cl._m28_susp)} cycles") check("MAINT-24 live shape: A charges 1475 W and the pause never comes", -1500.0 <= mean_a <= -1440.0 and not any(cl._m28_susp)) d1 = sc.digest(scen(a=CH, n=1, fc=None, peak=None)[0]) dd = sc.digest(scen(a=DRAIN_R, soc=(60.0, 30.0), peak=None, fc=None)[0]) dm = sc.digest(scen(a=CH, variant="modes_off", peak=None, fc=None)[0]) d2n = sc.digest(scen(a=CH, house_w=5000.0, peak=None, fc="grid")[0]) d1p = sc.digest(scen(a=CH, n=1, fc=None, house_w=6500.0)[0]) AT_BASE = { "n2_nobank": "f9053833891794ead8b3f44db4030ded8c4b43681e25e8a19882105d3593f20d", "n1_peak": "3e451c536bea8270469fffe578c5a7199431e0e7ba4902413406b0afe2bb0fd1", "n1": "02ecc3011c559df050ae5321ceb6b24cf79342ae35f1a289fbd1a749c86f1d10", "drain": "bfcf9a643dfffff21d0a8417426d2088377a8f64afbc544732b1635a77775d8a", "off": "6f4ec14578d8ed59fc7721485d037f3c1fb086abae7c52f28c9e9155cf2107b0", } got = {"n2_nobank": d2n, "n1_peak": d1p, "n1": d1, "drain": dd, "off": dm} for k in AT_BASE: print(f" {k} {'same' if got[k] == AT_BASE[k] else 'DIFFERS'} " f"{got[k]}") check("N=1 with A in maintenance charge and no forecast entity: equal to " "release/1.0 1b5f7c9", got["n1"] == AT_BASE["n1"]) check("N=1 in a real peak above a banked month peak (house 6500 W, the guard steps in): equal " "to 1b5f7c9, the hold is for N=2 and up", got["n1_peak"] == AT_BASE["n1_peak"]) check("N=2 in a real peak with no banked peak (house 5000 W over the aim " "3450 W): equal to 1b5f7c9, the hold is for a month aim above the cap", got["n2_nobank"] == AT_BASE["n2_nobank"]) check("the SAFETY-78 drain beside a charger: equal to 1b5f7c9", got["drain"] == AT_BASE["drain"]) check("battery modes off: equal to 1b5f7c9", got["off"] == AT_BASE["off"]) print() if fails: print(f"{len(fails)} FAILED:") for f in fails: print(f" - {f}") sys.exit(1) print("all checks passed")