239 lines
9.2 KiB
Python
239 lines
9.2 KiB
Python
"""add meter table and energy_cost_period.meter_id
|
|
|
|
Introduces the ``meter`` table (one row per physical meter installation epoch)
|
|
and a nullable FK column ``energy_cost_period.meter_id`` that attributes each
|
|
billing period to a specific physical meter.
|
|
|
|
**Backfill logic (§3.7 of the M7 design doc)**:
|
|
|
|
If the database already contains any ``dsmr_reading`` or
|
|
``energy_cost_period`` rows, one initial ``meter`` row is created:
|
|
|
|
label = "Initial meter"
|
|
commodity = "electricity"
|
|
started_at = earliest dsmr_reading.recorded_at
|
|
(or, if none, earliest energy_cost_period.period_start,
|
|
or, if still none, the migration timestamp)
|
|
ended_at = NULL (still active)
|
|
reason = "initial"
|
|
|
|
All existing ``energy_cost_period`` rows are then back-filled with that
|
|
initial meter's id.
|
|
|
|
**Idempotency**: the backfill is guarded with a check for any existing
|
|
``meter`` row whose ``reason = 'initial'`` and ``commodity = 'electricity'``
|
|
and ``ended_at IS NULL``, so repeating the upgrade does not create duplicate
|
|
meters or overwrite already-filled meter_id values.
|
|
|
|
**Audit (on-non-degraded periods only)**: after backfilling, the number of
|
|
non-degraded ``energy_cost_period`` rows with ``meter_id IS NULL`` must be
|
|
zero; if it is not, the migration raises a ``RuntimeError`` and rolls back.
|
|
|
|
**Data safety**: this migration is additive only — no existing rows are
|
|
deleted or overwritten; it only creates a new table, adds a nullable column,
|
|
and back-fills that column.
|
|
|
|
Revision ID: 20260625_13_meter_table
|
|
Revises: 20260624_12_dsmr_decouple_telegram_id
|
|
Create Date: 2026-06-25 00:00:00.000000
|
|
"""
|
|
|
|
from datetime import datetime, timezone
|
|
from typing import Sequence, Union
|
|
|
|
import sqlalchemy as sa
|
|
from alembic import op
|
|
|
|
revision: str = "20260625_13_meter_table"
|
|
down_revision: Union[str, None] = "20260624_12_dsmr_decouple_telegram_id"
|
|
branch_labels: Union[str, Sequence[str], None] = None
|
|
depends_on: Union[str, Sequence[str], None] = None
|
|
|
|
|
|
def upgrade() -> None:
|
|
# ------------------------------------------------------------------ #
|
|
# 1. Create the meter table. #
|
|
# ------------------------------------------------------------------ #
|
|
op.create_table(
|
|
"meter",
|
|
sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
|
|
sa.Column("label", sa.String(length=255), nullable=False),
|
|
sa.Column("commodity", sa.String(length=32), nullable=False),
|
|
sa.Column("started_at", sa.DateTime(timezone=True), nullable=False),
|
|
sa.Column("ended_at", sa.DateTime(timezone=True), nullable=True),
|
|
sa.Column("reason", sa.String(length=64), nullable=False),
|
|
sa.Column("note", sa.String(length=1024), nullable=True),
|
|
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
|
|
sa.PrimaryKeyConstraint("id"),
|
|
)
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# 2. Add meter_id column to energy_cost_period (nullable FK). #
|
|
# ------------------------------------------------------------------ #
|
|
with op.batch_alter_table("energy_cost_period", schema=None) as batch_op:
|
|
batch_op.add_column(
|
|
sa.Column("meter_id", sa.Integer(), nullable=True)
|
|
)
|
|
batch_op.create_foreign_key(
|
|
"fk_energy_cost_period_meter_id",
|
|
"meter",
|
|
["meter_id"],
|
|
["id"],
|
|
ondelete="RESTRICT",
|
|
)
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# 3. Backfill initial meter (idempotent). #
|
|
# ------------------------------------------------------------------ #
|
|
conn = op.get_bind()
|
|
|
|
# Check if there is already an initial meter (idempotency guard).
|
|
existing_initial = conn.execute(
|
|
sa.text(
|
|
"SELECT id FROM meter "
|
|
"WHERE reason = 'initial' AND commodity = 'electricity' AND ended_at IS NULL "
|
|
"LIMIT 1"
|
|
)
|
|
).fetchone()
|
|
|
|
if existing_initial is not None:
|
|
# Already backfilled — nothing to do.
|
|
return
|
|
|
|
# Determine whether there is any historical data to create a meter for.
|
|
has_readings = conn.execute(
|
|
sa.text("SELECT 1 FROM dsmr_reading LIMIT 1")
|
|
).fetchone()
|
|
has_periods = conn.execute(
|
|
sa.text("SELECT 1 FROM energy_cost_period LIMIT 1")
|
|
).fetchone()
|
|
|
|
if not has_readings and not has_periods:
|
|
# Empty database: no historical data, so no initial meter is needed.
|
|
# meter_id will remain NULL on any future rows until T02 service layer
|
|
# starts populating it.
|
|
return
|
|
|
|
# Determine started_at: earliest dsmr_reading.recorded_at, falling back to
|
|
# earliest energy_cost_period.period_start, and finally to now().
|
|
earliest_reading_row = conn.execute(
|
|
sa.text("SELECT MIN(recorded_at) AS ts FROM dsmr_reading")
|
|
).fetchone()
|
|
earliest_period_row = conn.execute(
|
|
sa.text("SELECT MIN(period_start) AS ts FROM energy_cost_period")
|
|
).fetchone()
|
|
|
|
started_at_value: datetime | None = None
|
|
if earliest_reading_row and earliest_reading_row[0] is not None:
|
|
# SQLite returns ISO strings for datetime columns; parse to datetime.
|
|
raw = earliest_reading_row[0]
|
|
started_at_value = _parse_sqlite_datetime(raw)
|
|
if started_at_value is None and earliest_period_row and earliest_period_row[0] is not None:
|
|
raw = earliest_period_row[0]
|
|
started_at_value = _parse_sqlite_datetime(raw)
|
|
if started_at_value is None:
|
|
started_at_value = datetime.now(tz=timezone.utc)
|
|
|
|
now_utc = datetime.now(tz=timezone.utc)
|
|
|
|
# Insert the initial meter row.
|
|
conn.execute(
|
|
sa.text(
|
|
"INSERT INTO meter (label, commodity, started_at, ended_at, reason, note, created_at) "
|
|
"VALUES (:label, :commodity, :started_at, NULL, :reason, NULL, :created_at)"
|
|
),
|
|
{
|
|
"label": "Initial meter",
|
|
"commodity": "electricity",
|
|
"started_at": _iso(started_at_value),
|
|
"reason": "initial",
|
|
"created_at": _iso(now_utc),
|
|
},
|
|
)
|
|
|
|
# Retrieve the newly created meter id.
|
|
meter_row = conn.execute(
|
|
sa.text(
|
|
"SELECT id FROM meter "
|
|
"WHERE reason = 'initial' AND commodity = 'electricity' AND ended_at IS NULL "
|
|
"LIMIT 1"
|
|
)
|
|
).fetchone()
|
|
assert meter_row is not None, "Initial meter row not found after insert"
|
|
meter_id: int = meter_row[0]
|
|
|
|
# Back-fill all existing energy_cost_period rows that have meter_id IS NULL.
|
|
conn.execute(
|
|
sa.text(
|
|
"UPDATE energy_cost_period SET meter_id = :mid WHERE meter_id IS NULL"
|
|
),
|
|
{"mid": meter_id},
|
|
)
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# 4. Audit: verify all non-degraded periods have a meter_id. #
|
|
# ------------------------------------------------------------------ #
|
|
unmatched_row = conn.execute(
|
|
sa.text(
|
|
"SELECT COUNT(*) FROM energy_cost_period "
|
|
"WHERE meter_id IS NULL AND degraded = 0"
|
|
)
|
|
).fetchone()
|
|
unmatched_count: int = unmatched_row[0] if unmatched_row else 0
|
|
|
|
if unmatched_count != 0:
|
|
raise RuntimeError(
|
|
f"Meter backfill audit failed: {unmatched_count} non-degraded "
|
|
"energy_cost_period row(s) still have meter_id IS NULL after backfill. "
|
|
"Migration aborted to protect data integrity."
|
|
)
|
|
|
|
|
|
def downgrade() -> None:
|
|
# Remove the FK column from energy_cost_period first (references meter).
|
|
with op.batch_alter_table("energy_cost_period", schema=None) as batch_op:
|
|
batch_op.drop_constraint("fk_energy_cost_period_meter_id", type_="foreignkey")
|
|
batch_op.drop_column("meter_id")
|
|
|
|
# Drop the meter table.
|
|
op.drop_table("meter")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _parse_sqlite_datetime(value: str | datetime) -> datetime:
|
|
"""Parse a SQLite datetime value into an aware UTC datetime.
|
|
|
|
SQLite stores datetimes as ISO 8601 strings. SQLAlchemy may return them
|
|
as plain strings or as naive datetimes (no tzinfo) depending on the driver
|
|
and column declaration. This helper normalises both forms to an aware UTC
|
|
``datetime``.
|
|
"""
|
|
if isinstance(value, datetime):
|
|
if value.tzinfo is None:
|
|
return value.replace(tzinfo=timezone.utc)
|
|
return value
|
|
# String form — strip trailing Z or +00:00 variants, then attach UTC.
|
|
s = str(value).strip()
|
|
for suffix in ("+00:00", "Z", " UTC"):
|
|
if s.endswith(suffix):
|
|
s = s[: -len(suffix)]
|
|
# SQLite uses space as the T separator in some formats.
|
|
s = s.replace(" ", "T")
|
|
try:
|
|
dt = datetime.fromisoformat(s)
|
|
except ValueError:
|
|
# Fallback: strip subseconds if present to handle unusual formats.
|
|
dt = datetime.strptime(s[:19], "%Y-%m-%dT%H:%M:%S")
|
|
return dt.replace(tzinfo=timezone.utc)
|
|
|
|
|
|
def _iso(dt: datetime) -> str:
|
|
"""Serialise a datetime to an ISO 8601 string for SQLite storage."""
|
|
if dt.tzinfo is not None:
|
|
dt = dt.astimezone(timezone.utc).replace(tzinfo=None)
|
|
return dt.strftime("%Y-%m-%dT%H:%M:%S")
|