admin/goodwe-addon
1
0
Fork 0
Code Issues Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

UX-54: a Forecast tab with the sun, the house and each battery's level, per hour #210

Merged
admin merged 8 commits from UX-54 into release/1.0 2026-10-27 21:07:00 +01:00
Conversation 7 Commits 8 Files Changed 17
+211 -55
3 changed files with 211 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit a0ace389a4 - Show all commits
goodwe_controller/DECISIONS/UX-54-forecast-tab.md
+10 -3
View File
@@ -48,9 +48,14 @@ battery's expected level with its blocks. It follows the add-on, as Home does
levels (`_mode_roles()` each quarter; the house from the profile, the sun
at 100 % trust, a GIVE role counted only where control's own discharge
envelope `packshare.envelope()` is open, the cut-out at the floor), charges
and gives in exactly the quarters the tab has a block in 11 cases
and gives in exactly the quarters the tab has a block in 13 cases
(`test_ux54.py` part 7, `own_walk()`), e.g. 25/45 % with the peak guard:
control gives 17:00-20:15, the block is 17:00-20:06. With a cut-out under
control gives 17:00-20:15, the block is 17:00-20:06; 30/80 % with a 7 kW
evening house, B held to its rating: control 17:00-19:45, the block
17:00-19:32. The floor point is placed in one cut, where the battery runs
out at the rate it gives (`_floor_cut()`, checked to the second against
times worked out by hand), so a build stays small: at most 14 floor cuts
and 28 level points per battery and day in those cases. With a cut-out under
the floor, or a house or sun that differs from the forecast and profile,
control differs (Limits).
- **The sun** is what `solarfc` reads. An hourly split (Solcast
@@ -64,7 +69,9 @@ battery's expected level with its blocks. It follows the add-on, as Home does
`daycurve.curve()` derives it: each battery has the role `modetrade.decide()`
gives (follow the house, hold, give, charge from the grid, rest), spare solar
goes to Solar first batteries first, a shortfall is met by those that may
give, both at `dayplan.EFF`, inside the plan's floor and ceiling, from the
give, shared as control shares it (by what each holds above the floor,
`packshare.weights()`, each at most its rating, the rest passed on), both
at `dayplan.EFF`, inside the plan's floor and ceiling, from the
level now. Spare solar is `trust x max(0, sun - house)` and what is stored
loses `EFF`, the plan's own `trust x EFF x max(0, sun - house)` (10 kWh of
sun, 2 kWh of house, 70 %: 5.32 kWh stored). A shortfall is not scaled by
goodwe_controller/app/daycurve.py
+44 -31
View File
@@ -13,10 +13,13 @@ first or no price plan is on; inside a give run it covers the house (GIVE);
else it holds (HOLD, may charge from spare solar, gives nothing). Spare
solar (sun less the house's use) goes to Solar first batteries first, then
the rest, by room. A shortfall comes from the batteries that may give,
those in a give run first, by what they hold above the floor. Charging and
giving lose `dayplan.EFF`, as the plan counts. A level stays inside the
floor and ceiling the plan uses, and a battery already outside them does
not move further out.
those in a give run first, by what they hold above the floor, each at most
its rating, the rest passed on (as control shares it, `packshare.weights()`
and the allocator's fill). A battery that empties inside a step gets a
point there (`_floor_cut()`), as one that reaches its grid level does.
Charging and giving lose `dayplan.EFF`, as the plan counts. A level stays
inside the floor and ceiling the plan uses, and a battery already outside
them does not move further out.
`SolarToday` keeps the site's measured solar per hour, in memory, from the
reading the control cycle already has. It is shown and never read back
@@ -91,38 +94,41 @@ def _spread(amount, caps, weights):
def _give_caps(roles, level, by, floor, hrs):
"""`(first, rest)`: what each battery that may give can give in a step
of `hrs` hours (its rating, or what it holds above the floor), those in a
give run first."""
caps = {n: min(by[n].rating_kw * hrs,
max(0.0, level[n] - floor) / 100.0
* by[n].capacity_kwh * EFF)
"""`(first, rest, held)`: what each battery that may give can give in a
step of `hrs` hours (its rating, or what it holds above the floor), those
in a give run first, and what each holds above the floor (as given)."""
held = {n: max(0.0, level[n] - floor) / 100.0 * by[n].capacity_kwh * EFF
for n, r in roles.items() if r[2]}
caps = {n: min(by[n].rating_kw * hrs, h) for n, h in held.items()}
first = {n: c for n, c in caps.items() if roles[n][0] == GIVE}
return first, {n: c for n, c in caps.items() if n not in first}
return first, {n: c for n, c in caps.items() if n not in first}, held
def _floor_cut(a, z, net, roles, level, by, floor, hrs):
"""Where in the step `[a, z)` the first battery that gives reaches the
floor, or None. A shortfall is spread over a step at a steady rate and
shared by what each can give, so a group that holds less than its part
of the shortfall, one of them short of its rating, is empty at
`held / part` of the step."""
left, cut = -net, None
for grp in _give_caps(roles, level, by, floor, hrs):
held = sum(grp.values())
if held > EPS and left > held + EPS and any(
c < by[n].rating_kw * hrs * (1.0 - 1e-9)
for n, c in grp.items()):
t = a + (z - a) * (held / left)
if a + MIN_CUT < t < z - MIN_CUT and (cut is None or t < cut):
cut = t
left = max(0.0, left - held)
floor, or None. The shortfall comes at a steady rate, `-net / hrs` kW.
It is shared as control shares it (`packshare.weights()`: by the energy
each holds above its cut-out, here the floor), each at most its rating,
what one cannot take passed to the others, those in a give run first.
A battery giving `v` kW that holds `h` kWh is empty at `h / v` hours; a
cut there makes the step come out exact. A battery already at the floor
holds nothing, gives nothing and never sets a cut."""
rate, cut = -net / hrs, None
*groups, held = _give_caps(roles, level, by, floor, hrs)
for grp in groups:
rates, rate = _spread(rate, {n: by[n].rating_kw for n in grp},
{n: held[n] for n in grp})
for n, v in rates.items():
if held[n] > EPS and v > EPS:
t = a + timedelta(hours=held[n] / v)
if a + MIN_CUT < t < z - MIN_CUT and (cut is None
or t < cut):
cut = t
return cut
def curve(batts, start, end, sun_kwh, use_kwh, *, floor_pct, ceiling_pct,
trust=1.0):
trust=1.0, stats=None):
"""`{name: [(t, pct), ...]}` from `start` (each battery's level now) to
`end`, a point at every hour edge, every block edge and where a battery
reaches its grid level or, giving, the floor. A battery's list
@@ -130,7 +136,9 @@ def curve(batts, start, end, sun_kwh, use_kwh, *, floor_pct, ceiling_pct,
`stop_at` comes. `sun_kwh` and `use_kwh` give kWh for the span `[a, b)`.
Spare solar is `trust x max(0, sun - use)`, and what is stored loses
`EFF`: `dayplan.plan()`'s own `trust x EFF x max(0, sun - house)`. A
shortfall is the house's use less the whole sun."""
shortfall is the house's use less the whole sun. `stats`, a dict or
None: gets `floor_cuts` (the floor points placed) and `capped` (True when
`MAX_FLOOR_CUTS` stopped one)."""
floor, ceil = float(floor_pct), float(ceiling_pct)
cuts = {end}
for a, _b, _h in _hour_slices(start, end):
@@ -143,7 +151,7 @@ def curve(batts, start, end, sun_kwh, use_kwh, *, floor_pct, ceiling_pct,
level = {b.name: float(b.soc) for b in batts}
out = {b.name: [(start, level[b.name])] for b in batts}
alive = {b.name for b in batts}
floor_cuts = 0
floor_cuts, capped = 0, False
i = 0
while i < len(ts) - 1:
a, z = ts[i], ts[i + 1]
@@ -185,7 +193,9 @@ def curve(batts, start, end, sun_kwh, use_kwh, *, floor_pct, ceiling_pct,
# A battery that gives all it holds inside the step gets a point
# where it reaches the floor, the same way.
cut = _floor_cut(a, z, net, roles, level, by, floor, hrs) \
if net < -EPS and floor_cuts < MAX_FLOOR_CUTS else None
if net < -EPS else None
if cut is not None and floor_cuts >= MAX_FLOOR_CUTS:
capped, cut = True, None
if cut is not None:
floor_cuts += 1
ts.insert(i + 1, cut)
@@ -208,16 +218,19 @@ def curve(batts, start, end, sun_kwh, use_kwh, *, floor_pct, ceiling_pct,
for n, v in got.items():
level[n] += v * EFF / by[n].capacity_kwh * 100.0
elif net < -EPS:
first, rest = _give_caps(roles, level, by, floor, hrs)
first, rest, held = _give_caps(roles, level, by, floor, hrs)
left = -net
for grp in (first, rest):
got, left = _spread(left, grp, dict(grp))
got, left = _spread(left, grp, {n: held[n] for n in grp})
for n, v in got.items():
level[n] -= v / EFF / by[n].capacity_kwh * 100.0
for b in batts:
if b.name in alive:
out[b.name].append((z, level[b.name]))
i += 1
if stats is not None:
stats["floor_cuts"] = stats.get("floor_cuts", 0) + floor_cuts
stats["capped"] = stats.get("capped", False) or capped
return out
goodwe_controller/test_ux54.py
+157 -21
View File
@@ -43,6 +43,7 @@ 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
@@ -324,6 +325,64 @@ check("a battery already above the ceiling does not charge and one below the "
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)
@@ -903,14 +962,65 @@ check("F2: the same with the batteries part full and nearly full at 22:00",
# 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. 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.
def own_walk(soc, *, guard=None, sun=(0.0, 0.0)):
# 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
@@ -919,7 +1029,7 @@ def own_walk(soc, *, guard=None, sun=(0.0, 0.0)):
c.solar_trust_pct = 100
c.solar_fc = fc = S.forecast(now0, sun[0], sun[1])
c._mode_roles(now0, None)
out = tab(c, now0)
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}
@@ -959,15 +1069,14 @@ def own_walk(soc, *, guard=None, sun=(0.0, 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
for i in givers}
tot = sum(hold.values())
* 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:
part = min(-net * hold[i.name] / tot,
i.max_w / 1000.0 * 0.25) if tot > 0.0 else 0.0
i.soc = max(floor_, i.soc - part / dayplan.EFF
i.soc = max(floor_, i.soc - got[i.name] / dayplan.EFF
/ i.capacity_kwh * 100.0)
t += Q
return out, seen, d0
return out, seen, d0, cost
def own_match(out, seen, d0):
@@ -998,7 +1107,7 @@ def give_span(seen, d0, name):
own = {}
for key, soc, g, sun in (("20/20", (20.0, 20.0), None, (0.0, 0.0)),
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)),
@@ -1010,19 +1119,46 @@ for key, soc, g, sun in (("20/20", (20.0, 20.0), None, (0.0, 0.0)),
(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))):
o_w, s_w, d_w = own_walk(soc, guard=g, sun=sun)
own[key] = (o_w, s_w, d_w, own_match(o_w, s_w, d_w))
("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])
"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 11 cases it charges and gives in exactly the quarters "
"levels): in all 13 cases it charges and gives in exactly the quarters "
"the tab has a block of that kind, on both batteries",
all(not v[3] for v in own.values()))
o_g, s_g, d_g, _b = own["25/45 guard"]
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 "
@@ -1032,7 +1168,7 @@ check("...25/45 % with the peak guard: control gives 17:00-20:15 on both "
"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"]
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)