diff --git a/app/main.py b/app/main.py index f4e2b3c..aeb6edd 100644 --- a/app/main.py +++ b/app/main.py @@ -2,6 +2,7 @@ import logging import os from contextlib import asynccontextmanager from pathlib import Path +from typing import Callable from fastapi import FastAPI, HTTPException, Request from fastapi.responses import FileResponse @@ -38,6 +39,7 @@ from app.services.modbus_poll import poll_all_enabled_devices, BASE_POLL_TICK_SE from app.services.ha_discovery import publish_discovery, publish_states from app.services.tibber_prices import refresh_prices from app.services.energy_cost import compute_closed_periods +from app.services.meter_cost import compute_closed_periods as compute_closed_meter_cost_periods from app.services.warmtelink_worker import warmtelink_worker_manager from app.services.timezone import local_tz from scripts.app_db_adopt import AppDatabaseAdoptionError, validate_app_runtime_db @@ -112,22 +114,44 @@ def _run_scheduled_energy_cost() -> None: does not crash the scheduler or affect the other background jobs. """ session_local = get_session_local() - session: Session = session_local() - try: - compute_closed_periods(session) - # After billing periods are computed, push fresh energy-cost state values - # to MQTT/HA. publish_states is internally guarded by _should_publish - # (MQTT disabled / not connected → no-op), so this never raises due to - # unconfigured MQTT and does not block the billing job. + + def run_scope(label: str, operation: Callable[[Session], None]) -> None: + """Run one best-effort scope in an isolated transaction/session.""" + session: Session | None = None try: - from app.services.ha_discovery import publish_states - publish_states(session) + session = session_local() + operation(session) except Exception: - logger.exception("_run_scheduled_energy_cost: publish_states failed (non-fatal)") - except Exception: - logger.exception("_run_scheduled_energy_cost: unexpected error") - finally: - session.close() + logger.exception("_run_scheduled_energy_cost: %s failed", label) + if session is not None: + try: + session.rollback() + except Exception: + # A failed cleanup must not replace the operation/factory + # error or prevent the following independent scope. + logger.exception("_run_scheduled_energy_cost: %s rollback failed", label) + finally: + if session is not None: + try: + session.close() + except Exception: + # Sessions are intentionally isolated; close failures are + # diagnostic only and must remain best-effort too. + logger.exception("_run_scheduled_energy_cost: %s close failed", label) + + # Electricity, thermal and HA publishing must not share failed transaction + # state or accidentally commit each other's partially-flushed changes. + run_scope("electricity computation", compute_closed_periods) + run_scope("thermal computation", compute_closed_meter_cost_periods) + + def publish(session: Session) -> None: + # publish_states is internally guarded by _should_publish (MQTT + # disabled / disconnected -> no-op), but gets a clean Session anyway. + from app.services.ha_discovery import publish_states + + publish_states(session) + + run_scope("publish_states (non-fatal)", publish) def _run_scheduled_ha_state_publish() -> None: diff --git a/app/services/meter_cost.py b/app/services/meter_cost.py new file mode 100644 index 0000000..010ce87 --- /dev/null +++ b/app/services/meter_cost.py @@ -0,0 +1,288 @@ +"""Thermal (WarmteLink) 15-minute cost ledger. + +This module deliberately does not share the electricity ledger: thermal has +two independently-bound cumulative domains and Decimal database columns. +""" + +from __future__ import annotations + +import logging +from datetime import UTC, date, datetime, time, timedelta +from decimal import Decimal +from typing import Any + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.models.energy import Meter, MeterCostPeriod +from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel, WarmteLinkReading +from app.services.contracts import active_contract_version_at, active_contract_versions +from app.services.energy_cost import floor_to_quarter +from app.services import timezone as timezone_service + +logger = logging.getLogger(__name__) + +_PERIOD = timedelta(minutes=15) +_FRESHNESS = timedelta(seconds=120) +_LIMITS = {"heating": Decimal("0.1"), "hot_water": Decimal("1")} +_ACCEPTED_QUALITIES = {"valid", "unverifiable"} +_SETTLEMENT_TIME = time(1, 5) + + +def _utc(value: datetime) -> datetime: + return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC) + + +def _decimal(value: Any) -> Decimal: + return value if isinstance(value, Decimal) else Decimal(str(value)) + + +def _existing(session: Session, commodity: str, start: datetime) -> MeterCostPeriod | None: + return session.execute( + select(MeterCostPeriod).where( + MeterCostPeriod.commodity == commodity, MeterCostPeriod.period_start == start + ) + ).scalar_one_or_none() + + +def _meter_at(session: Session, commodity: str, instant: datetime) -> Meter | None: + meters = session.execute( + select(Meter).where( + Meter.commodity == commodity, + Meter.started_at <= instant, + (Meter.ended_at.is_(None)) | (Meter.ended_at > instant), + ) + ).scalars().all() + return meters[0] if len(meters) == 1 else None + + +def _binding_at( + session: Session, meter: Meter, instant: datetime +) -> tuple[MeterSourceBinding, MeterSourceChannel] | None: + rows = session.execute( + select(MeterSourceBinding, MeterSourceChannel) + .join(MeterSourceChannel, MeterSourceChannel.id == MeterSourceBinding.channel_id) + .join(MeterSource, MeterSource.id == MeterSourceChannel.source_id) + .where( + MeterSourceBinding.meter_id == meter.id, + MeterSourceBinding.started_at <= instant, + (MeterSourceBinding.ended_at.is_(None)) | (MeterSourceBinding.ended_at > instant), + MeterSource.kind == "warmtelink_serial", + ) + ).all() + return rows[0] if len(rows) == 1 else None + + +def _reading_at( + session: Session, + channel: MeterSourceChannel, + meter: Meter, + binding: MeterSourceBinding, + target: datetime, +) -> WarmteLinkReading | None: + """Choose the nearest accepted reading inside this cumulative domain. + + Freshness alone is insufficient: a frame immediately before a meter or + source hand-off belongs to a different cumulative register and must never + be used as the other side of a delta. + """ + window_start, window_end = target - _FRESHNESS, target + _FRESHNESS + rows = session.execute( + select(WarmteLinkReading) + .where( + WarmteLinkReading.channel_id == channel.id, + WarmteLinkReading.recorded_at >= window_start, + WarmteLinkReading.recorded_at <= window_end, + ) + ).scalars().all() + domain_start = max(_utc(meter.started_at), _utc(binding.started_at)) + domain_ends = (meter.ended_at, binding.ended_at) + domain_end = min((_utc(value) for value in domain_ends if value is not None), default=None) + accepted = [ + row for row in rows + if row.quality in _ACCEPTED_QUALITIES + and _utc(row.recorded_at) >= domain_start + and (domain_end is None or _utc(row.recorded_at) < domain_end) + ] + if not accepted: + return None + return min( + accepted, + key=lambda row: (abs((_utc(row.recorded_at) - target).total_seconds()), _utc(row.recorded_at)), + ) + + +def _degrade( + session: Session, + commodity: str, + start: datetime, + end: datetime, + existing: MeterCostPeriod | None, + reason: str, + meter_id: int | None = None, + binding_id: int | None = None, +) -> None: + now = datetime.now(UTC) + fields = dict( + period_end=end, meter_id=meter_id, source_binding_id=binding_id, + contract_version_id=None, quantity=Decimal("0"), cost=Decimal("0"), currency="EUR", + cost_breakdown={}, pricing_snapshot={}, quality="invalid", degraded=True, + degraded_reason=reason, updated_at=now, + ) + if existing is None: + session.add(MeterCostPeriod(commodity=commodity, period_start=start, created_at=now, **fields)) + else: + for key, value in fields.items(): + setattr(existing, key, value) + + +def compute_period( + session: Session, commodity: str, period_start: datetime, *, overwrite: bool = False +) -> bool: + """Compute one closed thermal period, recording every unsafe input as degraded.""" + if commodity not in _LIMITS: + raise ValueError("commodity must be heating or hot_water") + start = floor_to_quarter(_utc(period_start)) + end = start + _PERIOD + existing = _existing(session, commodity, start) + if existing is not None and not existing.degraded and not overwrite: + return False + + meter0, meter1 = _meter_at(session, commodity, start), _meter_at(session, commodity, end) + if meter0 is None: + _degrade(session, commodity, start, end, existing, "missing_or_ambiguous_meter") + return True + if meter1 is None or meter1.id != meter0.id: + _degrade(session, commodity, start, end, existing, "cross_meter_epoch", meter0.id) + return True + bound0, bound1 = _binding_at(session, meter0, start), _binding_at(session, meter1, end) + if bound0 is None or bound1 is None: + _degrade(session, commodity, start, end, existing, "missing_or_ambiguous_binding", meter0.id) + return True + binding, channel = bound0 + if bound1[0].id != binding.id or bound1[1].id != channel.id: + _degrade(session, commodity, start, end, existing, "cross_source_binding", meter0.id, binding.id) + return True + first = _reading_at(session, channel, meter0, binding, start) + last = _reading_at(session, channel, meter0, binding, end) + if first is None or last is None: + _degrade(session, commodity, start, end, existing, "missing_stale_or_invalid_reading", meter0.id, binding.id) + return True + delta = _decimal(last.value) - _decimal(first.value) + if delta < 0: + _degrade(session, commodity, start, end, existing, "negative_delta", meter0.id, binding.id) + return True + if delta > _LIMITS[commodity]: + _degrade(session, commodity, start, end, existing, "delta_limit_exceeded", meter0.id, binding.id) + return True + version = active_contract_version_at(session, start, scope="thermal") + if version is None: + _degrade(session, commodity, start, end, existing, "missing_contract", meter0.id, binding.id) + return True + + values = {key: _decimal(value) for key, value in version.values["variable"].items()} + if commodity == "heating": + breakdown = {"heating": delta * values["heating"]} + else: + breakdown = { + key: delta * values[key] + for key in ("hot_water_heating", "hot_water", "hot_water_tax") + } + cost = sum(breakdown.values(), Decimal("0")) + now = datetime.now(UTC) + fields = dict( + period_end=end, meter_id=meter0.id, source_binding_id=binding.id, + contract_version_id=version.id, quantity=delta, cost=cost, currency=version.contract.currency, + cost_breakdown=breakdown, pricing_snapshot=dict(version.values), + quality="valid" if first.quality == last.quality == "valid" else "unverifiable", + degraded=False, degraded_reason=None, updated_at=now, + ) + if existing is None: + session.add(MeterCostPeriod(commodity=commodity, period_start=start, created_at=now, **fields)) + else: + for key, value in fields.items(): + setattr(existing, key, value) + return True + + +def compute_closed_periods(session: Session, *, now: datetime | None = None) -> int: + """Retry incomplete thermal rows and fill recent closed periods without touching good rows.""" + now = _utc(now or datetime.now(UTC)) + first = floor_to_quarter(now - timedelta(days=7)) + written = 0 + cursor = first + while cursor + _PERIOD <= now: + for commodity in ("heating", "hot_water"): + if compute_period(session, commodity, cursor): + written += 1 + cursor += _PERIOD + session.commit() + return written + + +def recompute_range(session: Session, start: datetime, end: datetime) -> int: + cursor, end = floor_to_quarter(_utc(start)), _utc(end) + now, written = datetime.now(UTC), 0 + while cursor < end: + if cursor + _PERIOD <= now: + for commodity in ("heating", "hot_water"): + if compute_period(session, commodity, cursor, overwrite=True): + written += 1 + cursor += _PERIOD + session.commit() + return written + + +def _settled_end_date(now: datetime) -> date: + local_now = timezone_service.to_local(now) + return local_now.date() if local_now.timetz().replace(tzinfo=None) >= _SETTLEMENT_TIME else local_now.date() - timedelta(days=1) + + +def summarize(session: Session, start: datetime, end: datetime, *, now: datetime | None = None) -> dict[str, Any]: + """Return thermal variable/fixed totals; standing is charged once per contract/day.""" + start, end = _utc(start), _utc(end) + rows = session.execute(select(MeterCostPeriod).where( + MeterCostPeriod.period_start >= start, MeterCostPeriod.period_start < end + )).scalars().all() + good = [row for row in rows if not row.degraded] + variable = sum((_decimal(row.cost) for row in good), Decimal("0")) + breakdown: dict[str, Decimal] = {key: Decimal("0") for key in ( + "heating", "hot_water_heating", "hot_water", "hot_water_tax")} + for row in good: + for key, value in row.cost_breakdown.items(): + breakdown[key] = breakdown.get(key, Decimal("0")) + _decimal(value) + + # A summary is half-open. ``end`` at local midnight has no overlap with + # that next local date, and an empty/reversed range owns no standing day. + final_day = timezone_service.local_date(end - timedelta(microseconds=1)) + final_day = min(final_day, _settled_end_date(now or datetime.now(UTC))) + day = timezone_service.local_date(start) + fixed = Decimal("0") + versions = active_contract_versions(session, scope="thermal") + while start < end and day <= final_day: + day_start = datetime.combine(day, time.min, tzinfo=timezone_service.local_tz()).astimezone(UTC) + next_day_start = datetime.combine( + day + timedelta(days=1), time.min, tzinfo=timezone_service.local_tz() + ).astimezone(UTC) + local_day_seconds = Decimal(str((next_day_start - day_start).total_seconds())) + # A rate revision part-way through a local date is attributable only + # to its effective interval. This preserves one contract-level daily + # charge while correctly handling first-version and intra-day changes. + for version in versions: + segment_start = max(day_start, _utc(version.effective_from)) + version_end = _utc(version.effective_to) if version.effective_to is not None else next_day_start + segment_end = min(next_day_start, version_end) + if segment_start >= segment_end: + continue + values = version.values["standing"] + annual = sum((_decimal(values.get(key, "0")) for key in ( + "heating_network", "metering", "delivery_set", "hot_water_network", "other" + )), Decimal("0")) + fraction = Decimal(str((segment_end - segment_start).total_seconds())) / local_day_seconds + fixed += annual / Decimal("365") * fraction + day += timedelta(days=1) + return { + "currency": good[0].currency if good else "EUR", "variable_cost": variable, + "fixed_cost": fixed, "total_cost": variable + fixed, "breakdown": breakdown, + "period_count": len(good), "degraded_count": len(rows) - len(good), + } diff --git a/docs/design/m8-warmtelink-energy.md b/docs/design/m8-warmtelink-energy.md index 7daa72e..afbdc5c 100644 --- a/docs/design/m8-warmtelink-energy.md +++ b/docs/design/m8-warmtelink-energy.md @@ -850,7 +850,7 @@ T01～T06 先把现有 DSMR 安全迁到统一 source/binding；T07～T11 再接 ### M8-T15 — Thermal 15 分钟成本引擎与调度 [structural] -- **Status**: `todo` +- **Status**: `done` - **Depends**: M8-T14 - **Context**: 基于两个独立累计 Meter 生成可审计 variable 账本，并在 summary 层只计一次固定费。 diff --git a/tests/test_energy_cost.py b/tests/test_energy_cost.py index 2a62f99..6fad5cb 100644 --- a/tests/test_energy_cost.py +++ b/tests/test_energy_cost.py @@ -32,7 +32,7 @@ from pathlib import Path import pytest from alembic import command from alembic.config import Config -from sqlalchemy import create_engine, select +from sqlalchemy import create_engine, event, select from sqlalchemy.orm import Session from app.integrations.pricing.strategies import PeriodDeltas # noqa: F401 @@ -57,6 +57,28 @@ from app.services.energy_cost import ( from app.services import timezone as tz_module +def test_scheduler_isolates_electricity_and_thermal_cost_scopes(monkeypatch: pytest.MonkeyPatch) -> None: + """A thermal/electricity failure must not suppress the other ledger or HA publish.""" + from app import main + + calls: list[str] = [] + + class DummySession: + def rollback(self) -> None: + calls.append("rollback") + + def close(self) -> None: + calls.append("close") + + monkeypatch.setattr(main, "get_session_local", lambda: lambda: DummySession()) + monkeypatch.setattr(main, "compute_closed_periods", lambda _session: (_ for _ in ()).throw(ValueError())) + monkeypatch.setattr(main, "compute_closed_meter_cost_periods", lambda _session: calls.append("thermal")) + from app.services import ha_discovery + monkeypatch.setattr(ha_discovery, "publish_states", lambda _session: calls.append("publish")) + main._run_scheduled_energy_cost() + assert calls == ["rollback", "close", "thermal", "close", "publish", "close"] + + # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @@ -82,6 +104,163 @@ def energy_db(tmp_path: Path): engine.dispose() +@pytest.mark.parametrize("failing_scope", ["electricity", "thermal"]) +def test_scheduler_uses_clean_sessions_after_real_flush_failure( + energy_db: Session, monkeypatch: pytest.MonkeyPatch, failing_scope: str +) -> None: + """A DB failed-transaction in either ledger cannot poison the next scope or HA.""" + from app import main + from app.services import ha_discovery + + assert energy_db.bind is not None + sessions: list[Session] = [] + calls: list[str] = [] + + def factory() -> Session: + session = Session(energy_db.bind) + sessions.append(session) + return session + + def flush_failure(session: Session) -> None: + session.add(EnergyContract()) + session.flush() + + def successful(name: str): + def operation(session: Session) -> None: + session.execute(select(EnergyContract)).all() + calls.append(name) + + return operation + + monkeypatch.setattr(main, "get_session_local", lambda: factory) + monkeypatch.setattr( + main, "compute_closed_periods", + flush_failure if failing_scope == "electricity" else successful("electricity"), + ) + monkeypatch.setattr( + main, "compute_closed_meter_cost_periods", + flush_failure if failing_scope == "thermal" else successful("thermal"), + ) + monkeypatch.setattr(ha_discovery, "publish_states", lambda session: successful("ha")(session)) + + main._run_scheduled_energy_cost() + + assert calls == (["thermal", "ha"] if failing_scope == "electricity" else ["electricity", "ha"]) + assert len({id(session) for session in sessions}) == 3 + + +@pytest.mark.parametrize("failing_scope", ["electricity", "thermal"]) +def test_scheduler_isolates_real_commit_failure_and_rolls_back_write( + energy_db: Session, monkeypatch: pytest.MonkeyPatch, failing_scope: str +) -> None: + """A commit exception leaves no partial write and the later scopes still run.""" + from app import main + from app.services import ha_discovery + + assert energy_db.bind is not None + sessions: list[Session] = [] + calls: list[str] = [] + + def factory() -> Session: + session = Session(energy_db.bind) + sessions.append(session) + if len(sessions) == (1 if failing_scope == "electricity" else 2): + event.listen(session, "before_commit", lambda _session: (_ for _ in ()).throw(RuntimeError("commit"))) + return session + + def write(name: str): + def operation(session: Session) -> None: + now = datetime.now(UTC) + session.add(EnergyContract( + name=name, kind="manual", scope="electricity", active=False, currency="EUR", + created_at=now, updated_at=now, + )) + session.commit() + calls.append(name) + + return operation + + monkeypatch.setattr(main, "get_session_local", lambda: factory) + monkeypatch.setattr(main, "compute_closed_periods", write("electricity")) + monkeypatch.setattr(main, "compute_closed_meter_cost_periods", write("thermal")) + monkeypatch.setattr(ha_discovery, "publish_states", lambda _session: calls.append("ha")) + main._run_scheduled_energy_cost() + + assert calls == (["thermal", "ha"] if failing_scope == "electricity" else ["electricity", "ha"]) + persisted = Session(energy_db.bind) + try: + assert persisted.scalars(select(EnergyContract.name)).all() == ( + ["thermal"] if failing_scope == "electricity" else ["electricity"] + ) + finally: + persisted.close() + + +@pytest.mark.parametrize("failure", ["factory", "rollback", "close"]) +def test_scheduler_lifecycle_failures_are_non_fatal_and_keep_later_scopes_independent( + monkeypatch: pytest.MonkeyPatch, failure: str +) -> None: + """Factory and cleanup errors are logged locally, never allowed to escape.""" + from app import main + from app.services import ha_discovery + + calls: list[str] = [] + created = 0 + + class DummySession: + def rollback(self) -> None: + calls.append("rollback") + if failure == "rollback": + raise RuntimeError("rollback") + + def close(self) -> None: + calls.append("close") + if failure == "close": + raise RuntimeError("close") + + def factory() -> DummySession: + nonlocal created + created += 1 + if failure == "factory" and created == 1: + raise RuntimeError("factory") + return DummySession() + + def electricity(_session: DummySession) -> None: + if failure in {"rollback", "close"}: + raise RuntimeError("operation") + calls.append("electricity") + + monkeypatch.setattr(main, "get_session_local", lambda: factory) + monkeypatch.setattr(main, "compute_closed_periods", electricity) + monkeypatch.setattr(main, "compute_closed_meter_cost_periods", lambda _session: calls.append("thermal")) + monkeypatch.setattr(ha_discovery, "publish_states", lambda _session: calls.append("ha")) + main._run_scheduled_energy_cost() + + assert "thermal" in calls and "ha" in calls + assert created == 3 + + +def test_scheduler_swallows_ha_failure_after_two_successful_scopes(monkeypatch: pytest.MonkeyPatch) -> None: + from app import main + from app.services import ha_discovery + + calls: list[str] = [] + + class DummySession: + def rollback(self) -> None: + calls.append("rollback") + + def close(self) -> None: + calls.append("close") + + monkeypatch.setattr(main, "get_session_local", lambda: lambda: DummySession()) + monkeypatch.setattr(main, "compute_closed_periods", lambda _session: calls.append("electricity")) + monkeypatch.setattr(main, "compute_closed_meter_cost_periods", lambda _session: calls.append("thermal")) + monkeypatch.setattr(ha_discovery, "publish_states", lambda _session: (_ for _ in ()).throw(RuntimeError("ha"))) + main._run_scheduled_energy_cost() + assert calls == ["electricity", "close", "thermal", "close", "rollback", "close"] + + # --------------------------------------------------------------------------- # Data-builder helpers # --------------------------------------------------------------------------- diff --git a/tests/test_meter_cost.py b/tests/test_meter_cost.py new file mode 100644 index 0000000..abc539f --- /dev/null +++ b/tests/test_meter_cost.py @@ -0,0 +1,402 @@ +"""M8-T15 tests for the binding-aware thermal ledger.""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from decimal import Decimal +from pathlib import Path +from zoneinfo import ZoneInfo + +import pytest +from alembic import command +from alembic.config import Config +from sqlalchemy import create_engine, select +from sqlalchemy.orm import Session + +from app.models.energy import EnergyContract, EnergyContractVersion, Meter, MeterCostPeriod +from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel, WarmteLinkReading +from app.services import timezone as timezone_service +from app.services.meter_cost import compute_closed_periods, compute_period, recompute_range, summarize + + +@pytest.fixture() +def db(tmp_path: Path) -> Session: + url = f"sqlite:///{tmp_path / 'meter_cost.db'}" + cfg = Config("alembic_app.ini") + cfg.set_main_option("sqlalchemy.url", url) + command.upgrade(cfg, "head") + engine = create_engine(url) + session = Session(engine) + yield session + session.close() + engine.dispose() + + +T0 = datetime(2026, 6, 23, 10, tzinfo=UTC) + +VALUES = { + "variable": {"heating": "20.0", "hot_water_heating": "4.0", "hot_water": "2.0", "hot_water_tax": "1.0"}, + "standing": {"heating_network": "365", "metering": "73", "delivery_set": "0", "hot_water_network": "0", "other": "0"}, +} + + +def _contract(session: Session, *, values: dict = VALUES, start: datetime = T0 - timedelta(days=1)) -> EnergyContractVersion: + now = datetime.now(UTC) + contract = EnergyContract(name="thermal", kind="district_heating", scope="thermal", active=True, + currency="EUR", created_at=now, updated_at=now) + session.add(contract) + session.flush() + version = EnergyContractVersion(contract_id=contract.id, effective_from=start, values=values, created_at=now) + session.add(version) + session.flush() + return version + + +def _domain(session: Session, commodity: str, *, start: datetime = T0 - timedelta(days=1), end: datetime | None = None): + now = datetime.now(UTC) + meter = Meter(label=commodity, commodity=commodity, started_at=start, ended_at=end, + reason="initial", created_at=now) + source = MeterSource(name=f"{commodity}-source", kind="warmtelink_serial", enabled=True, config={}, + status="online", created_at=now, updated_at=now) + session.add_all((meter, source)) + session.flush() + unit = "GJ" if commodity == "heating" else "m3" + channel = MeterSourceChannel(source_id=source.id, channel_key=commodity, label=commodity, + suggested_commodity=commodity, unit=unit, latest_quality="valid", + created_at=now, updated_at=now) + session.add(channel) + session.flush() + binding = MeterSourceBinding(meter_id=meter.id, channel_id=channel.id, started_at=start, ended_at=end, + created_at=now, updated_at=now) + session.add(binding) + session.flush() + return meter, channel, binding + + +def _reading(session: Session, channel: MeterSourceChannel, at: datetime, value: str, quality: str = "valid") -> None: + session.add(WarmteLinkReading(channel_id=channel.id, recorded_at=at, received_at=at, value=Decimal(value), + unit=channel.unit, quality=quality, equipment_fingerprint="test")) + + +def _setup_good(session: Session) -> None: + _contract(session) + _, heating, _ = _domain(session, "heating") + _, water, _ = _domain(session, "hot_water") + _reading(session, heating, T0, "10.000") + _reading(session, heating, T0 + timedelta(minutes=15), "10.050") + _reading(session, water, T0, "20.000") + _reading(session, water, T0 + timedelta(minutes=15), "20.200") + session.commit() + + +def _row(session: Session, commodity: str) -> MeterCostPeriod: + return session.execute(select(MeterCostPeriod).where(MeterCostPeriod.commodity == commodity)).scalar_one() + + +def _row_at(session: Session, commodity: str, start: datetime) -> MeterCostPeriod: + return session.execute(select(MeterCostPeriod).where( + MeterCostPeriod.commodity == commodity, MeterCostPeriod.period_start == start + )).scalar_one() + + +def _version( + session: Session, contract: EnergyContract, *, start: datetime, values: dict +) -> EnergyContractVersion: + version = EnergyContractVersion( + contract_id=contract.id, effective_from=start, values=values, created_at=datetime.now(UTC) + ) + session.add(version) + session.flush() + return version + + +def test_two_commodity_decimal_breakdown_and_scheduler_idempotency(db: Session) -> None: + _setup_good(db) + assert compute_period(db, "heating", T0) + assert compute_period(db, "hot_water", T0) + db.commit() + heating, water = _row(db, "heating"), _row(db, "hot_water") + assert heating.quantity == Decimal("0.050000") + assert heating.cost == Decimal("1.000000000") + assert water.cost == Decimal("1.400000000") + assert {key: Decimal(value) for key, value in water.cost_breakdown.items()} == { + "hot_water_heating": Decimal("0.8"), "hot_water": Decimal("0.4"), "hot_water_tax": Decimal("0.2") + } + assert not compute_period(db, "heating", T0) + + +@pytest.mark.parametrize("commodity,start,end,reason", [ + ("heating", "10", "9.99", "negative_delta"), + ("heating", "10", "10.101", "delta_limit_exceeded"), + ("hot_water", "10", "11.001", "delta_limit_exceeded"), +]) +def test_bad_deltas_are_degraded(db: Session, commodity: str, start: str, end: str, reason: str) -> None: + _contract(db) + _, channel, _ = _domain(db, commodity) + _reading(db, channel, T0, start) + _reading(db, channel, T0 + timedelta(minutes=15), end) + db.commit() + compute_period(db, commodity, T0) + db.commit() + row = _row(db, commodity) + assert row.degraded and row.degraded_reason == reason and row.contract_version_id is None + + +def test_freshness_quality_binding_and_contract_fail_closed(db: Session) -> None: + _contract(db) + meter, channel, binding = _domain(db, "heating") + _reading(db, channel, T0 - timedelta(seconds=121), "1") + _reading(db, channel, T0 + timedelta(minutes=15), "1.01") + db.commit() + compute_period(db, "heating", T0) + assert _row(db, "heating").degraded_reason == "missing_stale_or_invalid_reading" + # Recompute sees invalid quality independently at the end boundary. + db.query(WarmteLinkReading).delete() + _reading(db, channel, T0, "1") + _reading(db, channel, T0 + timedelta(minutes=15), "1.01", "invalid") + db.commit() + recompute_range(db, T0, T0 + timedelta(minutes=15)) + assert _row(db, "heating").degraded_reason == "missing_stale_or_invalid_reading" + binding.ended_at = T0 + timedelta(minutes=15) + db.commit() + recompute_range(db, T0, T0 + timedelta(minutes=15)) + assert _row(db, "heating").degraded_reason == "missing_or_ambiguous_binding" + assert meter.id is not None + + +@pytest.mark.parametrize( + ("first_quality", "last_quality", "accepted", "period_quality"), + [ + ("valid", "valid", True, "valid"), + ("valid", "unverifiable", True, "unverifiable"), + ("unverifiable", "valid", True, "unverifiable"), + ("unverifiable", "unverifiable", True, "unverifiable"), + ("invalid", "valid", False, "invalid"), + ("valid", "invalid", False, "invalid"), + ], +) +def test_accepted_reading_quality_is_preserved_without_promotion( + db: Session, first_quality: str, last_quality: str, accepted: bool, period_quality: str +) -> None: + _contract(db) + _, channel, _ = _domain(db, "heating") + _reading(db, channel, T0, "10", first_quality) + _reading(db, channel, T0 + timedelta(minutes=15), "10.05", last_quality) + db.commit() + assert compute_period(db, "heating", T0) + db.commit() + row = _row(db, "heating") + assert row.degraded is not accepted + assert row.quality == period_quality + if not accepted: + assert row.degraded_reason == "missing_stale_or_invalid_reading" + + +def test_readings_must_be_inside_binding_and_meter_windows(db: Session) -> None: + _contract(db) + meter, channel, binding = _domain(db, "heating", start=T0) + _reading(db, channel, T0 - timedelta(seconds=60), "10") + _reading(db, channel, T0 + timedelta(minutes=15), "10.05") + db.commit() + compute_period(db, "heating", T0) + assert _row(db, "heating").degraded_reason == "missing_stale_or_invalid_reading" + + # A candidate after a closed binding is equally outside the cumulative domain. + binding.started_at = T0 - timedelta(days=1) + binding.ended_at = T0 + timedelta(minutes=15, seconds=30) + meter.started_at = T0 - timedelta(days=1) + db.query(WarmteLinkReading).delete() + _reading(db, channel, T0, "10") + _reading(db, channel, T0 + timedelta(minutes=15, seconds=60), "10.05") + db.commit() + recompute_range(db, T0, T0 + timedelta(minutes=15)) + assert _row(db, "heating").degraded_reason == "missing_stale_or_invalid_reading" + + +def test_freshness_is_independent_and_inclusive_with_nearest_candidate(db: Session) -> None: + _contract(db) + _, channel, _ = _domain(db, "heating") + _reading(db, channel, T0 - timedelta(seconds=120), "10") + _reading(db, channel, T0 + timedelta(minutes=15, seconds=120), "10.05") + db.commit() + compute_period(db, "heating", T0) + db.commit() + row = _row(db, "heating") + assert not row.degraded and row.quantity == Decimal("0.050000") + + +def test_closed_scheduler_retries_degraded_but_not_normal(db: Session) -> None: + _contract(db) + _, channel, _ = _domain(db, "heating") + # The first closed scheduler pass records a degraded row. Supplying the + # missing boundary inputs later must let its next pass repair that row. + db.commit() + assert compute_closed_periods(db, now=T0 + timedelta(minutes=16)) > 0 + assert _row_at(db, "heating", T0).degraded + _reading(db, channel, T0, "10") + _reading(db, channel, T0 + timedelta(minutes=15), "10.05") + db.commit() + assert compute_closed_periods(db, now=T0 + timedelta(minutes=16)) > 0 + normal = _row_at(db, "heating", T0) + assert not normal.degraded + frozen = (normal.cost, dict(normal.pricing_snapshot), normal.updated_at) + end_reading = db.execute(select(WarmteLinkReading).where( + WarmteLinkReading.channel_id == channel.id, + WarmteLinkReading.recorded_at == T0 + timedelta(minutes=15), + )).scalar_one() + end_reading.value = Decimal("10.090") + version = db.get(EnergyContractVersion, normal.contract_version_id) + assert version is not None + version.values = {**version.values, "variable": {**version.values["variable"], "heating": "99"}} + db.commit() + compute_closed_periods(db, now=T0 + timedelta(minutes=16)) + db.expire_all() + unchanged = db.execute(select(MeterCostPeriod).where(MeterCostPeriod.id == normal.id)).scalar_one() + assert (unchanged.cost, unchanged.pricing_snapshot, unchanged.updated_at) == frozen + + +def test_explicit_recompute_overwrites_and_source_switch_degrades(db: Session) -> None: + _setup_good(db) + compute_period(db, "heating", T0) + db.commit() + row = _row(db, "heating") + old_cost = row.cost + channel = db.get(MeterSourceChannel, row.source_binding.channel_id) + end_reading = db.execute(select(WarmteLinkReading).where( + WarmteLinkReading.channel_id == channel.id, + WarmteLinkReading.recorded_at == T0 + timedelta(minutes=15), + )).scalar_one() + end_reading.value = Decimal("10.080") + db.commit() + assert recompute_range(db, T0, T0 + timedelta(minutes=15)) == 2 + assert _row(db, "heating").cost != old_cost + # A hand-off precisely at the right boundary cannot form a single domain. + binding = _row(db, "heating").source_binding + binding.ended_at = T0 + timedelta(minutes=15) + db.commit() + recompute_range(db, T0, T0 + timedelta(minutes=15)) + assert _row(db, "heating").degraded_reason == "missing_or_ambiguous_binding" + + +def test_real_source_binding_handoff_never_crosses_cumulative_registers(db: Session) -> None: + _contract(db) + meter, old_channel, old_binding = _domain(db, "heating") + handoff = T0 + timedelta(minutes=15) + old_binding.ended_at = handoff + new_meter, new_channel, new_binding = _domain(db, "heating", start=handoff, end=handoff) + # Keep the same Meter: this is a source/channel/binding handoff, not a meter swap. + new_binding.meter_id = meter.id + new_binding.ended_at = None + assert new_meter.id != meter.id + _reading(db, old_channel, T0, "1000") + _reading(db, old_channel, handoff - timedelta(seconds=1), "1000.01") + _reading(db, new_channel, handoff, "7") + _reading(db, new_channel, handoff + timedelta(seconds=1), "7.01") + db.commit() + compute_period(db, "heating", T0) + assert _row(db, "heating").degraded_reason == "cross_source_binding" + + +def test_real_meter_epoch_handoff_never_crosses_cumulative_registers(db: Session) -> None: + _contract(db) + handoff = T0 + timedelta(minutes=15) + _, old_channel, _ = _domain(db, "heating", end=handoff) + _, new_channel, _ = _domain(db, "heating", start=handoff) + _reading(db, old_channel, T0, "1000") + _reading(db, old_channel, handoff - timedelta(seconds=1), "1000.01") + _reading(db, new_channel, handoff, "2") + _reading(db, new_channel, handoff + timedelta(seconds=1), "2.01") + db.commit() + compute_period(db, "heating", T0) + assert _row(db, "heating").degraded_reason == "cross_meter_epoch" + + +def test_summary_fixed_once_per_contract_day_and_dst(db: Session, monkeypatch: pytest.MonkeyPatch) -> None: + _setup_good(db) + compute_period(db, "heating", T0) + compute_period(db, "hot_water", T0) + db.commit() + monkeypatch.setattr(timezone_service, "local_tz", lambda: ZoneInfo("Europe/Amsterdam")) + result = summarize(db, T0, T0 + timedelta(days=1), now=datetime(2026, 6, 24, 1, 6, tzinfo=UTC)) + assert result["variable_cost"] == Decimal("2.400000000") + assert result["fixed_cost"] == Decimal("2.4") # two local days, each charged once not per commodity + # DST local day is charged once too, despite being 23 hours long. + db.execute(select(EnergyContractVersion)).scalar_one().effective_from = datetime(2026, 3, 1, tzinfo=UTC) + db.commit() + dst_start = datetime(2026, 3, 28, 23, tzinfo=UTC) + dst = summarize(db, dst_start, dst_start + timedelta(days=2), now=datetime(2026, 3, 31, tzinfo=UTC)) + assert dst["fixed_cost"] == Decimal("3.6") + + +def test_summary_is_half_open_at_local_midnight_and_settlement(db: Session, monkeypatch: pytest.MonkeyPatch) -> None: + _contract(db) + monkeypatch.setattr(timezone_service, "local_tz", lambda: ZoneInfo("Europe/Amsterdam")) + local = ZoneInfo("Europe/Amsterdam") + start = datetime(2026, 6, 24, 0, tzinfo=local).astimezone(UTC) + end = datetime(2026, 6, 25, 0, tzinfo=local).astimezone(UTC) + assert summarize(db, start, start, now=end + timedelta(hours=2))["fixed_cost"] == Decimal("0") + assert summarize(db, end, start, now=end + timedelta(hours=2))["fixed_cost"] == Decimal("0") + before = datetime(2026, 6, 24, 1, 4, 59, tzinfo=local).astimezone(UTC) + after = datetime(2026, 6, 24, 1, 5, tzinfo=local).astimezone(UTC) + assert summarize(db, start, end, now=before)["fixed_cost"] == Decimal("0") + assert summarize(db, start, end, now=after)["fixed_cost"] == Decimal("1.2") + + +def test_summary_versions_cover_first_day_intra_day_and_cross_day(db: Session, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(timezone_service, "local_tz", lambda: ZoneInfo("Europe/Amsterdam")) + local = ZoneInfo("Europe/Amsterdam") + day = datetime(2026, 6, 24, tzinfo=local) + first = _contract(db, start=(day + timedelta(hours=6)).astimezone(UTC)) + first.values = {**VALUES, "standing": {**VALUES["standing"], "heating_network": "365"}} + second_values = {**VALUES, "standing": {**VALUES["standing"], "heating_network": "730"}} + switch = (day + timedelta(hours=18)).astimezone(UTC) + first.effective_to = switch + _version(db, first.contract, start=switch, values=second_values) + db.commit() + start = day.astimezone(UTC) + next_day = (day + timedelta(days=1)).astimezone(UTC) + following = (day + timedelta(days=2)).astimezone(UTC) + # Initial-version day only owns 18h: V1 owns 12h, V2 6h; on the following + # day V2 owns the full daily fixed amount. + assert summarize(db, start, next_day, now=following + timedelta(hours=2))["fixed_cost"] == Decimal("1.15") + assert summarize(db, start, following, now=following + timedelta(hours=2))["fixed_cost"] == Decimal("3.35") + + +def test_normal_row_has_full_audit_snapshot_and_single_commodity_scheduler(db: Session) -> None: + version = _contract(db) + meter, channel, binding = _domain(db, "heating") + _reading(db, channel, T0, "10") + _reading(db, channel, T0 + timedelta(minutes=15), "10.05") + db.commit() + compute_closed_periods(db, now=T0 + timedelta(minutes=16)) + row = _row_at(db, "heating", T0) + assert (row.meter_id, row.source_binding_id, row.contract_version_id) == (meter.id, binding.id, version.id) + assert row.pricing_snapshot == VALUES + assert row.currency == "EUR" and { + key: Decimal(value) for key, value in row.cost_breakdown.items() + } == {"heating": Decimal("1.0")} + assert not row.degraded + assert _row_at(db, "hot_water", T0).degraded + + +@pytest.mark.parametrize(("commodity", "end_value"), [("heating", "10.1"), ("hot_water", "11")]) +def test_delta_limit_is_inclusive_at_exact_boundary(db: Session, commodity: str, end_value: str) -> None: + _contract(db) + _, channel, _ = _domain(db, commodity) + _reading(db, channel, T0, "10") + _reading(db, channel, T0 + timedelta(minutes=15), end_value) + db.commit() + compute_period(db, commodity, T0) + assert not _row(db, commodity).degraded + + +@pytest.mark.parametrize("day", [datetime(2026, 3, 29), datetime(2026, 10, 25)]) +def test_summary_dst_local_midnights_charge_one_daily_rate( + db: Session, monkeypatch: pytest.MonkeyPatch, day: datetime +) -> None: + _contract(db, start=datetime(2026, 1, 1, tzinfo=UTC)) + monkeypatch.setattr(timezone_service, "local_tz", lambda: ZoneInfo("Europe/Amsterdam")) + local = ZoneInfo("Europe/Amsterdam") + start = day.replace(tzinfo=local).astimezone(UTC) + end = (day + timedelta(days=1)).replace(tzinfo=local).astimezone(UTC) + assert summarize(db, start, end, now=end + timedelta(hours=2))["fixed_cost"] == Decimal("1.2")