"""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, 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 (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 MIN_CUT = timedelta(seconds=1) # no step is cut closer to its ends MAX_FLOOR_CUTS = 400 # a bound on the floor cuts of one curve 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 _give_caps(roles, level, by, floor, hrs): """`(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}, 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. 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, 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 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)`. 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. `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): 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} floor_cuts, capped = 0, False i = 0 while i < len(ts) - 1: a, z = ts[i], ts[i + 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) if net > 0.0: net *= trust 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) by = {b.name: b for b in batts} # A battery that reaches its level inside the step gets a point # there: the step is cut and read again, so the line is exact. cut = None for n, to in want.items(): b = by[n] room = (to - level[n]) / 100.0 * b.capacity_kwh rate = b.rating_kw * hrs * EFF if EPS < room < rate * (1.0 - 1e-9): t_r = a + (z - a) * (room / rate) if a < t_r < z and (cut is None or t_r < cut): cut = t_r if cut is not None: ts.insert(i + 1, cut) continue roles = {b.name: _role(b, a, bool(want), b.name in want) for b in batts if b.name in alive} # 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 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) continue 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: first, rest, held = _give_caps(roles, level, by, floor, hrs) left = -net for grp in (first, rest): 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 # --------------------------------------------------------------------------- # # 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: """The start of `t`'s local hour, in UTC. Taken back from the UTC instant, never from the wall clock, so each of the two 02:00 hours of the autumn change has its own (`dayplan._hour_slices()`, SAFETY-123).""" lt = _local(t) return t.astimezone(timezone.utc) - timedelta( minutes=lt.minute, seconds=lt.second, microseconds=lt.microsecond)