Claude dev agent
902524ae5e
Cheap and expensive hours as runs of consecutive firm price slots
(priceruns.py); each battery's role per tick from its mode (modetrade.py):
grid charging for Leftover solar / Cheap hours only, never Solar first;
holding for the expensive runs; selling per battery (pack_sell) above the
spread and the floor. The site trade is sized inside the modes (loop bound)
and the peak guard's headroom; a run edge reseeds the loop. Cheap hours only
and selling are enabled on the Batteries screen and in the HA select
(appended). The site Why row no longer says "held back" when the peak limit
pushes power up (F2). Modes off: byte-identical to 7a00f12 on the trade path.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
111 lines
3.7 KiB
Python
111 lines
3.7 KiB
Python
#!/usr/bin/env python3
|
|
"""OPT-11 P3: with battery modes off, is every command byte-identical to
|
|
release/1.0 7a00f12, on the paths P3 changed?
|
|
|
|
python3 tools/opt11p3_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) plus a day-ahead price series and `arbitrage_enabled`,
|
|
so the site trade P3 now skips with modes on (`arbitrage.decide()`, its
|
|
claim, the sizing to the guard's headroom) runs on every cycle, with a buy
|
|
period, a sell period and periods in between, and a house that goes over the
|
|
limit. Each row hashes safety79's fields plus the trade's action, watts and
|
|
sentence. Uses only the API 7a00f12 has, so the same file runs on both trees;
|
|
test_modes_p3.py part 1 holds this tree to the digests printed on 7a00f12.
|
|
"""
|
|
|
|
import functools
|
|
import hashlib
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from datetime import timedelta
|
|
|
|
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
|
|
from app.price import PricePoint, PriceProduct, PriceRead, brussels # noqa
|
|
from app.static_tariff import TariffPlan # noqa: E402
|
|
|
|
DUR = 5400.0
|
|
|
|
|
|
def prices():
|
|
"""15-minute firm prices: the cheapest period 15-30 min into the run,
|
|
the dearest 75-90 min in, and ones in between."""
|
|
pts = []
|
|
for i in range(-8, 40):
|
|
s = rig.BASE + timedelta(minutes=15 * i)
|
|
p = 0.08 if i == 1 else 0.52 if i == 5 else 0.20 + 0.003 * (i % 7)
|
|
pts.append(PricePoint(start=brussels(s), eur_per_kwh=p,
|
|
provisional=False))
|
|
return TariffPlan(PriceRead.series(pts, PriceProduct.DAY_AHEAD,
|
|
"generic_dayahead", rig.BASE), None,
|
|
"provider")
|
|
|
|
|
|
PLAN = prices()
|
|
|
|
|
|
def _feed(c, t):
|
|
c.tariff_plan, c.tariff_at = PLAN, rig.BASE + timedelta(seconds=t)
|
|
s79._feed(c, t)
|
|
|
|
|
|
def load(t):
|
|
"""Under the limit, over it, a surplus, a big evening load."""
|
|
k = int(t // 600) % 5
|
|
return (900.0, 3200.0, -2200.0, 1500.0, 4200.0)[k]
|
|
|
|
|
|
def soc(u, t):
|
|
return 55.0 if u == "a" else 45.0
|
|
|
|
|
|
ARB = {"arbitrage_enabled": True, "cycle_cost_threshold_eur": 0.05}
|
|
SCENARIOS = {
|
|
"n1": dict(n=1),
|
|
"n2": dict(n=2),
|
|
# The option present and false: an install that saved it off.
|
|
"n1_off": dict(n=1, extra={"battery_modes_enabled": False}),
|
|
"n2_off": dict(n=2, extra={"battery_modes_enabled": False}),
|
|
}
|
|
|
|
|
|
def run(n, *, extra=None, dur=DUR):
|
|
opts = dict(s79.P1, **ARB, **(extra or {}))
|
|
trade = []
|
|
|
|
def on_cycle(c, t):
|
|
a = c.arb
|
|
trade.append(None if a is None else (a.action, a.power_w, a.reason))
|
|
db = os.path.join(tempfile.mkdtemp(prefix="p3id-"), "tariff.db")
|
|
real = rig.Controller
|
|
rig.Controller = functools.partial(real, peak_db_path=db)
|
|
try:
|
|
rows, c = rig.run(n, extra=opts, dur=dur, load=load, soc=soc,
|
|
cut=lambda u, t: False, before=_feed,
|
|
on_cycle=on_cycle)
|
|
finally:
|
|
rig.Controller = real
|
|
return [r + (x,) for r, x in zip(rows, trade)], 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]
|
|
acts = sorted({str(r[-1][0]) if r[-1] else "None" for r in rows})
|
|
print(f"{k}: {len(rows)} cycles, trade actions {acts}")
|
|
for k, v in digests().items():
|
|
print(f"{k} {v}")
|