Claude
eb10537aee
A Home Assistant solar forecast (Forecast.Solar, Solcast or Open-Meteo
shapes) and the add-on's own hourly house use decide, before each cheap
run, whether the sun will fill each battery before the expensive hours.
Leftover solar then buys from the grid only the shortfall; Cheap hours only
skips the part the sun covers; Solar first never grid-charges. No, stale or
failed forecast: P3 option A exactly (12 digests pinned from 5a4cdc2).
Replay (16 recorded days, forecast errors from Elia's day-ahead BE solar):
year-earlier prices P4 19.15 vs P2 19.33 / option A 19.44 / old trade 18.14;
real 2026 prices P4 29.19 vs 32.26 / 32.01 / 31.40 EUR net.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
113 lines
4.7 KiB
Python
113 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
|
"""OPT-11 P4: with no usable solar forecast, is every command byte-identical
|
|
to release/1.0 5a4cdc2 (P3, owner option A)?
|
|
|
|
python3 tools/opt11p4_identity.py # one digest per scenario
|
|
|
|
tools/opt11p3_identity.py's harness (the REAL Controller on a P1 meter, a
|
|
cheap slot, a dear slot, a house over the limit, `arbitrage_enabled`), now
|
|
with battery modes ON and each battery's mode set in the store, so P3's run
|
|
code runs on every cycle. Each row adds the trade and the roles. Scenarios:
|
|
* `ls`, `ch`, `ls_ls`, `ch_ls`: no `pv_forecast_entity` (the day plan never
|
|
runs); N=1 and N=2.
|
|
* `*_fc_down`: `pv_forecast_entity` set but the sensor reads `unavailable`,
|
|
with a ready house profile in the store (the plan falls back); 5a4cdc2
|
|
ignores both keys, so the digest must match the scenario without them.
|
|
Uses only the API 5a4cdc2 has; test_modes_p4.py part 1 holds this tree to the
|
|
digests printed on 5a4cdc2.
|
|
"""
|
|
|
|
import functools
|
|
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
|
|
from app.store import unit_key # noqa: E402
|
|
|
|
FC = {"pv_forecast_entity": "sensor.energy_production_today_remaining",
|
|
"pv_forecast_tomorrow_entity": "sensor.energy_production_tomorrow"}
|
|
|
|
|
|
async def _unavailable(self, e):
|
|
return {"state": "unavailable", "attributes": {},
|
|
"last_updated": "2026-09-20T11:00:00+00:00"}
|
|
|
|
|
|
rig._Hass.state = _unavailable
|
|
|
|
SCENARIOS = {
|
|
"n1_ls": dict(n=1, modes={"inverter-a": "leftover_solar"}),
|
|
"n1_ch": dict(n=1, modes={"inverter-a": "cheap_hours"}),
|
|
"n2_sf_ls": dict(n=2, modes={"inverter-a": "solar_first",
|
|
"inverter-b": "leftover_solar"}),
|
|
"n2_ch_ls": dict(n=2, modes={"inverter-a": "cheap_hours",
|
|
"inverter-b": "leftover_solar"}),
|
|
"n2_ls_ls": dict(n=2, modes={"inverter-a": "leftover_solar",
|
|
"inverter-b": "leftover_solar"}),
|
|
"n2_sf_ch": dict(n=2, modes={"inverter-a": "solar_first",
|
|
"inverter-b": "cheap_hours"}),
|
|
}
|
|
for _k, _v in list(SCENARIOS.items()):
|
|
SCENARIOS[_k + "_fc_down"] = dict(_v, extra=FC)
|
|
|
|
|
|
def run(n, *, modes, extra=None, dur=p3.DUR):
|
|
opts = dict(s79.P1, **p3.ARB, battery_modes_enabled=True,
|
|
**(extra or {}))
|
|
if n == 2:
|
|
# A capacity on each battery, so a day plan could size a level;
|
|
# small, so the cheap run is the one cheap slot.
|
|
opts["inverters"] = [
|
|
{"name": f"inverter-{u}", "setpoint_entity": f"number.sp_{u}",
|
|
"soc_entity": f"sensor.soc_{u}", "batt_entity": f"sensor.batt_{u}",
|
|
"max_w": 2500, "bms_soc_floor": 10, "capacity_kwh": 1.0}
|
|
for u in "ab"]
|
|
store = rig._Store()
|
|
# At N=1 the one battery is named after the site; the key is per name.
|
|
for name, mode in modes.items():
|
|
store.data[unit_key(name, "pack_mode")] = mode
|
|
if extra:
|
|
# A ready house profile (a day, 600 W, 3 kW 17:00-22:00), so only
|
|
# the missing forecast stops a plan here. 5a4cdc2 ignores the key.
|
|
store.data["house_profile"] = {"2026-09-19": {
|
|
str(h): (3000.0 if 17 <= h < 22 else 600.0) for h in range(24)}}
|
|
trade = []
|
|
|
|
def on_cycle(c, t):
|
|
a, tick = c.arb, getattr(c, "mode_tick", None)
|
|
trade.append((None if a is None else (a.action, a.power_w, a.reason),
|
|
None if tick is None else
|
|
tuple(sorted(tick.roles.items()))))
|
|
db = os.path.join(tempfile.mkdtemp(prefix="p4id-"), "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=p3.load, soc=p3.soc,
|
|
cut=lambda u, t: False, before=p3._feed,
|
|
on_cycle=on_cycle, store=store)
|
|
finally:
|
|
rig.Controller = real
|
|
return [r + (x,) for r, x in zip(rows, trade)], c
|
|
|
|
|
|
def digests() -> dict:
|
|
return {k: p3.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][0]) if r[-1][0] else "None"
|
|
for r in rows})
|
|
roles = sorted({x for r in rows if r[-1][1]
|
|
for x in (y for _n, y in r[-1][1])})
|
|
print(f"{k}: {len(rows)} cycles, trade {acts}, roles {roles}")
|
|
for k, v in digests().items():
|
|
print(f"{k} {v}")
|