diff --git a/alembic_app/versions/20260822_17_warmtelink_readings.py b/alembic_app/versions/20260822_17_warmtelink_readings.py new file mode 100644 index 0000000..eaa609c --- /dev/null +++ b/alembic_app/versions/20260822_17_warmtelink_readings.py @@ -0,0 +1,52 @@ +"""add normalized WarmteLink scalar reading history + +Revision ID: 20260822_17_warmtelink_readings +Revises: 20260822_16_dsmr_source_adoption +Create Date: 2026-08-22 00:00:00.000000 + +The upgrade is additive: existing business rows are neither changed nor +removed. The downgrade is schema-only and is exercised only on isolated test +databases. +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + + +revision: str = "20260822_17_warmtelink_readings" +down_revision: Union[str, None] = "20260822_16_dsmr_source_adoption" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "warmtelink_reading", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("channel_id", sa.Integer(), nullable=False), + sa.Column("recorded_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("received_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("value", sa.Numeric(precision=15, scale=3), nullable=False), + sa.Column("unit", sa.String(length=32), nullable=False), + sa.Column("quality", sa.String(length=32), nullable=False), + sa.Column("equipment_fingerprint", sa.String(length=64), nullable=False), + sa.CheckConstraint( + "quality IN ('valid', 'invalid', 'unverifiable')", + name="ck_warmtelink_reading_quality", + ), + sa.ForeignKeyConstraint( + ["channel_id"], ["meter_source_channel.id"], ondelete="RESTRICT" + ), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint( + "channel_id", "recorded_at", name="uq_warmtelink_reading_channel_recorded_at" + ), + ) + op.create_index("ix_warmtelink_reading_recorded_at", "warmtelink_reading", ["recorded_at"]) + + +def downgrade() -> None: + op.drop_index("ix_warmtelink_reading_recorded_at", table_name="warmtelink_reading") + op.drop_table("warmtelink_reading") diff --git a/app/models/meter_source.py b/app/models/meter_source.py index bd51997..0807be5 100644 --- a/app/models/meter_source.py +++ b/app/models/meter_source.py @@ -4,10 +4,12 @@ from __future__ import annotations import uuid as _uuid from datetime import datetime +from decimal import Decimal from typing import TYPE_CHECKING from sqlalchemy import ( Boolean, + CheckConstraint, DateTime, ForeignKey, Index, @@ -93,12 +95,48 @@ class MeterSourceChannel(Base): bindings: Mapped[list["MeterSourceBinding"]] = relationship( back_populates="channel", cascade="save-update, merge" ) + warmtelink_readings: Mapped[list["WarmteLinkReading"]] = relationship( + back_populates="channel", cascade="save-update, merge" + ) __table_args__ = ( UniqueConstraint("source_id", "channel_key", name="uq_meter_source_channel_source_key"), ) +class WarmteLinkReading(Base): + """One accepted scalar cumulative reading from a WarmteLink channel.""" + + __tablename__ = "warmtelink_reading" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + channel_id: Mapped[int] = mapped_column( + ForeignKey("meter_source_channel.id", ondelete="RESTRICT"), nullable=False + ) + # These timestamps retain their UTC-aware application semantics. SQLite + # stores them without an offset, so callers must always supply aware UTC. + recorded_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + received_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + # SQLite's ORM Numeric path can exactly round-trip this 12-integer-digit + # range at scale 3. That is ample for a long-lived cumulative meter while + # retaining the protocol's 0.001 resolution without float conversion. + value: Mapped[Decimal] = mapped_column(Numeric(15, 3), nullable=False) + unit: Mapped[str] = mapped_column(String(32), nullable=False) + quality: Mapped[str] = mapped_column(String(32), nullable=False) + equipment_fingerprint: Mapped[str] = mapped_column(String(64), nullable=False) + + channel: Mapped["MeterSourceChannel"] = relationship(back_populates="warmtelink_readings") + + __table_args__ = ( + CheckConstraint( + "quality IN ('valid', 'invalid', 'unverifiable')", + name="ck_warmtelink_reading_quality", + ), + UniqueConstraint("channel_id", "recorded_at", name="uq_warmtelink_reading_channel_recorded_at"), + Index("ix_warmtelink_reading_recorded_at", "recorded_at"), + ) + + class MeterSourceBinding(Base): """Connect one source channel to one physical meter for a half-open window.""" diff --git a/docs/design/m8-warmtelink-energy.md b/docs/design/m8-warmtelink-energy.md index 8b6910c..7d4b2e4 100644 --- a/docs/design/m8-warmtelink-energy.md +++ b/docs/design/m8-warmtelink-energy.md @@ -577,7 +577,7 @@ T01～T06 先把现有 DSMR 安全迁到统一 source/binding；T07～T11 再接 ### M8-T08 — 建立 WarmteLink 标量读数表 -- **Status**: `todo` +- **Status**: `done` - **Depends**: M8-T07 - **Context**: WarmteLink 与 DSMR payload 结构不同，建立可精确计算的 Decimal 标量历史表。 diff --git a/scripts/app_db_adopt.py b/scripts/app_db_adopt.py index f9c3101..beb9943 100644 --- a/scripts/app_db_adopt.py +++ b/scripts/app_db_adopt.py @@ -15,7 +15,7 @@ if str(PROJECT_ROOT) not in sys.path: from app.config import get_settings -APP_BASELINE_REVISION = "20260822_16_dsmr_source_adoption" +APP_BASELINE_REVISION = "20260822_17_warmtelink_readings" class AppDatabaseAdoptionError(RuntimeError): diff --git a/tests/test_warmtelink_models.py b/tests/test_warmtelink_models.py new file mode 100644 index 0000000..5618533 --- /dev/null +++ b/tests/test_warmtelink_models.py @@ -0,0 +1,252 @@ +"""Schema and migration tests for the normalized WarmteLink reading table.""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from decimal import Decimal +from pathlib import Path + +import pytest +import sqlalchemy.exc +from alembic import command +from alembic.config import Config +from sqlalchemy import create_engine, event, inspect, text +from sqlalchemy.orm import Session + +from app.models.meter_source import MeterSourceChannel, WarmteLinkReading +from scripts.app_db_adopt import APP_BASELINE_REVISION + +REVISION_16 = "20260822_16_dsmr_source_adoption" +REVISION_17 = "20260822_17_warmtelink_readings" + + +def _config(database_url: str) -> Config: + config = Config("alembic_app.ini") + config.set_main_option("sqlalchemy.url", database_url) + return config + + +def _engine(database_url: str): + engine = create_engine(database_url, connect_args={"check_same_thread": False}) + + @event.listens_for(engine, "connect") + def _foreign_keys(connection, _record) -> None: + connection.execute("PRAGMA foreign_keys=ON") + + return engine + + +def _add_channel(connection, timestamp: datetime, channel_key: str = "heating-total") -> int: + source_id = connection.execute( + text("SELECT id FROM meter_source WHERE kind = 'dsmr_mqtt'") + ).scalar_one() + connection.execute( + text( + "INSERT INTO meter_source_channel " + "(uuid, source_id, channel_key, label, suggested_commodity, unit, device_type, " + "fingerprint, latest_value, latest_at, latest_quality, created_at, updated_at) " + "VALUES (:uuid, :source_id, :channel_key, 'WarmteLink channel', 'heating', 'GJ', " + "'warmtelink', NULL, NULL, NULL, NULL, :at, :at)" + ), + { + "uuid": f"{channel_key[:8]:0<8}-0000-4000-8000-000000000000", + "source_id": source_id, + "channel_key": channel_key, + "at": timestamp, + }, + ) + return int(connection.execute(text("SELECT last_insert_rowid()")).scalar_one()) + + +def _insert_reading(connection, channel_id: int, timestamp: datetime, value: Decimal) -> None: + connection.execute( + text( + "INSERT INTO warmtelink_reading " + "(channel_id, recorded_at, received_at, value, unit, quality, equipment_fingerprint) " + "VALUES (:channel_id, :recorded_at, :received_at, :value, 'GJ', 'unverifiable', :fingerprint)" + ), + { + "channel_id": channel_id, + "recorded_at": timestamp, + "received_at": timestamp, + "value": str(value), + "fingerprint": "a" * 64, + }, + ) + + +def test_empty_database_and_revision_16_upgrade_are_additive_and_idempotent(tmp_path: Path): + empty_url = f"sqlite:///{tmp_path / 'warmtelink_empty.db'}" + empty_config = _config(empty_url) + command.upgrade(empty_config, "head") + command.upgrade(empty_config, "head") + empty_engine = _engine(empty_url) + try: + with empty_engine.connect() as connection: + assert connection.execute(text("SELECT version_num FROM alembic_version")).scalar_one() == REVISION_17 + assert connection.execute(text("PRAGMA foreign_key_check")).all() == [] + finally: + empty_engine.dispose() + + database_url = f"sqlite:///{tmp_path / 'warmtelink_upgrade.db'}" + config = _config(database_url) + timestamp = datetime(2026, 8, 22, tzinfo=timezone.utc) + + command.upgrade(config, REVISION_16) + engine = _engine(database_url) + try: + with engine.begin() as connection: + channel_id = _add_channel(connection, timestamp) + before_counts = { + table: int(connection.execute(text(f"SELECT COUNT(*) FROM {table}")).scalar_one()) + for table in ("dsmr_reading", "meter_source", "meter_source_channel", "meter_source_binding") + } + assert connection.execute(text("SELECT version_num FROM alembic_version")).scalar_one() == REVISION_16 + assert channel_id > 0 + finally: + engine.dispose() + + command.upgrade(config, "head") + command.upgrade(config, "head") + engine = _engine(database_url) + try: + with engine.connect() as connection: + after_counts = { + table: int(connection.execute(text(f"SELECT COUNT(*) FROM {table}")).scalar_one()) + for table in before_counts + } + assert after_counts == before_counts + assert connection.execute(text("SELECT version_num FROM alembic_version")).scalar_one() == REVISION_17 + assert connection.execute(text("PRAGMA foreign_key_check")).all() == [] + finally: + engine.dispose() + + +def test_warmtelink_reading_constraints_indexes_and_decimal_round_trip(tmp_path: Path): + database_url = f"sqlite:///{tmp_path / 'warmtelink_constraints.db'}" + config = _config(database_url) + command.upgrade(config, "head") + timestamp = datetime(2026, 8, 22, 10, 30, tzinfo=timezone.utc) + engine = _engine(database_url) + try: + with engine.begin() as connection: + first_channel = _add_channel(connection, timestamp) + second_channel = _add_channel(connection, timestamp, "hot-water-total") + _insert_reading(connection, first_channel, timestamp, Decimal("12345.678")) + _insert_reading(connection, second_channel, timestamp, Decimal("4.125")) + + with Session(engine) as session: + reading = session.query(WarmteLinkReading).filter_by(channel_id=first_channel).one() + assert reading.value == Decimal("12345.678") + assert reading.recorded_at == timestamp + assert reading.received_at == timestamp + session.add( + WarmteLinkReading( + channel_id=first_channel, + recorded_at=timestamp + timedelta(minutes=1), + received_at=timestamp + timedelta(minutes=1), + value=Decimal("234.567"), + unit="GJ", + quality="valid", + equipment_fingerprint="c" * 64, + ) + ) + session.commit() + assert session.query(WarmteLinkReading).filter_by(value=Decimal("234.567")).one().value == Decimal( + "234.567" + ) + + # Numeric(15, 3) promises twelve integer digits. Exercise its + # declared upper boundary through the SQLite ORM path, rather than + # a raw SQL value that could bypass its Decimal result processor. + boundary = Decimal("999999999999.999") + session.add( + WarmteLinkReading( + channel_id=second_channel, + recorded_at=timestamp + timedelta(minutes=1), + received_at=timestamp + timedelta(minutes=1), + value=boundary, + unit="GJ", + quality="valid", + equipment_fingerprint="d" * 64, + ) + ) + session.commit() + session.expire_all() + assert ( + session.query(WarmteLinkReading) + .filter_by(channel_id=second_channel, recorded_at=timestamp + timedelta(minutes=1)) + .one() + .value + == boundary + ) + + with pytest.raises(sqlalchemy.exc.IntegrityError): + with engine.begin() as connection: + _insert_reading(connection, first_channel, timestamp, Decimal("5.000")) + with pytest.raises(sqlalchemy.exc.IntegrityError): + with engine.begin() as connection: + connection.execute( + text( + "INSERT INTO warmtelink_reading " + "(channel_id, recorded_at, received_at, value, unit, quality, equipment_fingerprint) " + "VALUES (:channel_id, :at, :at, 1.000, 'GJ', 'estimated-valid', :fingerprint)" + ), + {"channel_id": first_channel, "at": timestamp, "fingerprint": "b" * 64}, + ) + with pytest.raises(sqlalchemy.exc.IntegrityError): + with engine.begin() as connection: + connection.execute(text("DELETE FROM meter_source_channel WHERE id = :id"), {"id": first_channel}) + with pytest.raises(sqlalchemy.exc.IntegrityError): + with engine.begin() as connection: + connection.execute( + text("DELETE FROM meter_source WHERE id = (SELECT source_id FROM meter_source_channel WHERE id = :id)"), + {"id": first_channel}, + ) + + inspector = inspect(engine) + columns = {column["name"]: column for column in inspector.get_columns("warmtelink_reading")} + assert columns["value"]["type"].precision == 15 + assert columns["value"]["type"].scale == 3 + assert "raw_telegram" not in columns + assert "equipment_identifier" not in columns + assert "raw_identifier" not in columns + assert {tuple(item["column_names"]) for item in inspector.get_unique_constraints("warmtelink_reading")} >= { + ("channel_id", "recorded_at"), + } + foreign_key = inspector.get_foreign_keys("warmtelink_reading")[0] + assert foreign_key["constrained_columns"] == ["channel_id"] + assert foreign_key["referred_table"] == "meter_source_channel" + assert foreign_key["options"]["ondelete"] == "RESTRICT" + assert "ix_warmtelink_reading_recorded_at" in { + index["name"] for index in inspector.get_indexes("warmtelink_reading") + } + with engine.connect() as connection: + assert connection.execute(text("PRAGMA foreign_key_check")).all() == [] + finally: + engine.dispose() + + +def test_warmtelink_reading_model_uses_restrictive_relationship_and_aware_columns(): + relationship = MeterSourceChannel.warmtelink_readings.property + assert "delete" not in relationship.cascade + assert "delete-orphan" not in relationship.cascade + assert WarmteLinkReading.__table__.c.recorded_at.type.timezone is True + assert WarmteLinkReading.__table__.c.received_at.type.timezone is True + assert APP_BASELINE_REVISION == REVISION_17 + + +def test_warmtelink_reading_downgrade_is_schema_only_on_temporary_database(tmp_path: Path): + database_url = f"sqlite:///{tmp_path / 'warmtelink_downgrade.db'}" + config = _config(database_url) + command.upgrade(config, "head") + command.downgrade(config, REVISION_16) + + engine = _engine(database_url) + try: + with engine.connect() as connection: + assert "warmtelink_reading" not in set(inspect(engine).get_table_names()) + assert connection.execute(text("SELECT version_num FROM alembic_version")).scalar_one() == REVISION_16 + assert connection.execute(text("PRAGMA foreign_key_check")).all() == [] + finally: + engine.dispose()