"""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, if it is at most `KEPT_MAX_AGE_S` old when used. 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 the bell starts (`BELL_FROM_H`, 06:00 local), today's (OPT-34: at most `KEPT_MORNING_MAX_AGE_S` old, it must have been usable at midnight, and no fresh today reading may have replaced it since). DECISIONS/OPT-27-last-fresh-forecast-for-the-night.md, DECISIONS/OPT-34-the-night-charge-keeps-its-level.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 # local midnight, and a reading older than this when used is not kept (a # morning figure must not drive the night). KEPT_MAX_AGE_S = 12 * 3600.0 # OPT-34: the evening's "tomorrow" figure, once midnight has passed, is today's # until the bell starts (`BELL_FROM_H`): nothing has been produced yet, so it # is still the whole day. A 17:05 reading is 12 h 55 min old at 06:00, so this # case may be older than `KEPT_MAX_AGE_S`, up to this, but only a figure that # was usable at midnight (at most `KEPT_MAX_AGE_S` old then), so one that has # not been used all evening does not come back at 00:00. KEPT_MORNING_MAX_AGE_S = 15 * 3600.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, since: datetime | None = None) -> 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). `since` is the moment a state-only "left today" figure counts from (OPT-23): `t0` itself when not given, so a span that starts later than now would be spread over what is left of the bell from its own start. Pass now for every span of a plan, and the spans add up to the figure. An hourly split ignores it.""" if t1 <= t0: return 0.0 since = t0 if since is None else since 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, since) 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. Nothing of it is before that moment. start = max(now, day.day0) a = max(a, start) if b <= a: return 0.0 h_from = _hours_since(day.day0, start) 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 the bell starts (OPT-34). 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, at most `KEPT_MAX_AGE_S` old. `kept` is true when (b) was needed, or `as_today` (OPT-34: a tomorrow figure standing in for today, after midnight and until the bell starts, `KEPT_MORNING_MAX_AGE_S`).""" rec = self._r.get((kind, entity, day0)) if rec is None: return None day, at = rec age = (now - at).total_seconds() if age <= FRESH_S and not as_today: return day, at, False if as_today: ok = (day0 <= now < _bell_start(day0) and age <= KEPT_MORNING_MAX_AGE_S and (day0 - at).total_seconds() <= KEPT_MAX_AGE_S) return (day, at, True) if ok else None if age > KEPT_MAX_AGE_S: return None # A tomorrow sensor is asked with day0 = the next midnight, which is # in the future all day, so the evening ends at that day's end. evening = _local_hour(now) >= BELL_TO_H and now < _bounds(day0)[1] return (day, at, True) if evening else None def replaced(self, entities, day0) -> bool: """OPT-34 F-1: every today sensor has given a fresh reading for `day0` (so after midnight, newer than any evening figure). The evening's tomorrow figure is then no stand-in for today any more, also once those readings are stale.""" return all(("today", e, day0) in self._r for e in entities) 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 _bell_start(day0): """The moment the bell starts on the local day that begins at `day0` (`BELL_FROM_H` local, so 06:00 on a 23 h or 25 h day too).""" here = day0.astimezone(BILLING_TZ) t = datetime(here.year, here.month, here.day, tzinfo=BILLING_TZ) \ + timedelta(hours=BELL_FROM_H) return t.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 the bell starts (`stand_in()`), unless fresh today readings have replaced them (`Hold.replaced()`). 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) and not hold.replaced( [ent for ent, _st in today], day0): 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