39 lines
1.5 KiB
Python
39 lines
1.5 KiB
Python
from __future__ import annotations
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timedelta
|
|
from typing import Callable, Mapping, Optional
|
|
from EnergyPrice import EnergyPriceBase
|
|
|
|
PriceLookup = Callable[[datetime], Optional[float]]
|
|
|
|
@dataclass
|
|
class RDNProvider(EnergyPriceBase):
|
|
"""
|
|
Skeleton for market-based energy cost (RDN).
|
|
Choose ONE feeding strategy:
|
|
1) price_lookup: callable(ts_local) -> PLN/kWh (net) or None
|
|
2) mapping: dict/Mapping[datetime->PLN/kWh] with tz-aware keys (Europe/Warsaw)
|
|
Optional: period controls how you round/align (default: 1h).
|
|
"""
|
|
price_lookup: PriceLookup | None = None
|
|
mapping: Mapping[datetime, float] | None = None
|
|
period: timedelta = timedelta(hours=1)
|
|
|
|
def _get_price(self, ts_local: datetime) -> float:
|
|
if self.price_lookup:
|
|
val = self.price_lookup(ts_local)
|
|
if val is None:
|
|
raise KeyError(f"No RDN price for {ts_local.isoformat()}")
|
|
return float(val)
|
|
if self.mapping is not None:
|
|
# align to exact period start (e.g., hour start)
|
|
aligned = ts_local.replace(minute=0, second=0, microsecond=0)
|
|
if aligned in self.mapping:
|
|
return float(self.mapping[aligned])
|
|
raise KeyError(f"No RDN price in mapping for {aligned.isoformat()}")
|
|
raise RuntimeError("RDNProvider needs price_lookup or mapping")
|
|
|
|
def rate(self, ts: datetime) -> float:
|
|
dt = self.to_local_dt(ts)
|
|
return self._get_price(dt)
|