tliu93/home-automation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 17 Wiki Activity

M8-T05: bind electricity costs to source bindings

Browse Source
This commit is contained in:
tliu93
2026-08-23 21:22:05 +02:00
parent 2e125dbd53
commit 1ea2f659e0
10 changed files with 1057 additions and 304 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/models/energy.py
+16 -12
View File
@@ -13,7 +13,17 @@ from __future__ import annotations
import uuid as _uuid
from datetime import datetime
from sqlalchemy import Boolean, DateTime, Float, ForeignKey, Integer, String, UniqueConstraint, event, text
from sqlalchemy import (
Boolean,
DateTime,
Float,
ForeignKey,
Integer,
String,
UniqueConstraint,
event,
text,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship, synonym
from sqlalchemy.types import JSON
@@ -53,9 +63,7 @@ class Meter(Base):
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
)
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)
@@ -207,15 +215,11 @@ class EnergyContractVersion(Base):
)
# Start of this version's validity window (inclusive, UTC).
effective_from: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False
)
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
)
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).
@@ -320,8 +324,8 @@ class EnergyCostPeriod(Base):
ForeignKey("meter.id", ondelete="RESTRICT"), nullable=True
)
# Nullable while M8 adopts historical DSMR rows. Future normal periods
# will point at the binding that supplied both cumulative endpoints.
# 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
)
app/schemas/energy.py
+8 -3
View File
@@ -95,13 +95,18 @@ class CostPeriodSchema(BaseModel):
export_revenue: float = Field(description="Revenue from electricity fed to grid (EUR).")
net_cost: float = Field(description="import_cost − export_revenue (EUR).")
currency: str = Field(description="ISO 4217 currency code.")
degraded: bool = Field(
description="True when the period was computed with incomplete data."
)
degraded: bool = Field(description="True when the period was computed with incomplete data.")
contract_version_id: int | None = Field(
default=None,
description="FK to the contract version used for this billing period (null when degraded).",
)
source_binding_id: int | None = Field(
default=None,
description=(
"FK to the source binding that supplied both cumulative endpoints "
"(null for legacy or degraded periods)."
),
)
model_config = {"from_attributes": True}
app/services/energy_cost.py
+94 -28
View File
@@ -37,7 +37,7 @@ Design notes
- **Register keys**: DSMR payload uses JSON strings like ``"20915.154"``
for cumulative kWh registers. ``register_at`` converts them to Decimal.
- **Degraded vs skip semantics**:
- *No meter coverage* (``meter_at`` returns None for t0): write a
- *No unique meter coverage* (no sole electricity meter at t0): write a
``degraded=True`` row with ``meter_id=None``.
- *Cross-meter boundary* (m0.id != m1.id for t0/t1): write a ``degraded=True``
row with ``meter_id=m0.id``; losing this one period at the swap boundary is
@@ -67,8 +67,8 @@ Meter-aware compute_period ordering rationale (M7-T03)
The order of checks inside ``compute_period`` is:
1. **Immutability guard** (existing non-degraded row, overwrite=False) → return False.
2. **Meter determination** (m0 = meter_at(t0), m1 = meter_at(t1)):
- No meter (m0 is None) → write degraded, meter_id=None.
2. **Meter determination** (m0/m1 each resolve to one electricity Meter):
- No unique meter (m0 is None) → write degraded, meter_id=None.
- Cross-meter boundary (m0.id != m1.id) → write degraded, meter_id=m0.id.
3. **Active contract version check** → skip (no write) if absent.
4. **Boundary register readings** within m0's window → write degraded if missing.
@@ -98,8 +98,8 @@ from app.integrations.pricing.strategies import (
get_strategy,
)
from app.models.energy import DsmrReading, EnergyCostPeriod, Meter
from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel
from app.services.contracts import active_contract_version_at, active_contract_versions
from app.services.meters import meter_at
from app.services.timezone import local_date, local_now
logger = logging.getLogger(__name__)
@@ -134,10 +134,10 @@ _MAX_DELTA_KWH = Decimal("100")
_SETTLEMENT_OFFSET = timedelta(hours=1, minutes=5)
# DSMR payload register keys (cumulative kWh, JSON string values).
_KEY_D1 = "electricity_delivered_1" # delivered low-tariff (dal / _1)
_KEY_D2 = "electricity_delivered_2" # delivered high-tariff (normal / _2)
_KEY_R1 = "electricity_returned_1" # returned low-tariff
_KEY_R2 = "electricity_returned_2" # returned high-tariff
_KEY_D1 = "electricity_delivered_1" # delivered low-tariff (dal / _1)
_KEY_D2 = "electricity_delivered_2" # delivered high-tariff (normal / _2)
_KEY_R1 = "electricity_returned_1" # returned low-tariff
_KEY_R2 = "electricity_returned_2" # returned high-tariff
# ---------------------------------------------------------------------------
@@ -174,6 +174,23 @@ def _existing_period(session: Session, t0: datetime) -> EnergyCostPeriod | None:
).scalar_one_or_none()
def _unique_electricity_meter_at(session: Session, boundary: datetime) -> Meter | None:
"""Return the sole electricity meter covering *boundary*, if one exists.
Billing must treat overlapping meter epochs as a structural ambiguity rather
than relying on ``meter_at``'s newest-started tie breaker. A cumulative
delta is safe only when exactly one electricity meter covers each endpoint.
"""
candidates = session.execute(
select(Meter).where(
Meter.commodity == "electricity",
Meter.started_at <= boundary,
(Meter.ended_at.is_(None)) | (Meter.ended_at > boundary),
)
).scalars().all()
return candidates[0] if len(candidates) == 1 else None
# ---------------------------------------------------------------------------
# register_at — boundary reading lookup (meter-aware)
# ---------------------------------------------------------------------------
@@ -183,6 +200,8 @@ def register_at(
session: Session,
boundary: datetime,
meter: Meter,
*,
meter_source_id: int | None = None,
) -> dict[str, Decimal] | None:
"""Return the four cumulative kWh register values at *boundary*, within *meter*'s window.
@@ -226,7 +245,7 @@ def register_at(
"""
# Build the meter-window constraints: [started_at, ended_at).
meter_lower = meter.started_at # DsmrReading.recorded_at >= meter.started_at
meter_upper = meter.ended_at # DsmrReading.recorded_at < meter.ended_at (if set)
meter_upper = meter.ended_at # DsmrReading.recorded_at < meter.ended_at (if set)
stmt = (
select(DsmrReading)
@@ -240,6 +259,8 @@ def register_at(
# Apply the upper bound only when the meter is closed (ended_at is not None).
if meter_upper is not None:
stmt = stmt.where(DsmrReading.recorded_at < meter_upper)
if meter_source_id is not None:
stmt = stmt.where(DsmrReading.meter_source_id == meter_source_id)
row: DsmrReading | None = session.execute(stmt).scalar_one_or_none()
@@ -273,6 +294,33 @@ def register_at(
}
def _binding_at(
session: Session, boundary: datetime, meter: Meter
) -> tuple[MeterSourceBinding, int] | None:
"""Resolve the sole DSMR binding for *meter* at one period boundary.
Costing must not infer a cumulative domain from whichever reading happens
to be latest. A binding anchors both the physical meter epoch and its
source stream. Any missing or overlapping binding is therefore
deliberately unresolvable.
"""
candidates = session.execute(
select(MeterSourceBinding, MeterSourceChannel.source_id)
.join(MeterSourceChannel, MeterSourceChannel.id == MeterSourceBinding.channel_id)
.join(MeterSource, MeterSource.id == MeterSourceChannel.source_id)
.where(
MeterSourceBinding.meter_id == meter.id,
MeterSourceBinding.started_at <= boundary,
(MeterSourceBinding.ended_at.is_(None)) | (MeterSourceBinding.ended_at > boundary),
MeterSource.kind == "dsmr_mqtt",
)
).all()
if len(candidates) != 1:
return None
binding, source_id = candidates[0]
return binding, source_id
# ---------------------------------------------------------------------------
# compute_period — single 15-minute period
# ---------------------------------------------------------------------------
@@ -302,9 +350,9 @@ def compute_period(session: Session, t0: datetime, *, overwrite: bool = False) -
Side-effects
------------
- Inserts or updates an ``EnergyCostPeriod`` row keyed on ``period_start=t0``.
- If no meter covers t0 (``meter_at`` returns None for t0): inserts/updates
a degraded row with ``meter_id=None``.
- If the period spans a meter boundary (``meter_at(t0).id != meter_at(t1).id``):
- If no unique meter covers t0: inserts/updates a degraded row with
``meter_id=None``.
- If the period spans a meter boundary (m0.id != m1.id):
inserts/updates a degraded row with ``meter_id=m0.id`` (D5 decision).
- If readings are missing at either boundary within the meter window:
inserts/updates a degraded row with ``meter_id=m0.id``.
@@ -332,7 +380,7 @@ def compute_period(session: Session, t0: datetime, *, overwrite: bool = False) -
# is corrected and a recompute_range is triggered.
#
# Ordering rationale:
# 1. No meter (m0 is None) → degraded(meter_id=None): no epoch for t0.
# 1. No unique meter (m0 is None) → degraded(meter_id=None): no unambiguous epoch for t0.
# 2. Cross-meter boundary (m0.id != m1.id) → degraded(meter_id=m0.id): D5.
# 3. (Single meter, proceed) → contract check → readings → delta guard → price.
#
@@ -341,13 +389,13 @@ def compute_period(session: Session, t0: datetime, *, overwrite: bool = False) -
# first, a missing-contract skip would silently discard the cross-table
# evidence; once a contract is added and recompute runs, the engine would
# incorrectly use cross-table reads.
m0 = meter_at(session, t0)
m1 = meter_at(session, t1)
m0 = _unique_electricity_meter_at(session, t0)
m1 = _unique_electricity_meter_at(session, t1)
if m0 is None:
# No meter epoch covers t0 — degraded with no meter attribution.
# No unambiguous meter epoch covers t0 — degraded with no attribution.
logger.debug(
"compute_period(%s): no active meter at t0 — writing degraded (meter_id=None).",
"compute_period(%s): no unique active meter at t0 — writing degraded (meter_id=None).",
t0.isoformat(),
)
_upsert_degraded(session, t0, now, existing, meter_id=None)
@@ -366,6 +414,16 @@ def compute_period(session: Session, t0: datetime, *, overwrite: bool = False) -
_upsert_degraded(session, t0, now, existing, meter_id=m0.id)
return True
# Both endpoints must resolve to the same binding and source before a
# cumulative subtraction is permitted. This is checked before contract
# lookup so structural inconsistencies remain visible as degraded rows.
bound0 = _binding_at(session, t0, m0)
bound1 = _binding_at(session, t1, m1)
if bound0 is None or bound1 is None or bound0[0].id != bound1[0].id or bound0[1] != bound1[1]:
_upsert_degraded(session, t0, now, existing, meter_id=m0.id)
return True
binding, meter_source_id = bound0
# --- Active contract version at t0 ---
# If there is no active contract covering t0, skip the period entirely.
# We do not write a degraded row — there is no meaningful state to recover
@@ -378,8 +436,8 @@ def compute_period(session: Session, t0: datetime, *, overwrite: bool = False) -
return False
# --- Boundary readings within m0's meter window ---
start_regs = register_at(session, t0, m0)
end_regs = register_at(session, t1, m0)
start_regs = register_at(session, t0, m0, meter_source_id=meter_source_id)
end_regs = register_at(session, t1, m0, meter_source_id=meter_source_id)
if start_regs is None or end_regs is None:
# Missing readings within the meter window → degraded with m0 attribution.
@@ -421,9 +479,7 @@ def compute_period(session: Session, t0: datetime, *, overwrite: bool = False) -
except TibberPriceNotFoundError:
# Missing Tibber price → skip the period; it will be retried once the
# price arrives (e.g. after the next Tibber refresh job runs).
logger.debug(
"compute_period(%s): no Tibber price found — skipping.", t0.isoformat()
)
logger.debug("compute_period(%s): no Tibber price found — skipping.", t0.isoformat())
return False
# --- Upsert the billing record ---
@@ -445,6 +501,7 @@ def compute_period(session: Session, t0: datetime, *, overwrite: bool = False) -
existing.pricing = pricing
existing.contract_version_id = version.id
existing.meter_id = m0.id
existing.source_binding_id = binding.id
existing.degraded = False
existing.computed_at = now
else:
@@ -461,6 +518,7 @@ def compute_period(session: Session, t0: datetime, *, overwrite: bool = False) -
pricing=pricing,
contract_version_id=version.id,
meter_id=m0.id,
source_binding_id=binding.id,
degraded=False,
computed_at=now,
)
@@ -520,6 +578,7 @@ def _upsert_degraded(
existing.pricing = {}
existing.contract_version_id = None
existing.meter_id = meter_id
existing.source_binding_id = None
existing.degraded = True
existing.computed_at = now
else:
@@ -536,6 +595,7 @@ def _upsert_degraded(
pricing={},
contract_version_id=None,
meter_id=meter_id,
source_binding_id=None,
degraded=True,
computed_at=now,
)
@@ -756,12 +816,16 @@ def summarize(session: Session, start: datetime, end: datetime) -> dict[str, Any
end_utc = _as_utc(end)
# --- Fetch all EnergyCostPeriod rows in [start, end) ---
rows = session.execute(
select(EnergyCostPeriod).where(
EnergyCostPeriod.period_start >= start_utc,
EnergyCostPeriod.period_start < end_utc,
rows = (
session.execute(
select(EnergyCostPeriod).where(
EnergyCostPeriod.period_start >= start_utc,
EnergyCostPeriod.period_start < end_utc,
)
)
).scalars().all()
.scalars()
.all()
)
good_rows = [r for r in rows if not r.degraded]
degraded_rows = [r for r in rows if r.degraded]
@@ -863,7 +927,9 @@ def summarize(session: Session, start: datetime, end: datetime) -> dict[str, Any
version_segments: list[tuple[_date, _date | None, dict]] = []
for v in versions:
v_start_local = local_date(_as_utc(v.effective_from))
v_end_local = local_date(_as_utc(v.effective_to)) if v.effective_to is not None else None
v_end_local = (
local_date(_as_utc(v.effective_to)) if v.effective_to is not None else None
)
version_segments.append((v_start_local, v_end_local, v.values or {}))
for v_start, v_end_excl, v_values in version_segments: