Claude
5d27f8496f
At two inverters the drain pins its pack and the site target is split around the pin. Below the pin the other pack had to charge, the allocator refused the opposite directions, and the site sat at 0 W with an ARBITER line every cycle (house below or above the drain rate, or a grid charge). main.py now files a site limit `drain` (lo = what the pin places, at P_MAINTENANCE) at N>=2 while a pack drains, so the other pack only adds power in the same direction. A drain that cannot give power waits with one log line. With battery modes on, no grid charge while a pack drains, and the other battery's card says why. N=1 is byte-identical. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
111 lines
4.0 KiB
Python
111 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
"""SAFETY-78: at one inverter, is a maintenance drain byte-identical to before?
|
|
|
|
python3 tools/safety78_identity.py # prints one digest per scenario
|
|
|
|
tools/safety79_identity.py's harness (the REAL Controller on a P1 meter, so
|
|
the peak guard runs) with the battery draining for its monthly cycle from the
|
|
first cycle, under a house below, at and above the drain rate, a solar
|
|
surplus and a load over the peak limit. Each row hashes safety79's fields,
|
|
the modes' roles and the trade, the action log's newest line, and every claim
|
|
main.py hands `resolve()`, so a claim or a log line added at N=1 that changes
|
|
no watt still moves the digest. The `_modes` scenarios run battery modes on,
|
|
Cheap hours only, over opt11p3_identity's prices. Uses only the API release/1.0 5a4cdc2 has, so the same file runs on
|
|
both trees; test_safety78.py part 5 holds this tree to the digests printed
|
|
there.
|
|
"""
|
|
|
|
import functools
|
|
import hashlib
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
import opt11_identity as rig # noqa: E402
|
|
import opt11p3_identity as p3 # noqa: E402
|
|
import safety79_identity as s79 # noqa: E402
|
|
import app.main as m # noqa: E402
|
|
from app import packmode # noqa: E402
|
|
from app.store import unit_key # noqa: E402
|
|
|
|
DUR = 2400.0
|
|
DRAIN = {"maintenance_enabled": True, "maintenance_start_hour": 11}
|
|
|
|
|
|
def load(t):
|
|
"""Below the 2000 W drain, at it, above it, a surplus, over the limit."""
|
|
k = int(t // 480)
|
|
return (1200.0, 2000.0, 3000.0, -1500.0, 6000.0)[min(k, 4)]
|
|
|
|
|
|
def store_cheap(n):
|
|
s = rig._Store()
|
|
for u in ("inverter-a", "inverter-b")[:n]:
|
|
s.data[unit_key(u, packmode.MODE_KEY)] = packmode.CHEAP_HOURS
|
|
return s
|
|
|
|
|
|
SCENARIOS = {
|
|
"n1_drain": dict(n=1),
|
|
"n1_drain_modes": dict(n=1, modes=True),
|
|
}
|
|
|
|
|
|
def run(n, *, modes=False, dur=DUR):
|
|
extra = dict(s79.P1, **DRAIN)
|
|
if modes:
|
|
extra.update(p3.ARB, battery_modes_enabled=True)
|
|
seen, extra_rows = [], []
|
|
real_resolve = m.resolve
|
|
|
|
def spy(claims, **kw):
|
|
seen.append(tuple((c.source, c.priority, c.kind, c.value, c.lo, c.hi,
|
|
c.reason, c.actuator) for c in claims))
|
|
return real_resolve(claims, **kw)
|
|
|
|
def on_cycle(c, t):
|
|
g, tick, a = c.peak_guard, c.mode_tick, c.arb
|
|
extra_rows.append((
|
|
seen[-1] if seen else None,
|
|
None if g is None else (g.engaged, g.floor_w, g.reason),
|
|
None if tick is None else (tuple(sorted(tick.roles.items())),
|
|
tick.action, tick.why),
|
|
None if a is None else (a.action, a.power_w, a.reason),
|
|
[mm.phase for mm in c.maints],
|
|
len(c.events), c.events[0] if c.events else None))
|
|
db = os.path.join(tempfile.mkdtemp(prefix="s78-"), "tariff.db")
|
|
real = rig.Controller
|
|
rig.Controller = functools.partial(real, peak_db_path=db)
|
|
m.resolve = spy
|
|
try:
|
|
rows, c = rig.run(n, extra=extra, dur=dur, load=load, soc=s79.soc,
|
|
cut=lambda u, t: False,
|
|
before=p3._feed if modes else s79._feed,
|
|
on_cycle=on_cycle,
|
|
store=store_cheap(n) if modes else None)
|
|
finally:
|
|
rig.Controller = real
|
|
m.resolve = real_resolve
|
|
return [r + (x,) for r, x in zip(rows, extra_rows)], c
|
|
|
|
|
|
def digest(rows) -> str:
|
|
return hashlib.sha256(repr(rows).encode()).hexdigest()
|
|
|
|
|
|
def digests() -> dict:
|
|
return {k: digest(run(**v)[0]) for k, v in SCENARIOS.items()}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if "--summary" in sys.argv:
|
|
for k, v in SCENARIOS.items():
|
|
rows = run(**v)[0]
|
|
print(f"{k}: {len(rows)} cycles, phases "
|
|
f"{sorted({str(r[-1][-1]) for r in rows})}, targets "
|
|
f"{sorted({r[1] for r in rows})[:6]}")
|
|
for k, v in digests().items():
|
|
print(f"{k} {v}")
|