glenn schrooyenandClaude
c68fd275cc
G1: remove the false 'Fixed tariff not used' notification. One INFO line at start and an info note beside the setting (EN/NL) when price_service has a non-blank static_profile; the old notification id is dismissed at start. DOCS, migration step 5 and DECISIONS rule 7 now say what the code does. S05: whitespace-only static_profile counts as not set. G2: maintenance_enabled read through MaintConfig.from_options(). Co-Authored-By: Claude <noreply@anthropic.com> Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Bdqr7PiaHcBnzPituu2yTC
229 lines
9.4 KiB
Python
229 lines
9.4 KiB
Python
"""PRICE-01 - where the prices come from when the first source does not answer.
|
|
|
|
A provider that names `FALLBACKS` (app/price.py) is read through this module
|
|
instead of straight from `load_prices()`. The order is fixed:
|
|
|
|
1. the provider itself (the owner's price server)
|
|
2. each id in its `FALLBACKS` (energy-charts.info)
|
|
3. the prices the history store kept from the last fetch that worked
|
|
|
|
A step counts only if its answer holds a price for NOW. A server that answers
|
|
200 with yesterday's prices is as useless to a battery as one that is down, so
|
|
it is skipped, with the same log line.
|
|
|
|
Every other provider never comes here (`uses_chain()` is False for it), so
|
|
`generic_dayahead`, `manual_csv`, `custom_formula` and `off` take the road they
|
|
always took.
|
|
|
|
THE GATE. `gate_for()` is the maintenance charge rule for a site on such a
|
|
provider: the current 15-minute price is at or below today's average. It is the
|
|
rule the Home Assistant pair (`price_now_entity`, `price_avg_entity`) gave, read
|
|
from the provider's own prices, so moving a site from the pair to the provider
|
|
does not change WHEN the charge runs. DECISIONS/PRICE-01-own-price-server.md
|
|
says how equal, and where it is not.
|
|
"""
|
|
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
|
|
from .price import (AUTH_REFUSED, PROVIDERS, PriceRead, brussels,
|
|
price_option, provider_id, scrub)
|
|
|
|
_LOG = logging.getLogger("goodwe.price.chain")
|
|
|
|
# The four places a plan can come from, best first. A higher number is further
|
|
# down the chain.
|
|
STEP_OWN, STEP_PUBLIC, STEP_STORED, STEP_NONE = 0, 1, 2, 3
|
|
|
|
# What a step is called in a status line.
|
|
SOURCE_WORDS = {
|
|
"price_service": "the price service",
|
|
"generic_dayahead": "energy-charts",
|
|
"price_history": "saved prices",
|
|
}
|
|
|
|
|
|
def uses_chain(opts: dict) -> bool:
|
|
"""Does the selected provider ask for the fallback chain?"""
|
|
cls = PROVIDERS.get(provider_id(opts))
|
|
return bool(cls is not None and (cls.FALLBACKS or cls.KEEPS_LAST_GOOD))
|
|
|
|
|
|
def fixed_tariff_beside_service(opts: dict) -> bool:
|
|
"""PRICE-01: a fixed tariff is set beside `price_service`. Not a fault: it
|
|
stays the plan's last fallback (`resolve_tariff`), and only the cheap window
|
|
and the gate ignore it. A blank or whitespace-only value is not set."""
|
|
return (provider_id(opts) == "price_service"
|
|
and bool(str(price_option(opts, "static_profile")).strip()))
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Resolved:
|
|
"""What the chain found. `published` is the series BEFORE the yesterday
|
|
fill and the window, which is what the day's average has to be read from:
|
|
the plan itself starts at the current hour."""
|
|
|
|
read: PriceRead
|
|
step: int
|
|
reasons: tuple = ()
|
|
|
|
@property
|
|
def published(self):
|
|
return self.read.match(series=lambda r: r, no_data=lambda _r: None)
|
|
|
|
|
|
class PriceWatch:
|
|
"""Which step the prices come from, and one log line per change of it.
|
|
|
|
The line is written when the step CHANGES and not before. Nothing repeats
|
|
while a source stays down, and nothing is written for a refresh that
|
|
changes nothing. The reasons (a status code, a timeout) go on a DEBUG line
|
|
beside it, with the token taken out.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.step = None
|
|
self.auth_refused = False
|
|
|
|
def note(self, step: int, reasons=(), opts=None) -> None:
|
|
old, self.step = self.step, step
|
|
self.auth_refused = any(AUTH_REFUSED in r for r in reasons)
|
|
if old == step:
|
|
return
|
|
if reasons:
|
|
_LOG.debug("Price sources said: %s.",
|
|
scrub("; ".join(reasons), opts))
|
|
if step == STEP_OWN:
|
|
if old is None:
|
|
_LOG.info("Prices come from the price service.")
|
|
else:
|
|
_LOG.info("The price service answers again.")
|
|
elif step == STEP_PUBLIC:
|
|
if old is not None and old > step:
|
|
_LOG.info("Prices come from energy-charts again.")
|
|
elif self.auth_refused:
|
|
_LOG.warning("The price service refused the token, so prices "
|
|
"come from energy-charts.")
|
|
else:
|
|
_LOG.warning("The price service does not answer, so prices "
|
|
"come from energy-charts.")
|
|
elif step == STEP_STORED:
|
|
if old is not None and old > step:
|
|
_LOG.info("The add-on uses the prices it saved again.")
|
|
else:
|
|
_LOG.warning("No price source answers, so the add-on uses "
|
|
"the prices it saved.")
|
|
else:
|
|
_LOG.warning("No price source answers, and no saved price covers "
|
|
"now.")
|
|
|
|
|
|
async def resolve(opts: dict, since: datetime, until: datetime, transport,
|
|
history, stamp: datetime, watch: PriceWatch,
|
|
load) -> Resolved:
|
|
"""Walk the chain and return the first step that has a price for `stamp`.
|
|
|
|
`load` is the caller's `load_prices`, passed in so that `static_tariff.py`
|
|
stays the one module that fetches prices. Never raises
|
|
for a source condition: a read that gets nowhere is a `no_data` whose
|
|
reason names every step, so the §11.4 row applies exactly as for a single
|
|
provider that failed.
|
|
"""
|
|
pid = provider_id(opts)
|
|
cls = PROVIDERS[pid]
|
|
reasons = []
|
|
for step, sid in enumerate((pid,) + tuple(cls.FALLBACKS)):
|
|
# A fallback is asked for ITS OWN feed. The operator's feed override is
|
|
# for the first provider and would send energy-charts' request to a
|
|
# URL it was never meant for.
|
|
o = opts if step == 0 else dict(opts, price_provider=sid,
|
|
price_feed_url="")
|
|
read = await load(o, since, until, transport)
|
|
if read.match(series=lambda r: r.covers(stamp), no_data=lambda _r: False):
|
|
# One fallback is shipped. Any further one would also be "public".
|
|
where = STEP_OWN if step == 0 else STEP_PUBLIC
|
|
watch.note(where, reasons, opts)
|
|
return Resolved(read, where, tuple(reasons))
|
|
reasons.append(read.match(
|
|
series=lambda r: f"{sid}: it has no price for now",
|
|
no_data=lambda r: r.reason))
|
|
if cls.KEEPS_LAST_GOOD:
|
|
kept = history.stored(stamp)
|
|
if kept.match(series=lambda r: r.covers(stamp),
|
|
no_data=lambda _r: False):
|
|
watch.note(STEP_STORED, reasons, opts)
|
|
return Resolved(kept, STEP_STORED, tuple(reasons))
|
|
reasons.append(kept.match(
|
|
series=lambda r: "price_history: it has no price for now",
|
|
no_data=lambda r: r.reason))
|
|
watch.note(STEP_NONE, reasons, opts)
|
|
# The reason is short on purpose: it is printed on the status page and in
|
|
# a WARNING every refresh. What each step said is on the DEBUG line.
|
|
return Resolved(PriceRead.no_data(
|
|
"no price source has a price for now"), STEP_NONE,
|
|
tuple(scrub(r, opts) for r in reasons))
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# The maintenance charge gate, read from the provider's prices
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
@dataclass(frozen=True)
|
|
class Gate:
|
|
"""What the 30 s tick takes from the standing plan.
|
|
|
|
`covered` is whether a published price covers now. When it is False,
|
|
`cheap` is False and the charge waits: no price is not a cheap price.
|
|
`since` is when the cover ended, for the status line, or None when nothing
|
|
was ever published.
|
|
"""
|
|
|
|
cheap: bool
|
|
covered: bool
|
|
since: datetime | None
|
|
words: str
|
|
|
|
|
|
NO_PLAN = Gate(False, False, None, "No prices yet")
|
|
|
|
|
|
def _round5(value: float) -> float:
|
|
# The Home Assistant template rounded the average to 5 decimals, and a
|
|
# price to 5 as well. Both sides are compared as it did, so a quarter that
|
|
# sits within 0.00001 EUR/kWh of the average gets the same answer.
|
|
return round(float(value), 5)
|
|
|
|
|
|
def gate_for(published, now: datetime) -> Gate:
|
|
"""The charge gate for `now`, from the series the chain published.
|
|
|
|
THE RULE (owner's decision, PRICE-01): the current 15-minute price is at or
|
|
below the average of today's prices, today being the Brussels day. This is
|
|
what `now_p <= avg_p` was on the Home Assistant pair. It is NOT
|
|
`TariffPlan.cheap_now()`, which is strictly below the mean of the firm
|
|
hours of the whole plan, and it is not meant to be.
|
|
|
|
`published` is a series or None. With no price for now there is no gate
|
|
answer: the charge waits.
|
|
"""
|
|
if published is None:
|
|
return NO_PLAN
|
|
if not published.covers(now):
|
|
return Gate(False, False, published.last_end_before(now),
|
|
"No price for now")
|
|
today = brussels(now).date()
|
|
todays = [p.eur_per_kwh for p in published.points
|
|
if not p.provisional and brussels(p.start).date() == today]
|
|
here = published.at(now)
|
|
if here.provisional or not todays:
|
|
return Gate(False, False, published.last_end_before(now),
|
|
"No price for now")
|
|
avg = sum(todays) / len(todays)
|
|
cheap = _round5(here.eur_per_kwh) <= _round5(avg)
|
|
source = SOURCE_WORDS.get(published.provider, published.provider)
|
|
return Gate(cheap, True, None,
|
|
f"Prices from {source}, {here.eur_per_kwh:.4f} EUR/kWh now, "
|
|
f"{'cheap' if cheap else 'not cheap'} against today's average "
|
|
f"of {avg:.4f}")
|