"""Controller methods for trading: mode runs, peak floors, pack actions. Moved unchanged from app/main.py (MAINT-21). DECISIONS/MAIN-orchestration.md.""" import dataclasses import logging from datetime import datetime, timedelta, timezone from . import arbitrage from .arbiter import ( ACTUATOR_SITE, Claim, P_MAINTENANCE, P_MAINT_SHAPE, P_PEAK, SET, pre_resolve, ) from . import allocator from . import dayplan from . import modetrade from . import packmode from . import priceruns from . import solarfc from . import control as _ctl from . import messages from .maintenance import CHARGE, DRAIN, HOLD, IDLE, MaintConfig, POWER_MOVING from .mqtt import MODE_LABELS, USE_OFF, USE_ON, unit_slugs, unit_state_key from . import peakguard from . import packshare from .price import Absence from .price import required_spread as price_required_spread from .price import uncertainty_of as price_uncertainty_of from .price_chain import gate_for from .static_tariff import static_gated from .ctl_inputs import TARIFF_MAX_AGE_S _LOG = logging.getLogger("goodwe") # OPT-11 P3: how long the last cheap run stays the price stored energy is # valued at: the rest of that day. REF_CHEAP_KEEP = timedelta(hours=12) # OPT-15: the longest a run edge holds the loop at E waiting for the meter to # show the step (control.PLANT_SETTLE_S). EDGE_HOLD_S = _ctl.PLANT_SETTLE_S # OPT-11 P4: store key of the house profile, and how often the forecast sensors # are read (hourly at most). HOUSE_PROFILE_KEY = "house_profile" # OPT-34: the level of a night grid charge that has started, so it does not # drop when the forecast goes (`Controller._latch_charge()`). CHARGE_LATCH_KEY = "night_charge_latch" # OPT-34: a latch outlives its last cheap run by one price slot, since the next # run is planned at the slot edge. CHARGE_LATCH_GRACE_S = 900.0 def _night_limit(born): """OPT-34: a run that starts after this cannot carry the latch on: the first bell start (`solarfc.BELL_FROM_H`, local) after `born`.""" d0, d1 = solarfc._bounds(born) bell = solarfc._bell_start(d0) return bell if bell > born else solarfc._bell_start(d1) def pack_energy_of(inv): """OPT-09. ONE roster entry as the view `packshare` consumes. One builder for two callers (`pack_energy` needs the roster, `cycle()` one entry): written twice they would drift. A module function so inverter.py need not import `packshare`.""" return packshare.PackEnergy(inv.name, inv.soc, inv.bms_soc_floor, inv.capacity_kwh, inv.max_w) # MAINT-13: a battery power reading at or below minus this many W is a pack # that is taking charge. MAINT_CHARGING_W = 50.0 class TradeMixin: def trade_power_w(self, floor_pct, ceiling_pct) -> tuple: """OPT-09. `(charge_w, discharge_w)`: how hard a trade may be SIZED: what the site can push AND STILL HAVE SOMEWHERE TO PUT. The bounds are arbitrage's own (`arbitrage_soc_floor` / `_ceiling`), NOT the BMS cut-out that `packshare.envelope()` uses (SAFETY-25): different questions. It sizes the ask, it does not route it. A pack whose SoC did not read counts on neither side. Neither side may be zero on a trading tick (`arbitrage.decide()` refuses `IDLE_NO_SIZE`: a 0 W claim is a park order). DECISIONS/MAIN-orchestration.md.""" # The reduction lives in `packshare`; `test_packshare.py` part 9 calls # the SAME function (a transcription in a check is a copy that drifts). return packshare.trade_power_w(self.pack_energy, floor_pct, ceiling_pct) def peak_aim_w(self) -> float: """SAFETY-79. `peak_cap_w` less `peak_margin_w`: what the forecast checks aim at. Reports and the bill use the real cap.""" return float(self.o.get("peak_cap_w", 3500)) - self.peak_margin_w @property def pack_energy(self) -> list: """OPT-09. Each pack as the FOUR numbers a split needs; the fourth is `capacity_kwh`, which is why this is not `packs`. EVERY unit, live or not, for `packs`'s reason (the allocator drops dead ones itself).""" return [pack_energy_of(inv) for inv in self.inverters] def pack_weights(self, target_w): """OPT-09. The comfort weights for a site target of `target_w`. A method handed to `resolve()`, not a dict computed before it (energy headroom depends on the target), and read fresh on every call, never cached.""" base = packshare.weights(self.pack_energy, target_w) if not self.modes_on: return base # OPT-11: the solar order. While the site charges, Leftover solar # batteries weigh 0, so Solar first fill first (comfort, like the rest # here; permission is the mode claim). w = packmode.charge_weights(base, self.pack_modes(), target_w) # P3: in a price run, the batteries whose run it is carry it first. tick = self.mode_tick return w if tick is None else modetrade.weights(tick.roles, w, target_w) def pack_mode(self, inv) -> "packmode.PackMode": """OPT-11. This battery's on/off switch and mode, read from the store.""" return packmode.read(getattr(self.store, "data", None) or {}, inv.name) def pack_modes(self) -> dict: return {inv.name: self.pack_mode(inv) for inv in self.inverters} def loop_bound(self, claims): """OPT-11. `(lo, hi)`: what the batteries may place if the loop decides: the sum over live units of each unit's hard limits with its mode applied (`allocator.loop_bound()`), from THIS cycle's unit claims. None when no mode is filed (maintenance) or the range leaves out 0 W (a fault `resolve()` reports).""" if len(self.inverters) <= 1: modes = [c for c in claims if c.source == "mode"] if not modes: return None lo = max(c.lo for c in modes) hi = min(c.hi for c in modes) return (lo, hi) if lo <= 0.0 <= hi else None bounds = {u: pre_resolve([c for c in claims if c.actuator == u]) for u in self.units} return allocator.loop_bound(bounds, live=self.live_units()) # --- OPT-11 P3: price runs, per battery ------------------------------- def _mode_packs(self, maint_results=None) -> list: """Each battery as `modetrade.Pack`, in roster order.""" live = set(self.live_units()) out = [] for i, inv in enumerate(self.inverters): pm = self.pack_mode(inv) idle = (maint_results is None or maint_results[i].phase == IDLE) refused = (inv.refused if len(self.inverters) > 1 else self.batt_refused) or 0 # OPT-15: a charge refusal's probe (SAFETY-73, N>1) as it stood # when last decided; its pin is filed after this, in cycle(). probing = (len(self.inverters) > 1 and refused < 0 and inv.probe_from is not None) out.append(modetrade.Pack( inv.name, pm.use, pm.mode, pm.sell and pm.can_sell, float(inv.max_w), inv.soc, inv.capacity_kwh, live=inv.name in live, idle=idle, refused=int(refused), probing=probing)) return out def _wx_track(self, now) -> None: """TARIFF-24, once per cycle, before the plan: with a forecast read, remember when the run now under way (cheap, give or sell) ends.""" if self.solar_fc is None: return rp = self.mode_runs ends = [] if rp is not None and rp.on: ends = [r.end for r in (rp.cheap, rp.sell) if r is not None and r.holds(now)] g = rp.give_at(now) if g is not None: ends.append(g.end) self._wx_free_to = max(ends) if ends else None def weather_held(self, now) -> bool: """TARIFF-24: no forecast now, but a run planned with one is under way: it keeps its gap to its end (OPT-34: a night charge that keeps its level too); the wider gap applies from the next run. Pure; Home asks the same (`homeview.trade_verdict()`). (DECISIONS/MAIN-orchestration-archive.md)""" return self.solar_fc is None and ( (self._wx_free_to is not None and now < self._wx_free_to) or (self._charge_latch is not None and now < self._charge_latch[0])) def _price_runs(self, now, packs, floor_pct, ceiling_pct): """The kept run plan, rebuilt at most once per price slot (15 min) or when the prices or a battery's choice change; a run under way is kept to its end (`priceruns.keep_started()`). None with no fresh firm prices: then no battery holds or trades. (DECISIONS/MAIN-orchestration-archive.md)""" plan = self.fresh_tariff_plan(now) if plan is None or not plan.has_prices: self.mode_runs, self._runs_key = None, None self.day_plan = None return None slot = int(now.timestamp() // 900) key = (id(plan), slot, tuple((p.name, p.use, p.mode, p.sell) for p in packs)) held = self.weather_held(now) if self.fc_entity is not None: # P4: a forecast arriving or going rebuilds the plan at once (and # so does the end of a run that held the gap, TARIFF-24). key += (self.solar_fc is None, held) if key != self._runs_key: done_c = self._run_done.get("charge", (None, set()))[1] done_s = self._run_done.get("sell", (None, set()))[1] e = modetrade.energy(packs, floor_pct, ceiling_pct, done_c, done_s) required = price_required_spread( arbitrage.threshold_eur_kwh(self.o), price_uncertainty_of(plan.read.product) | arbitrage.weather_uncertainty(True, self.solar_fc, held)) ref = self._ref_cheap if ref is not None and now - ref.end > REF_CHEAP_KEEP: ref = self._ref_cheap = None new = priceruns.plan( plan.read.points, now, required_eur_kwh=required, charge_kwh=e[0], charge_kw=e[1], sell_kwh=e[2], sell_kw=e[3], ref_eur_kwh=None if ref is None else ref.mean_eur_kwh) if self.fc_entity is not None: new = self._day_plan(now, packs, floor_pct, ceiling_pct, new, plan, required, ref, done_c, e) self.mode_runs = priceruns.keep_started(self.mode_runs, new, now) self._runs_key = key runs = self.mode_runs # The cheapest cheap run under way lately is the price the stored # energy was bought at; it stays the reference for REF_CHEAP_KEEP # after it ends, unless a cheaper one comes. ref = self._ref_cheap if runs is not None and runs.cheap is not None \ and runs.cheap.holds(now) and ( ref is None or now - ref.end > REF_CHEAP_KEEP or runs.cheap.mean_eur_kwh <= ref.mean_eur_kwh): self._ref_cheap = runs.cheap return runs def _day_plan(self, now, packs, floor_pct, ceiling_pct, runs, plan, required, ref, done_c, e): """OPT-11 P4: the day plan (`dayplan.plan()`) on the runs just made, its levels held by the hysteresis (`dayplan.hold()`), then the runs made again with each battery's shortfall as the charge. With no plan (no or stale forecast, no house profile yet) the levels are dropped: P3 option A. (DECISIONS/MAIN-orchestration-archive.md)""" dp = dayplan.plan( self.solar_fc, None if self.house_profile is None else self.house_profile.estimator(), priceruns.keep_started(self.mode_runs, runs, now), now, packs, trust_pct=self.solar_trust_pct, floor_pct=floor_pct, ceiling_pct=ceiling_pct, aim_w=self.peak_aim_w(), slots=priceruns.firm_slots(plan.read.points, now), spread=required) self.day_plan = dp if dp is None: self._plan_targets = None return runs self._plan_targets = dayplan.hold( self._plan_targets, dp.targets, {p.name: p.capacity_kwh for p in packs}) e2 = modetrade.energy(packs, floor_pct, ceiling_pct, done_c, (), targets=self._plan_targets) return priceruns.plan( plan.read.points, now, required_eur_kwh=required, charge_kwh=e2[0], charge_kw=e2[1], sell_kwh=e[2], sell_kw=e[3], ref_eur_kwh=None if ref is None else ref.mean_eur_kwh) def plan_targets(self): """P4: each battery's planned level, or None (no day plan). Dropped at once when the forecast goes (stale or missing), except the level of a night charge that has started (OPT-34): it keeps that level.""" if self.solar_fc is None: lt = self._charge_latch if lt is None or datetime.now(timezone.utc) >= lt[0]: return None return dict(lt[1]) return self._plan_targets def _latch_charge(self, now, runs, targets, packs=None) -> None: """OPT-34, once per cycle after `modetrade.decide()`: a started night grid charge keeps its planned level when the forecast goes mid-charge (`plan_targets()` gives the latched levels, `weather_held()` the narrow price gap). The latch lasts through every cheap run plus one price slot (`CHARGE_LATCH_GRACE_S`), is dropped when `now` passes its end or the battery leaves its mode, and a charge that never started has none. DECISIONS/OPT-34-the-night-charge-keeps-its-level.md""" old = lt = self._charge_latch tick = self.mode_tick run = None if runs is None else runs.cheap in_run = run is not None and run.holds(now) grace = timedelta(seconds=CHARGE_LATCH_GRACE_S) if lt is not None and now >= lt[0]: lt = None if lt is not None and packs is not None: ok = {p.name for p in packs if p.use and p.mode in packmode.PLAN_CHARGE_MODES} lv = {n: v for n, v in lt[1].items() if n in ok} lt = (lt[0], lv, lt[2]) if lv else None # A run may carry the latch on only if it starts by the night's bell. carry = in_run and (lt is None or run.start <= _night_limit(lt[2])) if in_run and self.solar_fc is not None: # Rebuilt from the plan: a battery already latched follows its # level up or down whatever its role this tick (a fresh figure may # lower it below the SoC), one that left the plan is dropped, and # a new one comes in only while it charges from the grid. tg = targets or {} lv = {n: float(tg[n]) for n in (lt[1] if lt else {}) if n in tg} if tick is not None and tick.action == modetrade.CHARGE: lv.update({n: float(tg[n]) for n, r in tick.roles.items() if r == modetrade.CHARGE and n in tg}) if not lv: lt = None elif lt is None: lt = (run.end + grace, lv, now) else: lt = (max(run.end + grace, lt[0]) if carry else lt[0], lv, lt[2]) elif lt is not None and carry: lt = (max(lt[0], run.end + grace), lt[1], lt[2]) if lt != old: self._charge_latch = lt self.store.set(CHARGE_LATCH_KEY, None if lt is None else { "until": lt[0].isoformat(), "levels": lt[1], "born": lt[2].isoformat()}) def _mode_roles(self, now, maint_results): """OPT-11 P3: each battery's role and range this tick. Grid charging and selling need a fresh plan that `TariffPlan.arbitrage()` allows with the MEASURED peak guard (None refuses: nothing measures the quarter), exactly the gate `arbitrage.decide()` has. (DECISIONS/MAIN-orchestration-archive.md)""" floor_pct, ceiling_pct = arbitrage.soc_bounds(self.o) packs = self._mode_packs(maint_results) self._wx_track(now) runs = self._price_runs(now, packs, floor_pct, ceiling_pct) guard = arbitrage.guard_state(self.peak_guard) plan = self.fresh_tariff_plan(now) charge_ok = sell_ok = False trade_why = "not_covered" def allowed(active): return plan.arbitrage( peak_guard_active=active, min_spread_eur_kwh=arbitrage.threshold_eur_kwh(self.o), uncertainty=arbitrage.weather_uncertainty( True, self.solar_fc, self.weather_held(now)), now=now).match( allowed=lambda _v: True, refused=lambda _v: False) if guard is not None and plan is not None and runs is not None: # A charge asks the guard (section 9.1 interlock); a sell only # needs it to have measured the quarter, since discharge is never # capped by it (a big discharge alone engages it on export). sell_ok = allowed(False) charge_ok = sell_ok and allowed(guard) trade_why = (None if charge_ok else "peak" if sell_ok else "not_worth") self.mode_gate = None if charge_ok else trade_why price_now = None if plan is not None: try: pt = plan.read.match(series=lambda r: r.at(now), no_data=lambda _r: None) price_now = (None if pt is None or pt.provisional else float(pt.eur_per_kwh)) except Absence: price_now = None house = (None if self.grid is None or self.site_batt_w is None else float(self.grid) + float(self.site_batt_w)) if self.house_profile is not None and \ self.house_profile.add(now, house): self.store.set(HOUSE_PROFILE_KEY, self.house_profile.dump()) targets = self.plan_targets() if targets is None: done_c = self._run_mark("charge", runs and runs.cheap, packs, lambda p: p.soc is not None and p.soc >= ceiling_pct) else: # P4: out of the run at its planned level. A level the # hysteresis moved opens the run to it again (a new mark key). def full(p): to = modetrade.charge_to(p, ceiling_pct, targets) return p.soc is not None and (to is None or p.soc >= to) done_c = self._run_mark("charge", runs and runs.cheap, packs, full, tuple(sorted(targets.items()))) done_s = self._run_mark("sell", runs and runs.sell, packs, lambda p: p.soc is not None and p.soc <= floor_pct) quarter = int(now.timestamp() // 900) was = self.mode_tick is not None and \ self.mode_tick.action == modetrade.CHARGE self.mode_tick = modetrade.decide( runs, now, packs, charge_ok=charge_ok, sell_ok=sell_ok, trade_why=trade_why, price_now=price_now, headroom_w=self.arb_headroom_w, floor_pct=floor_pct, ceiling_pct=ceiling_pct, house_w=house, charging_now=was, blocked=self._charge_blocked_q == quarter, done_charge=done_c, done_sell=done_s, targets=targets, draining=any(r.phase == DRAIN for r in maint_results or ())) self._latch_charge(now, runs, targets, packs) # A charge the guard or the headroom stopped stays off this period. # OPT-15: only a MEASURED stop latches; the first unmeasured headroom # in a quarter stops one tick, a second latches. # (DECISIONS/MAIN-orchestration-archive.md) if was and self.mode_tick.action != modetrade.CHARGE: why = self.mode_tick.why gap = why == "headroom" and self.arb_headroom_w is None if why == "peak" or (why == "headroom" and not gap) or ( gap and self._charge_gap_q == quarter): self._charge_blocked_q = quarter elif gap: self._charge_gap_q = quarter return self.mode_tick def _run_mark(self, kind, run, packs, out, extra=None) -> set: """The batteries that left run `run` (full, or at the floor); they stay out until it ends, so a level edge does not re-arm them. `extra` (P4: the planned levels) is part of the run's key.""" key = None if not run else (run.start, run.end) if key is not None and extra is not None: key += (extra,) was_key, done = self._run_done.get(kind, (None, set())) if key != was_key: done = set() if key is not None: done |= {p.name for p in packs if out(p)} self._run_done[kind] = (key, done) return done def _mode_trade(self, claims, now) -> bool: """OPT-11 P3: file the site trade the roles ask for, sized inside the modes and the hard limits and, for a charge, the billed quarter's headroom (SAFETY-45, aim cap - margin, SAFETY-79), so no battery that its mode or a refusal holds is asked for watts it may not give. (DECISIONS/MAIN-orchestration-archive.md)""" tick = self.mode_tick if tick is None: return False if tick.action == modetrade.CHARGE: # OPT-20: the ask carries PROBE_W for exactly the probes whose pin # is filed this tick (`decide()` ran before the pin, so they can # differ by one cycle; left alone the allocator moves the pin's # watts to another battery). # (DECISIONS/MAIN-orchestration-archive.md) pins = _ctl.PROBE_W * sum( 1 for c in claims if c.source == "refused" and c.kind == "set" and c.value < 0) if pins != tick.probe_w: tick = self.mode_tick = dataclasses.replace( tick, ask_w=tick.ask_w + tick.probe_w - pins, probe_w=pins) size = modetrade.size(tick, self.loop_bound(claims) if tick.action else None, self.arb_headroom_w) runs = self.mode_runs if tick.action and (size == 0.0 or runs is None): # Nothing the batteries may place (a refusal, a hard limit): no # trade, and the roles say so rather than "charging". back = {modetrade.CHARGE: modetrade.HOLD, modetrade.REST: modetrade.HOUSE, modetrade.SELL: modetrade.GIVE} tick = self.mode_tick = modetrade.Tick( tick.ranges, {n: back.get(r, r) for n, r in tick.roles.items()}, None, 0.0, "size") if size == 0.0 or runs is None: reason = ("battery modes: no price run now" if tick.why is None else f"battery modes: no trade ({tick.why})") self.arb = arbitrage.ArbitrageDecision( action=None, power_w=None, reason=reason, spread_eur_kwh=None, required_eur_kwh=None, price_now_eur_kwh=None) else: charge = tick.action == modetrade.CHARGE run = runs.cheap if charge else runs.sell who = ", ".join(n for n, r in sorted(tick.roles.items()) if r == tick.action) act = arbitrage.CHARGE if charge else arbitrage.DISCHARGE self.arb = arbitrage.ArbitrageDecision( action=act, power_w=size, reason=(f"{act}: battery modes, " f"{'cheap' if charge else 'sell'} run " f"{run.start.isoformat()} to {run.end.isoformat()} " f"at {run.mean_eur_kwh:.4f} EUR/kWh; " f"{abs(size):.0f} W on {who}"), spread_eur_kwh=None, required_eur_kwh=runs.level_eur_kwh, price_now_eur_kwh=None) claims.append(arbitrage.claim(self.arb)) if self.arb.action != self._arb_action: self._arb_action = self.arb.action self.log_event(self.arb.explain(), hint=messages.hint( "trade_charge" if self.arb.action == arbitrage.CHARGE else "trade_give" if self.arb.action == arbitrage.DISCHARGE else "trade_off")) return self.arb.action is not None def _reseed_run_edge(self, claims, now) -> None: """OPT-11 P3: a mode trade just ended, so the loop takes over from a command no loop chose: it starts at the house's own point E, not at the trade's watts, and OPT-15 HOLDS it at E (`_edge_grid()`) until the meter shows the step or `EDGE_HOLD_S` passes. DECISIONS/OPT-11-battery-modes.md""" self._edge = None batt = self.site_batt_w if batt is None or self.grid is None: return want = float(batt) + float(self.grid) - self.tuning.target_grid_w bound = self.loop_bound(claims) if bound is not None: want = max(bound[0], min(bound[1], want)) want = float(round(max(-self.site_max_w, min(self.site_max_w, want)))) self.target = self.i_w = want if want != float(batt): self._edge = (want, 1.0 if want > float(batt) else -1.0, now + timedelta(seconds=EDGE_HOLD_S), float(batt)) def _edge_grid(self, end, now): """OPT-15: the grid reading the loop acts on this tick. While a run edge is held, the target itself (no error, the command stays at E); the hold ends when the meter reaches the target, on `end`, or after `EDGE_HOLD_S`. (DECISIONS/MAIN-orchestration-archive.md)""" e = self._edge if e is None: return self.grid held_w, sign, until, last_b = e tg, db = self.tuning.target_grid_w, self.tuning.deadband_w b = self.site_batt_w moving = (b is not None and sign * (float(b) - last_b) >= db and sign * (held_w - float(b)) >= db and now < until + timedelta(seconds=EDGE_HOLD_S)) if (end or (now >= until and not moving) or sign * (float(self.grid) - tg) < db): self._edge = None return self.grid self._edge = (held_w, sign, until, last_b if b is None else float(b)) self.i_w = held_w return tg def _charge_track(self, claims, maint_results) -> None: """SAFETY-93, N>=2: a charge ends when its battery leaves CHARGE or can no longer be placed (stale link, refusal). Kept until the loop runs. SAFETY-95: a drain likewise; its refusal (SAFETY-73) is read from the battery (+1 = will not give). (DECISIONS/MAIN-orchestration-archive.md)""" if len(self.inverters) < 2: return live = set(self.live_units()) now = {CHARGE: set(), DRAIN: set()} for inv, res in zip(self.inverters, maint_results): u = self.actuator_of("maintenance", inv) if res.phase in now and u in live and not pre_resolve( [c for c in claims if c.actuator == u]).contradiction \ and not (res.phase == DRAIN and (inv.refused or 0) > 0): now[res.phase].add(inv.name) self._charge_ended |= self._charging_was - now[CHARGE] self._charging_was = frozenset(now[CHARGE]) self._drain_ended |= self._draining_was - now[DRAIN] self._draining_was = frozenset(now[DRAIN]) def _reseed_charge_end(self, maint_results) -> None: """SAFETY-93, N>=2: when a battery's maintenance charge ends, the loop starts from what the others stood at plus that battery's new command (its HOLD pin, or its last discharge if it is free; a charge stops); else the charge's watts land on the others (B 0 -> -1600 W in one cycle). DECISIONS/SAFETY-93-charge-end.md""" ended, self._charge_ended = self._charge_ended, frozenset() if len(self.inverters) < 2 or not ended: return for inv, res in zip(self.inverters, maint_results): if inv.name not in ended: continue now_w = (float(res.setpoint_w or 0.0) if res.owns_setpoint else max(float(inv.target), 0.0)) if not inv.is_live(datetime.now(timezone.utc))[0]: now_w = 0.0 # SAFETY-96: out of the split, it is sent 0 W self.target += now_w - float(inv.target) # SAFETY-96: it settles from its charge to that command. self._loop_start[inv.name] = now_w self._settle[inv.name] = self._now_mono self.i_w = self.target def _reseed_drain_end(self, maint_results, live) -> None: """SAFETY-95, N>=2: when a battery's maintenance drain ends, its drain leaves the loop's start point, as a charge does (SAFETY-93): the start is what the others stood at plus its new command (its pin, else 0 W), never above last cycle's start; else the drain's floor lands on the others (B 0 -> +1600 W, exporting). DECISIONS/SAFETY-95-drain-end.md""" ended, self._drain_ended = self._drain_ended, frozenset() if len(self.inverters) < 2 or not ended: return hi = self.target for inv, res in zip(self.inverters, maint_results): if inv.name not in ended: continue now_w = (float(res.setpoint_w or 0.0) if res.owns_setpoint and res.phase != DRAIN else min(float(inv.target), 0.0)) if inv.name not in live: now_w = 0.0 # SAFETY-96: out of the split, it is sent 0 W else: self._drain_tail.add(inv.name) self.target += now_w - float(inv.target) self._loop_start[inv.name] = now_w self._settle[inv.name] = self._now_mono lo, hi = sorted((self.target, hi)) # A dropped battery's power is the meter's, like the house's. here = float(self.grid) - self.tuning.target_grid_w + sum( float(inv.batt) for inv in self.inverters if inv.name in live) self.target = self.i_w = min(hi, max(lo, here)) def _drain_grid_w(self, live) -> float: """SAFETY-95: the grid the loop regulates. While a battery settles from its ended drain (`_held_out_w`, at most PLANT_SETTLE_DOWN_S) the batteries count at the site command, so the loop does not chase the drain's tail. SAFETY-103: a reading can be up to a register period (10 s) older than the meter, so the house point is held and moved only as far as the meter proves. DECISIONS/SAFETY-103-old-readings-after-a-drain.md""" grid = float(self.grid) self._drain_tail &= set(self._settle) & set(live) if not self._drain_tail: self._drain_house = None return grid aim = self.tuning.target_grid_w lo = hi = grid - aim for inv in self.inverters: if inv.name not in live: continue r = float(inv.batt) span = self._read_span.get(inv.name) if (span is None or span[2] != inv.batt_fresh.changed_mono or span[2] == self._now_mono): span = (r, r) # A battery that refuses one way (SAFETY-73) is not heading # that way, whatever it was asked: that side is its reading. if (inv.refused or 0) > 0: span = (min(span[0], r), r) elif (inv.refused or 0) < 0: span = (r, max(span[1], r)) lo, hi = lo + span[0], hi + span[1] held = (float(self.target) if self._drain_house is None else self._drain_house) self._drain_house = min(hi, max(lo, held)) return self._drain_house - float(self.target) + aim def _track_reads(self) -> None: """SAFETY-103, N>=2: per battery, the range its power can be in now: its last changed reading and every command on its wire since (a lagging battery moves from one toward the others, never beyond).""" if len(self.inverters) < 2: return for inv in self.inverters: if inv.batt is None: continue cmd, since = inv.standing_w(), inv.batt_fresh.changed_mono lo, hi, was = self._read_span.get(inv.name, (0.0, 0.0, None)) if was is None or since != was: lo = hi = float(inv.batt) self._read_span[inv.name] = (min(lo, cmd), max(hi, cmd), since) def _reseed_drop(self, live) -> None: """SAFETY-96, N>=2: a battery that drops out of the split (stale link) leaves the loop's start point, so the others take over at one slew step a cycle instead of taking its share at once. DECISIONS/SAFETY-96-dropped-battery.md""" was, self._live_was = self._live_was, frozenset(live) if len(self.inverters) < 2 or was is None or was == live: return for inv in self.inverters: if inv.name in self._charge_ended | self._drain_ended: continue # SAFETY-93/95's edge: `_reseed_*_end()` if inv.name in was and inv.name not in live: now_w = 0.0 elif inv.name in live and inv.name not in was: now_w = float(inv.batt) # The split is made again, so every battery's command steps. self._settle.update( {u: self._now_mono for u in live}) else: continue self.target += now_w - float(inv.target) self._loop_start[inv.name] = now_w self.i_w = self.target def _held_out_w(self, live) -> float: """SAFETY-96, N>=2: what to add to the site sum so that a battery the loop cannot move right now counts at its command, not at its reading. Its real power still reaches the loop through the meter: neither lost nor counted twice. DECISIONS/SAFETY-96-dropped-battery.md""" if len(self.inverters) < 2: return 0.0 out = 0.0 for inv in self.inverters: cmd = self._loop_start.get(inv.name, float(inv.target)) since = self._settle.get(inv.name) age = (None if since is None or self._now_mono is None else self._now_mono - since) if since is not None and ( age is None or age >= _ctl.PLANT_SETTLE_DOWN_S or (age > 0 and abs(float(inv.batt) - cmd) < self.tuning.deadband_w)): del self._settle[inv.name] if (inv.name not in live or inv.rejected_since is not None or inv.name in self._settle): out += cmd - float(inv.batt) return out def _mode_fault_log(self, resolution) -> None: """OPT-11: say once, per battery, that its mode was dropped as a fault. A mode range without 0 W would force a direction, which no mode may do, so it is never applied (allocator.unit_interval()); main.py never builds one. (DECISIONS/MAIN-orchestration-archive.md)""" alloc = resolution.allocation now = {u for u, _ in getattr(alloc, "mode_faults", ())} if resolution.mode_fault is not None and self.inverters: now.add(self.inverters[0].name) for inv in self.inverters: if inv.name in now and inv.name not in self._mode_faults: _LOG.error("Battery mode ignored for %s. Software fault " "MODE_RANGE.", inv.name) self.log_event(f"battery mode ignored on {inv.name}, a " f"software fault", hint=messages.hint( "mode_fault", unit=inv.name)) self._mode_faults = now def _pack_action(self, what, unit, value, now) -> dict: """OPT-11: switch one battery on or off, set its mode, or (P3) turn its selling on or off. Store state, not an option: it applies next cycle with no restart. Switching a battery off also stops its maintenance cycle. (DECISIONS/MAIN-orchestration-archive.md)""" if not self.modes_on: return {"error": messages.pack_refused("off")} inv = next((i for i in self.inverters if i.name == unit), None) if inv is None: return {"error": messages.pack_refused("unit")} if what == "pack_use": if not isinstance(value, bool): return {"error": messages.pack_refused("value")} inv.store_set(packmode.USE_KEY, value) self.log_event(f"{inv.name} switched {'on' if value else 'off'} " f"by operator", hint=messages.hint("pack_on" if value else "pack_off", unit=inv.name)) if not value: self._maint_stop_off(self.maints[self.inverters.index(inv)], now) elif what == "pack_sell": if not packmode.SELLING_ENABLED: return {"error": messages.pack_refused("sell_off")} if not isinstance(value, bool): return {"error": messages.pack_refused("value")} inv.store_set(packmode.SELL_KEY, value) self.log_event(f"{inv.name} selling switched " f"{'on' if value else 'off'} by operator", hint=messages.hint("pack_sell", unit=inv.name)) else: if value not in packmode.MODES: return {"error": messages.pack_refused("value")} inv.store_set(packmode.MODE_KEY, value) self.log_event(f"{inv.name} mode set to {value.replace('_', ' ')} " f"by operator", hint=messages.hint("pack_mode", unit=inv.name)) pm = self.pack_mode(inv) out = {"unit": inv.name, "use": pm.use, "mode": pm.mode} if what == "pack_sell": out["sell"] = pm.sell return out async def _maint_unit_action(self, what, unit, now) -> dict: """MAINT-11: start or abort the maintenance cycle of ONE named battery. The site keeps ONE maintenance slot: a start while any battery holds it is refused (no queue). (DECISIONS/MAIN-orchestration-archive.md)""" inv = next((i for i in self.inverters if i.name == unit), None) if inv is None: return self._maint_refusal("unit") machine = self.maints[self.inverters.index(inv)] if what == "maint_start": holder = next((i for i, m in zip(self.inverters, self.maints) if m.phase in POWER_MOVING), None) if holder is not None: return self._maint_refusal("busy", holder.name) if self.modes_on and not self.pack_mode(inv).use: return self._maint_refusal("off", inv.name) for msg in machine.force_start(now): self.log_event(msg, hint=messages.hint( "maintenance", phase=machine.phase)) return {"unit": inv.name, "phase": machine.phase} if machine.phase not in POWER_MOVING: return self._maint_refusal("idle", inv.name) for msg in machine.abort(now, "operator"): self.log_event(msg, hint=messages.hint("maintenance", phase=machine.phase)) await self.command_site_zero() return {"unit": inv.name, "phase": machine.phase} @staticmethod def _maint_refusal(code, unit="") -> dict: return {"error": messages.maint_refused(code, unit), "code": code, "unit": unit} def mode_values(self) -> dict: """OPT-11 P2: each battery's mode and switch, for their HA state topics. Read from the store, so they are right after a restart. Empty while battery modes are off.""" if not self.modes_on: return {} out = {} names = [inv.name for inv in self.inverters] for inv, slug in zip(self.inverters, unit_slugs(names)): pm = self.pack_mode(inv) out[unit_state_key(slug, "mode")] = MODE_LABELS.get(pm.mode, pm.mode) out[unit_state_key(slug, "use")] = USE_ON if pm.use else USE_OFF return out def _drain_floor(self, claims, maint_results): """SAFETY-78: the site `lo` for a pinned maintenance drain, or None. At N>=2 the drain pins its pack and the site target is split around the pin; below the pin a sibling would have to charge (opposite directions, refused, 0 W, the drain never moves), so the site target may not go below what the pin places. At N=1 the drain is the site SET itself: nothing filed. A draining pack with nothing to give adds no floor (logged once).""" if len(self.inverters) < 2: return None live = set(self.live_units()) floor, waits = 0.0, set() for inv, res in zip(self.inverters, maint_results): if res.phase != DRAIN or not res.owns_setpoint \ or not (res.setpoint_w or 0.0) > 0.0: continue u = self.actuator_of("maintenance", inv) b = pre_resolve([c for c in claims if c.actuator == u]) p = (None if u not in live or b.contradiction or b.pinned_by != "maintenance" else allocator.pin_w(b)) if p is not None and p > 0.0: floor += p else: waits.add(inv.name) for name in sorted(waits - self._drain_waits): self.log_event(f"maintenance: {name} cannot give power now, so " f"its monthly cycle waits", hint=messages.hint("maint_waits", unit=name)) self._drain_waits = waits return floor if floor > 0.0 else None def _peak_on_able(self, decision, maint_results): """SAFETY-87: at N>=2, `decision` placed on the batteries that can still give: live, not refusing, above their own reserve. `able_w` is their rating plus the drain a held battery gives anyway. DECISIONS/SAFETY-87-guard-uses-able-batteries.md""" if len(self.inverters) < 2: return decision floor = float(self.o.get( "peak_soc_floor", peakguard.PEAK_OPTION_DEFAULTS["peak_soc_floor"])) ok, held = peakguard.reserve_split(self.packs, floor) if not held: return decision live = set(self.live_units()) able = [p for p, inv in zip(self.packs, self.inverters) if p in ok and inv.name in live and not (inv.refused or 0) > 0] if not able: return dataclasses.replace(decision, held=held) drain = sum(min(float(r.setpoint_w), i.rating_w()) for i, r in zip(self.inverters, maint_results) if i.name in held and r.phase == DRAIN and r.owns_setpoint and (r.setpoint_w or 0.0) > 0.0) names = {p.name for p in able} return peakguard.place_on_able( decision, held, able_w=drain + sum(i.rating_w() for i in self.inverters if i.name in names), soc=min(float(p.soc) for p in able), soc_floor=peakguard.binding_soc_floor(able, floor)) def _peak_landing(self, decision): """SAFETY-126: `decision` marked `within_aim` while the guard wants power but the batteries at 0 W would end the quarter within `ALARM_MISS_W` of the aim (`peakguard.misses_aim()`). The claim and the commands are unchanged; only the alarms and the "Why" read it. DECISIONS/SAFETY-126-landing-error-is-not-an-overrun.md""" batt, prev = self.site_batt_w, self._land_batt_prev self._land_batt_prev = batt if not (decision.engaged and decision.blind is None and (decision.wanted_w or 0.0) > 0.0): self._peak_needed = False return decision if batt is not None and prev is not None: batt = min(float(batt), float(prev)) self._peak_needed = self._peak_needed or ( self._peak_proj is not None and peakguard.misses_aim( self._peak_proj, batt_w=batt, target_w=decision.target_w)) return (decision if self._peak_needed else dataclasses.replace(decision, within_aim=True)) def _peak_reserve(self) -> frozenset: """SAFETY-87: the batteries that give nothing for the peak this cycle: under their reserve while the others carry a peak floor.""" g = self.peak_guard if g is None or g.able_w is None or not (g.floor_w or 0.0) > 0.0: return frozenset() return frozenset(g.held) def _held_back(self) -> frozenset: """SAFETY-91, N>=2: the batteries at or under their own peak reserve while another can still give (live, not refusing). One held back stays so until its SoC reads more than 1 point above its reserve. DECISIONS/SAFETY-91-held-battery-last.md""" if len(self.inverters) < 2: return frozenset() floor = float(self.o.get( "peak_soc_floor", peakguard.PEAK_OPTION_DEFAULTS["peak_soc_floor"])) ok, held = peakguard.reserve_split(self.packs, floor) held = set(held) | { p.name for p in ok if p.name in self._held_was and float(p.soc) <= peakguard.pack_floor_pct(p, floor) + 1.0} live = set(self.live_units()) if not held or not any( p.name not in held and inv.name in live and not (inv.refused or 0) > 0 for p, inv in zip(self.packs, self.inverters)): self._held_was = frozenset() else: self._held_was = frozenset(held) return self._held_was def _drain_rise(self, claims, maint_results) -> None: """SAFETY-91, N>=2: while a battery is held back (`_held_back()`), a draining battery's pin becomes a floor at its drain rate, so it rises towards its rating before the held one gives. A site stop still decides alone. DECISIONS/SAFETY-91-held-battery-last.md""" held = self._held_back() if not held or any(c.kind == SET and c.actuator == ACTUATOR_SITE and c.priority > P_PEAK for c in claims): return live = set(self.live_units()) for inv, res in zip(self.inverters, maint_results): u = self.actuator_of("maintenance", inv) if res.phase != DRAIN or not res.owns_setpoint \ or inv.name in held or u not in live: continue b = pre_resolve([c for c in claims if c.actuator == u]) p = (None if b.contradiction or b.pinned_by != "maintenance" else allocator.pin_w(b)) if p is None or not p > 0.0: continue for i, c in enumerate(claims): if c.source == "maintenance" and c.kind == SET \ and c.actuator == u: claims[i] = Claim.limit("maintenance", P_MAINTENANCE, lo=p, reason=c.reason, actuator=u) def _fill_peak_gaps(self, proj, now, reading): """SAFETY-90: `proj` with a meter gap's uncounted seconds counted at `peakguard.gap_w()` of the imports seen just before and just after it, and those before this run's first reading likewise at the import seen first (shown as not measured). DECISIONS/SAFETY-90-peak-guard-after-a-gap.md.""" self.peak_gap_s = self.peak_unmeasured_s = 0.0 avg = self.p1.averager since = avg.held_since if proj is None or proj.block_start is None or since is None: return proj if self._unseen is None and self.p1.last is not None: self._unseen = [proj.block_start, max(0.0, ( since - proj.block_start).total_seconds() - avg.elapsed_s), since, None] seen = (None if self.grid is None or self.site_batt_w is None else float(self.grid) + max(0.0, float(self.site_batt_w))) if proj.block_start != self._gap_block: self._gap_block, self._gap_w = proj.block_start, None self._gap_filled_s = self._gap_held_ws = 0.0 cap = float(self.o.get("peak_cap_w", 3500)) ceiling = peakguard.quarter_ceiling_w(cap, reading)[0] unseen_on = (self._unseen is not None and self._unseen[0] == proj.block_start) if unseen_on and self._unseen[3] is None and seen is not None: # While the add-on was down the batteries were at 0 W, so the # seconds before its first reading count like a gap, at the # import seen first; a restored reading's held part is topped up. self._unseen[3] = peakguard.gap_w(ceiling, seen) if self._run_pre is not None: s0, g0 = self._run_pre h = (min(s0 + timedelta(seconds=avg.max_hold_s), self._unseen[2]) - max(s0, proj.block_start)).total_seconds() self._unseen_held_ws = max(0.0, h) * max( 0.0, self._unseen[3] - max(0.0, g0)) kw = dict(tail_s=(now - since).total_seconds(), max_hold_s=avg.max_hold_s, unseen_s=self._unseen[1] if unseen_on else 0.0, unseen_w=(self._unseen[3] if unseen_on and self._unseen[3] is not None else ceiling)) _p, gap_s, _u = peakguard.fill_gaps(proj, gap_w=0.0, **kw) if gap_s > self._gap_filled_s + peakguard.GAP_TOL_S: self._gap_w = peakguard.gap_w(ceiling, self._gap_w, self._gap_seen_w, seen) if self._gap_pre is not None: # The averager credited the gap's first max_hold_s at the # reading before it; top that part up to gap_w as well. s0, g0 = self._gap_pre h = (min(s0 + timedelta(seconds=avg.max_hold_s), since) - max(s0, proj.block_start)).total_seconds() self._gap_held_ws += max(0.0, h) * max( 0.0, self._gap_w - max(0.0, g0)) self.log_event(f"PEAK_GUARD_GAP: the meter missed {gap_s:.0f} s " f"of this 15-minute period, counted at " f"{self._gap_w:.0f} W", hint=messages.hint("peak_gap")) self._gap_filled_s = max(self._gap_filled_s, gap_s) if seen is not None: self._gap_seen_w = seen if self.grid is not None: self._gap_pre = (since, float(self.grid)) out, self.peak_gap_s, self.peak_unmeasured_s = peakguard.fill_gaps( proj, gap_w=self._gap_w if self._gap_w is not None else ceiling, held_ws=self._gap_held_ws + (self._unseen_held_ws if unseen_on else 0.0), **kw) return out def _peak_over_drain(self, claims, maint_results) -> None: """SAFETY-83: at N=1 a peak floor above the drain rate outranks it: the drain is the site SET at P_MAINTENANCE, above P_PEAK, so the floor did not bind. Refiled at P_MAINTENANCE where it asks for more; P_SAFETY still outranks. N>=2: the `drain` floor and `_drain_over_reserve()`.""" if len(self.inverters) != 1: self._drain_over_reserve(claims, maint_results) return res = maint_results[0] if res.phase != DRAIN or not res.owns_setpoint: return rate = float(res.setpoint_w or 0.0) for i, c in enumerate(claims): if c.source == peakguard.SOURCE and c.lo > rate: claims[i] = dataclasses.replace(c, priority=P_MAINTENANCE) def _held_pin(self, claims, u): """SAFETY-87: what held unit `u` places anyway (its own pin), or None when that pin charges, so the site is never split both ways.""" b = pre_resolve([c for c in claims if c.actuator == u]) p = None if b.contradiction else allocator.pin_w(b) return None if p is not None and p < 0.0 else max(p or 0.0, 0.0) def _drain_over_reserve(self, claims, maint_results) -> None: """SAFETY-87, N>=2: while a battery is held at its peak reserve, a draining battery that may still give is raised from its drain rate towards its rating for the peak floor, after the others. A site stop still decides alone. DECISIONS/SAFETY-87-guard-uses-able-batteries.md""" reserve = self._peak_reserve() floor = max((c.lo for c in claims if c.source == peakguard.SOURCE), default=0.0) if not reserve or not floor > 0.0 or any( c.kind == SET and c.actuator == ACTUATOR_SITE and c.priority > P_PEAK for c in claims): return live = set(self.live_units()) drains, others = [], 0.0 for inv, res in zip(self.inverters, maint_results): u = self.actuator_of("maintenance", inv) if u not in live: continue if inv.name in reserve: p = self._held_pin(claims, u) if p is None: return others += p continue b = pre_resolve([c for c in claims if c.actuator == u]) if b.contradiction: continue p = allocator.pin_w(b) if p is not None and p < 0.0: return # a pack pinned to charge: never split both ways if res.phase == DRAIN and b.pinned_by == "maintenance" \ and p is not None and p > 0.0: drains.append((u, p, allocator.hi_w(b))) others += allocator.hi_w(b) if p is None else p give = floor - others for u, p, hi in drains: if give <= 0.0: break up = max(p, min(hi, p + give)) give -= up - p for i, c in enumerate(claims): if c.source == "maintenance" and c.kind == SET \ and c.actuator == u: claims[i] = dataclasses.replace(c, value=up) def _peak_over_hold(self, claims, maint_results): """SAFETY-86: a peak floor outranks a maintenance HOLD (0 W). N=1: refiled at P_MAINTENANCE as for a drain. N>=2: the holding pack's pin is raised to the floor minus what the others may place, and the floor is returned as the site `hold` ceiling. A site stop still decides alone. DECISIONS/SAFETY-86-peak-over-hold.md.""" hold = {self.actuator_of("maintenance", inv) for inv, res in zip(self.inverters, maint_results) if res.phase == HOLD and res.owns_setpoint} peak = [i for i, c in enumerate(claims) if c.source == peakguard.SOURCE and c.lo > 0.0] if not hold or not peak: return if len(self.inverters) == 1: for i in peak: claims[i] = dataclasses.replace(claims[i], priority=P_MAINTENANCE) return if any(c.kind == SET and c.actuator == ACTUATOR_SITE and c.priority > P_PEAK for c in claims): return live = set(self.live_units()) reserve = self._peak_reserve() # SAFETY-87: they give nothing others = 0.0 for inv in self.inverters: u = self.actuator_of("maintenance", inv) if u in hold or u not in live: continue if inv.name in reserve: # only its own pin, never both ways p = self._held_pin(claims, u) if p is None: return others += p continue b = pre_resolve([c for c in claims if c.actuator == u]) if b.contradiction: continue p = allocator.pin_w(b) if p is not None and p < 0.0: return # a pack pinned to charge: never split both ways others += allocator.hi_w(b) if p is None else p floor = max(claims[i].lo for i in peak) give = floor - others if give <= 0.0: return for i, c in enumerate(claims): if c.source == "maintenance" and c.kind == SET \ and c.actuator in hold: claims[i] = dataclasses.replace(c, value=give) return floor def _charge_headroom(self, floor: float, claims) -> float: """SAFETY-84: a cheap-window charge pulls no more than the quarter's room under the peak aim (`charge_room_w()`), so it yields just enough to the peak guard; at N>=2 nothing under a site stop (its 0 W would need the other pack to give). SAFETY-125: at N=1 too; the stop rule stays N>=2 only. DECISIONS/SAFETY-125-grid-charge-under-the-month-peak.md""" if len(self.inverters) > 1 and any( c.kind == SET and c.actuator == ACTUATOR_SITE and c.priority > P_MAINT_SHAPE for c in claims): return 0.0 room = peakguard.charge_room_w(self.peak_guard, self._peak_proj) floor = floor if room is None else min(floor, room) self.room_charge = self.room_charge or floor > 0.0 return floor def _charge_ceiling(self, claims): """SAFETY-84: the site `hi` for a cheap-window charge at N>=2, or None: the charging pack's own `hi` (P_MAINT_SHAPE) makes it charge; above that the other pack would have to give (opposite directions, refused, 0 W). Not filed at N=1.""" if len(self.inverters) < 2: return None live = set(self.live_units()) ceiling = 0.0 for inv in self.inverters: u = self.actuator_of("maintenance", inv) b = pre_resolve([c for c in claims if c.actuator == u]) if u in live and not b.contradiction and b.hi < 0.0: ceiling += allocator.hi_w(b) return ceiling if ceiling < 0.0 else None def _maint_stop_off(self, machine, now) -> None: """OPT-11: stop this battery's maintenance cycle because it is off.""" if machine.phase == IDLE: return for msg in machine.abort(now, "battery switched off"): self.log_event(msg, hint=messages.hint("maint_off", unit=machine.name)) def _provider_gate(self, now: datetime) -> None: """PRICE-01: `cheap`, the gate and the gap from the standing plan. No price for now is a closed gate, so the charge waits rather than run unpriced; the gap flag stays False until the first refresh has finished. (DECISIONS/MAIN-orchestration-archive.md)""" plan = self.fresh_tariff_plan(now) published = None if plan is None else plan.published if published is not None: self._last_published = published gate = gate_for(published, now) self.cheap, self._price_gated = gate.cheap, True self.tariff_words = gate.words # Owner, 2026-10-14: the words say 'maintenance waits', so they show # only while the maintenance charge is on. self.price_gap = ((not gate.covered) and self.tariff_at is not None and MaintConfig.from_options(self.o).enabled) last = self._last_published if gate.covered or last is None: self.price_gap_since = None elif last.covers(now): # The last series still covers now, but no refresh has made it # current for TARIFF_MAX_AGE_S: the refresh task has stopped. The # gap began when the plan was retired. self.price_gap_since = ( None if self.tariff_at is None else self.tariff_at + timedelta(seconds=TARIFF_MAX_AGE_S)) else: self.price_gap_since = last.last_end_before(now) def _static_gated(self, o: dict, now: datetime) -> bool: """`static_gated()`, worked out once an hour: it builds the same month-long plan as `static_status()` (about 20 ms with a profile), and whether a fixed tariff is set up does not change from tick to tick (options change only with a restart).""" key = (id(o), now.strftime("%Y-%m-%d %H")) if key != self._gated_key: self._gated_key, self._gated = key, static_gated(o, now) return self._gated def _charge_gate_open(self, inv) -> bool: """MAINT-13: may this pack charge now, for the maintenance timeout? Closed only when a price gate exists, it is shut, and the pack is not really taking charge. (DECISIONS/MAIN-orchestration-archive.md)""" if not self._price_gated or self.cheap: return True batt = inv.batt return batt is not None and float(batt) <= -MAINT_CHARGING_W