"""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, 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)) @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: _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}")