home-automation/alembic_app/versions/20260822_16_dsmr_source_adoption.py

"""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"])