home-automation/app/models/energy.py

"""SQLAlchemy models for the energy pricing and DSMR metering subsystem.

Six tables:
  - meter:                    physical electricity meter lifecycle epoch.
  - dsmr_reading:             raw DSMR telegram blobs (10-second down-sampled).
  - energy_contract:          contract head (manual or tibber, one active at a time).
  - energy_contract_version:  versioned pricing values; append-only for auditability.
  - tibber_price:             cached Tibber 15-minute spot prices (immutable).
  - energy_cost_period:       computed 15-minute billing periods (immutable snapshot).
"""

from __future__ import annotations

import uuid as _uuid
from datetime import datetime, timezone
from decimal import Decimal
from typing import Any
from sqlalchemy import (
    Boolean,
    CheckConstraint,
    DateTime,
    Float,
    ForeignKey,
    Index,
    Integer,
    Numeric,
    String,
    UniqueConstraint,
    event,
    text,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship, synonym, validates
from sqlalchemy.types import JSON, TypeDecorator

from app.db import Base
from app.models.meter_source import MeterSourceBinding


def _uuid4_str() -> str:
    return str(_uuid.uuid4())


def _decimal_json(value: Any) -> Any:
    """Make auditable JSON portable without admitting binary numeric values."""
    if isinstance(value, Decimal):
        return format(value, "f")
    if isinstance(value, float) or (isinstance(value, int) and not isinstance(value, bool)):
        raise ValueError("JSON amounts and quantities must be decimal strings, not numeric JSON values")
    if isinstance(value, dict):
        return {key: _decimal_json(child) for key, child in value.items()}
    if isinstance(value, list):
        return [_decimal_json(child) for child in value]
    return value


def _validate_fixed_decimal(value: Decimal, precision: int, scale: int, field: str) -> Decimal:
    if not isinstance(value, Decimal):
        raise ValueError(f"{field} must be a Decimal, not a binary float or other numeric type")
    if not value.is_finite():
        raise ValueError(f"{field} must be finite")
    if -value.as_tuple().exponent > scale:
        raise ValueError(f"{field} exceeds scale {scale}")
    integer_digits = 0 if value.is_zero() else max(value.copy_abs().adjusted() + 1, 0)
    if integer_digits > precision - scale:
        raise ValueError(f"{field} exceeds precision {precision},{scale}")
    return value


class ExactDecimal(TypeDecorator[Decimal]):
    """Fixed-point Decimal which uses SQLite TEXT, never a binary float."""

    impl = Numeric
    cache_ok = True

    def __init__(self, precision: int, scale: int) -> None:
        self.precision = precision
        self.scale = scale
        super().__init__(precision=precision, scale=scale)

    def load_dialect_impl(self, dialect):
        if dialect.name == "sqlite":
            return dialect.type_descriptor(String(self.precision + 2))
        return dialect.type_descriptor(Numeric(self.precision, self.scale, asdecimal=True))

    def process_bind_param(self, value: Decimal | None, dialect) -> Decimal | str | None:
        if value is None:
            return None
        value = _validate_fixed_decimal(value, self.precision, self.scale, "decimal value")
        if dialect.name == "sqlite":
            return format(value, f".{self.scale}f")
        return value

    def process_result_value(self, value: Decimal | str | None, _dialect) -> Decimal | None:
        return None if value is None else Decimal(value)


class DecimalJSON(TypeDecorator[dict]):
    """JSON which serializes Decimal values as strings on every write path."""

    impl = JSON
    cache_ok = True

    def process_bind_param(self, value: Any, _dialect) -> Any:
        return None if value is None else _decimal_json(value)


class UTCDateTime(TypeDecorator[datetime]):
    """UTC timestamps that preserve instant identity on SQLite and other dialects."""

    impl = DateTime(timezone=True)
    cache_ok = True

    def __init__(self, field: str) -> None:
        self.field = field
        super().__init__()

    def process_bind_param(self, value: datetime | None, _dialect) -> datetime | None:
        return None if value is None else _normalise_utc_period(value, self.field)

    def process_result_value(self, value: datetime | None, _dialect) -> datetime | None:
        if value is None:
            return None
        if value.tzinfo is None or value.utcoffset() is None:
            return value.replace(tzinfo=timezone.utc)
        return value.astimezone(timezone.utc)


def _require_aware_period(start: datetime, end: datetime) -> None:
    if start.tzinfo is None or start.utcoffset() is None:
        raise ValueError("period_start must be timezone-aware")
    if end.tzinfo is None or end.utcoffset() is None:
        raise ValueError("period_end must be timezone-aware")
    if end <= start:
        raise ValueError("period_end must be after period_start")


def _normalise_utc_period(value: datetime, field: str) -> datetime:
    if value.tzinfo is None or value.utcoffset() is None:
        raise ValueError(f"{field} must be timezone-aware")
    return value.astimezone(timezone.utc)


class Meter(Base):
    """One physical electricity meter's installation epoch.

    A ``meter`` record represents the period ``[started_at, ended_at)`` during
    which a particular physical meter was installed and active.  Replacing a meter
    (swap, home move, etc.) is modelled by closing the current record
    (``ended_at = swap_timestamp``) and opening a new one
    (``started_at = swap_timestamp``).

    **Invariant**: for each ``commodity`` there is at most one active meter
    (``ended_at IS NULL``) at any point in time.  The service layer enforces
    this — no DB-level constraint is added to keep the migration simple and to
    allow the application to return a meaningful error message.

    ``commodity`` defaults to ``"electricity"``; the field is a free-form string
    (no CHECK constraint) so future commodities (``gas``, ``heating``) can be
    added without a schema change.

    ``reason`` captures why this epoch started — one of ``initial``,
    ``meter_swap``, ``home_move``, or ``other`` — stored as a plain string so
    the application layer controls the allowed set.
    """

    __tablename__ = "meter"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)

    # Stable internal identity — used as HA Discovery unique_id anchor.
    uuid: Mapped[str] = mapped_column(String(36), unique=True, nullable=False, default=_uuid4_str)

    # Human-readable label for this physical meter (e.g. address, serial, tariff zone).
    label: Mapped[str] = mapped_column(String(255), nullable=False)

    # Energy commodity this meter measures.  Defaults to "electricity".
    commodity: Mapped[str] = mapped_column(String(32), nullable=False, default="electricity")

    # UTC timestamp when this meter epoch starts (inclusive).  May be in the past
    # (retroactive declaration); effective billing start = max(started_at, data start).
    started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)

    # UTC timestamp when this meter epoch ends (exclusive).  NULL = currently active.
    ended_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)

    # Why this epoch was created.  Application-layer validation enforces the
    # allowed set; no CHECK constraint to keep migrations simple.
    reason: Mapped[str] = mapped_column(String(64), nullable=False)

    # Free-form note (e.g. location, physical meter id, reason details).
    note: Mapped[str | None] = mapped_column(String(1024), nullable=True)

    # UTC timestamp of when this row was created.
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)

    # Relationship to cost periods attributed to this meter epoch (not loaded eagerly).
    cost_periods: Mapped[list["EnergyCostPeriod"]] = relationship(
        back_populates="meter", cascade="save-update, merge"
    )

    source_bindings: Mapped[list["MeterSourceBinding"]] = relationship(
        back_populates="meter", cascade="save-update, merge"
    )


class DsmrReading(Base):
    """One down-sampled DSMR telegram stored as a full JSON blob.

    Identity & idempotency are **independent of the DSMR Reader's telegram id**
    (that field overflows and must be manually reset to zero — a known DSMR
    quirk — so relying on it for uniqueness risks silently dropping new data).
    The table's own autoincrement ``id`` PK is the stable internal identity, and
    ``(meter_source_id, recorded_at)`` is the UNIQUE de-duplication key: each
    configured P1 source emits at most one telegram per timestamp, while
    different sources may legitimately emit at the same instant.

    ``recorded_at`` is a real column (not inside the payload) so time-range
    queries are efficient.  The entire telegram frame is stored verbatim in
    ``payload``; no field allow-list is applied so future commodities (gas,
    heating, three-phase) are accommodated without a schema change.
    """

    __tablename__ = "dsmr_reading"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)

    # UTC timestamp of the sample.  Idempotency is per configured source, so
    # distinct P1 sources may legitimately emit at the same instant.
    recorded_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)

    # Telegram's own id (DSMR Reader assigns it).  Stored only as a reference /
    # debugging aid — NOT used for uniqueness or idempotency (it overflows and
    # gets reset to zero).  Nullable because some DSMR sources may not emit one.
    telegram_id: Mapped[int | None] = mapped_column(Integer, nullable=True)

    # Compatibility for the pre-M8 ingest implementation.  This is an ORM
    # alias only; the physical database column is ``telegram_id``.
    source_id = synonym("telegram_id")

    # The configured source is the durable identity of the cumulative reading
    # stream.  It is non-null after the revision-16 historical adoption.
    meter_source_id: Mapped[int] = mapped_column(
        ForeignKey("meter_source.id", ondelete="RESTRICT"), nullable=False, index=True
    )

    # Full telegram frame as a JSON object; values are typically JSON strings
    # (e.g. "20915.154") — callers must cast to Decimal before arithmetic.
    payload: Mapped[dict] = mapped_column(JSON, nullable=False)

    __table_args__ = (
        UniqueConstraint(
            "meter_source_id", "recorded_at", name="uq_dsmr_reading_source_recorded_at"
        ),
    )


@event.listens_for(DsmrReading, "before_insert")
def _supply_legacy_dsmr_source(_mapper, connection, target: DsmrReading) -> None:
    """Keep the pre-T04 single-source writer working during the schema handoff."""
    if target.meter_source_id is None:
        target.meter_source_id = connection.execute(
            text("SELECT id FROM meter_source WHERE kind = 'dsmr_mqtt' ORDER BY id LIMIT 1")
        ).scalar_one()


class EnergyContract(Base):
    """Contract head: a named energy contract with a chosen pricing strategy.

    ``kind`` determines which price strategy is used (``manual`` for fixed
    dual-tariff rates entered by the user, ``tibber`` for dynamic API prices).
    A contract belongs to an energy ``scope`` (currently electricity; thermal
    profiles are reserved for the next milestone). Only one contract may be
    ``active`` per scope; the service layer enforces mutual exclusion. Specific
    pricing values live in ``EnergyContractVersion``
    so that price changes can be tracked without modifying historical records.
    """

    __tablename__ = "energy_contract"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)

    # Human-readable label; freely editable by the user.
    name: Mapped[str] = mapped_column(String(255), nullable=False)

    # Strategy selector: "manual" or "tibber".  Application-layer validation
    # enforces the allowed set; no DB CHECK constraint is added to keep the
    # migration simple and the strategy registry extensible.
    kind: Mapped[str] = mapped_column(String(32), nullable=False)

    # Billing domain. The service registry derives this from ``kind`` so API
    # callers cannot move a pricing strategy into an incompatible domain.
    scope: Mapped[str] = mapped_column(
        String(32), nullable=False, default="electricity", index=True
    )

    # Whether this is the currently active contract (at most one should be True).
    active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)

    # ISO 4217 currency code for all monetary values in this contract.
    currency: Mapped[str] = mapped_column(String(8), nullable=False, default="EUR")

    # Audit timestamps.
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)

    # Relationship to versions (back-reference; not loaded eagerly).
    versions: Mapped[list["EnergyContractVersion"]] = relationship(
        back_populates="contract", cascade="save-update, merge"
    )


class EnergyContractVersion(Base):
    """One time-bounded version of an energy contract's pricing values.

    Pricing changes are modelled as new versions (append-only); existing versions
    are never modified so that historical ``EnergyCostPeriod`` records remain
    fully auditable.  ``effective_to`` is ``NULL`` for the currently open version.
    """

    __tablename__ = "energy_contract_version"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)

    # FK to the parent contract.  RESTRICT prevents deletion of a contract that
    # still has versioned pricing rows attached to it.
    contract_id: Mapped[int] = mapped_column(
        ForeignKey("energy_contract.id", ondelete="RESTRICT"), nullable=False
    )

    # Start of this version's validity window (inclusive, UTC).
    effective_from: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)

    # End of this version's validity window (exclusive, UTC).  NULL means open-ended
    # (i.e. this is the most recent / current version).
    effective_to: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)

    # Pricing values as a JSON object conforming to the profile structure for
    # ``contract.kind`` (validated by the application layer against the YAML profile).
    values: Mapped[dict] = mapped_column(JSON, nullable=False)

    # Creation timestamp.
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)

    # Relationship back to the parent contract.
    contract: Mapped["EnergyContract"] = relationship(back_populates="versions")

    # Relationship to cost periods that reference this version.
    cost_periods: Mapped[list["EnergyCostPeriod"]] = relationship(
        back_populates="contract_version", cascade="save-update, merge"
    )


class TibberPrice(Base):
    """Cached Tibber 15-minute spot price point (immutable once fetched).

    ``starts_at`` is unique so that upserts are idempotent.  Past prices are
    never overwritten; the fetch job only adds rows for future time slots.
    """

    __tablename__ = "tibber_price"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)

    # UTC start of the 15-minute slot; unique so upsert is idempotent.
    starts_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, unique=True
    )

    # Resolution label as returned by the Tibber API (e.g. "QUARTER_HOURLY").
    resolution: Mapped[str] = mapped_column(String(32), nullable=False)

    # Price components in the contract currency (all include VAT, user-facing).
    energy: Mapped[float] = mapped_column(Float, nullable=False)
    tax: Mapped[float] = mapped_column(Float, nullable=False)
    total: Mapped[float] = mapped_column(Float, nullable=False)

    # Tibber price level (e.g. "NORMAL", "CHEAP", "EXPENSIVE"); may be absent.
    level: Mapped[str | None] = mapped_column(String(32), nullable=True)

    # ISO 4217 currency code as returned by the API.
    currency: Mapped[str] = mapped_column(String(8), nullable=False)

    # UTC timestamp of when this row was fetched/inserted.
    fetched_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)


class EnergyCostPeriod(Base):
    """Computed billing record for one 15-minute metering period (immutable snapshot).

    Each row captures the per-register kWh deltas, the resulting import cost and
    export revenue, and a full snapshot of the pricing values used so that the
    calculation is fully auditable and reproducible without re-querying the
    contract version.  Rows are written once and never modified; explicit
    recomputation via the API is the only way to overwrite a period.
    """

    __tablename__ = "energy_cost_period"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)

    # UTC start of the 15-minute period; unique so upsert is idempotent.
    period_start: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, unique=True
    )

    # Per-register kWh deltas for the period (end minus start of cumulative registers).
    # _1 = dal/low-tariff, _2 = normal/high-tariff (NL convention).
    d1_kwh: Mapped[float] = mapped_column(Float, nullable=False)  # delivered low
    d2_kwh: Mapped[float] = mapped_column(Float, nullable=False)  # delivered high
    r1_kwh: Mapped[float] = mapped_column(Float, nullable=False)  # returned low
    r2_kwh: Mapped[float] = mapped_column(Float, nullable=False)  # returned high

    # Computed monetary amounts for the period (in ``currency``).
    import_cost: Mapped[float] = mapped_column(Float, nullable=False)
    export_revenue: Mapped[float] = mapped_column(Float, nullable=False)
    net_cost: Mapped[float] = mapped_column(Float, nullable=False)

    # ISO 4217 currency code matching the contract.
    currency: Mapped[str] = mapped_column(String(8), nullable=False)

    # Full snapshot of the pricing inputs used during computation.  This makes
    # each row self-contained and auditable even if the contract is later changed.
    pricing: Mapped[dict] = mapped_column(JSON, nullable=False)

    # FK to the exact contract version whose values were used.  RESTRICT prevents
    # deletion of a version that has cost records attached.  Nullable to support
    # periods computed in ``degraded`` mode (missing price data).
    contract_version_id: Mapped[int | None] = mapped_column(
        ForeignKey("energy_contract_version.id", ondelete="RESTRICT"), nullable=True
    )

    # FK to the meter epoch this period belongs to.  RESTRICT prevents deletion of
    # a meter that still has attributed cost periods.  Nullable for backwards
    # compatibility (pre-M7 rows) and degraded periods where the meter was not
    # determinable.
    meter_id: Mapped[int | None] = mapped_column(
        ForeignKey("meter.id", ondelete="RESTRICT"), nullable=True
    )

    # Nullable for historical and degraded rows.  Every new normal period
    # points at the one binding that supplied both cumulative endpoints.
    source_binding_id: Mapped[int | None] = mapped_column(
        ForeignKey("meter_source_binding.id", ondelete="RESTRICT"), nullable=True
    )

    # True when the period was computed with incomplete data (missing readings or
    # missing price); serves as a flag for later recomputation.
    degraded: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)

    # UTC timestamp of when this row was computed/inserted.
    computed_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)

    # Relationship back to the contract version.
    contract_version: Mapped["EnergyContractVersion | None"] = relationship(
        back_populates="cost_periods"
    )

    # Relationship back to the meter epoch.
    meter: Mapped["Meter | None"] = relationship(back_populates="cost_periods")

    source_binding: Mapped["MeterSourceBinding | None"] = relationship(
        back_populates="cost_periods"
    )


class MeterCostPeriod(Base):
    """Auditable commodity-scoped ledger row for one half-open metering period.

    A degraded row intentionally permits missing audit links; services must
    still require them before writing a normal row.
    """

    __tablename__ = "meter_cost_period"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    commodity: Mapped[str] = mapped_column(String(32), nullable=False)
    period_start: Mapped[datetime] = mapped_column(UTCDateTime("period_start"), nullable=False)
    period_end: Mapped[datetime] = mapped_column(UTCDateTime("period_end"), nullable=False)
    meter_id: Mapped[int | None] = mapped_column(
        ForeignKey("meter.id", ondelete="RESTRICT"), nullable=True
    )
    source_binding_id: Mapped[int | None] = mapped_column(
        ForeignKey("meter_source_binding.id", ondelete="RESTRICT"), nullable=True
    )
    contract_version_id: Mapped[int | None] = mapped_column(
        ForeignKey("energy_contract_version.id", ondelete="RESTRICT"), nullable=True
    )

    # SQLite reliably round-trips at most fifteen significant decimal digits.
    # WarmteLink itself reports 0.001 units, so nine cost fractional digits
    # retain a six-place tariff times that source precision without float loss.
    quantity: Mapped[Decimal] = mapped_column(ExactDecimal(15, 6), nullable=False)
    cost: Mapped[Decimal] = mapped_column(ExactDecimal(15, 9), nullable=False)
    currency: Mapped[str] = mapped_column(String(8), nullable=False)
    cost_breakdown: Mapped[dict] = mapped_column(DecimalJSON(), nullable=False, default=dict)
    pricing_snapshot: Mapped[dict] = mapped_column(DecimalJSON(), nullable=False, default=dict)
    quality: Mapped[str] = mapped_column(String(32), nullable=False, default="valid")
    degraded: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
    degraded_reason: Mapped[str | None] = mapped_column(String(255), nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)

    meter: Mapped["Meter | None"] = relationship()
    source_binding: Mapped["MeterSourceBinding | None"] = relationship()
    contract_version: Mapped["EnergyContractVersion | None"] = relationship()

    __table_args__ = (
        CheckConstraint(
            "degraded OR (meter_id IS NOT NULL AND source_binding_id IS NOT NULL "
            "AND contract_version_id IS NOT NULL)",
            name="ck_meter_cost_period_normal_audit_links",
        ),
        CheckConstraint("period_end > period_start", name="ck_meter_cost_period_positive_interval"),
        CheckConstraint(
            "NOT degraded OR (degraded_reason IS NOT NULL AND length(trim(degraded_reason)) > 0)",
            name="ck_meter_cost_period_degraded_reason",
        ),
        UniqueConstraint("commodity", "period_start", name="uq_meter_cost_period_commodity_start"),
        Index("ix_meter_cost_period_commodity_start", "commodity", "period_start"),
        Index("ix_meter_cost_period_source_binding_id", "source_binding_id"),
    )

    @validates("cost_breakdown", "pricing_snapshot")
    def _validate_decimal_json(self, _key: str, value: dict) -> dict:
        return _decimal_json(value)

    @validates("quantity", "cost")
    def _validate_fixed_decimal(self, key: str, value: Decimal) -> Decimal:
        precision, scale = (15, 6) if key == "quantity" else (15, 9)
        return _validate_fixed_decimal(value, precision, scale, key)

    @validates("period_start", "period_end")
    def _normalise_period(self, key: str, value: datetime) -> datetime:
        return _normalise_utc_period(value, key)


@event.listens_for(MeterCostPeriod, "before_insert")
@event.listens_for(MeterCostPeriod, "before_update")
def _validate_meter_cost_period(_mapper, _connection, target: MeterCostPeriod) -> None:
    _require_aware_period(target.period_start, target.period_end)
    if not target.degraded and (
        target.meter_id is None
        or target.source_binding_id is None
        or target.contract_version_id is None
    ):
        raise ValueError("normal meter cost periods require meter, binding, and contract version")
    if target.degraded and not target.degraded_reason:
        raise ValueError("degraded meter cost periods require a degraded_reason")


# Index on recorded_at for efficient time-range queries on DSMR readings.
# (The ORM-level index=True on recorded_at already creates ix_dsmr_reading_recorded_at;
# no composite index is needed for single-meter deployments.)

# Index on period_start is covered by the unique constraint (SQLite creates an
# implicit index for UNIQUE columns), so no additional index is required.

# Index on starts_at for TibberPrice is covered by the unique constraint similarly.