glenn schrooyenandClaude Opus 5.5
9b1fa758ad
New list options pv_forecast_entities and pv_forecast_tomorrow_entities
(same shape as pv_entities); the single-entity keys are unchanged.
- Each member is read in its own unit (Wh/kWh/MWh) in one gather and the
day is summed. Hourly splits are merged when every member has one, else
the total goes on the bell.
- A member missing, unavailable, stale, non-energy or bad makes that
day's forecast unusable (no partial sum). A caution row names it by
position; the action log names the entity once per change.
- Single key and list both set for one day: refused at start.
- A one-member list behaves exactly like the single key (test_opt17.py);
the single-key path is byte-identical to e36565f over 24 scenarios.
- Supervisor shape measured: tools/supervisor_shape_probe.py section J.
DECISIONS/OPT-17-several-forecast-entities.md has the rules.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Bdqr7PiaHcBnzPituu2yTC
351 lines
13 KiB
Python
351 lines
13 KiB
Python
"""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-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
|
|
|
|
# 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
|
|
|
|
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
|
|
|
|
|
|
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, None)` or `(None, why)` for one tomorrow sensor."""
|
|
t, why = _day(state, day1, now, today=False)
|
|
if t is None:
|
|
return 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, STALE
|
|
return t, 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, _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)
|
|
|
|
|
|
def read_sum(today, tomorrow, now: datetime):
|
|
"""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)."""
|
|
if not today:
|
|
return None, NOT_SET, ()
|
|
day0, day1 = _bounds(now)
|
|
gaps, days, ats, why0 = [], [], [], None
|
|
for i, (ent, st) in enumerate(today, 1):
|
|
d, at, why = _today(st, day0, now)
|
|
if d is None:
|
|
gaps.append(("today", i, ent, why))
|
|
why0 = why0 or why
|
|
else:
|
|
days.append(d)
|
|
ats.append(at)
|
|
tdays = []
|
|
for i, (ent, st) in enumerate(tomorrow, 1):
|
|
t, why = _tomorrow(st, day0, day1, now)
|
|
if t is None:
|
|
gaps.append(("tomorrow", i, ent, why))
|
|
else:
|
|
tdays.append(t)
|
|
gaps = tuple(gaps)
|
|
if why0 is not None:
|
|
return None, why0, gaps
|
|
tday = _sum(tdays) if tomorrow and len(tdays) == len(tomorrow) else None
|
|
# The oldest member's report is when the sum was last fully known.
|
|
return Forecast(_sum(days), tday, min(ats)), None, gaps
|