goodwe-addon/goodwe_controller/app/solarfc.py

"""OPT-11 P4: the solar forecast, read from a Home Assistant sensor.

Pure: HA state dicts and the time in, a `Forecast` (or None and why) out.
main.py reads the entities; dayplan.py uses the answer. FCST-01 replaces
this later with the add-on's own forecast (owner answer, 2026-09-27).

Two options name the sensors: `pv_forecast_entity`, the solar energy still
to come TODAY, and `pv_forecast_tomorrow_entity`, TOMORROW's (optional; it
is only needed for a cheap run before midnight). Supported, by the entities
they create (DOCS.md lists them):

* Forecast.Solar (HA core): `sensor.energy_production_today_remaining` and
  `sensor.energy_production_tomorrow`. The state only; no hourly split.
* Solcast PV Forecast (HACS `solcast_solar`):
  `sensor.solcast_pv_forecast_forecast_remaining_today` and
  `..._forecast_tomorrow`, with `detailedForecast` (or `detailedHourly`):
  `[{period_start, pv_estimate}]`, kW per period.
* Open-Meteo Solar Forecast (HACS `open_meteo_solar_forecast`):
  `sensor.energy_production_today_remaining` / `..._tomorrow`, with
  `wh_period`: `{period start: Wh}`.

Any other sensor whose state is energy in Wh, kWh or MWh works too, as a
state only. The unit is read, never guessed (the rule `pv.watts()` has).
Without an hourly split, the day's energy is spread over a fixed bell from
06:00 to 20:00 local time (`BELL_FROM_H`, `BELL_TO_H`).

A reading older than `FRESH_S`, or from before local midnight, is stale,
and a stale or missing forecast is no forecast: the day plan is off and
every mode acts as in P3 option A.

OPT-27: but the last fresh reading of each sensor is kept (`Hold`), so the
night plan is not made blind. Forecast.Solar's sensors stop changing in the
evening and api.forecast.solar is sometimes out for a poll. A kept reading
stands in for its sensor (a) for up to `FRESH_S` when the sensor is missing
or unavailable, as a sensor that merely stopped updating is already
tolerated, and (b) from the end of the bell (`BELL_TO_H`) until local
midnight + `KEPT_UNTIL_H` however old it is. Never beyond that, and never
for another day: a today figure is only today's, a tomorrow figure is
tomorrow's until midnight and then, until midnight + 3 h, today's.
DECISIONS/OPT-27-last-fresh-forecast-for-the-night.md.

OPT-17: each day may instead name a list (`pv_forecast_entities`,
`pv_forecast_tomorrow_entities`), summed by `read_sum()`. One member that is
not usable makes that day's sum unusable, never a partial sum.
DECISIONS/OPT-17-several-forecast-entities.md.
"""

import math
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone

from .hass import BAD_STATES
from .tariff import BILLING_TZ

KWH_PER = {"Wh": 0.001, "kWh": 1.0, "MWh": 1000.0}

# Stale after 3 h. The integrations above refresh at least hourly, so a
# 2 h limit would flag a site whose hourly poll was missed once.
FRESH_S = 3 * 3600.0

# The bell a state-only forecast is spread over, in local hours.
BELL_FROM_H, BELL_TO_H = 6.0, 20.0

# OPT-27: a kept reading serves the night plan from the end of the bell until
# this many hours after local midnight (the card: "local midnight + 3 h").
KEPT_UNTIL_H = 3.0
SAVE_EVERY_S = 900.0

# Why there is no forecast (the page words are in ui_text `home_words`).
NOT_SET, MISSING, UNIT, STALE, BAD = (
    "not_set", "missing", "unit", "stale", "bad")

# OPT-17: (single key, list key) per day. A set list replaces the single key;
# the validator refuses both set for the same day.
TODAY_KEYS = ("pv_forecast_entity", "pv_forecast_entities")
TOMORROW_KEYS = ("pv_forecast_tomorrow_entity",
                 "pv_forecast_tomorrow_entities")


@dataclass(frozen=True)
class Day:
    """One day's forecast: `kwh` still to come (today) or in all (tomorrow),
    and the hourly split if the sensor has one: `((start, end, kwh), ...)`."""
    day0: datetime              # local midnight, as UTC
    kwh: float
    periods: tuple = ()


@dataclass(frozen=True)
class Forecast:
    today: Day
    tomorrow: Day | None
    at: datetime                # when the sensor last reported
    # OPT-27: a reading kept past `FRESH_S` is in it (the evening's figure at
    # night). The screens say so. Last, so `Forecast(today, tomorrow, at)` is
    # as before.
    kept: bool = False

    def between(self, t0: datetime, t1: datetime) -> float | None:
        """kWh the sun should give between `t0` (now or later) and `t1`.
        None when the span reaches the sunny hours of a day with no
        forecast (tomorrow, when its sensor is not set)."""
        if t1 <= t0:
            return 0.0
        last = self.tomorrow or self.today
        if t1 > last.day0 + timedelta(days=1, hours=BELL_FROM_H):
            return None
        total = 0.0
        for day in (self.today, self.tomorrow):
            if day is None:
                continue
            a = max(t0, day.day0)
            b = min(t1, day.day0 + timedelta(days=1))
            if b > a:
                total += _day_between(day, a, b, t0)
        return total


def _local_hour(t: datetime) -> float:
    lt = t.astimezone(BILLING_TZ)
    return lt.hour + lt.minute / 60.0 + lt.second / 3600.0


def _bell(h: float) -> float:
    """Cumulative share of a day's sun by local hour `h` (0..1)."""
    if h <= BELL_FROM_H:
        return 0.0
    if h >= BELL_TO_H:
        return 1.0
    x = (h - BELL_FROM_H) / (BELL_TO_H - BELL_FROM_H)
    return (1.0 - math.cos(math.pi * x)) / 2.0


def _hours_since(day0, t):
    return (t - day0).total_seconds() / 3600.0


def _day_between(day: Day, a: datetime, b: datetime, now: datetime) -> float:
    if day.periods:
        out = 0.0
        for s, e, kwh in day.periods:
            lo, hi = max(a, s), min(b, e)
            if hi > lo:
                out += kwh * (hi - lo).total_seconds() / (e - s).total_seconds()
        return out
    # State only: `kwh` is what is left from `max(now, day0)` to the day's
    # end, spread over the bell.
    h_from = _hours_since(day.day0, max(now, day.day0))
    left = 1.0 - _bell(h_from)
    if left <= 1e-9:
        return 0.0
    share = (_bell(_hours_since(day.day0, b))
             - _bell(_hours_since(day.day0, a))) / left
    return day.kwh * max(0.0, share)


def _parse_time(v):
    try:
        t = datetime.fromisoformat(str(v).replace("Z", "+00:00"))
    except (TypeError, ValueError):
        return None
    return t if t.tzinfo else t.replace(tzinfo=timezone.utc)


def _periods(attrs: dict, day0: datetime) -> tuple:
    """The hourly split, if the sensor has one, for the day at `day0`."""
    d1 = day0 + timedelta(days=1)
    out = []
    for key in ("detailedForecast", "detailedHourly"):
        rows = attrs.get(key)
        if isinstance(rows, list) and rows:
            starts = []
            for r in rows:
                if not isinstance(r, dict):
                    continue
                t = _parse_time(r.get("period_start"))
                try:
                    kw = float(r.get("pv_estimate"))
                except (TypeError, ValueError):
                    continue
                if t is not None and math.isfinite(kw) and kw >= 0:
                    starts.append((t, kw))
            starts.sort()
            for i, (t, kw) in enumerate(starts):
                step = (starts[i + 1][0] - t if i + 1 < len(starts)
                        else timedelta(minutes=30 if key ==
                                       "detailedForecast" else 60))
                if step <= timedelta(0) or step > timedelta(hours=1):
                    step = timedelta(minutes=30 if key == "detailedForecast"
                                     else 60)
                if day0 <= t < d1:
                    out.append((t, t + step,
                                kw * step.total_seconds() / 3600.0))
            if out:
                return tuple(out)
    wh = attrs.get("wh_period")
    if isinstance(wh, dict) and wh:
        rows = []
        for k, v in wh.items():
            t = _parse_time(k)
            try:
                e = float(v)
            except (TypeError, ValueError):
                continue
            if t is not None and math.isfinite(e) and e >= 0:
                rows.append((t, e / 1000.0))
        rows.sort()
        for i, (t, kwh) in enumerate(rows):
            nxt = rows[i + 1][0] if i + 1 < len(rows) else t + timedelta(
                hours=1)
            if nxt - t > timedelta(hours=1) or nxt <= t:
                nxt = t + timedelta(hours=1)
            if day0 <= t < d1:
                out.append((t, nxt, kwh))
    return tuple(out)


def _day(state, day0, now, *, today):
    """`(Day, None)` or `(None, why)` from one sensor's state dict."""
    if state is None:
        return None, MISSING
    raw = str(state.get("state", "")).strip()
    if raw.lower() in BAD_STATES:
        return None, MISSING
    try:
        v = float(raw)
    except ValueError:
        return None, BAD
    attrs = state.get("attributes") or {}
    unit = str(attrs.get("unit_of_measurement") or "").strip()
    if unit not in KWH_PER:
        return None, UNIT
    kwh = v * KWH_PER[unit]
    if not math.isfinite(kwh) or kwh < 0:
        return None, BAD
    periods = _periods(attrs, day0)
    if periods and today:
        # The split is the whole day; keep what is still to come, scaled so
        # it adds up to the state (the state is what the integration says
        # is left).
        left = sum(k * max(0.0, min(1.0, (e - max(s, now)).total_seconds()
                                    / (e - s).total_seconds()))
                   for s, e, k in periods)
        if left > 1e-6:
            f = kwh / left
            periods = tuple((s, e, k * f) for s, e, k in periods)
    return Day(day0, kwh, periods), None


def _reported(state):
    ts = [_parse_time(state.get(k)) for k in
          ("last_reported", "last_updated", "last_changed")]
    ts = [t for t in ts if t is not None]
    return max(ts) if ts else None


class Hold:
    """OPT-27: the last fresh reading of each forecast sensor, per day it is
    for. `read_sum()` remembers every fresh reading here and, when a sensor
    has none now, asks it (`stand_in()`).

    Keyed `(kind, entity, day0)`: `kind` is "today" (energy still to come on
    `day0`) or "tomorrow" (all of `day0`), and `day0` is the day the figure is
    FOR. A today figure therefore never answers for another day, and a
    tomorrow figure answers for its own day only: tomorrow until midnight,
    then today until midnight + `KEPT_UNTIL_H`. Only the newest reading of a
    key is kept, so the next fresh reading ends the hold at once.

    `dump()` / `load()` are what the controller keeps in its store, so an
    evening restart does not lose it (`load()` drops what is bad).
    `changed` is set when `dump()` would differ."""

    def __init__(self, data=None):
        self._r = {}
        self._saved = {}        # key -> the `at` last flagged for saving
        self.changed = False
        if isinstance(data, dict):
            self.load(data)

    def remember(self, kind, entity, day: "Day", at: datetime) -> None:
        key = (kind, entity, day.day0)
        old = self._r.get(key)
        if old is not None and at <= old[1]:
            return
        self._r[key] = (day, at)
        # A sensor that reports every minute with the same figure must not
        # rewrite the store every minute: the saved age may lag by
        # `SAVE_EVERY_S` (the bounds are hours).
        mark = self._saved.get(key)
        if old is None or old[0].kwh != day.kwh or mark is None \
                or (at - mark).total_seconds() >= SAVE_EVERY_S:
            self._saved[key] = at
            self.changed = True

    def forget_before(self, day0: datetime) -> None:
        """Drop what is for a day before `day0` (the local midnight now)."""
        gone = [k for k in self._r if k[2] < day0]
        for k in gone:
            del self._r[k]
            self._saved.pop(k, None)
        if gone:
            self.changed = True

    def stand_in(self, kind, entity, day0, now, *, as_today=False):
        """`(Day, at, kept)` for a sensor that has no fresh reading, or None.

        The kept reading must be for `day0`, and either (a) at most `FRESH_S`
        old, or (b) used at night: from the end of the bell to local midnight,
        or within `KEPT_UNTIL_H` after it. `kept` is true when (b) was needed,
        or `as_today` (a tomorrow figure standing in for today, after
        midnight)."""
        rec = self._r.get((kind, entity, day0))
        if rec is None:
            return None
        day, at = rec
        if (now - at).total_seconds() <= FRESH_S and not as_today:
            return day, at, False
        after_midnight = now < day0 + timedelta(hours=KEPT_UNTIL_H)
        evening = _local_hour(now) >= BELL_TO_H and now < day0 + timedelta(
            days=1)
        if as_today:
            return (day, at, True) if after_midnight else None
        return (day, at, True) if (evening or after_midnight) else None

    def dump(self) -> dict:
        return {"v": 1, "r": [
            [kind, ent, d.day0.isoformat(), d.kwh, at.isoformat(),
             [[s.isoformat(), e.isoformat(), k] for s, e, k in d.periods]]
            for (kind, ent, _d0), (d, at) in sorted(
                self._r.items(), key=lambda kv: (kv[0][0], kv[0][1],
                                                 kv[0][2]))]}

    def load(self, data: dict) -> None:
        rows = data.get("r") if isinstance(data, dict) else None
        if not isinstance(rows, list):
            return
        for row in rows:
            try:
                kind, ent, d0, kwh, at, per = row
                day0, at = _parse_time(d0), _parse_time(at)
                kwh = float(kwh)
                periods = tuple((_parse_time(s), _parse_time(e), float(k))
                                for s, e, k in per)
                ok = (kind in ("today", "tomorrow") and isinstance(ent, str)
                      and day0 is not None and at is not None
                      and math.isfinite(kwh) and kwh >= 0
                      and all(s is not None and e is not None and e > s
                              and math.isfinite(k) and k >= 0
                              for s, e, k in periods))
            except (TypeError, ValueError):
                continue
            if ok:
                self._r[(kind, ent, day0)] = (Day(day0, kwh, periods), at)
                self._saved[(kind, ent, day0)] = at


def entities(opts: dict, *, tomorrow=False) -> tuple:
    """The forecast entities one day names, in configured order: the list
    if it has any entry, else the single key, else (). A repeated id is
    dropped, as `pv.entities()` does: it would count one forecast twice."""
    one, many = TOMORROW_KEYS if tomorrow else TODAY_KEYS
    listed = [e for e in (str(x or "").strip()
                          for x in (opts.get(many) or [])) if e]
    if listed:
        return tuple(dict.fromkeys(listed))
    e = str(opts.get(one) or "").strip()
    return (e,) if e else ()


def _bounds(now):
    here = now.astimezone(BILLING_TZ)
    d0 = datetime(here.year, here.month, here.day, tzinfo=BILLING_TZ)
    nd = d0.date() + timedelta(days=1)
    d1 = datetime(nd.year, nd.month, nd.day, tzinfo=BILLING_TZ)
    return d0.astimezone(timezone.utc), d1.astimezone(timezone.utc)


def _today(state, day0, now):
    """`(Day, at, None)` or `(None, None, why)` for one today sensor."""
    today, why = _day(state, day0, now, today=True)
    if today is None:
        return None, None, why
    at = _reported(state)
    # A reading from before local midnight is yesterday's "left today" (0
    # after sunset) until the integration refreshes: stale, not a dark day.
    if at is None or at < day0 or (now - at).total_seconds() > FRESH_S \
            or (at - now).total_seconds() > 300:
        return None, None, STALE
    return today, at, None


def _tomorrow(state, day0, day1, now):
    """`(Day, at, None)` or `(None, None, why)` for one tomorrow sensor."""
    t, why = _day(state, day1, now, today=False)
    if t is None:
        return None, None, why
    t_at = _reported(state)
    if t_at is None or t_at < day0 or (now - t_at).total_seconds() > FRESH_S:
        return None, None, STALE
    return t, t_at, None


def read(today_state, tomorrow_state, now: datetime, *,
         configured=True):
    """`(Forecast, None)` or `(None, why)`. `tomorrow_state` may be None
    (the option is not set, or the sensor is missing): the forecast then
    covers today only."""
    if not configured:
        return None, NOT_SET
    day0, day1 = _bounds(now)
    today, at, why = _today(today_state, day0, now)
    if today is None:
        return None, why
    tomorrow = None
    if tomorrow_state is not None:
        tomorrow, _at, _w = _tomorrow(tomorrow_state, day0, day1, now)
    return Forecast(today, tomorrow, at), None


def _sum(days):
    """One Day from several members' Days. The hourly split is kept only if
    every member has one: the splits are then joined (`_day_between()`
    adds overlapping periods). Else the total goes on the bell."""
    if len(days) == 1:
        return days[0]
    periods = ()
    if all(d.periods for d in days):
        periods = tuple(sorted(p for d in days for p in d.periods))
    return Day(days[0].day0, sum(d.kwh for d in days), periods)


# Why a sensor may be covered by a kept reading: it has none now, or it is
# too old. A bad value or a wrong unit is a set-up fault and is never covered.
_COVERABLE = (MISSING, STALE)


def read_sum(today, tomorrow, now: datetime, hold: Hold | None = None):
    """OPT-17: `(Forecast or None, why, gaps)` over lists of
    `(entity, state)`, one per configured member, in configured order.

    Each member is read exactly as `read()` reads one sensor, in its own
    unit. A member that is missing, unavailable, stale, not in an energy
    unit or bad makes that DAY unusable, as one such sensor does today: no
    today means no forecast, no tomorrow means today only. A partial sum
    would under-count the sun and mislead the day plan.

    `gaps` is `((day, position, entity, why), ...)` for every unusable
    member (`day` "today" or "tomorrow", `position` 1-based), so the page
    and the log can name it. With one member per day the forecast is what
    `read()` gives (test_opt17.py proves it).

    OPT-27: with a `hold`, every fresh reading is remembered in it, and a
    member that is missing or stale is covered by its last fresh reading
    where `Hold.stand_in()` allows it (the forecast then has `kept` set when
    that took a reading older than `FRESH_S`). After midnight, if the today
    side is still short, the tomorrow figures read before midnight are
    today's, until midnight + `KEPT_UNTIL_H`. Without a `hold` this is
    exactly as before."""
    if not today:
        return None, NOT_SET, ()
    day0, day1 = _bounds(now)
    if hold is not None:
        hold.forget_before(day0)
    gaps, days, ats, why0, kept = [], [], [], None, False
    for i, (ent, st) in enumerate(today, 1):
        d, at, why = _today(st, day0, now)
        if d is not None and hold is not None:
            hold.remember("today", ent, d, at)
        elif hold is not None and why in _COVERABLE:
            r = hold.stand_in("today", ent, day0, now)
            if r is not None:
                d, at, k = r
                kept = kept or k
        if d is None:
            gaps.append(("today", i, ent, why))
            why0 = why0 or why
        else:
            days.append(d)
            ats.append(at)
    tdays, tats, tkept = [], [], False
    for i, (ent, st) in enumerate(tomorrow, 1):
        t, t_at, why = _tomorrow(st, day0, day1, now)
        if t is not None and hold is not None:
            hold.remember("tomorrow", ent, t, t_at)
        elif hold is not None and why in _COVERABLE:
            r = hold.stand_in("tomorrow", ent, day1, now)
            if r is not None:
                t, t_at, k = r
                tkept = tkept or k
        if t is None:
            gaps.append(("tomorrow", i, ent, why))
        else:
            tdays.append(t)
            tats.append(t_at)
    if why0 is not None and hold is not None and tomorrow and all(
            g[3] in _COVERABLE for g in gaps if g[0] == "today"):
        # Past midnight the sensors may still show yesterday's "left today".
        # The figure each tomorrow sensor gave before midnight is for this
        # day; all of them are needed, as for any sum.
        subs = [hold.stand_in("tomorrow", ent, day0, now, as_today=True)
                for ent, _st in tomorrow]
        if all(r is not None for r in subs):
            days = [r[0] for r in subs]
            ats = [r[1] for r in subs]
            gaps = [g for g in gaps if g[0] != "today"]
            why0, kept = None, True
    gaps = tuple(gaps)
    if why0 is not None:
        return None, why0, gaps
    tday = _sum(tdays) if tomorrow and len(tdays) == len(tomorrow) else None
    kept = kept or (tday is not None and tkept)
    # The oldest member's report is when the sum was last fully known.
    return Forecast(_sum(days), tday, min(ats), kept), None, gaps