Claude
d89c093318
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019DBUfvndgHwaV339Fvd2gr
548 lines
24 KiB
Python
548 lines
24 KiB
Python
"""Runnable check for SAFETY-87: the peak guard uses every battery that can
|
|
still give. `python3 test_safety87.py`
|
|
|
|
At N>=2 the guard graded the EMPTIEST battery (SAFETY-32), so one battery under
|
|
the peak reserve stopped the guard for all of them: it asked for nothing and
|
|
raised the reserve alarm while the other battery was full. Now a battery at or
|
|
under its own reserve is HELD (it gives nothing for the peak, weight 0) and
|
|
the guard's need is placed on the others; the reserve alarm fires only when
|
|
they cannot hold the limit. Drives the REAL Controller (tools/
|
|
safety84_identity.py's harness). DECISIONS/SAFETY-87-guard-uses-able-batteries.md.
|
|
|
|
1. Maintenance charge at N=2 (A under the reserve, B able), house over the
|
|
aim: no reserve alarm, the quarter on the aim, A gives nothing for the
|
|
peak while B can carry it. When B cannot, the alarm fires and names A.
|
|
2. HOLD at N=2 (A full holding, B under the reserve): A gives, B gives
|
|
nothing, the quarter lands on the aim; at a banked month peak too.
|
|
3. Both under the reserve, or the other one refusing: the alarm still fires
|
|
and names the right battery.
|
|
4. The alarm at the quarter's edge: a one-cycle miss does not fire it.
|
|
5. place_on_able() at its boundaries, and the claim it files.
|
|
6. The allocator: a held battery gives last, after an Off one.
|
|
7. A held battery draining for maintenance counts for its drain.
|
|
8. The page: a peak alarm does not say "Not ready" or "0 W until the alarm
|
|
clears" while control goes on.
|
|
9. N=1 is byte-identical to release/1.0 b8ddfd7 (tools/safety87_identity.py).
|
|
|
|
Mutations (MUTATION.md, purged copies) are listed in the pull request.
|
|
"""
|
|
|
|
import asyncio
|
|
import dataclasses
|
|
import json
|
|
import logging
|
|
import os
|
|
import sys
|
|
import types
|
|
|
|
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 safety84_identity as s84 # noqa: E402
|
|
import safety87_identity as s87 # noqa: E402
|
|
from app import messages, packmode, peakguard # noqa: E402
|
|
from app.arbiter import (Claim, P_MODE, P_PEAK, P_SAFETY, # noqa: E402
|
|
UnitBounds, resolve)
|
|
from app.main import Controller # noqa: E402
|
|
from app.maintenance import DRAIN, HOLD, IDLE, MaintResult # noqa: E402
|
|
from app.store import unit_key # noqa: E402
|
|
|
|
fails = []
|
|
A, B = "inverter-a", "inverter-b"
|
|
AIM = 3450.0 # peak_cap_w 3500 - peak_margin_w 50
|
|
TOL = 10.0 # W over the aim a banked quarter may show (SAFETY-81)
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
def store(phase, *, b_off=False):
|
|
"""Battery A in `phase` from the first cycle (B idle, cycle done)."""
|
|
s = rig._Store()
|
|
s.data[unit_key(A, "phase")] = phase
|
|
s.data[unit_key(A, "phase_started")] = rig.BASE.isoformat()
|
|
s.data[unit_key(B, "last_completed")] = rig.BASE.isoformat()
|
|
if phase == IDLE:
|
|
s.data[unit_key(A, "last_completed")] = rig.BASE.isoformat()
|
|
if b_off:
|
|
s.data[unit_key(B, packmode.USE_KEY)] = False
|
|
return s
|
|
|
|
|
|
def run(house, phase, socs, *, cheap=False, n=2, **kw):
|
|
"""Rows, controller, the notifications raised and one record per cycle:
|
|
(floor_limited, held, floor_w, reserve set, alarm latched)."""
|
|
notes, seen = [], []
|
|
real_check = Controller.check_peak_floor_limited
|
|
|
|
async def spy(self):
|
|
await real_check(self)
|
|
g = self.peak_guard
|
|
seen.append(None if g is None else (
|
|
g.floor_limited, g.held, g.floor_w, self._peak_reserve(),
|
|
self._peak_floor_limited))
|
|
|
|
def before(c, t):
|
|
if not getattr(c.hass, "_s87", False):
|
|
real = c.hass.call
|
|
|
|
async def call(*a):
|
|
if a[:2] == ("persistent_notification", "create"):
|
|
notes.append(a[2])
|
|
return await real(*a)
|
|
c.hass.call, c.hass._s87 = call, True
|
|
return s84.feed(cheap)(c, t)
|
|
|
|
kw.setdefault("dur", 2700.0)
|
|
extra = dict({"maintenance_hold_min": 600}, **kw.pop("extra", {}))
|
|
Controller.check_peak_floor_limited = spy
|
|
try:
|
|
rows, c = s84.run(
|
|
n, load=house if callable(house) else lambda t: house,
|
|
cheap=cheap, store=kw.pop("store", None) or store(phase),
|
|
soc=lambda u, t: socs[u], before=before, extra=extra, **kw)
|
|
finally:
|
|
Controller.check_peak_floor_limited = real_check
|
|
return rows, c, notes, seen
|
|
|
|
|
|
def without_fix(fn):
|
|
"""The shipped b8ddfd7 behaviour: no battery is ever held back."""
|
|
real = Controller._peak_on_able
|
|
Controller._peak_on_able = lambda self, decision, results: decision
|
|
try:
|
|
return fn()
|
|
finally:
|
|
Controller._peak_on_able = real
|
|
|
|
|
|
def cmd(r):
|
|
return dict(r[2])
|
|
|
|
|
|
def banked(c):
|
|
return [b.offtake_avg_w for b in c.p1.blocks]
|
|
|
|
|
|
def on_aim(c, aim=AIM):
|
|
b = banked(c)
|
|
return len(b) >= 2 and all(aim - TOL <= v <= aim + TOL for v in b)
|
|
|
|
|
|
def reserve_notes(notes):
|
|
return [x for x in notes
|
|
if x["notification_id"] == messages.ID_PEAK_RESERVE]
|
|
|
|
|
|
def limited_events(c):
|
|
return sum(1 for e in c.events if "PEAK_GUARD_FLOOR_LIMITED" in e)
|
|
|
|
|
|
def show(tag, rows, c, notes, seen):
|
|
a = [cmd(r)["a"] for r in rows]
|
|
b = [cmd(r)["b"] for r in rows]
|
|
print(f" {tag}: banked {[round(v, 1) for v in banked(c)]} W, mean A "
|
|
f"{sum(a) / len(a):.0f} W, B {sum(b) / len(b):.0f} W, reserve "
|
|
f"alarms {len(reserve_notes(notes))}, held "
|
|
f"{sorted({h for s in seen if s for h in s[1]})}")
|
|
|
|
|
|
# ============================================================================ #
|
|
print("1. maintenance charge at N=2: A under the reserve, B able")
|
|
LOW_A = {"a": 11.0, "b": 60.0}
|
|
CH = run(4500.0, "charge", LOW_A, cheap=True)
|
|
CH_OLD = without_fix(lambda: run(4500.0, "charge", LOW_A, cheap=True))
|
|
show("house 4500", *CH)
|
|
show("house 4500, before the fix", *CH_OLD)
|
|
check("⭐ house 4500 W, B can carry the guard alone: no reserve alarm (no "
|
|
"notification, no PEAK_GUARD_FLOOR_LIMITED line) and every quarter on "
|
|
"the aim",
|
|
not reserve_notes(CH[2]) and limited_events(CH[1]) == 0
|
|
and on_aim(CH[1]))
|
|
check("...A is held (it gives nothing for the peak) on every cycle the guard "
|
|
"asks for power, and the guard asks for a positive floor",
|
|
any(s and s[2] and s[2] > 0.0 for s in CH[3])
|
|
and all(s[1] == (A,) and s[3] == {A} for s in CH[3]
|
|
if s and s[2] and s[2] > 0.0))
|
|
check("non-vacuous: before the fix the same run raised the reserve alarm "
|
|
"(the SAFETY-84 rig test's check 2/3 false alarm)",
|
|
len(reserve_notes(CH_OLD[2])) >= 1 and limited_events(CH_OLD[1]) >= 1)
|
|
_b_carry = [r for r, s in zip(CH[0], CH[3])
|
|
if s and s[3] == {A} and r[1] <= 2500.0 and r[1] > 0.0]
|
|
check("...and while B can carry the site target (at most its 2500 W) A is "
|
|
"not asked to give at all (weight 0 behind B)",
|
|
len(_b_carry) >= 3 and all(cmd(r)["a"] <= 0.0 for r in _b_carry))
|
|
|
|
CH65 = run(6500.0, "charge", LOW_A, cheap=True)
|
|
show("house 6500", *CH65)
|
|
_n65 = reserve_notes(CH65[2])
|
|
check("⭐ house 6500 W, B (2500 W) cannot hold the 3450 W aim alone: the "
|
|
"reserve alarm fires and names A, the battery under its reserve, and "
|
|
"says the others cannot hold the limit alone",
|
|
len(_n65) >= 1
|
|
and all("inverter-a is at its peak reserve" in x["message"]
|
|
and "inverter-b" not in x["message"]
|
|
and "cannot keep this 15-minute period under your 3 500 W peak "
|
|
"limit alone" in x["message"] for x in _n65))
|
|
check("...and while it is limited the guard still asks B for everything B "
|
|
"has (floor 2500 W), so B is never left idle",
|
|
any(s and s[0] and s[2] == 2500.0 for s in CH65[3])
|
|
and all(s[2] == 2500.0 for s in CH65[3] if s and s[0]))
|
|
|
|
# ============================================================================ #
|
|
print("\n2. HOLD at N=2: A full and holding, B under the reserve")
|
|
LOW_B = {"a": 99.0, "b": 15.0}
|
|
HD = run(5500.0, HOLD, LOW_B)
|
|
HD_OLD = without_fix(lambda: run(5500.0, HOLD, LOW_B))
|
|
show("house 5500", *HD)
|
|
show("house 5500, before the fix", *HD_OLD)
|
|
check("⭐ house 5500 W: A gives, B gives nothing, every quarter lands on the "
|
|
"aim, no reserve alarm, and A stays in its HOLD",
|
|
on_aim(HD[1]) and not reserve_notes(HD[2])
|
|
and limited_events(HD[1]) == 0
|
|
and all(cmd(r)["b"] == 0.0 for r in HD[0])
|
|
and max(cmd(r)["a"] for r in HD[0]) > 2000.0
|
|
and HD[1].maints[0].phase == HOLD)
|
|
check("non-vacuous: before the fix A sat at 0 W, B (at 15 %) gave through "
|
|
"the loop and the reserve alarm fired (SAFETY-86 rig case 7a)",
|
|
all(cmd(r)["a"] == 0.0 for r in HD_OLD[0])
|
|
and max(cmd(r)["b"] for r in HD_OLD[0]) > 2000.0
|
|
and len(reserve_notes(HD_OLD[2])) >= 1)
|
|
|
|
HD65 = run(6500.0, HOLD, LOW_B)
|
|
HD65_OLD = without_fix(lambda: run(6500.0, HOLD, LOW_B))
|
|
show("house 6500", *HD65)
|
|
show("house 6500, before the fix", *HD65_OLD)
|
|
_n = reserve_notes(HD65[2])
|
|
check("⭐ house 6500 W, A alone (2500 W) cannot hold the aim: A gives all it "
|
|
"has, B still gives nothing, and the reserve alarm fires ONCE naming B",
|
|
all(cmd(r)["b"] == 0.0 for r in HD65[0])
|
|
and max(cmd(r)["a"] for r in HD65[0]) == 2500.0
|
|
and len(_n) == 1 and "inverter-b is at its peak reserve" in
|
|
_n[0]["message"] and "inverter-a" not in _n[0]["message"])
|
|
check("...and the quarter is no worse than before the fix (4051 W in the "
|
|
"SAFETY-86 review), with B kept at its reserve",
|
|
max(banked(HD65[1])) <= max(banked(HD65_OLD[1])) + 1.0)
|
|
|
|
BK = run(6000.0, HOLD, LOW_B, month_peak=4000.0)
|
|
show("house 6000, month peak 4000", *BK)
|
|
check("⭐ under a banked month peak of 4000 W the aim is 3950 W (peak - "
|
|
"margin): A gives, B nothing, every quarter on 3950 W, no alarm",
|
|
on_aim(BK[1], 3950.0) and not reserve_notes(BK[2])
|
|
and all(cmd(r)["b"] == 0.0 for r in BK[0]))
|
|
|
|
# ============================================================================ #
|
|
print("\n3. a real loss of peak limiting still alarms")
|
|
BOTH = run(5500.0, IDLE, {"a": 15.0, "b": 12.0})
|
|
show("both under", *BOTH)
|
|
_nb = reserve_notes(BOTH[2])
|
|
check("⭐ both batteries under the reserve: the reserve alarm fires and "
|
|
"names both, the guard asks for nothing (claim lo 0 W, 'SoC floor')",
|
|
len(_nb) >= 1
|
|
and all("inverter-a and inverter-b" in x["message"] for x in _nb)
|
|
and all(not s[2] for s in BOTH[3] if s and s[0])
|
|
and any(c[0] == "peak" and c[4] == 0.0 and c[6] == "SoC floor"
|
|
for r in BOTH[0] for c in r[-1][0]))
|
|
|
|
|
|
_ctl2 = BOTH[1]
|
|
for _inv, _s, _r in zip(_ctl2.inverters, (15.0, 60.0), (0, 1)):
|
|
_inv.soc, _inv.refused = _s, _r # SAFETY-73: B refuses to give
|
|
_ref = _ctl2._peak_on_able(_lim_all := peakguard.PeakDecision(
|
|
engaged=True, floor_w=0.0, ceiling_w=3500.0,
|
|
ceiling_from=peakguard.CEILING_CAP, projected_w=4000.0, excess_w=550.0,
|
|
wanted_w=2000.0, target_w=3450.0, floor_limited=True, soc=15.0,
|
|
soc_floor=20.0, soc_floor_reason=peakguard.FLOOR_AT_SOC_FLOOR,
|
|
reason="PEAK_GUARD_FLOOR_LIMITED: x"),
|
|
[MaintResult(IDLE, False, None, False, [])] * 2)
|
|
check("⭐ A under the reserve and B refusing to give (SAFETY-73): nothing can "
|
|
"give for the peak, so it stays limited at 0 W (the alarm fires) and "
|
|
"only A is named (test_peakguard.py part 12h drives the notification)",
|
|
_ref.floor_limited and _ref.floor_w == 0.0 and _ref.held == (A,)
|
|
and _ref.able_w is None and _ctl2._peak_reserve() == frozenset())
|
|
_ctl2.inverters[1].refused = 0
|
|
_ctl2.live_units = lambda: (A,) # B dropped from the split
|
|
_drop = _ctl2._peak_on_able(_lim_all, [MaintResult(IDLE, False, None, False,
|
|
[])] * 2)
|
|
del _ctl2.live_units
|
|
check("...and the same with B dropped (not live): nothing can give, the "
|
|
"alarm state stands at 0 W, only A named",
|
|
_drop.floor_limited and _drop.floor_w == 0.0 and _drop.held == (A,)
|
|
and _drop.able_w is None)
|
|
|
|
|
|
async def _notify(ctl, socs, cutouts, refused):
|
|
"""The shipped check_peak_floor_limited() on `ctl`, its notification."""
|
|
sent = []
|
|
|
|
async def call(*a):
|
|
sent.append(a)
|
|
return True
|
|
for inv, so, cut, ref in zip(ctl.inverters, socs, cutouts, refused):
|
|
inv.soc, inv.bms_soc_floor, inv.refused = so, cut, ref
|
|
real, ctl.hass.call = ctl.hass.call, call
|
|
ctl.peak_guard = ctl._peak_on_able(dataclasses.replace(
|
|
_lim_all, soc=min(socs), soc_floor=peakguard.binding_soc_floor(
|
|
ctl.packs, 20.0)), [MaintResult(IDLE, False, None, False, [])] * 2)
|
|
ctl._peak_floor_limited = False
|
|
try:
|
|
await ctl.check_peak_floor_limited()
|
|
finally:
|
|
ctl.hass.call = real
|
|
for inv in ctl.inverters:
|
|
inv.bms_soc_floor, inv.refused = 10, 0
|
|
return [a[2]["message"] for a in sent
|
|
if a[2].get("notification_id") == messages.ID_PEAK_RESERVE]
|
|
|
|
|
|
# Cut-outs A 10 % / B 35 % (simulated; rule 8 refuses it, sim/inject.py does
|
|
# not run rule 8), A at 25 % refusing, B at 30 %. Only B is at its own floor;
|
|
# A is above its own but under the site's highest (35 %).
|
|
_msg = asyncio.run(_notify(_ctl2, (25.0, 30.0), (10, 35), (1, 0)))
|
|
print(f" differing cut-outs: {_msg}")
|
|
check("⭐ the alarm names the battery at ITS OWN floor (B, 30 % <= 35 %), not "
|
|
"every battery under the site's highest floor (A at 25 % is above its "
|
|
"own 20 %)",
|
|
len(_msg) == 1 and "inverter-b is at its peak reserve" in _msg[0]
|
|
and "inverter-a" not in _msg[0])
|
|
|
|
|
|
# ============================================================================ #
|
|
print("\n4. the reserve alarm at the quarter's edge")
|
|
def spiky(t):
|
|
"""A 5000 W house (B alone can hold the aim) with a 5 s spike to
|
|
11000 W every 150 s, which B alone cannot."""
|
|
return 11000.0 if t % 150 < 5 else 5000.0
|
|
|
|
|
|
LOW_A_IDLE = {"a": 15.0, "b": 60.0}
|
|
SP = run(spiky, IDLE, LOW_A_IDLE)
|
|
_flips = sum(1 for s in SP[3] if s and s[0])
|
|
show("spiky house", *SP)
|
|
print(f" {_flips} cycles floor-limited (B alone short during a spike)")
|
|
check("⭐ a 5 s spike that B alone cannot cover makes the guard limited for a "
|
|
"cycle, and that does not fire the alarm: the miss must hold for "
|
|
"ALARM_HOLD_S, as for 'Load too high'. No quarter goes over the aim",
|
|
_flips >= 5 and not reserve_notes(SP[2])
|
|
and not any(s[4] for s in SP[3] if s)
|
|
and len(banked(SP[1])) >= 2 and max(banked(SP[1])) <= AIM)
|
|
|
|
|
|
def ungated(fn):
|
|
real = Controller._reserve_short
|
|
Controller._reserve_short = lambda self, g: True
|
|
try:
|
|
return fn()
|
|
finally:
|
|
Controller._reserve_short = real
|
|
|
|
|
|
_ug = ungated(lambda: run(spiky, IDLE, LOW_A_IDLE))
|
|
check("non-vacuous: without the hold the same run raises the alarm",
|
|
len(reserve_notes(_ug[2])) >= 1)
|
|
|
|
# ============================================================================ #
|
|
print("\n5. place_on_able() at its boundaries")
|
|
_lim = peakguard.PeakDecision(
|
|
engaged=True, floor_w=0.0, ceiling_w=3500.0,
|
|
ceiling_from=peakguard.CEILING_CAP, projected_w=4000.0, excess_w=550.0,
|
|
wanted_w=2000.0, target_w=3450.0, floor_limited=True, soc=15.0,
|
|
soc_floor=20.0, soc_floor_reason=peakguard.FLOOR_AT_SOC_FLOOR,
|
|
reason="PEAK_GUARD_FLOOR_LIMITED: x")
|
|
|
|
|
|
def place(d=_lim, held=("b",), able_w=2000.0, soc=60.0, floor=20.0):
|
|
return peakguard.place_on_able(d, held, able_w=able_w, soc=soc,
|
|
soc_floor=floor)
|
|
|
|
|
|
_eq, _short = place(), place(able_w=1999.0)
|
|
check("⭐ the others can give exactly the need (2000 W): not limited, floor "
|
|
"2000 W, the claim reads 'quarter peak' (the page says 'the peak "
|
|
"limit')",
|
|
not _eq.floor_limited and _eq.engaged and _eq.floor_w == 2000.0
|
|
and _eq.held == ("b",) and _eq.able_w == 2000.0
|
|
and peakguard.claim(_eq).lo == 2000.0
|
|
and peakguard.claim(_eq).reason == "quarter peak"
|
|
and _eq.reason.startswith("PEAK_GUARD: "))
|
|
check("...1 W short: limited, but the claim still asks them for all they "
|
|
"have (1999 W), not 0 W",
|
|
_short.floor_limited and _short.floor_w == 1999.0
|
|
and peakguard.claim(_short).lo == 1999.0
|
|
and peakguard.claim(_short).reason == "quarter peak"
|
|
and _short.reason.startswith("PEAK_GUARD_FLOOR_LIMITED: "))
|
|
check("...the others' own permission is graded: at or under their floor, or "
|
|
"unread, nothing is asked (0 W, 'SoC floor')",
|
|
all(place(soc=s).floor_w == 0.0 and place(soc=s).floor_limited
|
|
and peakguard.claim(place(soc=s)).reason == "SoC floor"
|
|
for s in (20.0, 19.0, None)))
|
|
check("...the roster's verdict is kept as decide() echoed it (soc 15 %, "
|
|
"floor 20 %), and a decision that was not limited is only labelled",
|
|
_eq.soc == 15.0 and _eq.soc_floor == 20.0
|
|
and all(place(d=d) == dataclasses.replace(d, held=("b",))
|
|
for d in (dataclasses.replace(
|
|
_lim, floor_limited=False, engaged=False,
|
|
floor_w=None),
|
|
dataclasses.replace(_lim, wanted_w=None))))
|
|
check("...and at N=1 the floor-limited claim is unchanged: 0 W, 'SoC floor'",
|
|
peakguard.claim(_lim).lo == 0.0
|
|
and peakguard.claim(_lim).reason == "SoC floor")
|
|
|
|
|
|
|
|
def three(socs, cuts):
|
|
"""`_peak_on_able()` on a stub roster of three (SIMULATED packs)."""
|
|
names = ("x", "y", "z")
|
|
invs = [types.SimpleNamespace(name=n, refused=0, rating_w=lambda: 2500.0)
|
|
for n in names]
|
|
stub = types.SimpleNamespace(
|
|
inverters=invs, o={"peak_soc_floor": 20},
|
|
packs=[peakguard.Pack(n, so, c) for n, so, c in zip(names, socs, cuts)],
|
|
live_units=lambda: names)
|
|
return Controller._peak_on_able(stub, _lim, [MaintResult(
|
|
IDLE, False, None, False, [])] * 3)
|
|
|
|
|
|
# x 25 % (cut-out 10 -> floor 20), y 40 % (cut-out 30), z 15 % (held). x and
|
|
# y each clear their own floor, but 25 % is not above y's 30 %: the pair is
|
|
# graded min SoC against max floor (SAFETY-32), so nothing is asked.
|
|
_t3 = three((25.0, 40.0, 15.0), (10, 30, 10))
|
|
_t3ok = three((35.0, 40.0, 15.0), (10, 30, 10))
|
|
check("⭐ three batteries, z held: the two others are graded as a pair, the "
|
|
"EMPTIEST of them (25 %) against the HIGHEST of their floors (30 %), "
|
|
"so nothing is asked; at 35 % they are asked for the need",
|
|
_t3.held == ("z",) and _t3.floor_limited and _t3.floor_w == 0.0
|
|
and _t3ok.held == ("z",) and not _t3ok.floor_limited
|
|
and _t3ok.floor_w == 2000.0)
|
|
|
|
# ============================================================================ #
|
|
print("\n6. the allocator: a held battery gives last")
|
|
|
|
|
|
def split(target, reserve, modes=(), floor=1500.0):
|
|
cl = [Claim.set("loop", 10, target),
|
|
Claim.limit("supervised", P_SAFETY, -2500.0, 2500.0, actuator=A),
|
|
Claim.limit("supervised", P_SAFETY, -2500.0, 2500.0, actuator=B),
|
|
Claim.limit("peak", P_PEAK, lo=min(target, floor))]
|
|
cl += [Claim.limit("mode", P_MODE, 0.0, 0.0, "off", actuator=u)
|
|
for u in modes]
|
|
r = resolve(cl, units=(A, B), live=(A, B), weights={A: 1.0, B: 3.0},
|
|
reserve=reserve)
|
|
return {u: round(v) for u, v in r.shares.items()}
|
|
|
|
|
|
check("⭐ B held: A carries the site target first (B's weight 3.0 ignored) "
|
|
"and B only takes what A cannot: 1500 -> A 1500 / B 0, 4000 -> A 2500 "
|
|
"/ B 1500",
|
|
split(1500.0, {B}) == {A: 1500, B: 0}
|
|
and split(4000.0, {B}) == {A: 2500, B: 1500})
|
|
check("...no one held: the weights decide, as before",
|
|
split(1500.0, set()) == {A: 375, B: 1125})
|
|
check("...charging is not touched: a held battery may still charge first",
|
|
split(-2000.0, {B}) == split(-2000.0, set()))
|
|
check("⭐ A Off and B under the reserve, a 3000 W floor B cannot place: the "
|
|
"Off battery gives first (2500 W) and B only the last 500 W - not an "
|
|
"equal 1500/1500 split",
|
|
split(3000.0, {B}, modes=(A,), floor=3000.0) == {A: 2500, B: 500})
|
|
check("...OPEN, recorded: when B alone CAN place the floor, OPT-11 keeps the "
|
|
"Off battery resting and B gives it (1500 W). See the DECISIONS entry",
|
|
split(1500.0, {B}, modes=(A,)) == {A: 0, B: 1500})
|
|
check("...and with B Off and nobody held, Off still gives way to the other "
|
|
"battery, as before (OPT-11)",
|
|
split(1500.0, set(), modes=(B,)) == {A: 1500, B: 0})
|
|
|
|
# ============================================================================ #
|
|
print("\n7. a held battery draining for maintenance")
|
|
_ctl = HD[1]
|
|
|
|
|
|
def able_w(results, socs=(15.0, 60.0)):
|
|
for inv, s in zip(_ctl.inverters, socs):
|
|
inv.soc, inv.refused = s, 0
|
|
d = _ctl._peak_on_able(dataclasses.replace(
|
|
_lim, wanted_w=6000.0), results)
|
|
return d.able_w, d.held, d.floor_w
|
|
|
|
|
|
_idle = [MaintResult(IDLE, False, None, False, [])] * 2
|
|
_drain = [MaintResult(DRAIN, True, 2000.0, False, []),
|
|
MaintResult(IDLE, False, None, False, [])]
|
|
print(f" A at 15 % idle: {able_w(_idle)}; draining 2000 W: "
|
|
f"{able_w(_drain)}")
|
|
check("⭐ A under the reserve but draining 2000 W for its monthly cycle: "
|
|
"that drain still counts (able 4500 W = B 2500 + 2000), so B is not "
|
|
"asked to carry it; idle, only B's 2500 W",
|
|
able_w(_idle) == (2500.0, (A,), 2500.0)
|
|
and able_w(_drain) == (4500.0, (A,), 4500.0))
|
|
|
|
# ============================================================================ #
|
|
print("\n8. the page while a peak alarm stands and control goes on")
|
|
_rows65, _c65 = CH65[0], CH65[1]
|
|
_c65._peak_floor_limited = True
|
|
_c65.peak_guard = dataclasses.replace(
|
|
_lim, floor_w=2500.0, held=(A,), able_w=2500.0)
|
|
_pay = _c65.status_payload()
|
|
_txt = json.dumps(_pay, ensure_ascii=False)
|
|
_row = [r for r in _pay["checks"] if r.get("id") == "peak_guard"]
|
|
print(f" banner: {_pay['banner']}; row: {_row[0]['text'] if _row else '-'}")
|
|
check("⭐ the banner is the alarm's title without 'Not ready', and nothing "
|
|
"on the page says the inverters go to 0 W or that an alarm stopped "
|
|
"control",
|
|
_pay["banner"] == messages.T_PEAK_RESERVE and _pay["level"] == "bad"
|
|
and "Not ready" not in _txt and "until the alarm clears" not in _txt
|
|
and "An alarm stopped control" not in _txt)
|
|
check("...the row names A and says the others cannot hold the limit alone, "
|
|
"in three short sentences",
|
|
len(_row) == 1 and _row[0]["text"].startswith(
|
|
"inverter-a battery at its peak reserve. The others cannot hold "
|
|
"the 3 500 W peak limit alone."))
|
|
_c65._peak_floor_limited = False
|
|
_blip = [r for r in _c65.checks() if r.get("id") == "peak_guard"]
|
|
_c65._peak_floor_limited = True
|
|
check("...and while the alarm is not latched (a one-cycle miss) the row is "
|
|
"the ordinary amber guard row, not the red reserve row",
|
|
len(_blip) == 1 and _blip[0]["warn"] and not _blip[0].get("runs")
|
|
and "peak reserve" not in _blip[0]["text"])
|
|
_c65.auto = False
|
|
_stopped = _c65.status_payload()
|
|
_c65.auto = True
|
|
check("...a stopped controller still says so: the peak alarm does not take "
|
|
"the banner from it", _stopped["banner"] != messages.T_PEAK_RESERVE
|
|
and _stopped["level"] != "bad")
|
|
|
|
# ============================================================================ #
|
|
print("\n9. N=1 is byte-identical to release/1.0 b8ddfd7")
|
|
PINNED = { # tools/safety87_identity.py on b8ddfd7
|
|
"n1_under":
|
|
"d2395ef8d74fb23607ad864d5a4bf62c03044bbd9367e873675d7ad10e8c169f",
|
|
"n1_crossing":
|
|
"97f7321017d9d34ee17a40dfd4abf73d915f25b9faef086a483d2b5cd124b198",
|
|
"n1_above":
|
|
"45726a285aa84f40542d56482652397c3c1e212ba4fe2009f6cc52f537ccad52",
|
|
"n1_under_banked":
|
|
"a2876c7cc873bc4c974d2b4bd98e23bba2d007b84545587d2fe0b713b0a6249b",
|
|
"n1_charge_low":
|
|
"91336e3c0314985a3b00f314c79b5ffb5a740d218de3ad44f88a4db6e4235b32",
|
|
"n1_hold":
|
|
"0a1df347754c69166b20d04c15ee878bbcefb02abb824b9b452334f68f03e46d",
|
|
}
|
|
_now = s87.digests()
|
|
for k in PINNED:
|
|
print(f" {k} {'same' if _now[k] == PINNED[k] else 'DIFFERS'}")
|
|
check("⭐ every N=1 digest (claims, commands, the guard's decision and the "
|
|
"log) equals b8ddfd7's, under, crossing and above the reserve, in a "
|
|
"charge and a HOLD, and under a banked peak",
|
|
_now == PINNED)
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED: " + ", ".join(fails))
|
|
sys.exit(1)
|
|
print("all SAFETY-87 checks passed")
|