goodwe-addon/goodwe_controller/app/daycurve.py

"""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)