308 lines
14 KiB
Python
308 lines
14 KiB
Python
"""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")
|