glenn schrooyenandClaude Opus 5.5
43a52e7f60
At N>=2 a cheap-window maintenance charge files a site `hi` (`charge`, P_MAINT_SHAPE) equal to what the charging pack's own `hi` places, so the other pack never has to give against it. The charge is sized to the quarter's headroom under the peak aim so it yields to the peak guard, and to 0 W under a site safety stop. N=1 byte-identical (safety84_identity). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
120 lines
4.4 KiB
Python
120 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
"""SAFETY-84: at one inverter, is a maintenance charge byte-identical to before?
|
|
|
|
python3 tools/safety84_identity.py # prints one digest per scenario
|
|
|
|
safety78_identity's harness (the REAL Controller on a P1 meter, every claim
|
|
hashed) with the battery in its maintenance CHARGE phase from the first cycle,
|
|
on the owner's settings (limit 3500, margin 50, charge 2000 W). The house steps
|
|
below the charge rate, at it, under the aim, over it and into a solar surplus,
|
|
in a cheap window, outside one, and under a month peak banked at 4000 W. Uses
|
|
only the API release/1.0 9d5b7d1 has, so the same file runs on both trees;
|
|
test_safety84.py holds this tree to the digests printed there (N=1) and to the
|
|
new N=2 ones.
|
|
"""
|
|
|
|
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 safety79_identity as s79 # noqa: E402
|
|
import app.main as m # noqa: E402
|
|
from app import tariff # noqa: E402
|
|
|
|
DUR = 2400.0
|
|
OPT = {"peak_cap_w": 3500, "peak_margin_w": 50, "maintenance_enabled": True,
|
|
"maintenance_start_hour": 11, "maintenance_charge_w": 2000,
|
|
"price_now_entity": "sensor.pnow", "price_avg_entity": "sensor.pavg"}
|
|
|
|
|
|
def load(t):
|
|
"""Below the 2000 W charge, at it, under the aim, over it, a surplus."""
|
|
return (800.0, 2000.0, 3000.0, 5000.0, -1500.0)[min(int(t // 480), 4)]
|
|
|
|
|
|
def feed(cheap):
|
|
"""s79's P1 feed plus a price now under (cheap) or over the average."""
|
|
now = 0.05 if cheap else 0.30
|
|
|
|
def before(c, t):
|
|
if not getattr(c.hass, "_s84", False):
|
|
read = c.hass.number
|
|
|
|
async def number(e, invert=False):
|
|
if e == "sensor.pnow":
|
|
return now
|
|
if e == "sensor.pavg":
|
|
return 0.20
|
|
return await read(e, invert)
|
|
c.hass.number, c.hass._s84 = number, True
|
|
s79._feed(c, t)
|
|
return before
|
|
|
|
|
|
SCENARIOS = {
|
|
"n1_charge": dict(n=1),
|
|
"n1_charge_dear": dict(n=1, cheap=False),
|
|
"n1_charge_banked": dict(n=1, month_peak=4000.0),
|
|
"n2_charge": dict(n=2),
|
|
"n2_charge_dear": dict(n=2, cheap=False),
|
|
"n2_charge_banked": dict(n=2, month_peak=4000.0),
|
|
}
|
|
|
|
|
|
def run(n, *, cheap=True, month_peak=None, dur=DUR, load=load, soc=None,
|
|
before=None, cut=None, extra=None, store=None):
|
|
"""Rows (opt11_identity's fields plus every claim, the guard and the
|
|
phases) and the controller. `month_peak` banks that peak this month."""
|
|
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 = c.peak_guard
|
|
extra_rows.append((
|
|
seen[-1] if seen else None,
|
|
None if g is None else (g.engaged, g.floor_w, g.reason),
|
|
c._maint_shape_suspended, [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="s84-"), "tariff.db")
|
|
if month_peak is not None:
|
|
tariff.PeakStore(db)._db.execute(
|
|
"INSERT INTO monthly_peak VALUES (?, ?, ?, NULL)",
|
|
(tariff.month_key(rig.BASE), month_peak,
|
|
"2026-09-02T10:00:00+00:00")).connection.commit()
|
|
real = rig.Controller
|
|
rig.Controller = functools.partial(real, peak_db_path=db)
|
|
m.resolve = spy
|
|
try:
|
|
rows, c = rig.run(n, extra=dict(s79.P1, **OPT, **(extra or {})),
|
|
dur=dur, load=load, soc=soc or s79.soc,
|
|
cut=cut or (lambda u, t: False),
|
|
before=before or feed(cheap), on_cycle=on_cycle,
|
|
store=store or s79.store_charging(n))
|
|
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__":
|
|
for k, v in digests().items():
|
|
print(f"{k} {v}")
|