glenn schrooyenandClaude Opus 5.5
63dda4b0d3
Comment, docstring and DECISIONS only (same-code script OK; python -OO bytecode now identical for all six files). - peakguard.charge_room_w: one-line docstring + SAFETY-125 pointer (cut whole). - peakguard.shaping_suspended: keep 'the one place this conjunction is written; do not inline it'. - price_forecast module note: an hour with no price and no yesterday is left out, never filled (test_price.py 14d cites the module note for it). - Refill comment blocks the cut left ragged; DECISIONS pointers on one line. - FCST-04 note: _yesterday_of(), not _previous_day(); FCST-03 note: the 49 stronger cases are why the rule is a real change. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Bdqr7PiaHcBnzPituu2yTC
452 lines
20 KiB
Python
452 lines
20 KiB
Python
"""FCST-04 - the day the carry-forward carries, kept where a restart cannot reach.
|
|
|
|
`price_forecast.fill_from_yesterday()` needs a yesterday that survives a
|
|
restart; this module is that store and the one caller of the fill. Four rules:
|
|
|
|
1. It records what the PROVIDER published, never what the fill produced
|
|
(feeding a fill back in rots the numbers behind a true `provisional` flag).
|
|
`carry()` is the only public entry point and `_remember()` has exactly one
|
|
caller; test_price_history.py parses every shipped module for the name and
|
|
grades the order by driving it.
|
|
2. "Yesterday" is the Brussels day before the instant asked about, not the
|
|
newest record; any other date is `no_data` naming both dates.
|
|
3. A store that cannot answer says so in the type: `yesterday()` returns a
|
|
`PriceRead`, never an empty series.
|
|
4. Hours accumulate within a day (merge by instant, the newer price wins), so a
|
|
feed that serves only the remaining hours cannot overwrite a full day.
|
|
|
|
Rows are keyed by instant, not local hour (the DST rule stays in the fill). A
|
|
store that cannot persist records nothing (FCST-06). It does not extend the fill
|
|
to a feed that published nothing: that stays 11.4's static fallback.
|
|
DECISIONS/FCST-04-the-day-that-is-carried.md (section 5 has the store details)."""
|
|
|
|
import copy
|
|
import json
|
|
import logging
|
|
import os
|
|
import tempfile
|
|
from datetime import date, datetime, timedelta, timezone
|
|
|
|
from . import messages
|
|
from .price import PricePoint, PriceProduct, PriceRead, brussels
|
|
from .price_forecast import fill_from_yesterday
|
|
|
|
_LOG = logging.getLogger("goodwe.price.history")
|
|
|
|
# Brussels days kept: a sort-order cap, not a horizon (the fill needs only
|
|
# yesterday).
|
|
KEEP_DAYS = 3
|
|
|
|
SCHEMA = 1
|
|
|
|
|
|
def _bdate(instant: datetime) -> str:
|
|
"""The Brussels calendar date of an instant, as `YYYY-MM-DD` (not UTC: 23:30 UTC in July is tomorrow)."""
|
|
return brussels(instant).strftime("%Y-%m-%d")
|
|
|
|
|
|
def _yesterday_of(instant: datetime) -> str:
|
|
"""The Brussels date of the day BEFORE `instant`'s Brussels day.
|
|
Arithmetic on the local `date`, not the instant: subtracting 24 absolute hours
|
|
was wrong after the spring-forward. DECISIONS/FCST-04-the-day-that-is-carried.md."""
|
|
return (brussels(instant).date() - timedelta(days=1)).strftime("%Y-%m-%d")
|
|
|
|
|
|
def _point_of(row):
|
|
"""One stored row -> a `PricePoint`, or `None` if the row is not one.
|
|
|
|
The one validator for both the read and the write path (FCST-05): a malformed
|
|
row used to raise out of `carry()` and fail every refresh for good. Returns
|
|
`None` rather than raising; the caller decides how a bad row is reported."""
|
|
if not isinstance(row, dict):
|
|
return None
|
|
try:
|
|
start = datetime.fromisoformat(row["start"])
|
|
value = float(row["eur_per_kwh"])
|
|
flag = row["provisional"]
|
|
except (KeyError, TypeError, ValueError):
|
|
return None
|
|
if start.tzinfo is None or start.utcoffset() is None \
|
|
or not isinstance(flag, bool):
|
|
return None
|
|
return PricePoint(start=start, eur_per_kwh=value, provisional=flag)
|
|
|
|
|
|
def _held(rec) -> tuple:
|
|
"""`(points, whole hours they cover)` for a day record. `(0, 0)` otherwise.
|
|
Two numbers, each in its own unit (96 points over 24 h on a quarter-hourly
|
|
feed). Total, and counts only rows that are points."""
|
|
if not isinstance(rec, dict):
|
|
return (0, 0)
|
|
raw = rec.get("points")
|
|
if not isinstance(raw, (list, tuple)):
|
|
return (0, 0)
|
|
n, hours = 0, set()
|
|
for row in raw:
|
|
pt = _point_of(row)
|
|
if pt is None:
|
|
continue
|
|
n += 1
|
|
hours.add(pt.start.astimezone(timezone.utc)
|
|
.replace(minute=0, second=0, microsecond=0))
|
|
return (n, len(hours))
|
|
|
|
|
|
def _row_of(day: str, points: int, hours: int) -> str:
|
|
"""One day's cell of the status row, with the unit spelt out on each number."""
|
|
# UX-10: the day as the page writes dates ("26 Sep"), not ISO.
|
|
return (f"{points} price{'' if points == 1 else 's'} over {hours} "
|
|
f"hour{'' if hours == 1 else 's'} for {day_words(day)}")
|
|
|
|
|
|
def day_words(day) -> str:
|
|
"""`'2026-09-26'` -> `'26 Sep'`, as the page writes a date. A key that is
|
|
not a date is shown as it is, because this feeds a status row."""
|
|
try:
|
|
d = date.fromisoformat(str(day))
|
|
except ValueError:
|
|
return str(day)
|
|
return f"{d.day} {d:%b}"
|
|
|
|
|
|
def _bad_rows(rec) -> int:
|
|
"""How many rows of a day record are not points. Total; never raises."""
|
|
if not isinstance(rec, dict):
|
|
return 0
|
|
raw = rec.get("points")
|
|
if not isinstance(raw, (list, tuple)):
|
|
return 0
|
|
return sum(1 for row in raw if _point_of(row) is None)
|
|
|
|
|
|
class PriceHistory:
|
|
"""The published price series of the last few Brussels days, in /data.
|
|
`ok` is about the file, not the data: a healthy store with no record for
|
|
yesterday is `ok` and answers `no_data`. `ok`/`last_error` describe the file."""
|
|
|
|
def __init__(self, path: str = "/data/price_history.json"):
|
|
self.path = path
|
|
self.ok = False
|
|
self.last_error: str | None = None
|
|
self._days: dict = {}
|
|
self._open()
|
|
|
|
# the file
|
|
def _open(self) -> None:
|
|
parent = os.path.dirname(self.path)
|
|
# NOT created (see FCST-04 section 5): a cache on the ephemeral layer
|
|
# forgets silently.
|
|
if parent and not os.path.isdir(parent):
|
|
self.last_error = f"no durable volume at {parent}"
|
|
# INFO, not ERROR: a dev host or test container has no /data.
|
|
_LOG.info("No price history is kept (%s), so missing hours are "
|
|
"not filled from yesterday's prices.", self.last_error)
|
|
return
|
|
self.ok = True
|
|
self._load()
|
|
|
|
def _load(self) -> None:
|
|
try:
|
|
with open(self.path, "r", encoding="utf-8") as fh:
|
|
blob = json.load(fh)
|
|
except FileNotFoundError:
|
|
_LOG.info("no price history yet at %s - the first refresh writes "
|
|
"one, and the fill has nothing to carry until the day "
|
|
"after that", self.path)
|
|
return
|
|
except (ValueError, OSError) as err:
|
|
# A corrupt cache must not stop the controller nor be silently re-
|
|
# created: the record is discarded and the reason kept, so
|
|
# `yesterday()` reports it. `ValueError` is the family caught, not
|
|
# `json.JSONDecodeError` (one non-UTF-8 byte raised
|
|
# `UnicodeDecodeError` and the add-on did not start, FCST-05).
|
|
self.last_error = f"unreadable ({err})"
|
|
_LOG.warning("Price history at %s is %s, so it starts empty. "
|
|
"Missing hours cannot be filled in until a full day "
|
|
"is saved.",
|
|
self.path, self.last_error)
|
|
return
|
|
found = blob.get("version") if isinstance(blob, dict) else None
|
|
# The type is part of the version: `1.0 == 1` in Python.
|
|
if not isinstance(blob, dict) or isinstance(found, bool) \
|
|
or not isinstance(found, int) or found != SCHEMA:
|
|
self.last_error = f"version {found!r}, not {SCHEMA}"
|
|
_LOG.warning("Price history at %s has %s, so it is not used.",
|
|
self.path, self.last_error)
|
|
return
|
|
days = blob.get("days")
|
|
if not isinstance(days, dict):
|
|
self.last_error = "no days in it"
|
|
_LOG.warning("Price history at %s has %s, so it is not used.",
|
|
self.path, self.last_error)
|
|
return
|
|
self._days = self._usable_days(days)
|
|
|
|
def _usable_days(self, days: dict) -> dict:
|
|
"""The day records that are shaped like day records. The rest are dropped.
|
|
A broken cache is a short horizon, never a controller that will not start.
|
|
Rows inside a kept record are counted, not dropped (`_parse_points()` filters)."""
|
|
kept, dropped, malformed = {}, [], {}
|
|
for key, rec in days.items():
|
|
if not isinstance(key, str) or not isinstance(rec, dict) \
|
|
or not isinstance(rec.get("points"), list):
|
|
dropped.append(str(key))
|
|
continue
|
|
bad = _bad_rows(rec)
|
|
if bad:
|
|
malformed[key] = bad
|
|
kept[key] = rec
|
|
notes = []
|
|
if dropped:
|
|
notes.append(f"{messages.count(len(dropped), 'unusable day')} "
|
|
f"dropped, " + ", ".join(day_words(d)
|
|
for d in sorted(dropped)))
|
|
if malformed:
|
|
notes.append(
|
|
f"{messages.count(sum(malformed.values()), 'bad price')} in "
|
|
+ ", ".join(f"{day_words(d)} ({n})"
|
|
for d, n in sorted(malformed.items())))
|
|
if notes:
|
|
# Recorded, not silent: a cache that quietly discards half a file
|
|
# reads like a first day.
|
|
self.last_error = " and ".join(notes)
|
|
_LOG.warning("Price history at %s: %s. The add-on starts anyway, "
|
|
"and fills in only what the rest holds.",
|
|
self.path, self.last_error)
|
|
return kept
|
|
|
|
def _save(self) -> bool:
|
|
"""Write `self._days` out. True only if the file now holds it.
|
|
The one "can this store persist?" answer (FCST-06), asked after the attempt."""
|
|
if not self.ok:
|
|
return False
|
|
tmp = None
|
|
try:
|
|
fd, tmp = tempfile.mkstemp(dir=os.path.dirname(self.path))
|
|
with os.fdopen(fd, "w", encoding="utf-8") as fh:
|
|
json.dump({"version": SCHEMA, "days": self._days}, fh, indent=1)
|
|
# Atomic: a power cut mid-write must not leave a truncated file.
|
|
os.replace(tmp, self.path)
|
|
tmp = None # it IS the file now; do not unlink
|
|
except OSError as err:
|
|
self.last_error = f"could not write ({err})"
|
|
_LOG.error("could not persist price history: %s", err)
|
|
return False
|
|
finally:
|
|
if tmp is not None:
|
|
try:
|
|
os.unlink(tmp)
|
|
except OSError: # nothing further to try, and a
|
|
pass # failed cleanup is not a fault
|
|
return True
|
|
|
|
# the one public entry point
|
|
def carry(self, read: PriceRead, *, since: datetime, until: datetime,
|
|
stamp: datetime, remember: bool = True) -> PriceRead:
|
|
"""Remember what this read published, then 8.3-fill the missing hours.
|
|
Recorded FIRST, so the filled series exists only as the return value (rule 1).
|
|
Returns `fill_from_yesterday()`'s series, or `no_data` when no hour can be priced."""
|
|
if remember:
|
|
# PRICE-01: False only for a read that came from this store,
|
|
# already filed.
|
|
self._remember(read, stamp)
|
|
return fill_from_yesterday(read, yesterday=self.yesterday(stamp),
|
|
since=since, until=until, stamp=stamp)
|
|
|
|
# the halves, private because rule 1 is an ordering
|
|
def _remember(self, read: PriceRead, stamp: datetime) -> None:
|
|
"""File a provider read's points under their own Brussels days.
|
|
Private, called from exactly one place: no value-level defence can tell a
|
|
carried point from a published-provisional one, so the defence is the ordering."""
|
|
points = read.match(series=lambda r: r.points, no_data=lambda _r: ())
|
|
if not points:
|
|
return
|
|
product = read.match(series=lambda r: r.product.value,
|
|
no_data=lambda _r: None)
|
|
provider = read.match(series=lambda r: r.provider,
|
|
no_data=lambda _r: None)
|
|
|
|
# The state this read started from, kept so it can be put back
|
|
# (FCST-06): the merge runs in place and is undone if `_save()` did not
|
|
# take it. Deep copy, since the loop rebinds the day records.
|
|
before = copy.deepcopy(self._days)
|
|
|
|
touched = set()
|
|
junked = {}
|
|
for p in points:
|
|
day = _bdate(p.start)
|
|
rec = self._days.setdefault(
|
|
day, {"product": product, "provider": provider, "points": []})
|
|
# Rule 4: MERGE by instant; a later read updates the hours it
|
|
# carries. Every held row goes through `_point_of()` and the key is
|
|
# the validated instant, not its spelling (FCST-05). A row that is
|
|
# not a point is dropped and reported, and `_save()` writes the
|
|
# record back clean.
|
|
raw = rec.get("points")
|
|
held = {}
|
|
for q in (raw if isinstance(raw, list) else ()):
|
|
kept = _point_of(q)
|
|
if kept is None:
|
|
junked[day] = junked.get(day, 0) + 1
|
|
continue
|
|
held[kept.start] = q
|
|
held[p.start] = p.canonical()
|
|
order = sorted(held, key=lambda t: t.astimezone(timezone.utc))
|
|
rec["points"] = [held[t] for t in order]
|
|
rec["product"], rec["provider"] = product, provider
|
|
touched.add(day)
|
|
|
|
if junked:
|
|
# Said, not swallowed: `words` prints `last_error`. A failed
|
|
# `_save()` overwrites it and undoes the drop.
|
|
self.last_error = (
|
|
f"{messages.count(sum(junked.values()), 'bad price')} dropped "
|
|
f"from " + ", ".join(f"{day_words(d)} ({n})"
|
|
for d, n in sorted(junked.items())))
|
|
_LOG.warning("Price history at %s: %s. The file is saved without "
|
|
"them and the price update goes on.",
|
|
self.path, self.last_error)
|
|
self._prune(stamp)
|
|
if not self._save():
|
|
# A store that cannot persist records nothing (FCST-06):
|
|
# `_save()`'s answer decides what is kept, so `yesterday()` never
|
|
# answers a series the file does not hold.
|
|
self._days = before
|
|
return
|
|
_LOG.debug("price history: %d point(s) filed under %s; %d day(s) held",
|
|
len(points), ", ".join(sorted(touched)), len(self._days))
|
|
|
|
def _prune(self, stamp: datetime) -> None:
|
|
"""Drop records no lookup can reach: a sort-order cap plus two pins. Nothing is
|
|
deleted by arithmetic against the clock (a jumped clock would delete everything,
|
|
WITNESS-03). The pin of the two dates `yesterday()` can be asked for only
|
|
retains, so a far-future record cannot push yesterday out."""
|
|
if len(self._days) <= KEEP_DAYS:
|
|
return
|
|
keep = set(sorted(self._days)[-KEEP_DAYS:])
|
|
keep |= {_bdate(stamp), _yesterday_of(stamp)}
|
|
for day in [d for d in self._days if d not in keep]:
|
|
del self._days[day]
|
|
|
|
def yesterday(self, stamp: datetime) -> PriceRead:
|
|
"""The published series of the Brussels day BEFORE `stamp`'s, or no_data.
|
|
Computed, not looked up (rule 2)."""
|
|
if stamp.tzinfo is None or stamp.utcoffset() is None:
|
|
raise ValueError(
|
|
f"yesterday({stamp!r}): a naive datetime is not an instant, "
|
|
f"and 'which Brussels day was yesterday' is exactly the "
|
|
f"question a wall-clock reading cannot answer.")
|
|
want = _yesterday_of(stamp)
|
|
if self.last_error and not self._days:
|
|
return PriceRead.no_data(
|
|
f"the price history at {self.path} is {self.last_error}, so "
|
|
f"there are no prices from yesterday to fill in for "
|
|
f"{day_words(want)}. The plan covers only the hours the feed "
|
|
f"published")
|
|
rec = self._days.get(want)
|
|
if rec is None:
|
|
have = (", ".join(day_words(d) for d in sorted(self._days))
|
|
or "no days at all")
|
|
return PriceRead.no_data(
|
|
f"no published prices are kept for {day_words(want)}, only for "
|
|
f"{have}, so there are no prices from yesterday to fill in. A "
|
|
f"record for another day is not a yesterday")
|
|
points, bad = _parse_points(rec.get("points"))
|
|
if not points:
|
|
why = bad or "it is empty"
|
|
return PriceRead.no_data(
|
|
f"the {day_words(want)} record holds no usable price, {why}, "
|
|
f"and an empty series is not a free day")
|
|
try:
|
|
product = PriceProduct(rec.get("product"))
|
|
except ValueError:
|
|
return PriceRead.no_data(
|
|
f"the {want} record names product {rec.get('product')!r}, "
|
|
f"which is not a PriceProduct. The optimiser switches on that "
|
|
f"enum, so a record that cannot name one cannot be a series")
|
|
if bad:
|
|
_LOG.warning("Price history: %s dropped from the %s record, and "
|
|
"%s kept.", bad, day_words(want),
|
|
messages.count(len(points), "price"))
|
|
return PriceRead.series(points, product,
|
|
str(rec.get("provider") or "price_history"),
|
|
stamp)
|
|
|
|
def stored(self, stamp: datetime) -> PriceRead:
|
|
"""PRICE-01: every price this store holds, as one series, or no_data. The last
|
|
step of the price server's fallback (price_chain.py): what was PUBLISHED and
|
|
filed, never what a fill produced."""
|
|
if stamp.tzinfo is None or stamp.utcoffset() is None:
|
|
raise ValueError(f"stored({stamp!r}): a naive datetime is not an "
|
|
f"instant")
|
|
points = []
|
|
for day in sorted(self._days):
|
|
rec = self._days[day]
|
|
got, _bad = _parse_points(rec.get("points") if isinstance(
|
|
rec, dict) else None)
|
|
points.extend(got)
|
|
if not points:
|
|
why = self.last_error or "it holds no prices"
|
|
return PriceRead.no_data(
|
|
f"the price history at {self.path} gives nothing: {why}")
|
|
points.sort(key=lambda p: p.start.astimezone(timezone.utc))
|
|
return PriceRead.series(points, PriceProduct.DAY_AHEAD,
|
|
"price_history", stamp)
|
|
|
|
# observability
|
|
@property
|
|
def words(self) -> str:
|
|
"""One line for the status page. Says what is held, or why nothing is.
|
|
Must not raise, ever: it runs from `Controller.__init__` and every status
|
|
render, unguarded. Prints points and hours separately (FCST-05)."""
|
|
if not self.ok:
|
|
return f"unavailable - {self.last_error}"
|
|
if not self._days:
|
|
why = f" - {self.last_error}" if self.last_error else ""
|
|
return f"empty{why} - nothing to carry forward yet"
|
|
held = ", ".join(_row_of(d, *_held(self._days[d]))
|
|
for d in sorted(self._days, key=str))
|
|
# A file that partly loaded says so, or dropped days read as never
|
|
# written.
|
|
return held + (f" [{self.last_error}]" if self.last_error else "")
|
|
|
|
|
|
def _parse_points(raw) -> tuple:
|
|
"""Canonical dicts -> PricePoints, and a phrase naming what was dropped. A
|
|
malformed point is dropped, not defaulted: a lost field must not read as a free hour."""
|
|
if not isinstance(raw, list):
|
|
return (), "the `points` field is not a list"
|
|
out, dropped = [], 0
|
|
for row in raw:
|
|
# `_point_of()` is the one validator; this loop used to be a weaker
|
|
# one.
|
|
pt = _point_of(row)
|
|
if pt is None:
|
|
dropped += 1
|
|
continue
|
|
out.append(pt)
|
|
out.sort(key=lambda p: p.start.astimezone(timezone.utc))
|
|
return tuple(out), (messages.count(dropped, "bad price") if dropped else "")
|
|
|
|
|
|
class NoHistory:
|
|
"""The explicit "this caller keeps no history": a NAME, not `None`, so the
|
|
parameter can be required. It records nothing and carries nothing."""
|
|
|
|
ok = False
|
|
last_error = "no history is kept by this caller"
|
|
words = "not kept"
|
|
|
|
def carry(self, read: PriceRead, *, since, until, stamp,
|
|
remember: bool = True) -> PriceRead:
|
|
return read
|
|
|
|
def stored(self, stamp: datetime) -> PriceRead:
|
|
return PriceRead.no_data("this caller keeps no price history, so "
|
|
"there are no stored prices")
|
|
|
|
def yesterday(self, stamp: datetime) -> PriceRead:
|
|
return PriceRead.no_data(
|
|
"this caller keeps no price history, so there are no prices from "
|
|
"yesterday to fill in")
|