Conflicts: modetrade.decide() takes both the day plan (targets) and SAFETY-78's draining flag; the Why line keeps SAFETY-78's other_cycle ahead of the plan's word (test added). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
270 lines
12 KiB
Python
270 lines
12 KiB
Python
"""OPT-11 P3: each battery's role this tick, from its mode and the price runs.
|
|
|
|
Pure: no I/O, no clock. main.py hands in the run plan (`priceruns`), each
|
|
battery's state and the peak guard's answers, files each battery's range as
|
|
its `mode` claim (P_MODE), and files the site trade as the `arbitrage` claim
|
|
(P_ARBITRAGE), sized inside the modes (`size()`), only with battery modes on.
|
|
With them off, `arbitrage.decide()` runs exactly as before.
|
|
|
|
Roles (the page's Why line names them):
|
|
off Off: [0, 0].
|
|
house follows the house: charges from spare solar, gives power at any
|
|
hour (Solar first always; the other modes when there is no plan).
|
|
hold Leftover solar / Cheap hours outside a give run: may charge from
|
|
spare solar, gives nothing, so its energy is kept for the run.
|
|
give the same inside a give run: covers the house.
|
|
sell inside the sell run, with selling on: gives its rating, to the
|
|
house first and the rest to the grid.
|
|
charge inside the cheap run: charges from the grid.
|
|
rest another battery charges from the grid, so this one may not charge
|
|
(the owner's 2026-09-27 answer 3: while one battery charges from the
|
|
grid the others cannot give power either; the allocator never
|
|
splits a target both ways).
|
|
|
|
Rules that keep this safe (tests: test_modes_p3.py):
|
|
* every range holds 0 W (a mode may forbid a direction, never force one);
|
|
* a Solar first battery never charges from the grid;
|
|
* P4: with a day plan (`targets`), a Leftover solar battery charges from
|
|
the grid only up to its planned level, never above the ceiling; with
|
|
none it takes solar only (P3 option A). Cheap hours only fills to the
|
|
ceiling either way (`charge_to()`);
|
|
* grid charging needs a measured peak guard that is clear and a measured
|
|
headroom (SAFETY-45); the trade is sized to that headroom, whose aim is
|
|
`peak_cap_w - peak_margin_w` (SAFETY-79). It starts only at START_W or
|
|
more, and once the guard or the headroom stops it, it stays stopped for
|
|
the rest of that 15-minute period (`blocked`): a charge makes the other
|
|
batteries stop giving power, so a few watts of charge flipping on and
|
|
off would swing the house between them and the grid (measured on the
|
|
2026-09-09 replay: 13 flips in 4 minutes at a 3.5 kW house);
|
|
* selling needs selling on for that battery, the sell run, a price >= 0
|
|
now, and the battery above `arbitrage_soc_floor`; it stops there. The
|
|
peak guard never stops a sell (section 9.1: discharge is never capped);
|
|
it must only have measured the quarter;
|
|
* a battery out of a run (full, or at the floor) stays out until the run
|
|
ends, so the ceiling and the floor do not re-arm within one run;
|
|
* a battery that refuses a direction (SAFETY-52/73) takes no part in a
|
|
run in that direction: the refusal wins over the run;
|
|
* no grid charge while another battery drains for its monthly cycle
|
|
(`draining`, SAFETY-78): it would charge against the drain.
|
|
"""
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from .packmode import (CHEAP_HOURS, GRID_CHARGE_MODES, LEFTOVER_SOLAR,
|
|
PLAN_CHARGE_MODES, SOLAR_FIRST)
|
|
|
|
OFF, HOUSE, HOLD, GIVE, SELL, CHARGE, REST = (
|
|
"off", "house", "hold", "give", "sell", "charge", "rest")
|
|
PRICE_MODES = (LEFTOVER_SOLAR, CHEAP_HOURS)
|
|
|
|
# The smallest grid charge worth starting, in W. Under it the other batteries
|
|
# would stop covering the house for less than they give.
|
|
START_W = 500.0
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Pack:
|
|
"""One battery as this module needs it."""
|
|
name: str
|
|
use: bool
|
|
mode: str
|
|
sell: bool
|
|
rating_w: float
|
|
soc: float | None
|
|
capacity_kwh: float | None
|
|
live: bool = True
|
|
idle: bool = True # not in a maintenance phase
|
|
refused: int = 0 # SAFETY-52/73: +1 refuses to give, -1 to charge
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Tick:
|
|
ranges: dict # name -> (lo, hi), only for idle batteries
|
|
roles: dict # name -> role
|
|
action: str | None # "charge" | "sell" | None
|
|
ask_w: float # signed site ask before sizing (0 if none)
|
|
why: str | None # why no trade when a run is on, else None
|
|
|
|
|
|
def _room_kwh(p: Pack, pct: float) -> float:
|
|
"""kWh to `pct` (charging) or above it (selling); a battery with no
|
|
capacity set counts one hour at its rating, so a run is still sized."""
|
|
if p.capacity_kwh is None or p.soc is None:
|
|
return abs(p.rating_w) / 1000.0
|
|
return max(0.0, abs(pct - p.soc)) / 100.0 * float(p.capacity_kwh)
|
|
|
|
|
|
def charge_to(p: Pack, ceiling_pct, targets=None):
|
|
"""The level this battery may charge to from the grid, or None.
|
|
|
|
`targets` is the day plan (`dayplan.plan()`: name -> level), or None
|
|
when there is none (no forecast, stale, or not configured). With no plan
|
|
this is P3 option A exactly: Cheap hours to the ceiling, the others never.
|
|
With a plan, a Leftover solar battery charges only to its planned level
|
|
(the sun fills the rest); a battery the plan leaves out, Cheap hours only
|
|
included, gets the no-plan answer. Solar first never charges."""
|
|
if p.mode in GRID_CHARGE_MODES:
|
|
return float(ceiling_pct)
|
|
if targets is None or p.name not in targets \
|
|
or p.mode not in PLAN_CHARGE_MODES:
|
|
return None
|
|
t = targets[p.name]
|
|
return None if t is None else min(float(ceiling_pct), float(t))
|
|
|
|
|
|
def chargers(packs, ceiling_pct, done=(), targets=None) -> list:
|
|
"""Batteries that may charge from the grid in a cheap run: below the
|
|
level `charge_to()` allows them (P4: the day plan's)."""
|
|
out = []
|
|
for p in packs:
|
|
to = charge_to(p, ceiling_pct, targets)
|
|
if (to is not None and p.use and p.idle and p.live
|
|
and p.soc is not None and p.soc < to
|
|
and p.refused >= 0 and p.name not in done):
|
|
out.append(p)
|
|
return out
|
|
|
|
|
|
def sellers(packs, floor_pct, done=()) -> list:
|
|
"""Batteries that may sell in the sell run. Solar first never sells."""
|
|
return [p for p in packs
|
|
if p.use and p.idle and p.live and p.sell
|
|
and p.mode in PRICE_MODES
|
|
and p.soc is not None and p.soc > floor_pct
|
|
and p.refused <= 0 and p.name not in done]
|
|
|
|
|
|
def energy(packs, floor_pct, ceiling_pct, done_charge=(), done_sell=(),
|
|
targets=None):
|
|
"""`(charge_kwh, charge_kw, sell_kwh, sell_kw)` for `priceruns.plan()`.
|
|
With a day plan the charge is each battery's shortfall, so the cheap run
|
|
is only as long as the shortfall needs."""
|
|
ch = chargers(packs, ceiling_pct, done_charge, targets)
|
|
se = sellers(packs, floor_pct, done_sell)
|
|
return (sum(_room_kwh(p, charge_to(p, ceiling_pct, targets))
|
|
for p in ch),
|
|
sum(abs(p.rating_w) for p in ch) / 1000.0,
|
|
sum(_room_kwh(p, floor_pct) for p in se),
|
|
sum(abs(p.rating_w) for p in se) / 1000.0)
|
|
|
|
|
|
def decide(runs, now, packs, *, charge_ok, sell_ok, trade_why, price_now,
|
|
headroom_w, floor_pct, ceiling_pct, house_w, charging_now=False,
|
|
blocked=False, done_charge=(), done_sell=(),
|
|
targets=None, draining=False) -> Tick:
|
|
"""This tick's ranges, roles and site ask.
|
|
|
|
`runs` is the kept `priceruns.RunPlan` or None (no fresh prices).
|
|
`charge_ok` is True only when `TariffPlan.arbitrage()` allowed a trade
|
|
this tick with the MEASURED peak guard clear; `sell_ok` when it allowed
|
|
one with the guard measured (either state); else `trade_why` says why.
|
|
`headroom_w` is `peakguard.charge_headroom_w()` (None: unmeasured).
|
|
`house_w` is the measured house (grid + batteries), or None.
|
|
`charging_now` says last tick charged (then any headroom above 0 keeps
|
|
it going); `blocked` that the guard or the headroom stopped a charge in
|
|
this 15-minute period. `targets` is the day plan (`charge_to()`).
|
|
`draining` says a battery drains for its monthly cycle, so none may
|
|
charge from the grid (why "cycle").
|
|
"""
|
|
plan_on = runs is not None and runs.on
|
|
in_cheap = plan_on and runs.cheap.holds(now) and runs.charge_useful()
|
|
in_give = plan_on and runs.give_at(now) is not None
|
|
in_sell = plan_on and runs.sell is not None and runs.sell.holds(now)
|
|
|
|
why = None
|
|
ch = chargers(packs, ceiling_pct, done_charge, targets) if in_cheap \
|
|
else []
|
|
if ch and draining:
|
|
ch, why = [], "cycle"
|
|
charging = False
|
|
if ch:
|
|
need = 0.0 if charging_now else START_W
|
|
if blocked:
|
|
why = "headroom"
|
|
elif not charge_ok:
|
|
why = trade_why
|
|
elif headroom_w is None or not headroom_w > need:
|
|
why = "headroom"
|
|
else:
|
|
charging = True
|
|
se = sellers(packs, floor_pct, done_sell) if in_sell and not charging \
|
|
else []
|
|
selling = False
|
|
if se:
|
|
if not sell_ok:
|
|
why = why or trade_why
|
|
elif price_now is None or price_now < 0.0 or house_w is None:
|
|
why = why or "negative"
|
|
else:
|
|
selling = True
|
|
|
|
ranges, roles = {}, {}
|
|
ch_names = {p.name for p in ch} if charging else set()
|
|
se_names = {p.name for p in se} if selling else set()
|
|
for p in packs:
|
|
r = abs(float(p.rating_w))
|
|
if not p.use:
|
|
role, lo, hi = OFF, 0.0, 0.0
|
|
elif p.name in ch_names:
|
|
role, lo, hi = CHARGE, -r, 0.0
|
|
elif charging:
|
|
# Another battery charges from the grid: this one may not charge
|
|
# (it would be grid charging too), and cannot give (one sign).
|
|
role, lo, hi = REST, 0.0, (r if p.mode == SOLAR_FIRST
|
|
or not plan_on or in_give else 0.0)
|
|
elif p.name in se_names:
|
|
role, lo, hi = SELL, -r, r
|
|
elif p.mode == SOLAR_FIRST or not plan_on:
|
|
role, lo, hi = HOUSE, -r, r
|
|
elif in_give:
|
|
role, lo, hi = GIVE, -r, r
|
|
else:
|
|
role, lo, hi = HOLD, -r, 0.0
|
|
roles[p.name] = role
|
|
if p.idle:
|
|
ranges[p.name] = (lo, hi)
|
|
|
|
if charging:
|
|
ask = -sum(abs(p.rating_w) for p in ch)
|
|
return Tick(ranges, roles, CHARGE, ask, None)
|
|
if selling:
|
|
ask = sum(abs(p.rating_w) for p in se) + max(0.0, float(house_w))
|
|
return Tick(ranges, roles, SELL, ask, None)
|
|
return Tick(ranges, roles, None, 0.0,
|
|
why if (in_cheap or in_sell) else None)
|
|
|
|
|
|
def size(tick: Tick, bound, headroom_w) -> float:
|
|
"""The site trade in W, inside what the batteries may place with their
|
|
modes (`bound` = main's loop bound) and, for a charge, the billed
|
|
quarter's headroom. 0.0 means no trade (a 0 W trade is a park order and
|
|
is never filed)."""
|
|
if tick.action is None or bound is None:
|
|
return 0.0
|
|
lo, hi = bound
|
|
if tick.action == CHARGE:
|
|
cap = min(-float(tick.ask_w), -float(lo),
|
|
float(headroom_w) if headroom_w is not None else 0.0)
|
|
return -max(0.0, cap)
|
|
return max(0.0, min(float(tick.ask_w), float(hi)))
|
|
|
|
|
|
def weights(roles: dict, base, target_w: float):
|
|
"""Who carries the site target first (design 3.5). Selling: the sellers.
|
|
Giving power: batteries in a give run before Solar first. Charging from
|
|
the grid: only the chargers. Otherwise `base` unchanged. Comfort only;
|
|
the ranges above are the permission."""
|
|
names = sorted(roles)
|
|
first = None
|
|
if target_w > 0 and any(r == SELL for r in roles.values()):
|
|
first = {n for n in names if roles[n] == SELL}
|
|
elif target_w > 0 and any(r == GIVE for r in roles.values()):
|
|
first = {n for n in names if roles[n] in (GIVE, SELL)}
|
|
elif target_w < 0 and any(r == CHARGE for r in roles.values()):
|
|
first = {n for n in names if roles[n] == CHARGE}
|
|
if not first:
|
|
return base
|
|
w = {n: (1.0 if base is None else float(base.get(n, 1.0))) for n in names}
|
|
out = {n: (w[n] if n in first else 0.0) for n in names}
|
|
return out if any(v > 0.0 for v in out.values()) else base
|