glenn schrooyenandClaude
a0ace389a4
daycurve._floor_cut() set a cut from a battery already at the floor (its cap 0 is below its rating), so a give run with one battery empty and the other held to its rating was cut until MAX_FLOOR_CUTS ran out: 2,800 cuts and 460 ms a build, and give blocks back at the hour edge after that. The shortfall in a step is now shared as control shares it (by what each battery holds above the floor, packshare.weights(), each at most its rating, the rest passed on), in the step and in the cut. A battery giving v kW that holds h kWh is empty at h / v; one at the floor holds nothing and never sets a cut. The cut lands where the step then comes out exact, so it is placed once. curve(stats=) reports the floor cuts and whether the bound stopped one. test_ux54.py: _floor_cut() checked to the second against floor times worked out by hand (one battery, a group with one held to its rating, the second group, the split by what they hold), a battery at the floor sets no cut, the bound is reported; own_walk() with the reviewer's two cases (30/80 %, guard, 7 kW house; 2 x 3 kW, 3.5 kW house) and, per build, the floor cuts, the level points and the time. No decision, command, timing or option key changes. Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Bdqr7PiaHcBnzPituu2yTC
1463 lines
67 KiB
Python
1463 lines
67 KiB
Python
"""Runnable check for UX-54. `python3 test_ux54.py`
|
|
|
|
The Forecast tab: today's and tomorrow's solar, the house's expected use and
|
|
each battery's expected level, per hour. It reads the add-on and decides
|
|
nothing. Real `Controller`s in known situations (tools/ux54_scenarios.py).
|
|
|
|
1. The hourly shape: what each forecast integration gives, and that the tab
|
|
spreads a state-only total over the bell and says so.
|
|
2. The blocks are the plan's: a block is on the tab exactly where Home has
|
|
one, with its words; tomorrow's come from the same builders; the gate
|
|
(UX-53), modes off and every mode are followed.
|
|
3. The level curve (`daycurve.curve()`): inside floor and ceiling, the right
|
|
way in a charge and a give block, from spare solar and from a shortfall,
|
|
and a fuzz against the invariants.
|
|
4. A part that is not known yet is a note: modes off, no forecast, a lost
|
|
forecast, learning, no tomorrow, a monthly cycle, no level, stopped,
|
|
shadow.
|
|
5. Measured solar (`SolarToday`), and that keeping it can never cost a cycle.
|
|
6. The words, English and Dutch, and the page and its route. Reading it
|
|
asks Home Assistant for nothing and moves nothing in control.
|
|
|
|
Mutations (MUTATION.md, purged copies): listed in the pull request.
|
|
"""
|
|
|
|
import asyncio
|
|
import dataclasses
|
|
import json
|
|
import logging
|
|
import os
|
|
import random
|
|
import re
|
|
import sys
|
|
import time
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
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 ux54_scenarios as S # noqa: E402
|
|
from app import arbitrage, dayplan, dayview, homeview, modetrade # noqa: E402
|
|
from app import packmode, packshare, peakguard, solarfc, uitext # noqa: E402
|
|
from app.main import pack_energy_of # noqa: E402
|
|
from app.price import Uncertainty # noqa: E402
|
|
from app import daycurve # noqa: E402
|
|
from app.daycurve import Batt, SolarToday, curve # noqa: E402
|
|
from app.peakguard import PeakDecision # noqa: E402
|
|
|
|
fails = []
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
UTC = timezone.utc
|
|
DAY0 = datetime(2026, 10, 26, 23, 0, tzinfo=UTC) # Tue 27 Oct, local
|
|
KINDS = ("grid", "give", "cycle")
|
|
|
|
|
|
def at(h, m=0, s=5):
|
|
return DAY0 + timedelta(hours=h, minutes=m, seconds=s)
|
|
|
|
|
|
def home(c, now, lang="en"):
|
|
return homeview.build(c, level="ok", checks=[], now=now, lang=lang)
|
|
|
|
|
|
def tab(c, now, lang="en"):
|
|
return dayview.build(c, home(c, now, lang), now, lang)
|
|
|
|
|
|
def block_key(r):
|
|
return (r["kind"], r["from"], r["to"], r["start_h"], r["end_h"],
|
|
r["what"], r["why"], r["amount"])
|
|
|
|
|
|
def tab_blocks(out, day=0):
|
|
return [[block_key(r) for r in b["blocks"]]
|
|
for b in out["days"][day]["batteries"]]
|
|
|
|
|
|
def home_blocks(status, n):
|
|
return [[block_key(r) for r in status["plan"]["runs"]
|
|
if r["battery"] == i and r["kind"] in KINDS] for i in range(n)]
|
|
|
|
|
|
def notes_of(out, day=0):
|
|
return out["days"][day]["notes"]
|
|
|
|
|
|
def say(key, lang="en", **v):
|
|
with homeview.in_lang(lang):
|
|
return homeview.day_note(key, **v)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("1. the hourly shape")
|
|
# --------------------------------------------------------------------------- #
|
|
now = at(10, 7)
|
|
c = S.make("owner", now)
|
|
fc = c.solar_fc
|
|
out = tab(c, now)
|
|
d = out["days"][0]
|
|
fcs = [h["forecast_kwh"] for h in d["hours"]]
|
|
print(f" owner: {len(c.fc_entities)} + {len(c.fc_tomorrow_entities)} "
|
|
f"state-only sensors, today {fc.today.kwh:.2f} kWh left, tomorrow "
|
|
f"{fc.tomorrow.kwh:.2f}")
|
|
check("the owner's shape (three 'remaining' + three 'tomorrow' sensors) "
|
|
"carries no hourly split: the add-on sums them and uses its bell",
|
|
len(c.fc_entities) == 3 and len(c.fc_tomorrow_entities) == 3
|
|
and fc.today.periods == () and fc.tomorrow.periods == ()
|
|
and abs(fc.tomorrow.kwh - 16.0) < 0.01)
|
|
check("...so the tab says it is an estimate from a daily total, and fades "
|
|
"the bars", d["sun"]["estimate"] is True
|
|
and say("estimate") in notes_of(out)
|
|
and out["days"][1]["sun"]["estimate"] is True)
|
|
check("today the hours still to come add up to what the sensors say is left "
|
|
"(no hour before now)",
|
|
all(f is None for f in fcs[:10])
|
|
and abs(sum(f for f in fcs if f is not None) - fc.today.kwh) < 1e-3)
|
|
tm = [h["forecast_kwh"] for h in out["days"][1]["hours"]]
|
|
check("tomorrow's 24 hours add up to tomorrow's total, are 0 outside 06-20 "
|
|
"and peak at 13:00 (the bell)",
|
|
abs(sum(tm) - fc.tomorrow.kwh) < 0.01
|
|
and all(v == 0 for v in tm[:6] + tm[20:])
|
|
and tm.index(max(tm)) in (12, 13))
|
|
# The day plan reads the same forecast hour by hour with `between(s, e)`;
|
|
# for a state-only forecast that spreads what is left over each span's own
|
|
# tail, so its hours add up to more than the total (a finding in the pull
|
|
# request, not changed here: no control change).
|
|
plan_sum = sum(fc.between(s, e) for s, e, _h in dayplan._hour_slices(
|
|
now, solarfc._bounds(now)[1]))
|
|
print(f" dayplan's own per-hour sum: {plan_sum:.1f} kWh for {fc.today.kwh:.1f} kWh left")
|
|
check("the tab's hours use the span from now (`dayview.sun_kwh()`), not the "
|
|
"plan's per-hour call, and so keep the total",
|
|
abs(dayview.sun_kwh(fc, now, now, solarfc._bounds(now)[1])
|
|
- fc.today.kwh) < 1e-6)
|
|
|
|
cp = S.make("periods", now)
|
|
dp = tab(cp, now)["days"]
|
|
check("an hourly split (Open-Meteo's `wh_period`, Solcast's "
|
|
"`detailedForecast`) is used as it is, and the tab does not call it an "
|
|
"estimate",
|
|
cp.solar_fc.today.periods and cp.solar_fc.tomorrow.periods
|
|
and dp[0]["sun"]["estimate"] is False
|
|
and say("estimate") not in dp[0]["notes"]
|
|
and abs(sum(h["forecast_kwh"] for h in dp[1]["hours"])
|
|
- cp.solar_fc.tomorrow.kwh) < 1e-3)
|
|
# Solcast's shape: `detailedForecast` rows, kW per half hour.
|
|
d0 = S.day0_of(now)
|
|
rows = [{"period_start": (d0 + timedelta(minutes=30 * i)).isoformat(),
|
|
"pv_estimate": 4.0 if 20 <= i < 36 else 0.0} for i in range(48)]
|
|
st = {"state": "20.0", "attributes": {"unit_of_measurement": "kWh",
|
|
"detailedForecast": rows},
|
|
"last_updated": now.isoformat(), "last_reported": now.isoformat()}
|
|
sc = solarfc.read_sum([("sensor.solcast", st)], [], now)[0]
|
|
check("Solcast's `detailedForecast` gives a split too (kW per half hour)",
|
|
sc is not None and sc.today.periods and len(sc.today.periods) >= 16)
|
|
check("the forecast the tab reads is the one the controller holds "
|
|
"(`Controller.solar_fc`), read by `solarfc` as the plan reads it",
|
|
out["updated"] == homeview.hhmm(c.solar_fc.at))
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n2. the blocks are the plan's")
|
|
# --------------------------------------------------------------------------- #
|
|
SCN = [("owner", 10.1), ("dynamic", 10.1), ("dynamic", 0.3),
|
|
("dynamic", 18.5), ("sunny", 0.3), ("sunny", 18.5), ("off", 10.1),
|
|
("off", 16.9), ("cycle", 10.1), ("nofc", 10.1), ("stale", 0.3),
|
|
("learning", 0.3), ("periods", 18.5)]
|
|
bad, seen = [], {"grid": 0, "give": 0, "cycle": 0}
|
|
for name, hh in SCN:
|
|
now = at(hh)
|
|
c = S.make(name, now)
|
|
st = home(c, now)
|
|
out = tab(c, now)
|
|
n = len(c.inverters)
|
|
if tab_blocks(out) != home_blocks(st, n):
|
|
bad.append((name, hh))
|
|
for rows in tab_blocks(out):
|
|
for r in rows:
|
|
seen[r[0]] += 1
|
|
print(f" {len(SCN)} scenarios, blocks seen on the tab: {seen}")
|
|
check("today, every battery's blocks on the tab are Home's `runs` of that "
|
|
"battery (grid, give, cycle): same times, same words, no more, no less "
|
|
f"({bad})", not bad and all(v > 0 for v in seen.values()))
|
|
|
|
now = at(10, 7)
|
|
c = S.make("dynamic", now)
|
|
d0, d1 = homeview.local_day(now)
|
|
n0, n1 = homeview.local_day(d1 + timedelta(hours=2))
|
|
check("Home's own price blocks for today are `homeview.runs_for()` of "
|
|
"today: one builder, two days",
|
|
[block_key(r) for r in homeview.runs_for(c, d0, d1, now, False)]
|
|
== [block_key(r) for r in home(c, now)["plan"]["runs"]
|
|
if r["kind"] in ("grid", "give")])
|
|
tmr = tab(c, now)["days"][1]["batteries"]
|
|
check("tomorrow's blocks are the same builder over tomorrow's window, read "
|
|
"`ahead` (the level now does not filter a day that has not started)",
|
|
[[block_key(r) for r in b["blocks"]] for b in tmr]
|
|
== [[block_key(r) for r in homeview.runs_for(c, n0, n1, now, False, ahead=True)
|
|
if r["battery"] == i] for i in range(2)])
|
|
|
|
# Modes on: Solar first draws no price block; Leftover solar and Cheap hours do.
|
|
c_o = S.make("owner", at(10, 7))
|
|
o = tab(c_o, at(10, 7))
|
|
check("modes on, both batteries Solar first, a day/night tariff (the "
|
|
"owner's shape): no block on either battery, either day, and the "
|
|
"battery says it follows the house",
|
|
all(not b["blocks"] for dd in o["days"] for b in dd["batteries"])
|
|
and all(b["line"] == homeview.run_house_what(False) + " "
|
|
+ homeview.run_house_why(False)
|
|
for b in o["days"][0]["batteries"]))
|
|
c_d = S.make("dynamic", at(10, 7))
|
|
dd = tab(c_d, at(10, 7))["days"][0]["batteries"]
|
|
rp = c_d.mode_runs
|
|
check("Leftover solar and Cheap hours: a give block exactly on the give run "
|
|
"(17:00 to 21:00), on both",
|
|
all([(r["kind"], r["from"], r["to"]) for r in b["blocks"]]
|
|
== [("give", "17:00", "21:00")] for b in dd)
|
|
and rp.give[0].start == at(17, 0, 0) and rp.give[0].end == at(21, 0, 0))
|
|
c_n = S.make("dynamic", at(0, 18))
|
|
nb = tab(c_n, at(0, 18))["days"][0]["batteries"]
|
|
check("in the night before the cheap run both charge from the grid in it "
|
|
"(01:00 to 02:30), then give at 17:00",
|
|
all([r["kind"] for r in b["blocks"]] == ["grid", "give"] for b in nb)
|
|
and all(b["blocks"][0]["from"] == "01:00"
|
|
and b["blocks"][0]["to"] == "02:30" for b in nb))
|
|
# UX-53: no grid block while the gate refuses.
|
|
for why in ("peak", "not_worth"):
|
|
c_n.mode_gate = why
|
|
gb = tab(c_n, at(0, 18))["days"][0]["batteries"]
|
|
check(f"the trade gate refuses ({why}): no grid block on the tab, as on "
|
|
"Home, and the give block stays",
|
|
all([r["kind"] for r in b["blocks"]] == ["give"] for b in gb)
|
|
and tab_blocks(tab(c_n, at(0, 18))) == home_blocks(
|
|
home(c_n, at(0, 18)), 2))
|
|
c_n.mode_gate = None
|
|
|
|
|
|
def at_h(b, h):
|
|
"""A drawn level at hour `h` of the day, NaN where no point is drawn."""
|
|
return dict((round(x, 3), v) for x, v in b["levels"]).get(h, float("nan"))
|
|
|
|
|
|
check("the curve charges in a grid block: A (planned level 64 %) and B "
|
|
"(Cheap hours, to the end of its 01:30 run) are well above where they "
|
|
"were at 01:00, and stay put after the run",
|
|
all(at_h(b, 2.5) - at_h(b, 1.0) > 8.0
|
|
and abs(at_h(b, 3.0) - at_h(b, 2.5)) < 1e-6
|
|
for b in nb) and nb[0]["levels"][0][1] == 55.0)
|
|
# Modes off: the verdict's legs.
|
|
c_f = S.make("off", at(10, 7))
|
|
fb = tab(c_f, at(10, 7))
|
|
check("modes off, arbitrage on: the verdict's blocks are Home's, today, and "
|
|
"tomorrow's charge comes from the same builder; no level is drawn",
|
|
[b["kind"] for x in fb["days"][0]["batteries"] for b in x["blocks"]]
|
|
== ["give", "give"]
|
|
and [b["kind"] for x in fb["days"][1]["batteries"]
|
|
for b in x["blocks"]] == ["grid", "grid"]
|
|
and not any(x["levels"] for dd in fb["days"] for x in dd["batteries"]))
|
|
# A battery that is Off has no block and no curve.
|
|
c_d.store.data[S.unit_key("inverter-b", packmode.USE_KEY)] = False
|
|
off = tab(c_d, at(10, 7))["days"][0]["batteries"]
|
|
check("a battery switched Off: no block, no curve, and it says so in the "
|
|
"plan's words",
|
|
off[1]["off"] is True and not off[1]["blocks"] and not off[1]["levels"]
|
|
and off[1]["line"] == homeview.why_words("off")
|
|
and off[0]["levels"])
|
|
# The cycle is a block, and the curve stops where it starts.
|
|
c_c = S.make("cycle", at(10, 7))
|
|
cy = tab(c_c, at(10, 7))
|
|
a0 = cy["days"][0]["batteries"][0]
|
|
check("a monthly cycle waiting today is a block, the level is drawn only "
|
|
"until it starts, and a note says so",
|
|
[r["kind"] for r in a0["blocks"]] == ["give", "cycle"]
|
|
and a0["levels"][-1][0] <= 20.0 + 1e-6
|
|
and say("cycle", name="inverter-a") in notes_of(cy)
|
|
and not cy["days"][1]["batteries"][0]["levels"])
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n3. the level curve")
|
|
# --------------------------------------------------------------------------- #
|
|
T0 = DAY0 + timedelta(hours=9)
|
|
T1 = T0 + timedelta(hours=24)
|
|
|
|
|
|
def flat(kwh_per_h):
|
|
return lambda a, z: kwh_per_h * (z - a).total_seconds() / 3600.0
|
|
|
|
|
|
def run(b, sun, use, **kw):
|
|
return curve([b] if isinstance(b, Batt) else b, T0, T1, sun, use,
|
|
floor_pct=20.0, ceiling_pct=95.0, **kw)
|
|
|
|
|
|
def lv(out, name, h):
|
|
"""The level at `h` hours after T0 (NaN when there is no such point)."""
|
|
return dict(out[name]).get(T0 + timedelta(hours=h), float("nan"))
|
|
|
|
|
|
B1 = Batt("a", 10.0, 5.0, 50.0, True, False)
|
|
o1 = run(B1, flat(3.0), flat(1.0))
|
|
check("Solar first: 2 kWh of spare solar an hour fills it at 95 % "
|
|
"efficiency, 19 % an hour",
|
|
abs(lv(o1, "a", 1) - 69.0) < 1e-6 and abs(lv(o1, "a", 2) - 88.0) < 1e-6)
|
|
check("...stops at the ceiling (95 %), and stays there", abs(lv(o1, "a", 3)
|
|
- 95.0) < 1e-6
|
|
and max(v for _t, v in o1["a"]) <= 95.0 + 1e-9)
|
|
o2 = run(B1, flat(0.0), flat(1.0))
|
|
check("a shortfall of 1 kWh an hour is given at 95 % efficiency, 10.53 % an "
|
|
"hour", abs(lv(o2, "a", 1) - (50.0 - 100.0 / 9.5)) < 1e-6)
|
|
check("...stops at the floor (20 %), and stays there",
|
|
abs(lv(o2, "a", 6) - 20.0) < 1e-6
|
|
and min(v for _t, v in o2["a"]) >= 20.0 - 1e-9)
|
|
check("a battery already above the ceiling does not charge and one below the "
|
|
"floor does not give",
|
|
lv(run(Batt("a", 10.0, 5.0, 98.0, True, False), flat(3), flat(1)),
|
|
"a", 3) == 98.0
|
|
and lv(run(Batt("a", 10.0, 5.0, 12.0, True, False), flat(0), flat(1)),
|
|
"a", 3) == 12.0)
|
|
slow = run(Batt("a", 10.0, 1.0, 50.0, True, False), flat(5.0), flat(0.0))
|
|
check("the rating caps the rate: 1 kW charges 9.5 % of 10 kWh an hour",
|
|
abs(lv(slow, "a", 1) - 59.5) < 1e-6)
|
|
# Where a giving battery reaches the floor, worked out by hand. 10 kWh at
|
|
# 25 % over a 20 % floor holds 0.05 x 10 x 0.95 = 0.475 kWh to give.
|
|
|
|
|
|
def floor_at(out, name):
|
|
"""`(seconds after T0, level)` of the first point at the floor."""
|
|
return next(((round((t - T0).total_seconds(), 3), v) for t, v in out[name]
|
|
if abs(v - 20.0) < 1e-9), (None, None))
|
|
|
|
|
|
fc1 = run(Batt("a", 10.0, 5.0, 25.0, True, False), flat(0.0), flat(2.0))
|
|
check("one battery, 2 kW short: empty at 0.475 / 2 h = 14:15 (855 s), a point "
|
|
f"there at the floor ({floor_at(fc1, 'a')})",
|
|
floor_at(fc1, "a") == (855.0, 20.0))
|
|
fc2 = run([Batt("a", 10.0, 5.0, 25.0, True, False),
|
|
Batt("b", 20.0, 4.0, 80.0, True, False)], flat(0.0), flat(5.0))
|
|
check("two that may give, 5 kW short: shared by what they hold (0.475 : "
|
|
"11.4 kWh), B held to its 4 kW, so A gives 1 kW and is empty at "
|
|
f"28:30 (1710 s) ({floor_at(fc2, 'a')})",
|
|
floor_at(fc2, "a") == (1710.0, 20.0))
|
|
GV1 = ((T0, T0 + timedelta(hours=2)),)
|
|
fc3 = run([Batt("g", 20.0, 2.0, 80.0, False, True, (), GV1),
|
|
Batt("h", 10.0, 5.0, 25.0, True, False)], flat(0.0), flat(2.95))
|
|
check("the second group: G in a give run gives first, at its 2 kW; the 0.95 "
|
|
"kW left goes to H (Solar first), empty at 0.475 / 0.95 h = 30:00 "
|
|
f"(1800 s) ({floor_at(fc3, 'h')})",
|
|
floor_at(fc3, "h") == (1800.0, 20.0))
|
|
fc4 = run([Batt("a", 10.0, 5.0, 25.0, True, False),
|
|
Batt("b", 20.0, 5.0, 80.0, True, False)], flat(0.0), flat(3.0))
|
|
check("shared by what they hold, not by what they can give in a step: 3 kW "
|
|
"short, A (0.475 kWh) and B (11.4 kWh) drain in step and are both "
|
|
"empty when the 11.875 kWh is gone, at 3:57:30 (14250 s) "
|
|
f"({floor_at(fc4, 'a')}, {floor_at(fc4, 'b')})",
|
|
floor_at(fc4, "a") == (14250.0, 20.0)
|
|
and floor_at(fc4, "b") == (14250.0, 20.0))
|
|
fc5 = curve([Batt("a", 10.0, 5.0, 20.0, True, False),
|
|
Batt("b", 20.0, 2.0, 80.0, True, False)], T0,
|
|
T0 + timedelta(hours=4), flat(0.0), flat(7.0), floor_pct=20.0,
|
|
ceiling_pct=95.0)
|
|
check("a battery already at the floor sets no cut: A at 20 %, B held to its "
|
|
"2 kW for 4 h (it holds 11.4 kWh): one point per hour "
|
|
f"({len(fc5['a'])} points)",
|
|
len(fc5["a"]) == 5 and len(fc5["b"]) == 5)
|
|
st_cap, real_max = {}, daycurve.MAX_FLOOR_CUTS
|
|
daycurve.MAX_FLOOR_CUTS = 0
|
|
try:
|
|
fc6 = run(Batt("a", 10.0, 5.0, 25.0, True, False), flat(0.0), flat(2.0),
|
|
stats=st_cap)
|
|
finally:
|
|
daycurve.MAX_FLOOR_CUTS = real_max
|
|
st_ok = {}
|
|
run(Batt("a", 10.0, 5.0, 25.0, True, False), flat(0.0), flat(2.0),
|
|
stats=st_ok)
|
|
check("the curve says when its bound on floor points stopped one (`stats`): "
|
|
f"{st_cap} at a bound of 0, {st_ok} as shipped",
|
|
st_cap == {"floor_cuts": 0, "capped": True}
|
|
and st_ok == {"floor_cuts": 1, "capped": False}
|
|
and floor_at(fc6, "a") == (3600.0, 20.0))
|
|
# Roles.
|
|
GV = ((T0 + timedelta(hours=3), T0 + timedelta(hours=6)),)
|
|
hold = Batt("p", 10.0, 5.0, 60.0, False, True, (), GV)
|
|
oh = run(hold, flat(0.0), flat(1.0))
|
|
check("a price-mode battery holds outside a give run (no give), and gives "
|
|
"inside it",
|
|
lv(oh, "p", 3) == 60.0
|
|
and lv(oh, "p", 6) < 60.0 - 3 * 100.0 / 9.5 + 1e-6 + 0.01
|
|
and lv(oh, "p", 24) == lv(oh, "p", 6))
|
|
oh2 = run(hold, flat(3.0), flat(1.0))
|
|
check("...and charges from spare solar while it holds",
|
|
lv(oh2, "p", 3) > 60.0)
|
|
GR = ((T0 + timedelta(hours=2), T0 + timedelta(hours=4), 80.0),)
|
|
ch = Batt("c", 10.0, 5.0, 40.0, False, True, GR, ())
|
|
oc = run(ch, flat(0.0), flat(0.0))
|
|
check("a grid block charges to its level at the rating (5 kW x 95 %: 47.5 % "
|
|
"an hour on 10 kWh, so 80 % in the first hour), and stops there",
|
|
lv(oc, "c", 2) == 40.0 and abs(lv(oc, "c", 3) - 80.0) < 1e-6
|
|
and lv(oc, "c", 5) == 80.0)
|
|
reach = T0 + timedelta(hours=2, minutes=50, seconds=31.579)
|
|
check("a battery that reaches its grid level inside a step gets a point "
|
|
"there (80 % at 02:50:31, not at the next hour), so the line and the "
|
|
"block end are exact",
|
|
any(abs((t - reach).total_seconds()) < 1.0
|
|
and abs(v - 80.0) < 1e-6 for t, v in oc["c"]))
|
|
two = [Batt("a", 10.0, 5.0, 50.0, True, False),
|
|
Batt("b", 10.0, 5.0, 50.0, False, True)]
|
|
ot = run(two, flat(3.0), flat(1.0))
|
|
check("two batteries: Solar first takes the spare solar first; the other "
|
|
"gets what is left",
|
|
lv(ot, "a", 1) > lv(ot, "b", 1) - 1e-9
|
|
and abs(lv(ot, "a", 1) - 69.0) < 1e-6 and lv(ot, "b", 1) == 50.0)
|
|
big = run(two, flat(8.0), flat(0.0))
|
|
check("...and the other charges once Solar first is at its rating or full",
|
|
lv(big, "b", 1) > 50.0)
|
|
GR2 = ((T0 + timedelta(hours=2), T0 + timedelta(hours=4), 80.0),)
|
|
mix = [Batt("a", 40.0, 5.0, 50.0, True, True, (), ()),
|
|
Batt("b", 10.0, 5.0, 40.0, False, True, GR2, ())]
|
|
om = run(mix, flat(0.0), flat(2.0))
|
|
check("while one battery charges from the grid the others do not charge "
|
|
"(they rest) and Solar first may still give",
|
|
lv(om, "b", 3) > lv(om, "b", 2) and lv(om, "a", 3) < lv(om, "a", 2))
|
|
ends = run(B1, lambda a, z: None if a >= T0 + timedelta(hours=5)
|
|
else flat(3.0)(a, z), flat(1.0))
|
|
check("an unknown sun ends the curve there", ends["a"][-1][0]
|
|
== T0 + timedelta(hours=5) and len(ends["a"]) == 6)
|
|
stop = run(Batt("a", 10.0, 5.0, 50.0, True, False, (), (),
|
|
T0 + timedelta(hours=2)), flat(3.0), flat(1.0))
|
|
check("a monthly cycle ends the battery's curve at its start",
|
|
stop["a"][-1][0] == T0 + timedelta(hours=2))
|
|
check("the curve starts at the battery's level now, at `start`",
|
|
o1["a"][0] == (T0, 50.0) and o2["a"][0] == (T0, 50.0))
|
|
|
|
# A fuzz against the invariants.
|
|
rnd = random.Random(54)
|
|
bad_f = []
|
|
for i in range(300):
|
|
n = rnd.choice((1, 2, 3))
|
|
bs = []
|
|
for k in range(n):
|
|
gs = ()
|
|
gw = ()
|
|
if rnd.random() < 0.5:
|
|
a = rnd.randint(0, 20)
|
|
gs = ((T0 + timedelta(hours=a),
|
|
T0 + timedelta(hours=a + rnd.randint(1, 4)),
|
|
rnd.uniform(40, 95)),)
|
|
if rnd.random() < 0.5:
|
|
a = rnd.randint(0, 20)
|
|
gw = ((T0 + timedelta(hours=a),
|
|
T0 + timedelta(hours=a + rnd.randint(1, 4))),)
|
|
bs.append(Batt(f"b{k}", rnd.uniform(5, 25), rnd.uniform(1, 6),
|
|
rnd.uniform(20, 95), rnd.random() < 0.5,
|
|
rnd.random() < 0.6, gs, gw))
|
|
sun = rnd.uniform(0, 5)
|
|
use = rnd.uniform(0, 3)
|
|
out = run(bs, lambda a, z, s=sun: s * (z - a).total_seconds() / 3600.0,
|
|
lambda a, z, u=use: u * (z - a).total_seconds() / 3600.0)
|
|
for b in bs:
|
|
pts = out[b.name]
|
|
vals = [v for _t, v in pts]
|
|
if min(vals) < 20.0 - 1e-6 or max(vals) > 95.0 + 1e-6:
|
|
bad_f.append(("range", i, b.name))
|
|
for (t1, v1), (t2, v2) in zip(pts, pts[1:]):
|
|
hrs = (t2 - t1).total_seconds() / 3600.0
|
|
if abs(v2 - v1) / 100.0 * b.capacity_kwh > b.rating_kw * hrs \
|
|
/ dayplan.EFF * 1.0001 + 1e-9:
|
|
bad_f.append(("rate", i, b.name))
|
|
# holds, and not in a grid block or a give run: never gives
|
|
in_give = any(s <= t1 < e for s, e in b.give)
|
|
in_grid = any(s <= t1 < e for s, e, _ in b.grid)
|
|
if b.priced and not b.solar_first and not in_give \
|
|
and not in_grid and v2 < v1 - 1e-9:
|
|
bad_f.append(("hold gave", i, b.name))
|
|
# Energy: what the batteries gain never exceeds the sun's spare (net of
|
|
# the grid blocks) at 95 %.
|
|
check(f"a fuzz of 300 days of 1-3 random batteries: every level stays inside "
|
|
f"the floor and ceiling, no step beats the rating, a holding battery "
|
|
f"never gives ({bad_f[:3]})", not bad_f)
|
|
|
|
# On the controllers: inside floor and ceiling, the right way in blocks.
|
|
rows_ok, moves_ok, n_curves = True, True, 0
|
|
for name, hh in SCN:
|
|
now = at(hh)
|
|
c = S.make(name, now)
|
|
out = tab(c, now)
|
|
fl, ce = arbitrage.soc_bounds(c.o)
|
|
for day in out["days"]:
|
|
for b in day["batteries"]:
|
|
n_curves += bool(b["levels"])
|
|
for h_, v in b["levels"]:
|
|
rows_ok &= fl - 1e-6 <= v <= ce + 1e-6
|
|
for r in b["blocks"]:
|
|
inside = [(h_, v) for h_, v in b["levels"]
|
|
if r["start_h"] - 1e-6 <= h_ <= r["end_h"] + 1e-6]
|
|
if len(inside) >= 2 and r["kind"] == "grid":
|
|
moves_ok &= inside[-1][1] >= inside[0][1] - 1e-6
|
|
if len(inside) >= 2 and r["kind"] == "give":
|
|
moves_ok &= inside[-1][1] <= inside[0][1] + 1e-6
|
|
print(f" {n_curves} battery curves checked on the controllers")
|
|
check("on the controllers: every drawn level is inside the floor and the "
|
|
"ceiling", rows_ok and n_curves >= 10)
|
|
check("...and the level rises in a grid block and does not rise in a give "
|
|
"block", moves_ok)
|
|
c_s = S.make("sunny", at(10, 7))
|
|
ls = tab(c_s, at(10, 7))["days"][0]["batteries"][0]["levels"]
|
|
check("a sunny day with Leftover solar: the level climbs by noon and "
|
|
"reaches the ceiling by evening",
|
|
ls[0][1] == c_s.inverters[0].soc and max(v for _h, v in ls) > 85.0)
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n4. a part that is not known yet is a note")
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def no_charts(out):
|
|
return all(not b["levels"] for dd in out["days"]
|
|
for b in dd["batteries"])
|
|
|
|
|
|
c = S.make("off", at(10, 7))
|
|
o = tab(c, at(10, 7))
|
|
check("modes off: a note, no forecast, no house use, no curve",
|
|
say("modes_off") in notes_of(o) and no_charts(o)
|
|
and not o["days"][0]["sun"]["forecast"]
|
|
and not o["days"][0]["sun"]["house"])
|
|
o = tab(S.make("nofc", at(10, 7)), at(10, 7))
|
|
check("no forecast set up: a note, and the price blocks still show",
|
|
say("not_set") in notes_of(o) and no_charts(o)
|
|
and o["days"][0]["batteries"][0]["blocks"])
|
|
o = tab(S.make("stale", at(10, 7)), at(10, 7))
|
|
check("a forecast that is missing or too old: a note, no curve",
|
|
say("no_forecast") in notes_of(o) and no_charts(o))
|
|
o = tab(S.make("learning", at(10, 7)), at(10, 7))
|
|
check("learning (no house profile yet): a note, the sun is drawn, no house "
|
|
"use and no curve",
|
|
say("learning") in notes_of(o) and no_charts(o)
|
|
and o["days"][0]["sun"]["forecast"] and not o["days"][0]["sun"]["house"]
|
|
and all(h["house_kwh"] is None for h in o["days"][0]["hours"]))
|
|
SUN_FROM = f"{int(solarfc.BELL_FROM_H):02d}:00"
|
|
o = tab(S.make("notomorrow", at(10, 7)), at(10, 7))
|
|
tmr = o["days"][1]
|
|
check("no forecast for tomorrow: a note on that day only, naming the hour "
|
|
"the curve really stops (`solarfc.BELL_FROM_H`, the first hour of the "
|
|
"sun: its last point is there); today is whole",
|
|
say("no_tomorrow", at=SUN_FROM) in tmr["notes"]
|
|
and say("no_tomorrow", at=SUN_FROM) not in notes_of(o, 0)
|
|
and not tmr["sun"]["forecast"]
|
|
and abs(tmr["batteries"][0]["levels"][-1][0] - solarfc.BELL_FROM_H) < 1e-6
|
|
and o["days"][0]["batteries"][0]["levels"][-1][0] > 23.0)
|
|
c = S.make("owner", at(10, 7))
|
|
c.inverters[1].soc = None
|
|
o = tab(c, at(10, 7))
|
|
check("a battery with no level reading: a note naming it, no curve for it, "
|
|
"and the other still has one",
|
|
say("no_level", name="inverter-b") in notes_of(o)
|
|
and not o["days"][0]["batteries"][1]["levels"]
|
|
and o["days"][0]["batteries"][0]["levels"])
|
|
c = S.make("owner", at(10, 7))
|
|
c.auto = False
|
|
o = tab(c, at(10, 7))
|
|
check("control stopped: the plan's words, no curve, no block",
|
|
homeview.why_words("stopped") in notes_of(o) and no_charts(o)
|
|
and not o["running"])
|
|
c = S.make("owner", at(10, 7))
|
|
c.o["shadow_mode"] = "on"
|
|
o = tab(c, at(10, 7))
|
|
check("shadow mode: the plan's own shadow sentence is on the tab",
|
|
homeview.why_words("shadow") in notes_of(o))
|
|
c = S.make("owner", at(10, 7))
|
|
check("the levels count on the trust the plan counts on, and the tab says "
|
|
"so", say("trust", pct=70) in notes_of(tab(c, at(10, 7))))
|
|
c.solar_trust_pct = 100
|
|
full = tab(c, at(10, 7))["days"][0]["batteries"][0]["levels"]
|
|
c.solar_trust_pct = 30
|
|
low = tab(c, at(10, 7))["days"][0]["batteries"][0]["levels"]
|
|
check("...a higher trust fills the battery more (the curve uses it)",
|
|
max(v for _h, v in full) > max(v for _h, v in low))
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n5. measured solar")
|
|
# --------------------------------------------------------------------------- #
|
|
t0 = at(0, 0, 0)
|
|
sd = SolarToday()
|
|
for i in range(0, 3600, 5): # 2 kW for the hour 10:00-11:00
|
|
sd.add(t0 + timedelta(hours=10, seconds=i), 2000.0)
|
|
sd.add(t0 + timedelta(hours=11), 2000.0)
|
|
check("2 kW read all hour is 2.0 kWh for that hour",
|
|
abs(sd.kwh(t0 + timedelta(hours=10), t0 + timedelta(hours=11),
|
|
t0 + timedelta(hours=12)) - 2.0) < 0.02)
|
|
check("the hour in progress is what was made so far (half an hour: 1 kWh)",
|
|
abs(sd.kwh(t0 + timedelta(hours=10), t0 + timedelta(hours=11),
|
|
t0 + timedelta(hours=10, minutes=30)) - 1.0) < 0.05)
|
|
check("an hour with no reading, or under half covered, is not guessed",
|
|
sd.kwh(t0 + timedelta(hours=9), t0 + timedelta(hours=10),
|
|
t0 + timedelta(hours=12)) is None)
|
|
g = SolarToday()
|
|
g.add(t0 + timedelta(hours=10), 1000.0)
|
|
g.add(t0 + timedelta(hours=10, minutes=50), 1000.0) # a gap: 60 s counted
|
|
check("a gap in the readings counts for no more than a minute, so a "
|
|
"restart does not invent energy",
|
|
g.kwh(t0 + timedelta(hours=10), t0 + timedelta(hours=11),
|
|
t0 + timedelta(hours=11)) is None)
|
|
g.add(t0 + timedelta(hours=10, minutes=51), None)
|
|
g.add(t0 + timedelta(hours=10, minutes=52), 1000.0)
|
|
check("no reading ends an interval; a negative reading counts as 0",
|
|
g._last == (t0 + timedelta(hours=10, minutes=52), 1000.0))
|
|
nxt = t0 + timedelta(days=1, hours=1)
|
|
g.add(nxt, 500.0)
|
|
check("a new local day starts again", g.day != (t0).date()
|
|
and g.first == nxt and g._hours == {})
|
|
c = S.make("owner", at(10, 7))
|
|
o = tab(c, at(10, 7))
|
|
got = [h["measured_kwh"] for h in o["days"][0]["hours"]]
|
|
check("on the page: the hours already over have measured solar, the hours "
|
|
"to come do not, tomorrow has none",
|
|
got[9] is not None and got[11] is None and got[0] is not None
|
|
and all(h["measured_kwh"] is None for h in o["days"][1]["hours"]))
|
|
c = S.make("owner", at(10, 7))
|
|
c.solar_today = SolarToday()
|
|
c.solar_today.add(at(8, 0, 0), 1000.0)
|
|
o2 = tab(c, at(10, 7))
|
|
check("the add-on that started at 08:00 says measured solar counts from "
|
|
"then", say("measured_from", at="08:00") in notes_of(o2))
|
|
|
|
|
|
class _Hass:
|
|
def __init__(self):
|
|
self.reads = 0
|
|
|
|
async def state(self, e):
|
|
self.reads += 1
|
|
return {"entity_id": e, "state": "1500",
|
|
"attributes": {"unit_of_measurement": "W"}}
|
|
|
|
|
|
async def _read_pv_with(add):
|
|
c = S.make("owner", at(10, 7))
|
|
c.hass = _Hass()
|
|
c.solar_today.add = add
|
|
await c._read_pv()
|
|
return c.pv_w
|
|
|
|
|
|
def _boom(*_a):
|
|
raise RuntimeError("no")
|
|
|
|
|
|
def _survives():
|
|
try:
|
|
return asyncio.run(_read_pv_with(_boom)) == 1500.0
|
|
except Exception: # noqa: BLE001
|
|
return False
|
|
|
|
|
|
check("a fault in keeping the measured solar never costs the control "
|
|
"cycle its reading (`_read_pv` still returns the total)", _survives())
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n6. words, page, route; reading asks for nothing and moves nothing")
|
|
# --------------------------------------------------------------------------- #
|
|
en = uitext.words("en")
|
|
nl = uitext.words("nl")
|
|
check("every sentence of the tab has an English and a Dutch form, with the "
|
|
"same placeholders",
|
|
set(en["home_words"]["day"]) == set(nl["home_words"]["day"])
|
|
and all(re.findall(r"\{\w+\}", en["home_words"]["day"][k])
|
|
== re.findall(r"\{\w+\}", nl["home_words"]["day"][k])
|
|
for k in en["home_words"]["day"]))
|
|
now = at(10, 7)
|
|
c = S.make("owner", now)
|
|
oe, on = tab(c, now, "en"), tab(c, now, "nl")
|
|
check("Dutch: the notes, the totals and the blocks are in Dutch, with a "
|
|
"decimal comma",
|
|
on["days"][0]["line"].startswith("Nog verwachte zon vandaag: 14,4 kWh")
|
|
and say("estimate", "nl") in on["days"][0]["notes"]
|
|
and on["days"][1]["line"].startswith("Verwachte zon morgen: 16,0 kWh")
|
|
and oe["days"][0]["line"].startswith("Solar still to come today: 14.4"))
|
|
c = S.make("dynamic", at(0, 18))
|
|
nl_b = tab(c, at(0, 18), "nl")["days"][0]["batteries"][0]["blocks"]
|
|
check("...and so are the plan's words on a block",
|
|
nl_b[0]["what"].startswith("Laadt tot") and nl_b[0]["why"].startswith(
|
|
"Goedkoopste uren") and nl_b[1]["what"].startswith("Levert tot"))
|
|
check("the language goes back to English after a Dutch build",
|
|
homeview.pill("running") == en["home_words"]["pill"]["running"])
|
|
c = S.make("dynamic", at(10, 7))
|
|
sent = []
|
|
orig_state = c.hass.state
|
|
c.hass.state = lambda *a, **k: sent.append(a) or orig_state(*a, **k)
|
|
snap = lambda: repr((c.mode_tick, c.mode_runs, c._plan_targets, c.arb,
|
|
[(i.soc, i.batt, i.last_write) for i in c.inverters],
|
|
c.mode_gate, c.day_plan))
|
|
before = snap()
|
|
t_a = time.perf_counter()
|
|
for _ in range(10):
|
|
c.forecast_payload("en")
|
|
per = (time.perf_counter() - t_a) / 10 * 1000.0
|
|
print(f" forecast_payload: {per:.1f} ms per build on this machine")
|
|
check("building the tab asks Home Assistant for nothing and changes nothing "
|
|
"control holds (modes, runs, levels, decision, readings)",
|
|
not sent and snap() == before)
|
|
def _broken(c):
|
|
real = dayview.build
|
|
dayview.build = lambda *a, **k: 1 / 0
|
|
logging.disable(logging.CRITICAL)
|
|
try:
|
|
return c.forecast_payload("en")
|
|
finally:
|
|
dayview.build = real
|
|
logging.disable(logging.WARNING)
|
|
|
|
|
|
pay = c.forecast_payload("en")
|
|
check("`Controller.forecast_payload()` gives the tab's JSON, and a fault in "
|
|
"building it gives `ok: false` rather than raising",
|
|
pay.get("ok") is True and len(pay["days"]) == 2
|
|
and _broken(c) == {"ok": False})
|
|
PAGE_SRC = None
|
|
|
|
|
|
async def _serve():
|
|
from aiohttp import web as aioweb, ClientSession
|
|
from app import web
|
|
|
|
class _Sup:
|
|
async def info(self):
|
|
return {"options": {}, "schema": [], "translations": {}}
|
|
c = S.make("owner", datetime.now(UTC))
|
|
runner = aioweb.AppRunner(web.build_app(c, _Sup()))
|
|
await runner.setup()
|
|
site = aioweb.TCPSite(runner, "127.0.0.1", 0)
|
|
await site.start()
|
|
port = site._server.sockets[0].getsockname()[1]
|
|
res = {}
|
|
try:
|
|
async with ClientSession() as s:
|
|
async with s.get(f"http://127.0.0.1:{port}/forecast") as r:
|
|
res["en"] = (r.status, await r.json())
|
|
async with s.get(f"http://127.0.0.1:{port}/forecast?lang=nl-BE") as r:
|
|
res["nl"] = await r.json()
|
|
finally:
|
|
await runner.cleanup()
|
|
return res
|
|
|
|
|
|
try:
|
|
res = asyncio.run(_serve())
|
|
except Exception: # noqa: BLE001
|
|
res = None
|
|
check("GET forecast answers 200 with the tab's JSON, in English by default "
|
|
"and in Dutch with ?lang=nl",
|
|
res is not None and res["en"][0] == 200 and res["en"][1]["ok"] is True
|
|
and res["en"][1]["days"][0]["line"].startswith("Solar still to come")
|
|
and res["nl"]["days"][0]["line"].startswith("Nog verwachte zon"))
|
|
PAGE = open(os.path.join(HERE, "app", "static", "index.html"),
|
|
encoding="utf-8").read()
|
|
JS = open(os.path.join(HERE, "app", "static", "app.js"),
|
|
encoding="utf-8").read()
|
|
check("the page has the Forecast tab between Batteries and Settings, with "
|
|
"its own panel", re.findall(r'data-hash="(\w+)"', PAGE)[:4]
|
|
== ["home", "batteries", "forecast", "settings"]
|
|
and 'id="s-fc"' in PAGE and 'aria-labelledby="t-fc"' in PAGE)
|
|
check("it asks for `forecast?lang=` (relative, for ingress), only while its "
|
|
"screen shows, and sets data as text",
|
|
'fetch("forecast?lang="' in JS and "scheduleForecast" in JS
|
|
and "innerHTML" not in JS)
|
|
labels = set(re.findall(r'''\bt\("(forecast\.[\w.]+)"\)''', JS)) | set(
|
|
re.findall(r'data-t(?:-label)?="(forecast\.[\w.]+)"', PAGE))
|
|
miss = sorted(k for k in labels if not isinstance(
|
|
en["home"]["forecast"].get(k.split(".", 1)[1]), str)
|
|
or not isinstance(nl["home"]["forecast"].get(k.split(".", 1)[1]), str))
|
|
check(f"every label the screen asks for is in both languages ({len(labels)} "
|
|
f"keys; missing {miss})", len(labels) >= 12 and not miss)
|
|
CSS = open(os.path.join(HERE, "app", "static", "app.css"),
|
|
encoding="utf-8").read()
|
|
check("it is readable on a phone: columns flex to the width, the chart has "
|
|
"no fixed pixel width, and the controls are 44 px tall",
|
|
"min-width:0" in CSS and "width:100%" in CSS
|
|
and not re.search(r"\.fc[a-z]*\{[^}]*width:\s*\d{3,}px", CSS)
|
|
and "min-height:44px" in CSS)
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n7. round 2: a block exists where the add-on will act (F1, F2, F3)")
|
|
# --------------------------------------------------------------------------- #
|
|
Q = timedelta(minutes=15)
|
|
|
|
|
|
def level_at(b, h):
|
|
"""The curve's level at hour `h` of the day the list is for, straight
|
|
between its points (the first level before the curve starts, the last
|
|
after it ends)."""
|
|
pts = b["levels"]
|
|
for (h0, v0), (h1, v1) in zip(pts, pts[1:]):
|
|
if h0 <= h <= h1:
|
|
return v0 if h1 <= h0 else v0 + (v1 - v0) * (h - h0) / (h1 - h0)
|
|
return pts[0][1] if h < pts[0][0] else pts[-1][1]
|
|
|
|
|
|
def night_evening(h):
|
|
"""A cheap night and a dear evening, flat between: one cheap run, one give
|
|
run, nothing the plan of tomorrow evening would add before then."""
|
|
return 0.08 if 1.0 <= h < 5.0 else 0.34 if 17.0 <= h < 21.0 else 0.20
|
|
|
|
|
|
def cut_out_at_floor(c):
|
|
"""Each pack cuts out at the add-on's floor. Home, the plan and the curve
|
|
stop giving there; control stops a pack at its own cut-out (the `pack`
|
|
claim, `packshare.envelope()`). With a lower cut-out a battery in a give
|
|
run covers the house below the floor, which none of them draws
|
|
(DECISIONS, Limits)."""
|
|
floor_, _ceil = arbitrage.soc_bounds(c.o)
|
|
for inv in c.inverters:
|
|
inv.bms_soc_floor = floor_
|
|
|
|
|
|
def may_give(inv):
|
|
"""Control's own answer: is this pack asked to discharge at all?"""
|
|
return packshare.envelope(pack_energy_of(inv)).hi > 0.0
|
|
|
|
|
|
def sweep(name, now0, soc, *, guard=None, until=48.0, sun=None):
|
|
"""Walk the real control cycle (`_mode_roles()`) every 15 minutes from
|
|
`now0` to the end of tomorrow, each battery at the level the tab's curve
|
|
has at that time, and return the tab and the role seen per battery and
|
|
quarter. A GIVE role on a pack that control's discharge envelope closes
|
|
is seen as "closed": the pack is not asked to give. The tab is built
|
|
once, at `now0`."""
|
|
c = S.make(name, now0, soc=soc)
|
|
cut_out_at_floor(c)
|
|
c.tariff_plan = S.quarters(now0, night_evening, days=3)
|
|
c.tariff_at = now0
|
|
if guard is not None:
|
|
c.peak_guard = guard
|
|
if sun is not None:
|
|
c.solar_fc = S.forecast(now0, sun[0], sun[1])
|
|
c._mode_roles(now0, None)
|
|
out = tab(c, now0)
|
|
d0, _d1 = homeview.local_day(now0)
|
|
seen = {inv.name: {} for inv in c.inverters}
|
|
t = now0
|
|
end = d0 + timedelta(hours=until)
|
|
while t < end:
|
|
h = (t - d0).total_seconds() / 3600.0
|
|
for i, inv in enumerate(c.inverters):
|
|
day = out["days"][0 if h < 24 else 1]
|
|
b = day["batteries"][i]
|
|
inv.soc = level_at(b, h if h < 24 else h - 24) if b["levels"] \
|
|
else inv.soc
|
|
c.tariff_at = t
|
|
tick = c._mode_roles(t, None)
|
|
for inv in c.inverters:
|
|
role = tick.roles.get(inv.name)
|
|
seen[inv.name][t] = "closed" if role == modetrade.GIVE \
|
|
and not may_give(inv) else role
|
|
t += Q
|
|
return c, out, seen, d0
|
|
|
|
|
|
def role_at(seen, name, d0, h):
|
|
"""The role seen in the quarter that starts at hour `h` from `d0`."""
|
|
return next((r for t, r in seen[name].items()
|
|
if 0 <= (t - d0).total_seconds() / 3600.0 - h < 0.25), None)
|
|
|
|
|
|
def tomorrow_match(out, seen, d0, role, kind):
|
|
"""Per battery, tomorrow: control has `role` in exactly the quarters the
|
|
tab has a block of `kind` (no quarter outside a block, none of a block
|
|
without it), with no tolerance."""
|
|
for i, name in enumerate(seen):
|
|
blocks = [r for r in out["days"][1]["batteries"][i]["blocks"]
|
|
if r["kind"] == kind]
|
|
quarters = [(t, (t - d0).total_seconds() / 3600.0 - 24.0)
|
|
for t in seen[name] if (t - d0).total_seconds() >= 86400]
|
|
for t, h in quarters:
|
|
acting = seen[name][t] == role
|
|
inside = any(r["start_h"] - 1e-6 <= h < r["end_h"] - 1e-6
|
|
for r in blocks)
|
|
if acting != inside:
|
|
print(" ", name, role, "quarter", round(h, 2),
|
|
"control acts" if acting else "tab has a block",
|
|
[(r["start_h"], r["end_h"]) for r in blocks])
|
|
return False
|
|
return True
|
|
|
|
|
|
t22 = at(22, 0, 5)
|
|
c1, o1, s1, d01 = sweep("dynamic", t22, (20.0, 20.0))
|
|
tm1 = o1["days"][1]["batteries"]
|
|
print(" 22:00, both batteries at 20 %: tomorrow", [[(r["kind"], r["from"], r["to"]) for r in b["blocks"]] for b in tm1])
|
|
check("F1, modes on: at 22:00 with both batteries at 20 % tomorrow's give "
|
|
"block (17:00-21:00) and the night's grid block show on both",
|
|
all([(r["kind"], r["from"]) for r in b["blocks"]]
|
|
== [("grid", "01:00"), ("give", "17:00")]
|
|
and b["blocks"][1]["to"] == "21:00" for b in tm1))
|
|
check("F2: ...and control, walked every 15 minutes through tomorrow at the "
|
|
"level the curve has, gives (role GIVE) exactly in the give block and "
|
|
"charges (role CHARGE) only in the grid block, on both batteries",
|
|
tomorrow_match(o1, s1, d01, modetrade.GIVE, "give")
|
|
and tomorrow_match(o1, s1, d01, modetrade.CHARGE, "grid"))
|
|
check("...and today's blocks are still Home's runs there",
|
|
tab_blocks(o1) == home_blocks(home(c1, t22), 2))
|
|
guard = PeakDecision(engaged=True, floor_w=1000.0, ceiling_w=3500.0,
|
|
ceiling_from="cap", projected_w=3600.0)
|
|
c2, o2, s2, d02 = sweep("dynamic", t22, (20.0, 20.0), guard=guard)
|
|
tm2 = o2["days"][1]["batteries"]
|
|
fl2 = arbitrage.soc_bounds(c2.o)[0]
|
|
print(" peak guard, 20 %: level at 17:00", [level_at(b, 17.0) for b in tm2],
|
|
"floor", fl2, "give runs", [(homeview.hhmm(g.start), homeview.hhmm(g.end))
|
|
for g in c2.mode_runs.give])
|
|
check("F2, the UX-53 gate: while the peak guard holds control never "
|
|
"charges tomorrow and the tab draws no grid block. The give run starts "
|
|
"at 17:00 with the curve at the floor on both (the 3 kWh of sun is "
|
|
"less than the house uses), so no give block, and control's GIVE role "
|
|
"there has its discharge closed",
|
|
c2.mode_gate == "peak"
|
|
and any(homeview.hhmm(g.start) == "17:00" for g in c2.mode_runs.give)
|
|
and all(level_at(b, 17.0) == fl2 for b in tm2)
|
|
and all(not b["blocks"] for b in tm2)
|
|
and all(role_at(s2, n, d02, 41.0) == "closed" for n in s2)
|
|
and tomorrow_match(o2, s2, d02, modetrade.CHARGE, "grid")
|
|
and tomorrow_match(o2, s2, d02, modetrade.GIVE, "give"))
|
|
c2b, o2b, s2b, d02b = sweep("dynamic", t22, (60.0, 60.0), guard=guard)
|
|
check("...with the batteries at 60 % the give block stays and the grid "
|
|
"block does not",
|
|
all([r["kind"] for r in b["blocks"]] == ["give"]
|
|
for b in o2b["days"][1]["batteries"])
|
|
and tomorrow_match(o2b, s2b, d02b, modetrade.CHARGE, "grid")
|
|
and tomorrow_match(o2b, s2b, d02b, modetrade.GIVE, "give"))
|
|
# Low sun, a battery at the floor when the evening give run starts (no grid
|
|
# charge, the peak guard holds): Home's rule draws no give block for it, and
|
|
# control has no energy to give from. A little sun fills it a little: a small
|
|
# block, and control gives from it.
|
|
c4, o4, s4, d04 = sweep("dynamic", t22, (20.0, 20.0), guard=guard,
|
|
sun=(0.0, 0.0))
|
|
check("F1, cloudy day, no grid charge, both batteries at the floor at the "
|
|
"evening give run: no block on either, and control has no energy to "
|
|
"give from (the sweep sees none)",
|
|
all(not b["blocks"] for b in o4["days"][1]["batteries"])
|
|
and tomorrow_match(o4, s4, d04, modetrade.GIVE, "give"))
|
|
c5, o5, s5, d05 = sweep("dynamic", t22, (20.0, 20.0), guard=guard,
|
|
sun=(14.0, 14.0))
|
|
gv5 = [[r for r in b["blocks"] if r["kind"] == "give"]
|
|
for b in o5["days"][1]["batteries"]]
|
|
check("...a bright day fills them: a give block on both, and control gives "
|
|
"in exactly that block",
|
|
all(len(g) == 1 and g[0]["from"] == "17:00" for g in gv5)
|
|
and tomorrow_match(o5, s5, d05, modetrade.GIVE, "give"))
|
|
o3 = [sweep("dynamic", t22, soc) for soc in ((55.0, 40.0), (90.0, 70.0))]
|
|
check("F2: the same with the batteries part full and nearly full at 22:00",
|
|
all(tomorrow_match(x[1], x[2], x[3], modetrade.GIVE, "give")
|
|
and tomorrow_match(x[1], x[2], x[3], modetrade.CHARGE, "grid")
|
|
for x in o3))
|
|
|
|
|
|
# Control walked with its own model of the packs, not the curve's levels:
|
|
# every 15 minutes the real `_mode_roles()`; a battery in CHARGE charges at
|
|
# its rating up to `charge_to()`; spare sun goes to the batteries whose range
|
|
# may charge, by room, up to the ceiling; a shortfall is covered by those
|
|
# whose range may give and whose own discharge envelope is open, by what
|
|
# they hold above the cut-out (at the floor), each at most its rating, what
|
|
# one cannot take passed to the others (a water fill, as the allocator's
|
|
# `_fill()`). The house is the profile the tab reads, the sun the forecast
|
|
# at 100 % trust. Per battery and quarter tomorrow, control charges (gives)
|
|
# exactly where the tab has a grid (give) block, both ways, no tolerance.
|
|
# Each tab build is also measured: the curves' floor cuts (`stats`), the
|
|
# level points per battery and day, and the fastest of five builds.
|
|
def water(amount, caps, weights):
|
|
"""`amount` shared by `weights`, none above its cap, the rest passed on."""
|
|
got = {n: 0.0 for n in caps}
|
|
live = {n for n in caps if caps[n] > 1e-12}
|
|
left = amount
|
|
while left > 1e-12 and live:
|
|
w = sum(weights[n] for n in live)
|
|
if w <= 0.0:
|
|
break
|
|
over = {n for n in live
|
|
if left * weights[n] / w >= caps[n] - got[n] - 1e-12}
|
|
if not over:
|
|
for n in live:
|
|
got[n] += left * weights[n] / w
|
|
break
|
|
for n in over:
|
|
left -= caps[n] - got[n]
|
|
got[n] = caps[n]
|
|
live.discard(n)
|
|
return got
|
|
|
|
|
|
def built(c, now):
|
|
"""The tab, with `(floor cuts, capped, fastest of 5 builds in s)`."""
|
|
stats, real = {}, dayview.curve
|
|
|
|
def counted(*a, **k):
|
|
return real(*a, stats=stats, **k)
|
|
dayview.curve = counted
|
|
try:
|
|
out = tab(c, now)
|
|
cut = (stats.get("floor_cuts", 0), stats.get("capped", False))
|
|
st = home(c, now)
|
|
fast = []
|
|
for _ in range(5):
|
|
t0_ = time.perf_counter()
|
|
dayview.build(c, st, now)
|
|
fast.append(time.perf_counter() - t0_)
|
|
finally:
|
|
dayview.curve = real
|
|
return out, cut + (min(fast),)
|
|
|
|
|
|
def own_walk(soc, *, guard=None, sun=(0.0, 0.0), evening=None, max_w=None):
|
|
now0 = t22
|
|
c = S.make("dynamic", now0, soc=soc)
|
|
cut_out_at_floor(c)
|
|
if max_w is not None:
|
|
for inv in c.inverters:
|
|
inv.max_w = max_w
|
|
if evening is not None:
|
|
c.house_profile = S.profile(now0, evening=evening)
|
|
floor_, ceil_ = arbitrage.soc_bounds(c.o)
|
|
c.tariff_plan = S.quarters(now0, night_evening, days=3)
|
|
c.tariff_at = now0
|
|
if guard is not None:
|
|
c.peak_guard = guard
|
|
c.solar_trust_pct = 100
|
|
c.solar_fc = fc = S.forecast(now0, sun[0], sun[1])
|
|
c._mode_roles(now0, None)
|
|
out, cost = built(c, now0)
|
|
house_w, base_w = c.house_profile.estimator()
|
|
d0, _d1 = homeview.local_day(now0)
|
|
seen = {inv.name: {} for inv in c.inverters}
|
|
t = now0
|
|
while t < d0 + timedelta(hours=48):
|
|
c.tariff_at = t
|
|
tick = c._mode_roles(t, None)
|
|
targets = c.plan_targets()
|
|
packs = {p.name: p for p in c._mode_packs()}
|
|
house = max(house_w(S.brussels(t).hour), base_w) / 1000.0 * 0.25
|
|
net = (dayview.sun_kwh(fc, now0, t, t + Q) or 0.0) - house
|
|
takers, givers = [], []
|
|
for inv in c.inverters:
|
|
role = tick.roles.get(inv.name)
|
|
lo, hi = tick.ranges.get(inv.name, (0.0, 0.0))
|
|
act = None
|
|
if role == modetrade.CHARGE:
|
|
to = modetrade.charge_to(packs[inv.name], ceil_, targets)
|
|
step = min(inv.max_w / 1000.0 * 0.25 * dayplan.EFF,
|
|
max(0.0, to - inv.soc) / 100.0 * inv.capacity_kwh)
|
|
act = "grid" if step > 1e-9 else None
|
|
inv.soc += step / inv.capacity_kwh * 100.0
|
|
else:
|
|
if lo < 0.0 and inv.soc < ceil_:
|
|
takers.append(inv)
|
|
if hi > 0.0 and may_give(inv):
|
|
givers.append(inv)
|
|
act = "give" if role == modetrade.GIVE else None
|
|
seen[inv.name][t] = act
|
|
if net > 0.0:
|
|
room = {i.name: (ceil_ - i.soc) / 100.0 * i.capacity_kwh
|
|
/ dayplan.EFF for i in takers}
|
|
tot = sum(room.values())
|
|
for i in takers:
|
|
part = min(net * room[i.name] / tot, room[i.name],
|
|
i.max_w / 1000.0 * 0.25) if tot > 0.0 else 0.0
|
|
i.soc += part * dayplan.EFF / i.capacity_kwh * 100.0
|
|
elif net < 0.0:
|
|
hold = {i.name: max(0.0, i.soc - floor_) / 100.0 * i.capacity_kwh
|
|
* dayplan.EFF for i in givers}
|
|
got = water(-net, {i.name: min(hold[i.name], i.max_w / 1000.0
|
|
* 0.25) for i in givers}, hold)
|
|
for i in givers:
|
|
i.soc = max(floor_, i.soc - got[i.name] / dayplan.EFF
|
|
/ i.capacity_kwh * 100.0)
|
|
t += Q
|
|
return out, seen, d0, cost
|
|
|
|
|
|
def own_match(out, seen, d0):
|
|
"""`[(name, kind, hour, who)]`: the quarters tomorrow where control and
|
|
the tab disagree (who acts there alone)."""
|
|
bad = []
|
|
for i, name in enumerate(seen):
|
|
for kind in ("grid", "give"):
|
|
blocks = [r for r in out["days"][1]["batteries"][i]["blocks"]
|
|
if r["kind"] == kind]
|
|
for t, a in seen[name].items():
|
|
h = (t - d0).total_seconds() / 3600.0 - 24.0
|
|
if h < 0.0:
|
|
continue
|
|
inside = any(r["start_h"] - 1e-6 <= h < r["end_h"] - 1e-6
|
|
for r in blocks)
|
|
if (a == kind) != inside:
|
|
bad.append((name, kind, round(h, 2),
|
|
"control" if a == kind else "tab"))
|
|
return bad
|
|
|
|
|
|
def give_span(seen, d0, name):
|
|
hs = [(t - d0).total_seconds() / 3600.0 - 24.0
|
|
for t, a in seen[name].items()
|
|
if a == "give" and (t - d0).total_seconds() >= 86400]
|
|
return (round(min(hs), 2), round(max(hs) + 0.25, 2)) if hs else None
|
|
|
|
|
|
own = {}
|
|
for key, soc, g, sun, *more in (("20/20", (20.0, 20.0), None, (0.0, 0.0)),
|
|
("55/40", (55.0, 40.0), None, (0.0, 0.0)),
|
|
("90/70", (90.0, 70.0), None, (0.0, 0.0)),
|
|
("30/80", (30.0, 80.0), None, (0.0, 0.0)),
|
|
("60/60 guard", (60.0, 60.0), guard, (0.0, 0.0)),
|
|
("20/20 guard", (20.0, 20.0), guard, (0.0, 0.0)),
|
|
("30/30 guard", (30.0, 30.0), guard, (0.0, 0.0)),
|
|
("25/45 guard", (25.0, 45.0), guard, (0.0, 0.0)),
|
|
("20/20 guard 14 kWh", (20.0, 20.0), guard,
|
|
(14.0, 14.0)),
|
|
("20/20 guard 6 kWh", (20.0, 20.0), guard,
|
|
(6.0, 6.0)),
|
|
("20/20 8 kWh", (20.0, 20.0), None, (8.0, 8.0)),
|
|
# one battery empty, the other held to its rating
|
|
("30/80 guard 7 kW house", (30.0, 80.0), guard,
|
|
(0.0, 0.0), 7000.0, None),
|
|
("30/80 guard 2 x 3 kW, 3.5 kW house", (30.0, 80.0),
|
|
guard, (0.0, 0.0), 3500.0, 3000.0)):
|
|
evening_, max_w_ = (more + [None, None])[:2]
|
|
o_w, s_w, d_w, cost = own_walk(soc, guard=g, sun=sun, evening=evening_,
|
|
max_w=max_w_)
|
|
pts = max(len(b["levels"]) for d in o_w["days"] for b in d["batteries"])
|
|
own[key] = (o_w, s_w, d_w, own_match(o_w, s_w, d_w), cost, pts)
|
|
print(f" own walk {key}: tab",
|
|
[[(r["kind"], r["from"], r["to"]) for r in b["blocks"]]
|
|
for b in o_w["days"][1]["batteries"]],
|
|
"control gives", {n: give_span(s_w, d_w, n) for n in s_w},
|
|
"mismatched", len(own[key][3]), own[key][3][:4],
|
|
f"| floor cuts {cost[0]}, capped {cost[1]}, {pts} points, "
|
|
f"{cost[2] * 1000:.0f} ms")
|
|
check("F2, control walked with its own model of the packs (not the curve's "
|
|
"levels): in all 13 cases it charges and gives in exactly the quarters "
|
|
"the tab has a block of that kind, on both batteries",
|
|
len(own) == 13 and all(not v[3] for v in own.values()))
|
|
MAX_CUTS, MAX_POINTS, MAX_S = 40, 40, 0.2
|
|
big = {k: (v[4], v[5]) for k, v in own.items() if v[4][1]
|
|
or v[4][0] > MAX_CUTS or v[5] > MAX_POINTS or v[4][2] >= MAX_S}
|
|
check(f"...and every build stays small: at most {MAX_CUTS} floor cuts over "
|
|
f"all its curves, never the curve's bound (`MAX_FLOOR_CUTS`), at most "
|
|
f"{MAX_POINTS} level points per battery and day, and under {MAX_S:g} s "
|
|
f"(the fastest of five builds on this host) {big}",
|
|
all(v[4][0] <= MAX_CUTS and not v[4][1] and v[5] <= MAX_POINTS
|
|
and v[4][2] < MAX_S for v in own.values()))
|
|
o_7, s_7, d_7 = own["30/80 guard 7 kW house"][:3]
|
|
check("...30/80 % with the guard and a 7 kW evening house (A soon empty, B "
|
|
"held to its rating): control gives A 17:00-17:45 and B 17:00-19:45, "
|
|
"and B's give block ends in control's last giving quarter, not at 20:00",
|
|
give_span(s_7, d_7, "inverter-a") == (17.0, 17.75)
|
|
and give_span(s_7, d_7, "inverter-b") == (17.0, 19.75)
|
|
and 19.5 < o_7["days"][1]["batteries"][1]["blocks"][-1]["end_h"]
|
|
<= 19.75)
|
|
o_g, s_g, d_g, _b = own["25/45 guard"][:4]
|
|
check("...25/45 % with the peak guard: control gives 17:00-20:15 on both "
|
|
"(the packs empty in the 20:00 quarter), and the tab's give block "
|
|
"starts at 17:00 and ends inside that quarter, at the curve's floor "
|
|
"point",
|
|
all(give_span(s_g, d_g, n) == (17.0, 20.25) for n in s_g)
|
|
and all([(r["kind"], r["from"]) for r in b["blocks"]] == [("give",
|
|
"17:00")]
|
|
and 20.0 < b["blocks"][0]["end_h"] <= 20.25
|
|
for b in o_g["days"][1]["batteries"]))
|
|
o_6, s_6, d_6, _b = own["20/20 guard 6 kWh"][:4]
|
|
check("...20/20 % with the guard and 6 kWh of sun: control gives 17:00-17:15, "
|
|
"and the tab's give block ends inside that quarter",
|
|
all(give_span(s_6, d_6, n) == (17.0, 17.25) for n in s_6)
|
|
and all(len(b["blocks"]) == 1 and 17.0 < b["blocks"][0]["end_h"] <= 17.25
|
|
for b in o_6["days"][1]["batteries"]))
|
|
fl_g = arbitrage.soc_bounds(S.make("dynamic", t22).o)[0]
|
|
check("...and the curve has a point exactly at the floor where the block "
|
|
"ends (no step end)",
|
|
all([v for h, v in b["levels"] if abs(h - b["blocks"][0]["end_h"])
|
|
< 1e-3] == [fl_g] for b in o_g["days"][1]["batteries"]))
|
|
|
|
|
|
def off_prices(h):
|
|
if 23.0 <= h < 24.0:
|
|
return 0.05
|
|
if 7.0 <= h < 9.0:
|
|
return 0.40
|
|
return 0.20
|
|
|
|
|
|
def off_case(soc):
|
|
now = at(20, 0, 5)
|
|
c = S.make("off", now, soc=soc)
|
|
c.tariff_plan = S.quarters(now, off_prices)
|
|
c.tariff_at = now
|
|
return c, now, tab(c, now), home(c, now)
|
|
|
|
|
|
for soc in ((20.0, 20.0), (60.0, 60.0), (95.0, 95.0)):
|
|
c, now, out, st = off_case(soc)
|
|
tz = S.brussels(now).tzinfo
|
|
v, _why = homeview.trade_verdict(c, c.fresh_tariff_plan(now), now)
|
|
t07 = at(24 + 7, 0, 5)
|
|
c.tariff_at = t07
|
|
v7, _w = homeview.trade_verdict(c, c.fresh_tariff_plan(t07), t07)
|
|
t06 = at(24 + 6, 0, 5)
|
|
c.tariff_at = t06
|
|
v6, _w6 = homeview.trade_verdict(c, c.fresh_tariff_plan(t06), t06)
|
|
check(f"F1, modes off, a charge today at 23:00 and a give tomorrow at "
|
|
f"07:00, batteries at {soc[0]:.0f} %: Home says 'Tomorrow: gives "
|
|
"power at 07:00', the tab draws that give on both batteries, and "
|
|
"the control cycle's own verdict gives at 07:00 and not at 06:00",
|
|
any("07:00" in x for x in st["plan"]["lines"])
|
|
and [[(r["kind"], r["from"]) for r in b["blocks"]]
|
|
for b in out["days"][1]["batteries"]]
|
|
== [[("give", "07:00")]] * 2
|
|
and v7 is not None and v7.dearest.start.astimezone(tz).strftime(
|
|
"%H:%M") == "07:00" and v7.dearest.start <= t07
|
|
and v6 is not None and v6.dearest.start > t06)
|
|
|
|
# Modes off: no forecast is read and no house profile kept, so there is no
|
|
# curve, and tomorrow's blocks are Home's own count (`dayview._counted()`).
|
|
# Control is walked every 15 minutes from 22:00 to the end of tomorrow on a
|
|
# cloudy day (no sun), the packs moved by what control does:
|
|
# `arbitrage.decide()` with the inputs `Controller.cycle()` gives it (the
|
|
# emptiest pack, `trade_power_w()`, the guard's headroom), and between trades
|
|
# each pack above its cut-out covers its share of the house. The tab is
|
|
# built once, at 22:00. Each price extreme is one quarter, so a leg is one
|
|
# quarter: per battery, control charges and gives in exactly the quarters
|
|
# the tab has a block of that kind, in both directions.
|
|
HOUSE_W = 400.0
|
|
|
|
|
|
def one_quarter(cheap, dear):
|
|
"""Prices with one cheap and one dear quarter a day (hours), 0.20 else."""
|
|
return lambda h: 0.12 if h == cheap else 0.45 if h == dear else 0.20
|
|
|
|
|
|
def off_sweep(price_of, now0, soc):
|
|
c = S.make("off", now0, soc=soc)
|
|
cut_out_at_floor(c)
|
|
c.tariff_plan = S.quarters(now0, price_of, days=3)
|
|
c.tariff_at = now0
|
|
# the guard has measured the quarter: 6 kW of charging headroom
|
|
c.peak_guard = dataclasses.replace(c.peak_guard, wanted_w=-6000.0)
|
|
out = tab(c, now0)
|
|
floor_, ceil_ = arbitrage.soc_bounds(c.o)
|
|
d0, _d1 = homeview.local_day(now0)
|
|
seen = {inv.name: {} for inv in c.inverters}
|
|
t = now0
|
|
while t < d0 + timedelta(hours=48):
|
|
c.tariff_at = t
|
|
charge_w, discharge_w = c.trade_power_w(floor_, ceil_)
|
|
d = arbitrage.decide(
|
|
c.fresh_tariff_plan(t), t,
|
|
peak_guard_active=arbitrage.guard_state(c.peak_guard),
|
|
charge_headroom_w=peakguard.charge_headroom_w(c.peak_guard),
|
|
soc=c.pack_floor_soc,
|
|
min_spread_eur_kwh=arbitrage.threshold_eur_kwh(c.o),
|
|
uncertainty=frozenset({Uncertainty.NO_WEATHER}),
|
|
charge_w=charge_w, discharge_w=discharge_w,
|
|
soc_floor_pct=floor_, soc_ceiling_pct=ceil_)
|
|
able = [inv for inv in c.inverters if may_give(inv)]
|
|
for inv in c.inverters:
|
|
cap, act = float(inv.capacity_kwh), None
|
|
if d.action == arbitrage.CHARGE and inv.soc < ceil_:
|
|
act = "grid"
|
|
kw = abs(d.power_w) * inv.max_w / charge_w / 1000.0
|
|
inv.soc = min(ceil_, inv.soc + kw * 0.25 * dayplan.EFF
|
|
/ cap * 100.0)
|
|
elif d.action == arbitrage.DISCHARGE and inv.soc > floor_ \
|
|
and may_give(inv):
|
|
act = "give"
|
|
inv.soc = max(floor_, inv.soc - inv.max_w / 1000.0 * 0.25
|
|
/ dayplan.EFF / cap * 100.0)
|
|
elif d.action is None and inv in able:
|
|
inv.soc = max(floor_, inv.soc - HOUSE_W / len(able) / 1000.0
|
|
* 0.25 / dayplan.EFF / cap * 100.0)
|
|
seen[inv.name][t] = act
|
|
t += Q
|
|
return c, out, seen, d0
|
|
|
|
|
|
def acts(seen, d0, kind):
|
|
"""`{name: [hour, ...]}` tomorrow where control does `kind`."""
|
|
return {name: [round((t - d0).total_seconds() / 3600.0 - 24.0, 2)
|
|
for t, a in row.items() if a == kind
|
|
and (t - d0).total_seconds() >= 86400]
|
|
for name, row in seen.items()}
|
|
|
|
|
|
def off_match(out, seen, d0):
|
|
"""Per battery, tomorrow: control does `kind` in exactly the quarters
|
|
the tab has a block of that kind."""
|
|
for kind in ("grid", "give"):
|
|
got = acts(seen, d0, kind)
|
|
for i, name in enumerate(seen):
|
|
hs = [round(r["start_h"] + 0.25 * k, 2)
|
|
for r in out["days"][1]["batteries"][i]["blocks"]
|
|
if r["kind"] == kind
|
|
for k in range(round((r["end_h"] - r["start_h"]) / 0.25))]
|
|
if hs != got[name]:
|
|
print(" ", name, kind, "tab", hs, "control", got[name])
|
|
return False
|
|
return True
|
|
|
|
|
|
def kinds(out):
|
|
return [[(r["kind"], r["from"], r["to"]) for r in b["blocks"]]
|
|
for b in out["days"][1]["batteries"]]
|
|
|
|
|
|
# A: at the floor, a cheap quarter at 05:00, a dear one at 07:00.
|
|
ca, oa, sa, da = off_sweep(one_quarter(5.0, 7.0), t22, (20.0, 20.0))
|
|
print(" modes off A, 20 %, cheap 05:00, dear 07:00, no sun: tab", kinds(oa),
|
|
"control", acts(sa, da, "grid"), acts(sa, da, "give"))
|
|
check("F2, modes off: both batteries at the floor at 22:00, a charge at "
|
|
"05:00 before the 07:00 give: both blocks on both, and control charges "
|
|
"and gives in exactly those quarters",
|
|
kinds(oa) == [[("grid", "05:00", "05:15"), ("give", "07:00", "07:15")]]
|
|
* 2 and off_match(oa, sa, da))
|
|
# B: the cloudy-day give run. Nothing charges before the give.
|
|
cb, ob, sb, db = off_sweep(one_quarter(13.0, 7.0), t22, (20.0, 20.0))
|
|
print(" modes off B, 20 %, dear 07:00, cheap 13:00, no sun: tab", kinds(ob),
|
|
"control", acts(sb, db, "grid"), acts(sb, db, "give"))
|
|
v_b, _w = homeview.trade_verdict(cb, cb.fresh_tariff_plan(t22), t22)
|
|
check("F2, modes off, cloudy day: both batteries at the floor at 22:00 and "
|
|
"nothing charges before the verdict's 07:00 give: no give block on "
|
|
"either and control never gives; the 13:00 charge block shows and "
|
|
"control charges in exactly that quarter",
|
|
v_b is not None and homeview.hhmm(v_b.dearest.start) == "07:00"
|
|
and kinds(ob) == [[("grid", "13:00", "13:15")]] * 2
|
|
and all(not v for v in acts(sb, db, "give").values())
|
|
and off_match(ob, sb, db))
|
|
check("...and each battery says the add-on plans to give, not that it only "
|
|
"follows the house",
|
|
all(b["line"] == say("give_later")
|
|
for b in ob["days"][1]["batteries"]))
|
|
# D: the same day with B at 60 %: the site gives as one, and not while its
|
|
# emptiest pack is at the floor (`arbitrage.decide()` reads the emptiest).
|
|
cd_, od, sd, dd = off_sweep(one_quarter(13.0, 7.0), t22, (20.0, 60.0))
|
|
print(" modes off D, 20 % and 60 %, dear 07:00, cheap 13:00: tab", kinds(od),
|
|
"control", acts(sd, dd, "grid"), acts(sd, dd, "give"))
|
|
check("F2, modes off: one battery at the floor and one at 60 %: no give block "
|
|
"on either (the site does not give while its emptiest pack is at the "
|
|
"floor) and control gives on neither; the charge blocks match",
|
|
all(not [r for r in b if r[0] == "give"] for b in kinds(od))
|
|
and all(not v for v in acts(sd, dd, "give").values())
|
|
and off_match(od, sd, dd))
|
|
c_dn = S.make("off", t22, soc=(20.0, 60.0))
|
|
c_dn.tariff_plan = S.quarters(t22, one_quarter(13.0, 7.0), days=3)
|
|
c_dn.tariff_at = t22
|
|
od_nl = tab(c_dn, t22, "nl")
|
|
check("...and B, which has energy, is not told it has none: the line names "
|
|
"'a battery' (the emptiest one holds the site back), EN and NL",
|
|
all(b["line"] == say("give_later") for b in od["days"][1]["batteries"])
|
|
and say("give_later").endswith(
|
|
"No block shows while a battery has no energy to give.")
|
|
and all(b["line"] == say("give_later", "nl")
|
|
for b in od_nl["days"][1]["batteries"])
|
|
and say("give_later", "nl").endswith(
|
|
"Er verschijnt geen blok zolang een batterij niets kan leveren."))
|
|
# C: full at 22:00 and a cheap quarter at 01:00: the house uses some first,
|
|
# so control charges then, and the tab has that block.
|
|
cc, oc, sc, dc = off_sweep(one_quarter(1.0, 17.0), t22, (95.0, 95.0))
|
|
print(" modes off C, 95 %, cheap 01:00, dear 17:00, no sun: tab", kinds(oc),
|
|
"control", acts(sc, dc, "grid"), acts(sc, dc, "give"))
|
|
check("F2, modes off: both batteries at the ceiling at 22:00, a cheap "
|
|
"quarter at 01:00 and a dear one at 17:00: the house uses some first, "
|
|
"so control charges at 01:00 and gives at 17:00, in exactly the "
|
|
"quarters the tab has a block",
|
|
kinds(oc) == [[("grid", "01:00", "01:15"), ("give", "17:00", "17:15")]]
|
|
* 2 and off_match(oc, sc, dc))
|
|
|
|
# `trade_runs(ahead=f)` reads each leg at `f(name, kind, start)`.
|
|
n0_, n1_ = homeview.local_day(homeview.local_day(t22)[1] + timedelta(hours=2))
|
|
fl_b, ce_b = arbitrage.soc_bounds(cb.o)
|
|
cb.tariff_at = t22
|
|
|
|
|
|
def legs_at(lv):
|
|
return sorted((r["battery"], r["kind"], r["from"]) for r in
|
|
homeview.runs_for(cb, n0_, n1_, t22, False,
|
|
ahead=lambda _n, k, _s: lv.get(k)))
|
|
|
|
|
|
check("trade_runs(ahead=f): a give leg is read at the level f gives when it "
|
|
"starts (none at the floor), a charge too (none at the ceiling); f "
|
|
"None reads a charge from the floor and a give from the ceiling",
|
|
legs_at({"give": fl_b, "grid": None}) == [(0, "grid", "13:00"),
|
|
(1, "grid", "13:00")]
|
|
and legs_at({"give": 60.0, "grid": ce_b}) == [(0, "give", "07:00"),
|
|
(1, "give", "07:00")]
|
|
and legs_at({}) == sorted([(0, "give", "07:00"), (1, "give", "07:00"),
|
|
(0, "grid", "13:00"), (1, "grid", "13:00")]))
|
|
tm90 = o3[1][1]["days"][1]["batteries"]
|
|
check("F2: a battery already at its planned level when the run starts has no "
|
|
"grid block tomorrow (A at 90 %), the one below it has (B at 70 %); "
|
|
"control charged neither A nor B outside that",
|
|
[r["kind"] for r in tm90[0]["blocks"]] == ["give"]
|
|
and [r["kind"] for r in tm90[1]["blocks"]] == ["grid", "give"])
|
|
# Today: the give run is planned and Home draws no block for an empty battery.
|
|
c_s = S.make("sunny", at(10, 7), soc=(20.0, 20.0))
|
|
st_s = home(c_s, at(10, 7))
|
|
o_s = tab(c_s, at(10, 7))
|
|
check("today: with a give run in the plan and both batteries at the floor "
|
|
"Home draws no give block, the tab equals Home, and it says the "
|
|
"add-on plans to give, never 'No charge or give blocks planned'",
|
|
not [r for r in st_s["plan"]["runs"] if r["kind"] == "give"]
|
|
and c_s.mode_runs.give
|
|
and tab_blocks(o_s) == home_blocks(st_s, 2)
|
|
and all(b["line"] == say("give_later")
|
|
for b in o_s["days"][0]["batteries"]))
|
|
bad_no_blocks = []
|
|
for name, hh in SCN + [("sunny", 10.1), ("dynamic", 22.0)]:
|
|
for soc in ((20.0, 20.0), (55.0, 40.0), (95.0, 95.0)):
|
|
c = S.make(name, at(hh), soc=soc)
|
|
if not c.modes_on:
|
|
continue
|
|
o = tab(c, at(hh))
|
|
for k in (0, 1):
|
|
for b in o["days"][k]["batteries"]:
|
|
if b["line"] == say("no_blocks"):
|
|
d0_, d1_ = homeview.local_day(at(hh))
|
|
lo, hi = (d0_, d1_) if k == 0 else homeview.local_day(
|
|
d1_ + timedelta(hours=2))
|
|
if c.mode_runs is not None and c.mode_runs.on and any(
|
|
g.end > max(at(hh), lo) and g.start < hi
|
|
for g in c.mode_runs.give):
|
|
bad_no_blocks.append((name, hh, soc, k))
|
|
check("'No charge or give blocks planned.' is never said where the run plan "
|
|
f"has a give run in that day ({bad_no_blocks})", not bad_no_blocks)
|
|
|
|
# F3: trust as the plan counts it.
|
|
cap_big = 100.0
|
|
f3 = curve([Batt("a", cap_big, 50.0, 50.0, True, False)], T0,
|
|
T0 + timedelta(hours=1), flat(10.0), flat(2.0), floor_pct=20.0,
|
|
ceiling_pct=95.0, trust=0.7)
|
|
stored = (lv(f3, "a", 1) - 50.0) / 100.0 * cap_big
|
|
plan_sun = 0.7 * dayplan.EFF * max(0.0, 10.0 - 2.0)
|
|
print(f" 10 kWh sun, 2 kWh house, 70 %: curve stores {stored:.3f} kWh, plan {plan_sun:.3f}")
|
|
check("F3: 10 kWh of sun, 2 kWh of house and 70 % trust: the curve stores "
|
|
"5.32 kWh, as the plan counts trust x EFF x (sun - house), not 4.75",
|
|
abs(stored - 5.32) < 1e-6 and abs(plan_sun - 5.32) < 1e-9)
|
|
f3b = curve([Batt("a", cap_big, 50.0, 50.0, True, False)], T0,
|
|
T0 + timedelta(hours=1), flat(1.0), flat(2.0), floor_pct=20.0,
|
|
ceiling_pct=95.0, trust=0.7)
|
|
check("...a shortfall is not scaled by trust (the house needs all of it)",
|
|
abs((50.0 - lv(f3b, "a", 1)) / 100.0 * cap_big - 1.0 / dayplan.EFF)
|
|
< 1e-6)
|
|
c_t = S.make("owner", at(10, 7))
|
|
c_t.solar_trust_pct = 70
|
|
hi_ = tab(c_t, at(10, 7))["days"][0]["batteries"][0]["levels"]
|
|
check("on a controller the curve is built with the plan's trust "
|
|
"(`solar_trust_pct`)", hi_ and max(v for _h, v in hi_) > hi_[0][1])
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED")
|
|
sys.exit(1)
|
|
print("all passed")
|