288 lines
13 KiB
Python
288 lines
13 KiB
Python
"""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 main_files # noqa: E402
|
|
import opt11p3_replay as rp # noqa: E402
|
|
import opt20_scen as sc # noqa: E402
|
|
from app import 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:
|
|
now = main_files.live.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")
|