"""Pricing profile loader, validator, and contract-values checker. A *pricing profile* is a YAML file that describes the **structure** of an energy contract — which fields exist, their units, and which have defaults. Actual pricing values (the numbers the user fills in via the UI) are stored in the DB as ``EnergyContractVersion.values`` (a JSON blob) and must conform to the corresponding profile's structure. Profiles live in ``app/integrations/pricing/profiles/.yaml`` and are located at runtime relative to *this file* (not CWD), matching the pattern established by the Modbus profile loader. Design notes ------------ - **Purely data + functions** — no abstract base classes or inheritance. - Two Pydantic models — ``ManualProfile`` and ``TibberProfile`` — capture the different structures of the two supported kinds. A thin union dispatcher in ``load_profile`` picks the right one. - ``validate_values(kind, values)`` fills in fields that carry a ``default`` and raises ``ProfileValidationError`` for missing required fields or wrong types. - All errors are subclasses of built-ins so callers need not import this module just to catch them. """ from __future__ import annotations import logging from pathlib import Path from typing import Any, Optional import yaml from pydantic import BaseModel, ValidationError, model_validator logger = logging.getLogger(__name__) # Directory containing the YAML profiles, located relative to *this* file. _PROFILES_DIR = Path(__file__).parent / "profiles" # --------------------------------------------------------------------------- # Custom exceptions # --------------------------------------------------------------------------- class ProfileNotFoundError(FileNotFoundError): """Raised when the requested profile YAML file does not exist.""" class ProfileValidationError(ValueError): """Raised when a profile YAML fails Pydantic validation or when contract values do not conform to the profile structure.""" # --------------------------------------------------------------------------- # Leaf-node models (a single pricing field: unit + optional default) # --------------------------------------------------------------------------- class FieldSpec(BaseModel): """Specification for a single numeric pricing field.""" unit: str """Physical / monetary unit string (e.g. ``"EUR/kWh"``, ``"EUR/month"``).""" default: Optional[float] = None """Default value used when the field is absent from contract values. ``None`` means the field is required (no default).""" # --------------------------------------------------------------------------- # ManualProfile — fixed / variable dual-tariff contract structure # --------------------------------------------------------------------------- class ManualBuySpec(BaseModel): normal: FieldSpec # high-tariff buy price (delivered_2 registers) dal: FieldSpec # low-tariff buy price (delivered_1 registers) class ManualSellSpec(BaseModel): normal: FieldSpec # high-tariff sell / return price dal: FieldSpec # low-tariff sell / return price class ManualEnergySpec(BaseModel): dual_tariff: bool # always True for manual profiles buy: ManualBuySpec sell: ManualSellSpec energy_tax: FieldSpec # added to buy price; includes VAT ode: FieldSpec # currently merged into energy_tax; default 0 class ManualStandingSpec(BaseModel): network_fee: FieldSpec # EUR/month management_fee: FieldSpec # EUR/month class ManualCreditsSpec(BaseModel): heffingskorting: FieldSpec # EUR/year — energy-tax credit deducted at summary class ManualProfile(BaseModel): """Complete structure description for a ``manual`` pricing contract.""" kind: str label: str energy: ManualEnergySpec standing: ManualStandingSpec credits: ManualCreditsSpec @model_validator(mode="after") def _check_kind(self) -> "ManualProfile": if self.kind != "manual": raise ValueError(f"ManualProfile requires kind='manual', got {self.kind!r}") return self # --------------------------------------------------------------------------- # TibberProfile — dynamic Tibber API contract structure # --------------------------------------------------------------------------- class TibberEnergySpec(BaseModel): source: str # must be "tibber_api" energy_tax: FieldSpec # subtracted from total to derive sell price sell_fee: FieldSpec # verkoopvergoeding (feed-in fee); always subtracted from sell; default 0.0248 sell_adjust: FieldSpec # additional sell-price adjustment; default 0 class TibberStandingSpec(BaseModel): management_fee: FieldSpec # EUR/month; has a default network_fee: FieldSpec # EUR/month class TibberCreditsSpec(BaseModel): heffingskorting: FieldSpec # EUR/year class TibberProfile(BaseModel): """Complete structure description for a ``tibber`` pricing contract.""" kind: str label: str energy: TibberEnergySpec standing: TibberStandingSpec credits: TibberCreditsSpec @model_validator(mode="after") def _check_kind(self) -> "TibberProfile": if self.kind != "tibber": raise ValueError(f"TibberProfile requires kind='tibber', got {self.kind!r}") return self # A union type for type hints where either profile is acceptable. AnyProfile = ManualProfile | TibberProfile # Map kind → Pydantic model class used for validation. _PROFILE_MODELS: dict[str, type[ManualProfile] | type[TibberProfile]] = { "manual": ManualProfile, "tibber": TibberProfile, } # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- def load_profile(kind: str) -> AnyProfile: """Load and validate a pricing profile by kind name. The profile file is expected at ``app/integrations/pricing/profiles/.yaml``. Parameters ---------- kind: Profile kind without extension (``"manual"`` or ``"tibber"``). Returns ------- ManualProfile | TibberProfile Validated profile model. Raises ------ ProfileNotFoundError If ``profiles/.yaml`` does not exist. ProfileValidationError If the YAML is syntactically valid but fails schema validation. """ path = _PROFILES_DIR / f"{kind}.yaml" if not path.exists(): raise ProfileNotFoundError( f"Pricing profile '{kind}' not found (looked for {path})" ) with path.open("r", encoding="utf-8") as fh: raw = yaml.safe_load(fh) if not isinstance(raw, dict): raise ProfileValidationError( f"Profile '{kind}': expected a YAML mapping, got {type(raw).__name__}" ) # Choose the right Pydantic model based on the ``kind`` field in the YAML. yaml_kind = raw.get("kind", kind) model_cls = _PROFILE_MODELS.get(yaml_kind) if model_cls is None: raise ProfileValidationError( f"Profile '{kind}' has unknown kind={yaml_kind!r}; " f"supported: {list(_PROFILE_MODELS)}" ) try: return model_cls.model_validate(raw) except ValidationError as exc: raise ProfileValidationError( f"Profile '{kind}' failed validation: {exc}" ) from exc def list_profiles() -> list[dict[str, Any]]: """Return structural data for all available pricing profiles. Scans ``app/integrations/pricing/profiles/*.yaml``, loads each, and returns a list of dicts (Pydantic model dumps). Profiles that fail to load are skipped with a warning. Returns ------- list[dict[str, Any]] One entry per valid profile, suitable for JSON serialisation and front-end form rendering. """ results: list[dict[str, Any]] = [] if not _PROFILES_DIR.exists(): return results for path in sorted(_PROFILES_DIR.glob("*.yaml")): kind = path.stem try: profile = load_profile(kind) results.append(profile.model_dump()) except (ProfileNotFoundError, ProfileValidationError, Exception) as exc: logger.warning("Skipping malformed pricing profile '%s': %s", kind, exc) return results # --------------------------------------------------------------------------- # Contract-values validation # --------------------------------------------------------------------------- def _fill_defaults_manual(values: dict[str, Any], profile: ManualProfile) -> dict[str, Any]: """Return a copy of *values* with any ``ode`` default applied if missing.""" filled = dict(values) energy = dict(filled.get("energy", {})) # Apply default for ode (default=0) if absent. if "ode" not in energy and profile.energy.ode.default is not None: energy["ode"] = profile.energy.ode.default filled["energy"] = energy return filled def _fill_defaults_tibber(values: dict[str, Any], profile: TibberProfile) -> dict[str, Any]: """Return a copy of *values* with sell_fee, sell_adjust and management_fee defaults applied.""" filled = dict(values) energy = dict(filled.get("energy", {})) # Apply default for sell_fee (default=0.0248) if absent. if "sell_fee" not in energy and profile.energy.sell_fee.default is not None: energy["sell_fee"] = profile.energy.sell_fee.default # Apply default for sell_adjust (default=0) if absent. if "sell_adjust" not in energy and profile.energy.sell_adjust.default is not None: energy["sell_adjust"] = profile.energy.sell_adjust.default filled["energy"] = energy standing = dict(filled.get("standing", {})) # Apply default for management_fee if absent. if ( "management_fee" not in standing and profile.standing.management_fee.default is not None ): standing["management_fee"] = profile.standing.management_fee.default filled["standing"] = standing return filled def _require_numeric(section: str, key: str, container: dict[str, Any]) -> None: """Assert that *container[key]* exists and is a number; raise ProfileValidationError.""" if key not in container: raise ProfileValidationError( f"Contract values missing required field '{section}.{key}'" ) val = container[key] if not isinstance(val, (int, float)): raise ProfileValidationError( f"Contract values field '{section}.{key}' must be a number, " f"got {type(val).__name__!r}" ) def _validate_manual_values(values: dict[str, Any], profile: ManualProfile) -> dict[str, Any]: """Validate and fill-defaults for manual contract values. Returns the filled values dict on success. Raises ProfileValidationError on missing required fields or wrong types. """ filled = _fill_defaults_manual(values, profile) energy = filled.get("energy", {}) buy = energy.get("buy", {}) sell = energy.get("sell", {}) standing = filled.get("standing", {}) credits = filled.get("credits", {}) # Required energy.buy fields. _require_numeric("energy.buy", "normal", buy) _require_numeric("energy.buy", "dal", buy) # Required energy.sell fields. _require_numeric("energy.sell", "normal", sell) _require_numeric("energy.sell", "dal", sell) # Required energy fields. _require_numeric("energy", "energy_tax", energy) _require_numeric("energy", "ode", energy) # Required standing fields. _require_numeric("standing", "network_fee", standing) _require_numeric("standing", "management_fee", standing) # Required credits fields. _require_numeric("credits", "heffingskorting", credits) return filled def _validate_tibber_values(values: dict[str, Any], profile: TibberProfile) -> dict[str, Any]: """Validate and fill-defaults for tibber contract values. Returns the filled values dict on success. Raises ProfileValidationError on missing required fields or wrong types. """ filled = _fill_defaults_tibber(values, profile) energy = filled.get("energy", {}) standing = filled.get("standing", {}) credits = filled.get("credits", {}) # Required energy fields. _require_numeric("energy", "energy_tax", energy) _require_numeric("energy", "sell_fee", energy) _require_numeric("energy", "sell_adjust", energy) # Required standing fields. _require_numeric("standing", "management_fee", standing) _require_numeric("standing", "network_fee", standing) # Required credits fields. _require_numeric("credits", "heffingskorting", credits) return filled def validate_values(kind: str, values: dict[str, Any]) -> dict[str, Any]: """Validate a contract-values dict against the named profile structure. Fields that carry a ``default`` in the profile (e.g. ``ode``, ``sell_fee``, ``sell_adjust``, tibber ``management_fee``) are silently filled in when absent from *values*. Fields with no default that are absent, or fields whose value is not a number, cause a ``ProfileValidationError``. Parameters ---------- kind: Profile kind (``"manual"`` or ``"tibber"``). values: Contract values dict as stored in ``EnergyContractVersion.values``. Returns ------- dict[str, Any] A (possibly mutated) copy of *values* with defaults applied. Raises ------ ProfileNotFoundError If the profile YAML for *kind* does not exist. ProfileValidationError If required fields are missing or have wrong types. """ profile = load_profile(kind) if isinstance(profile, ManualProfile): return _validate_manual_values(values, profile) if isinstance(profile, TibberProfile): return _validate_tibber_values(values, profile) # Unreachable with current kinds, but guard for future extensions. raise ProfileValidationError(f"No validator implemented for kind={kind!r}")