"""adopt historical DSMR rows into the source and binding model Revision ID: 20260822_16_dsmr_source_adoption Revises: 20260822_15_meter_sources Create Date: 2026-08-22 00:00:00.000000 The upgrade is deliberately data-preserving: it creates one migration-owned DSMR source/channel, moves the telegram identifier to ``telegram_id``, and audits every reading and cost row before committing. Old ``app_config`` rows, payload JSON, and cost snapshots are never deleted or rewritten. """ from __future__ import annotations import uuid from datetime import datetime, timezone from typing import Sequence, Union import sqlalchemy as sa from alembic import op revision: str = "20260822_16_dsmr_source_adoption" down_revision: Union[str, None] = "20260822_15_meter_sources" branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None def _as_bool(value: str | None) -> bool: return value is not None and value.strip().lower() in {"1", "true", "yes", "on"} def _iso_now() -> str: return datetime.now(tz=timezone.utc).replace(tzinfo=None).isoformat(sep=" ") def _config(connection: sa.Connection) -> dict[str, object]: rows = connection.execute(sa.text("SELECT key, value FROM app_config")).all() values = {str(key): str(value) for key, value in rows} # An unconfigured historical DSMR installation needs a disabled identity, # not guessed connection details. Preserve every legacy value we model # when any legacy DSMR/MQTT configuration was explicitly present. legacy_keys = { "MQTT_BROKER_HOST", "MQTT_BROKER_PORT", "MQTT_USERNAME", "MQTT_PASSWORD", "MQTT_TLS_ENABLED", "DSMR_MQTT_TOPIC", "DSMR_TARIFF_TOPIC", "DSMR_SAMPLE_INTERVAL_S", } if not legacy_keys & values.keys(): return {} return { "broker_host": values.get("MQTT_BROKER_HOST", ""), "broker_port": int(values.get("MQTT_BROKER_PORT", "1883")), "username": values.get("MQTT_USERNAME", ""), "password": values.get("MQTT_PASSWORD", ""), "tls_enabled": _as_bool(values.get("MQTT_TLS_ENABLED")), "topic": values.get("DSMR_MQTT_TOPIC", "dsmr/json"), "tariff_topic": values.get("DSMR_TARIFF_TOPIC", "dsmr/meter-stats/electricity_tariff"), "sample_interval_s": int(values.get("DSMR_SAMPLE_INTERVAL_S", "10")), } def _count(connection: sa.Connection, table: str) -> int: return int(connection.execute(sa.text(f"SELECT COUNT(*) FROM {table}")).scalar_one()) def upgrade() -> None: connection = op.get_bind() readings_before = _count(connection, "dsmr_reading") costs_before = _count(connection, "energy_cost_period") sources_before = _count(connection, "meter_source") channels_before = _count(connection, "meter_source_channel") bindings_before = _count(connection, "meter_source_binding") now = _iso_now() # A source exists even without historical configuration/readings. It stays # disabled unless the old explicit DSMR switch was enabled, so no broker or # topic is guessed at runtime. config = _config(connection) source_result = connection.execute( sa.text( "INSERT INTO meter_source " "(uuid, name, kind, enabled, config, status, last_seen_at, last_error, created_at, updated_at) " "VALUES (:uuid, :name, 'dsmr_mqtt', :enabled, :config, 'unknown', NULL, NULL, :now, :now)" ), { "uuid": str(uuid.uuid4()), "name": "Migrated DSMR source", "enabled": _as_bool( connection.execute( sa.text("SELECT value FROM app_config WHERE key = 'DSMR_INGEST_ENABLED'") ).scalar_one_or_none() ), "config": __import__("json").dumps(config), "now": now, }, ) source_id = source_result.lastrowid if source_id is None: raise RuntimeError("DSMR source adoption failed to create a source") channel_result = connection.execute( sa.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, 'electricity-total', 'DSMR electricity total', 'electricity', " "'kWh', NULL, NULL, NULL, NULL, NULL, :now, :now)" ), {"uuid": str(uuid.uuid4()), "source_id": source_id, "now": now}, ) channel_id = channel_result.lastrowid if channel_id is None: raise RuntimeError("DSMR source adoption failed to create an electricity channel") if _count(connection, "meter_source") != sources_before + 1: raise RuntimeError("DSMR source adoption source row-count audit failed") if _count(connection, "meter_source_channel") != channels_before + 1: raise RuntimeError("DSMR source adoption channel row-count audit failed") # Rename/add while nullable, back-fill all rows, then make the FK non-null # and replace the legacy timestamp-only uniqueness in a SQLite batch rebuild. with op.batch_alter_table("dsmr_reading", schema=None) as batch_op: batch_op.alter_column("source_id", new_column_name="telegram_id") batch_op.add_column(sa.Column("meter_source_id", sa.Integer(), nullable=True)) connection.execute( sa.text("UPDATE dsmr_reading SET meter_source_id = :source_id WHERE meter_source_id IS NULL"), {"source_id": source_id}, ) with op.batch_alter_table("dsmr_reading", schema=None) as batch_op: batch_op.drop_constraint("uq_dsmr_reading_recorded_at", type_="unique") batch_op.alter_column("meter_source_id", existing_type=sa.Integer(), nullable=False) batch_op.create_foreign_key( "fk_dsmr_reading_meter_source_id", "meter_source", ["meter_source_id"], ["id"], ondelete="RESTRICT", ) batch_op.create_unique_constraint( "uq_dsmr_reading_source_recorded_at", ["meter_source_id", "recorded_at"] ) batch_op.create_index("ix_dsmr_reading_meter_source_id", ["meter_source_id"]) adopted_readings = int( connection.execute( sa.text("SELECT COUNT(*) FROM dsmr_reading WHERE meter_source_id = :source_id"), {"source_id": source_id}, ).scalar_one() ) if adopted_readings != readings_before: raise RuntimeError("DSMR source adoption reading source audit failed") # Bind each electricity meter only where it overlaps the actual DSMR data. data_window = connection.execute( sa.text("SELECT MIN(recorded_at), MAX(recorded_at) FROM dsmr_reading") ).one() expected_binding_count = 0 if data_window[0] is not None: meters = connection.execute( sa.text( "SELECT id, started_at, ended_at FROM meter WHERE commodity = 'electricity' " "ORDER BY started_at, id" ) ).all() for meter_id, started_at, ended_at in meters: # Intersect [meter start, meter end) with the inclusive historical # samples. A closed boundary at the final sample remains valid for # the preceding interval; an empty intersection gets no fake binding. if started_at > data_window[1] or (ended_at is not None and ended_at <= data_window[0]): continue expected_binding_count += 1 binding_start = max(started_at, data_window[0]) binding_end = ended_at connection.execute( sa.text( "INSERT INTO meter_source_binding " "(uuid, meter_id, channel_id, started_at, ended_at, created_at, updated_at) " "VALUES (:uuid, :meter_id, :channel_id, :started_at, :ended_at, :now, :now)" ), { "uuid": str(uuid.uuid4()), "meter_id": meter_id, "channel_id": channel_id, "started_at": binding_start, "ended_at": binding_end, "now": now, }, ) # A cost period may be linked only if exactly one binding covers both its # start and end. Historical boundary/unknown rows remain auditable but are # explicitly degraded instead of being silently attributed to a current meter. periods = connection.execute( sa.text("SELECT id, meter_id, period_start, degraded FROM energy_cost_period") ).all() resolvable_normal_periods: dict[int, int] = {} unresolved_period_ids: set[int] = set() for period_id, meter_id, period_start, degraded_before in periods: candidates = [] if meter_id is not None: candidates = connection.execute( sa.text( "SELECT id FROM meter_source_binding " "WHERE meter_id = :meter_id AND started_at <= :start " "AND (ended_at IS NULL OR julianday(ended_at) > julianday(:start, '+15 minutes'))" ), {"meter_id": meter_id, "start": period_start}, ).all() if len(candidates) == 1: if not degraded_before: resolvable_normal_periods[period_id] = candidates[0][0] connection.execute( sa.text("UPDATE energy_cost_period SET source_binding_id = :binding_id WHERE id = :id"), {"binding_id": candidates[0][0], "id": period_id}, ) else: unresolved_period_ids.add(period_id) connection.execute( sa.text( "UPDATE energy_cost_period SET degraded = 1, source_binding_id = NULL WHERE id = :id" ), {"id": period_id}, ) readings_after = _count(connection, "dsmr_reading") costs_after = _count(connection, "energy_cost_period") if readings_after != readings_before or costs_after != costs_before: raise RuntimeError("DSMR source adoption row-count audit failed") if _count(connection, "meter_source_binding") != bindings_before + expected_binding_count: raise RuntimeError("DSMR source adoption binding row-count audit failed") if int( connection.execute( sa.text("SELECT COUNT(*) FROM meter_source_binding WHERE channel_id = :channel_id"), {"channel_id": channel_id}, ).scalar_one() ) != expected_binding_count: raise RuntimeError("DSMR source adoption binding channel audit failed") for period_id, binding_id in resolvable_normal_periods.items(): bound, degraded = connection.execute( sa.text("SELECT source_binding_id, degraded FROM energy_cost_period WHERE id = :id"), {"id": period_id}, ).one() if bound != binding_id or degraded: raise RuntimeError("DSMR source adoption resolvable cost audit failed") if unresolved_period_ids: unresolved_count = int( connection.execute( sa.text( "SELECT COUNT(*) FROM energy_cost_period " "WHERE id IN :period_ids AND (degraded != 1 OR source_binding_id IS NOT NULL)" ).bindparams(sa.bindparam("period_ids", expanding=True)), {"period_ids": list(unresolved_period_ids)}, ).scalar_one() ) if unresolved_count: raise RuntimeError("DSMR source adoption unresolved cost audit failed") orphan_rows = connection.execute(sa.text("PRAGMA foreign_key_check")).all() if orphan_rows: raise RuntimeError("DSMR source adoption foreign-key audit failed") normal_unbound = int( connection.execute( sa.text("SELECT COUNT(*) FROM energy_cost_period WHERE degraded = 0 AND source_binding_id IS NULL") ).scalar_one() ) if normal_unbound: raise RuntimeError(f"DSMR source adoption left {normal_unbound} normal cost period(s) unbound") if _count(connection, "meter_source") < 1: raise RuntimeError("DSMR source adoption source audit failed") def downgrade() -> None: # Schema-only downgrade for isolated test databases. It intentionally does # not delete migration-created source/channel/binding rows. with op.batch_alter_table("dsmr_reading", schema=None) as batch_op: batch_op.drop_index("ix_dsmr_reading_meter_source_id") batch_op.drop_constraint("uq_dsmr_reading_source_recorded_at", type_="unique") batch_op.drop_constraint("fk_dsmr_reading_meter_source_id", type_="foreignkey") batch_op.drop_column("meter_source_id") batch_op.alter_column("telegram_id", new_column_name="source_id") batch_op.create_unique_constraint("uq_dsmr_reading_recorded_at", ["recorded_at"])