"""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 here: first the provider (the owner's price server), then each id in `FALLBACKS`, then 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 answering 200 with yesterday's prices is as useless as one that is down). Other providers never come here (`uses_chain()` is False). `gate_for()` is the maintenance charge rule for such a site: the current 15-minute price is at or below today's average, as the Home Assistant pair gave. 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 (nothing repeats while a source stays down). Reasons go on a DEBUG line, token removed.""" 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`, so static_tariff.py stays the one module that fetches. Never raises for a source condition: `no_data` names every step.""" 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 override is for # the first provider. 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) # Short on purpose: it is on the status page and in a WARNING every # refresh. 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` False means `cheap` is False and the charge waits (no price is not a cheap price). `since` is when the cover ended, or None.""" 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 both sides to 5 decimals; so do we, # for 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 current 15-minute price is at or below the average of today's (Brussels day) prices, as `now_p <= avg_p` was on the HA pair (owner, PRICE-01). Not `TariffPlan.cheap_now()`. With no price for now 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}")