"""UX-54: the expected battery level through the day, for the Forecast tab. Pure: batteries, the sun and the house's use in; a level per battery over time out. Display only: nothing here is read by control, and it never decides a block. The blocks are the plan's (`homeview.mode_runs()`); this only says where the level goes, hour by hour, if the add-on does what the plan says. Each step gives every battery a role, as `modetrade.decide()` does (the names are modetrade's): in a grid block it charges (CHARGE) and the others rest (REST, no charge); else it follows the house (HOUSE) if it is Solar 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. `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 (TEL-11 keeps site solar out of control). """ import math from datetime import datetime, timedelta, timezone from dataclasses import dataclass from .dayplan import EFF, _hour_slices, _local from .modetrade import CHARGE, GIVE, HOLD, HOUSE, REST EPS = 1e-9 GAP_S = 60.0 # a longer gap between readings is not counted MIN_COVER = 0.5 # share of the elapsed hour a reading must cover @dataclass(frozen=True) class Batt: """One battery that is on, with a level and a capacity.""" name: str capacity_kwh: float rating_kw: float soc: float # % now solar_first: bool # Solar first: follows the house all day priced: bool # a price plan is on and its mode is a price mode grid: tuple = () # ((start, end, to_pct), ...) grid blocks give: tuple = () # ((start, end), ...) give runs stop_at: datetime | None = None # a monthly cycle starts: no level after def _role(b, t, charging, own): """The battery's role at instant `t`; `own`: it charges from the grid; `charging`: some battery does. Returns `(role, may_charge, may_give)`.""" in_give = any(s <= t < e for s, e in b.give) if own: return CHARGE, False, False if charging: return REST, False, b.solar_first or not b.priced or in_give if b.solar_first or not b.priced: return HOUSE, True, True if in_give: return GIVE, True, True return HOLD, True, False def _spread(amount, caps, weights): """Split `amount` over the keys of `caps` by `weights`, none over its cap. Returns `(taken, left)`: what each took and what no one could.""" got = {k: 0.0 for k in caps} left = float(amount) live = {k for k, c in caps.items() if c > EPS and weights.get(k, 0.0) > 0} while left > EPS and live: total = sum(weights[k] for k in live) used, full = 0.0, set() for k in live: part = left * weights[k] / total room = caps[k] - got[k] if part >= room - EPS: part = room full.add(k) got[k] += part used += part left -= used if not full: break live -= full return got, max(0.0, left) def curve(batts, start, end, sun_kwh, use_kwh, *, floor_pct, ceiling_pct): """`{name: [(t, pct), ...]}` from `start` (each battery's level now) to `end`, a point at every hour edge and every block edge. A battery's list stops where `sun_kwh(a, b)` is None (the sun is not known) or its `stop_at` comes. `sun_kwh` and `use_kwh` give kWh for the span `[a, b)`.""" floor, ceil = float(floor_pct), float(ceiling_pct) cuts = {end} for a, _b, _h in _hour_slices(start, end): cuts.add(a) for b in batts: for s, e, *_ in b.grid + b.give: cuts.update(x for x in (s, e) if start < x < end) cuts.add(start) ts = sorted(cuts) 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} for a, z in zip(ts, ts[1:]): if not alive: break sun = sun_kwh(a, z) if sun is None: break for b in batts: if b.name in alive and b.stop_at is not None and a >= b.stop_at: alive.discard(b.name) net = sun - use_kwh(a, z) hrs = (z - a).total_seconds() / 3600.0 want = {} # name -> grid target % for b in batts: if b.name in alive: for s, e, to in b.grid: if s <= a < e and level[b.name] < to - EPS: want[b.name] = float(to) roles = {b.name: _role(b, a, bool(want), b.name in want) for b in batts if b.name in alive} by = {b.name: b for b in batts} for n, to in want.items(): # charging from the grid b = by[n] room = (to - level[n]) / 100.0 * b.capacity_kwh level[n] += min(b.rating_kw * hrs * EFF, room) \ / b.capacity_kwh * 100.0 if net > EPS: caps = {n: min(by[n].rating_kw * hrs, max(0.0, ceil - level[n]) / 100.0 * by[n].capacity_kwh / EFF) for n, r in roles.items() if r[1]} first = {n: c for n, c in caps.items() if by[n].solar_first} rest = {n: c for n, c in caps.items() if n not in first} left = net for grp in (first, rest): got, left = _spread(left, grp, dict(grp)) for n, v in got.items(): level[n] += v * EFF / by[n].capacity_kwh * 100.0 elif net < -EPS: caps = {n: min(by[n].rating_kw * hrs, max(0.0, level[n] - floor) / 100.0 * by[n].capacity_kwh * EFF) for n, r in roles.items() if r[2]} first = {n: c for n, c in caps.items() if roles[n][0] == GIVE} rest = {n: c for n, c in caps.items() if n not in first} left = -net for grp in (first, rest): got, left = _spread(left, grp, dict(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])) return out # --------------------------------------------------------------------------- # # Measured solar, per local hour # --------------------------------------------------------------------------- # class SolarToday: """The site's solar power, averaged per local hour of one local day. In memory only: a restart starts it again, and `first` says from when it has counted. Display only.""" def __init__(self): self.day = None self.first = None self._hours = {} # hour start (UTC) -> [W x s, seconds] self._last = None # (time, W) def add(self, now: datetime, w) -> None: """One reading (`None`: no reading, which ends the interval).""" day = _local(now).date() if day != self.day: self.day, self.first, self._hours, self._last = day, None, {}, None try: ok = w is not None and math.isfinite(float(w)) except (TypeError, ValueError): ok = False if not ok: self._last = None return w = max(0.0, float(w)) if self.first is None: self.first = now if self._last is not None and now > self._last[0]: dt = min(GAP_S, (now - self._last[0]).total_seconds()) key = _hour_start(self._last[0]) row = self._hours.setdefault(key, [0.0, 0.0]) row[0] += (self._last[1] + w) / 2.0 * dt row[1] += dt self._last = (now, w) def kwh(self, start: datetime, end: datetime, now: datetime): """kWh for the hour `[start, end)` up to `now`: the mean power it was read at over the part of the hour that has passed, or None when the readings cover under half of that part (or none has passed).""" if self.day != _local(now).date(): return None over = (min(end, now) - start).total_seconds() row = self._hours.get(start) if over <= 0 or row is None or row[1] < MIN_COVER * over: return None return row[0] / row[1] * over / 3.6e6 def _hour_start(t: datetime) -> datetime: lt = _local(t) return lt.replace(minute=0, second=0, microsecond=0).astimezone( timezone.utc)