"""Runnable check for SAFETY-107: a refusal held while another writer has the wire is learned once. `python3 test_safety107.py` A full pack under a maintenance charge in a cheap window took no charge. At one inverter SAFETY-52's "another writer" end then fired every cycle, and compute() learned the refusal again in the same cycle: "BATTERY REFUSES" and "BATTERY REFUSES ENDED: another writer" each second on the rig (MAINT-13 F5). Now another writer ends a refusal only if compute(), judging it afresh against that writer's ask, no longer finds it. DECISIONS/SAFETY-107. Drives the REAL Controller (read_inputs + cycle) on opt11_identity's rig. 1. The card's case: 15 min of a full pack under a maintenance charge, N=1 and N=2, at a 1, 2 and 5 s meter: at most 2 refusal lines, held throughout. 2. Other writers: a trade and the peak guard over a battery that gives nothing. One episode each, no flap. 3. A refusal still ends: a different writer really takes over, the battery delivers again, a writer asks the other way, stop. The probe clock. 4. Not covered, pinned: a perfectly flat reading under a writer. Mutations (MUTATION.md, purged copies) are listed in DECISIONS/SAFETY-107. """ import functools import logging import os import sys import tempfile 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 opt11_identity as rig # noqa: E402 import safety79_identity as s79 # noqa: E402 import app.main as app_main # noqa: E402 from app import control # noqa: E402 from app.arbiter import Claim, P_ARBITRAGE # noqa: E402 from app.main import Controller # noqa: E402 from named_absence import nth, present # noqa: E402 fails = [] QUARTER = 900.0 def check(name, cond): print(f" {'ok ' if cond else 'FAIL'} {name}") if not cond: fails.append(name) def run(n=1, *, plant, house, dur=QUARTER, tick=5.0, maint=True, cheap=lambda c, t: True, trade=lambda t: None, p1=False, cap=2500, auto=lambda t: True): """Rows (t, commands, site target, loop reason, refused, refused_at, unit refusals) and every action-log line as (t, line, hint). `plant(u, t, cmd)` is battery u's reading. `cheap(c, t)` opens the cheap window. `trade(t)` is a SET claim at P_ARBITRAGE added to the claim list (a writer above the loop), or None.""" log, rows, now = [], [], [0.0] real_power, real_tick = rig._Hass.power, rig.TICK real_log, real_resolve, real_c = (Controller.log_event, app_main.resolve, rig.Controller) def power(self, u): return float(plant(u, self.t, self.cmd[u])) def log_event(self, msg, hint=None): log.append((now[0], msg, hint)) return real_log(self, msg, hint) def resolve(claims, **kw): v = trade(now[0]) if v is not None: claims.append(Claim.set("arbitrage", P_ARBITRAGE, v, "trade")) return real_resolve(claims, **kw) def before(c, t): now[0] = t if not getattr(c.hass, "_s107", False): read = c.hass.number async def number(e, invert=False): if e == "sensor.pnow": return 0.05 if cheap(c, c.hass.t) else 0.30 if e == "sensor.pavg": return 0.20 return await read(e, invert) c.hass.number, c.hass._s107 = number, True if p1: s79._feed(c, t) c.auto = auto(t) def on_cycle(c, t): rows.append((t, tuple(c.hass.cmd[u] for u in "ab"[:n]), c.target, c.loop_reason, c.batt_refused, c.batt_refused_at, tuple(i.refused for i in c.inverters))) extra = {"price_now_entity": "sensor.pnow", "price_avg_entity": "sensor.pavg"} if maint: extra.update({"maintenance_enabled": True, "maintenance_start_hour": 11, "maintenance_charge_w": 2000}) if p1: extra.update(s79.P1, peak_cap_w=cap) db = os.path.join(tempfile.mkdtemp(prefix="s107-"), "tariff.db") rig._Hass.power, rig.TICK = power, tick Controller.log_event, app_main.resolve = log_event, resolve rig.Controller = functools.partial(real_c, peak_db_path=db) try: _, c = rig.run(n, extra=extra, dur=dur, load=house, soc=lambda u, t: 50.0, cut=lambda u, t: False, before=before, on_cycle=on_cycle, store=s79.store_charging(n) if maint else rig._Store()) finally: rig._Hass.power, rig.TICK = real_power, real_tick Controller.log_event, app_main.resolve = real_log, real_resolve rig.Controller = real_c return rows, log, c def full_a(u, t, cmd): """A takes no charge (a full pack), its reading jittering by a few watts as a real one does; B follows.""" return -1.0 - int(t) % 3 if u == "a" and cmd < 0 else round(cmd) def flat_a(u, t, cmd): """`full_a` with a perfectly flat reading (the stale-reading gate trips).""" return -1.0 if u == "a" and cmd < 0 else round(cmd) def empty_a(u, t, cmd): """A gives nothing (an empty pack), its reading jittering by a few watts (SAFETY-73's jitter shape, so the stale-reading gate stays out of it); B follows.""" return -4.0 - int(t) % 3 if u == "a" and cmd >= -4.0 else round(cmd) def lines(log, word="REFUSES"): return [x for x in log if word in x[1]] def ends(log): return [x for x in log if "REFUSES ENDED" in x[1]] def refused(r): """Row r holds a refusal (site at N=1, any unit at N>1).""" return bool(r[4]) or any(r[6]) def held_from(rows): """True when, once learned, the refusal is held on every later row.""" first = next((k for k, r in enumerate(rows) if refused(r)), None) return first is not None and all(refused(r) for r in rows[first:]) # ============================================================================== print("1. the card: a full pack under a maintenance charge, 15 min") # ============================================================================== for n, tick, house in ((n, k, h) for n in (1, 2) for k in (1.0, 2.0, 5.0) for h in (800.0, -1500.0)): rows, log, c = run(n, plant=full_a, house=lambda t: house, tick=tick) ref = lines(log) tag = f"N={n}, {tick:.0f} s meter, house {house:+.0f} W" print(f" {tag}: {len(ref)} refusal line(s): " + " / ".join(x[1][:60] for x in ref)) check(f"{tag}: at most 2 refusal lines in 15 min (rig: 548 + 548)", 1 <= len(ref) <= 2) check(f"{tag}: no end line", not ends(log)) check(f"{tag}: learned once and held to the end", held_from(rows)) # N=1: the charge's ceiling stays on the wire. N=2: SAFETY-73 closes A # in the refused direction, as before. check(f"{tag}: still in its charge, the wire as before", c.maints[0].phase == "charge" and all(r[1][0] == (-2000.0 if n == 1 else 0.0) for r in rows[-10:])) # The hint under the buttons is set at most once for the refusal. check(f"{tag}: the refusal hint is set once", sum(1 for x in ref if x[2] is not None) == 1) # The probe clock: while the charge holds the wire no probe is tried (the # loop's ask is not on the wire), so the action log stays quiet. rows, log, _ = run(1, plant=full_a, house=lambda t: 800.0, dur=3600.0) check("N=1, an hour: still one line, no probe tried under the charge", len(lines(log)) == 1 and not any(r[3] == control.PROBING for r in rows)) # ============================================================================== print("\n2. other writers: a trade and the peak guard, no flap") # ============================================================================== # A trade (a SET above the loop) asks a full pack to charge for 15 min. rows, log, _ = run(1, plant=full_a, house=lambda t: 800.0, maint=False, trade=lambda t: -1500.0) print(f" trade into a full pack: {len(lines(log))} line(s)") check("a trade into a full pack: learned once, held, no end", len(lines(log)) == 1 and not ends(log) and held_from(rows)) # The peak guard asks an empty pack to give while the house is over the limit. rows, log, c = run(1, plant=empty_a, house=lambda t: 4500.0, maint=False, p1=True) print(f" peak guard on an empty pack: {len(lines(log))} line(s), " f"guard floor {c.peak_guard.floor_w if c.peak_guard else None}") check("the peak guard on an empty pack: the guard holds the wire", c.peak_guard is not None and c.peak_guard.engaged) check("the peak guard on an empty pack: at most 2 lines, held to the end", 1 <= len(lines(log)) <= 2 and held_from(rows)) # ============================================================================== print("\n3. a refusal still ends") # ============================================================================== # a. Learned by the loop on an empty pack; at 300 s a trade really takes the # wire and asks it to give 2000 W. The refusal was learned against the # loop's ask: it ends, then is learned against the trade, once. rows, log, _ = run(1, plant=empty_a, house=lambda t: 2000.0, maint=False, trade=lambda t: 2000.0 if 300.0 <= t < 900.0 else None, dur=1200.0) e = ends(log) print(" " + " / ".join(f"{x[0]:.0f} s {x[1][:45]}" for x in lines(log))) check("a different writer takes over: the refusal ends, 'another writer'", len(e) == 1 and "another writer" in e[0][1] and 300.0 < e[0][0] < 320.0) check("...and is learned again against the trade, once, and held", len(lines(log)) == 3 and held_from([r for r in rows if r[0] > 400.0])) check("...and the loop keeps it when the trade leaves (no line)", all(x[0] < 900.0 for x in lines(log)) and refused(nth(rows, -1))) # The writer asks the other way, which the empty pack takes. rows, log, _ = run(1, plant=empty_a, house=lambda t: 2000.0, maint=False, trade=lambda t: -1500.0 if t >= 300.0 else None) e = ends(log) check("a writer that asks the other way ends it ('another writer')", len(e) == 1 and "another writer" in e[0][1] and present(nth(rows, -1)) and not refused(nth(rows, -1))) # b. The full pack starts taking charge again under the same writer. rows, log, _ = run(1, plant=lambda u, t, cmd: ( round(cmd) if t >= 400.0 else full_a(u, t, cmd)), house=lambda t: 800.0) e = ends(log) print(" " + " / ".join(f"{x[0]:.0f} s {x[1][:45]}" for x in lines(log))) check("the battery takes charge again: the refusal ends once, on its reading", len(e) == 1 and "the battery gives power" in e[0][1] and 400.0 <= e[0][0] < 430.0 and present(nth(rows, -1)) and not refused(nth(rows, -1))) # c. Stop and start still end it. rows, log, _ = run(1, plant=full_a, house=lambda t: 800.0, auto=lambda t: not 500.0 <= t < 520.0) e = ends(log) check("stopping control still ends it", any("control stopped" in x[1] for x in e)) # d. The probe clock. Learned under the charge; the cheap window closes when # the refusal is 910 s old, and the loop holds the wire (a surplus, so it # wants to charge). The first probe is a full PROBE_EVERY_S after the # window closed, as before: never at once. close = [None] def window(c, t): if close[0] is None and c.batt_refused: close[0] = t + 910.0 return close[0] is None or t < close[0] rows, log, _ = run(1, plant=full_a, house=lambda t: -1500.0, cheap=window, dur=3000.0) probes = [r[0] for r in rows if r[3] == control.PROBING] print(" " + " / ".join(f"{x[0]:.0f} s {x[1][:45]}" for x in lines(log))) print(f" window closed at {close[0]} s, first probe at " f"{probes[0] if probes else None} s") check("the refusal carries over to the loop with no line", len(lines(log)) == 1 and held_from(rows)) check("the first probe after the writer left comes PROBE_EVERY_S later", close[0] is not None and bool(probes) and probes[0] >= close[0] + control.PROBE_EVERY_S - 10.0) # e. A writer turns round on a dead pack with no gap. The evidence restarts # on the new sign (SAFETY-94), so the old refusal ends at once and the # other way is learned once, ~12 cycles later. rows, log, _ = run(1, plant=lambda u, t, cmd: ( -2.0 - int(t) % 3 if u == "a" else round(cmd)), house=lambda t: 2000.0, maint=False, trade=lambda t: 2000.0 if t < 300.0 else -2000.0) e = ends(log) print(" " + " / ".join(f"{x[0]:.0f} s {x[1][:45]}" for x in lines(log))) check("a writer turns round on a dead pack: one end, then the other way", len(lines(log)) == 3 and len(e) == 1 and 300.0 < e[0][0] < 400.0 and present(nth(rows, -1)) and nth(rows, -1)[4] == -1) # ============================================================================== print("\n4. not covered: a perfectly flat reading under a writer") # ============================================================================== # The stale-reading gate trips on a flat reading while the command moves, # stops the site and seeds the refusal (SAFETY-52 decision 3). After the stop # the writer's ask comes back fresh, compute() has no saturation history # against it yet, so the refusal ends at the first cycle and the next trip # seeds it again: one pair per gate cycle, as on release/1.0 (the commands # are the same; DECISIONS/SAFETY-107 "Not covered"). Pinned so a change shows. rows, log, _ = run(1, plant=flat_a, house=lambda t: -1500.0) trips = len(lines(log, "STALE inverter")) print(f" flat reading, house -1500 W: {trips} gate trips, " f"{len(lines(log))} refusal lines in 15 min") check("flat reading: no per-cycle flap, at most one pair per gate trip", trips >= 1 and len(lines(log)) <= 2 * trips) print() if fails: print(f"{len(fails)} FAILED: " + ", ".join(fails)) sys.exit(1) print("all SAFETY-107 checks passed")