530 lines
19 KiB
Python
530 lines
19 KiB
Python
"""Service layer for source/channel discovery and meter-source bindings.
|
|
|
|
All mutating functions receive a caller-owned :class:`~sqlalchemy.orm.Session`
|
|
and never commit. This lets HTTP handlers compose source and meter changes in
|
|
one transaction later without exposing any connection I/O here.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from sqlalchemy import or_, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.integrations.meter_sources import (
|
|
SourceProfileError,
|
|
get_source_profile,
|
|
merge_source_config,
|
|
validate_source_config,
|
|
)
|
|
from app.models.energy import Meter
|
|
from app.models.meter_source import (
|
|
MeterSource,
|
|
MeterSourceBinding,
|
|
MeterSourceChannel,
|
|
half_open_intervals_overlap,
|
|
)
|
|
|
|
|
|
class MeterSourceError(ValueError):
|
|
"""Base class for source-domain validation errors."""
|
|
|
|
|
|
class SourceNotFoundError(MeterSourceError):
|
|
"""Raised when the requested source does not exist."""
|
|
|
|
|
|
class ChannelNotFoundError(MeterSourceError):
|
|
"""Raised when the requested source channel does not exist."""
|
|
|
|
|
|
class MeterNotFoundError(MeterSourceError):
|
|
"""Raised when the requested meter does not exist."""
|
|
|
|
|
|
class BindingNotFoundError(MeterSourceError):
|
|
"""Raised when the requested binding does not exist."""
|
|
|
|
|
|
class BindingValidationError(MeterSourceError):
|
|
"""Raised for an incompatible unit, commodity, or invalid interval."""
|
|
|
|
|
|
class BindingOverlapError(BindingValidationError):
|
|
"""Raised when a meter or channel already has an overlapping binding."""
|
|
|
|
|
|
class SourceDeleteRestrictedError(MeterSourceError):
|
|
"""Raised when a source has retained channel, binding, or reading history."""
|
|
|
|
|
|
COMMODITY_UNITS = {"electricity": "kWh", "heating": "GJ", "hot_water": "m³"}
|
|
_UNSET = object()
|
|
|
|
|
|
def _utc_now() -> datetime:
|
|
return datetime.now(UTC)
|
|
|
|
|
|
def _as_utc(value: datetime) -> datetime:
|
|
return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)
|
|
|
|
|
|
def get_source(session: Session, source_id: int) -> MeterSource:
|
|
source = session.get(MeterSource, source_id)
|
|
if source is None:
|
|
raise SourceNotFoundError(f"Meter source {source_id} was not found.")
|
|
return source
|
|
|
|
|
|
def list_sources(session: Session, *, kind: str | None = None) -> list[MeterSource]:
|
|
statement = select(MeterSource).order_by(MeterSource.id)
|
|
if kind is not None:
|
|
get_source_profile(kind)
|
|
statement = statement.where(MeterSource.kind == kind)
|
|
return list(session.execute(statement).scalars())
|
|
|
|
|
|
def create_source(
|
|
session: Session,
|
|
*,
|
|
name: str,
|
|
kind: str,
|
|
config: dict[str, Any],
|
|
enabled: bool = True,
|
|
) -> MeterSource:
|
|
"""Add a source after validating its complete kind-specific config."""
|
|
now = _utc_now()
|
|
source = MeterSource(
|
|
name=name,
|
|
kind=kind,
|
|
enabled=enabled,
|
|
config=validate_source_config(kind, config),
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(source)
|
|
return source
|
|
|
|
|
|
def update_source(
|
|
session: Session,
|
|
source_id: int,
|
|
*,
|
|
name: str | None = None,
|
|
enabled: bool | None = None,
|
|
config_patch: dict[str, Any] | None = None,
|
|
) -> MeterSource:
|
|
"""Update source metadata and/or merge a partial source config without commit."""
|
|
source = get_source(session, source_id)
|
|
if name is not None:
|
|
source.name = name
|
|
if enabled is not None:
|
|
source.enabled = enabled
|
|
if config_patch is not None:
|
|
source.config = merge_source_config(source.kind, source.config, config_patch)
|
|
source.updated_at = _utc_now()
|
|
return source
|
|
|
|
|
|
def delete_source(session: Session, source_id: int) -> None:
|
|
"""Delete an entirely unused source; history is always retained instead."""
|
|
source = get_source(session, source_id)
|
|
has_channel = session.execute(
|
|
select(MeterSourceChannel.id).where(MeterSourceChannel.source_id == source.id).limit(1)
|
|
).scalar_one_or_none()
|
|
has_binding = session.execute(
|
|
select(MeterSourceBinding.id)
|
|
.join(MeterSourceChannel)
|
|
.where(MeterSourceChannel.source_id == source.id)
|
|
.limit(1)
|
|
).scalar_one_or_none()
|
|
if has_channel is not None or has_binding is not None:
|
|
raise SourceDeleteRestrictedError(
|
|
f"Meter source {source_id} has dependent channels, bindings, or readings."
|
|
)
|
|
session.delete(source)
|
|
|
|
|
|
def get_channel(session: Session, channel_id: int) -> MeterSourceChannel:
|
|
channel = session.get(MeterSourceChannel, channel_id)
|
|
if channel is None:
|
|
raise ChannelNotFoundError(f"Meter source channel {channel_id} was not found.")
|
|
return channel
|
|
|
|
|
|
def upsert_discovered_channel(
|
|
session: Session,
|
|
*,
|
|
source_id: int,
|
|
channel_key: str,
|
|
label: str,
|
|
unit: str,
|
|
suggested_commodity: str | None = None,
|
|
device_type: str | None = None,
|
|
fingerprint: str | None = None,
|
|
latest_value: Any = None,
|
|
latest_at: datetime | None = None,
|
|
latest_quality: str | None = None,
|
|
) -> MeterSourceChannel:
|
|
"""Idempotently create or refresh a discovered channel's metadata.
|
|
|
|
``suggested_commodity`` remains metadata only; this function never creates
|
|
a meter or a binding.
|
|
"""
|
|
source = get_source(session, source_id)
|
|
if unit not in get_source_profile(source.kind).allowed_units:
|
|
raise SourceProfileError(f"Unit {unit!r} is not allowed for source kind {source.kind!r}.")
|
|
channel = session.execute(
|
|
select(MeterSourceChannel).where(
|
|
MeterSourceChannel.source_id == source.id,
|
|
MeterSourceChannel.channel_key == channel_key,
|
|
)
|
|
).scalar_one_or_none()
|
|
now = _utc_now()
|
|
if channel is None:
|
|
channel = MeterSourceChannel(
|
|
source_id=source.id,
|
|
channel_key=channel_key,
|
|
label=label,
|
|
unit=unit,
|
|
suggested_commodity=suggested_commodity,
|
|
device_type=device_type,
|
|
fingerprint=fingerprint,
|
|
latest_value=latest_value,
|
|
latest_at=latest_at,
|
|
latest_quality=latest_quality,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(channel)
|
|
return channel
|
|
|
|
if channel.unit != unit:
|
|
binding_id = session.execute(
|
|
select(MeterSourceBinding.id)
|
|
.where(MeterSourceBinding.channel_id == channel.id)
|
|
.limit(1)
|
|
).scalar_one_or_none()
|
|
if binding_id is not None:
|
|
raise BindingValidationError(
|
|
f"Cannot change unit of bound channel {channel.id} from {channel.unit!r} to {unit!r}."
|
|
)
|
|
|
|
channel.label = label
|
|
channel.unit = unit
|
|
channel.suggested_commodity = suggested_commodity
|
|
channel.device_type = device_type
|
|
channel.fingerprint = fingerprint
|
|
channel.latest_value = latest_value
|
|
channel.latest_at = latest_at
|
|
channel.latest_quality = latest_quality
|
|
channel.updated_at = now
|
|
return channel
|
|
|
|
|
|
def list_bindings(
|
|
session: Session, *, meter_id: int | None = None, channel_id: int | None = None
|
|
) -> list[MeterSourceBinding]:
|
|
statement = select(MeterSourceBinding).order_by(MeterSourceBinding.started_at, MeterSourceBinding.id)
|
|
if meter_id is not None:
|
|
statement = statement.where(MeterSourceBinding.meter_id == meter_id)
|
|
if channel_id is not None:
|
|
statement = statement.where(MeterSourceBinding.channel_id == channel_id)
|
|
return list(session.execute(statement).scalars())
|
|
|
|
|
|
def _get_meter(session: Session, meter_id: int) -> Meter:
|
|
meter = session.get(Meter, meter_id)
|
|
if meter is None:
|
|
raise MeterNotFoundError(f"Meter {meter_id} was not found.")
|
|
return meter
|
|
|
|
|
|
def _validate_binding(
|
|
session: Session,
|
|
*,
|
|
meter_id: int,
|
|
channel_id: int,
|
|
started_at: datetime,
|
|
ended_at: datetime | None,
|
|
excluding_ids: set[int] | None = None,
|
|
) -> None:
|
|
meter = _get_meter(session, meter_id)
|
|
channel = get_channel(session, channel_id)
|
|
expected_unit = COMMODITY_UNITS.get(meter.commodity)
|
|
if expected_unit is None:
|
|
raise BindingValidationError(f"Commodity {meter.commodity!r} cannot be bound to a source channel.")
|
|
if channel.unit != expected_unit:
|
|
raise BindingValidationError(
|
|
f"Meter commodity {meter.commodity!r} requires unit {expected_unit!r}, "
|
|
f"but channel has {channel.unit!r}."
|
|
)
|
|
if ended_at is not None and _as_utc(ended_at) <= _as_utc(started_at):
|
|
raise BindingValidationError("Binding ended_at must be strictly after started_at.")
|
|
if _as_utc(started_at) < _as_utc(meter.started_at):
|
|
raise BindingValidationError("Binding must not start before its meter epoch.")
|
|
if meter.ended_at is None:
|
|
if ended_at is not None:
|
|
# A historical binding on an active epoch is valid, but it must be
|
|
# wholly within that epoch (whose upper bound is open).
|
|
pass
|
|
else:
|
|
meter_end = _as_utc(meter.ended_at)
|
|
if ended_at is None or _as_utc(ended_at) > meter_end:
|
|
raise BindingValidationError("Closed meter bindings must end within the meter epoch.")
|
|
|
|
excluded = excluding_ids or set()
|
|
candidates = session.execute(
|
|
select(MeterSourceBinding).where(
|
|
or_(
|
|
MeterSourceBinding.meter_id == meter_id,
|
|
MeterSourceBinding.channel_id == channel_id,
|
|
)
|
|
)
|
|
).scalars()
|
|
for existing in candidates:
|
|
if existing.id in excluded:
|
|
continue
|
|
if half_open_intervals_overlap(
|
|
_as_utc(started_at),
|
|
_as_utc(ended_at) if ended_at is not None else None,
|
|
_as_utc(existing.started_at),
|
|
_as_utc(existing.ended_at) if existing.ended_at is not None else None,
|
|
):
|
|
side = "meter" if existing.meter_id == meter_id else "channel"
|
|
raise BindingOverlapError(f"Binding overlaps existing {side} binding {existing.id}.")
|
|
|
|
|
|
def create_binding(
|
|
session: Session,
|
|
*,
|
|
meter_id: int,
|
|
channel_id: int,
|
|
started_at: datetime,
|
|
ended_at: datetime | None = None,
|
|
) -> MeterSourceBinding:
|
|
"""Create a compatible non-overlapping half-open source binding."""
|
|
now = _utc_now()
|
|
if _as_utc(started_at) > now or (ended_at is not None and _as_utc(ended_at) > now):
|
|
raise BindingValidationError("Binding boundaries must not be in the future.")
|
|
_validate_binding(
|
|
session,
|
|
meter_id=meter_id,
|
|
channel_id=channel_id,
|
|
started_at=started_at,
|
|
ended_at=ended_at,
|
|
)
|
|
binding = MeterSourceBinding(
|
|
meter_id=meter_id,
|
|
channel_id=channel_id,
|
|
started_at=started_at,
|
|
ended_at=ended_at,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(binding)
|
|
return binding
|
|
|
|
|
|
def create_binding_for_meter_swap(
|
|
session: Session,
|
|
*,
|
|
old_meter_id: int | None,
|
|
new_meter_id: int,
|
|
channel_id: int,
|
|
started_at: datetime,
|
|
) -> MeterSourceBinding:
|
|
"""Create a binding during a physical meter swap, handing off one channel if safe.
|
|
|
|
A channel is transferable only when exactly one of its bindings covered the
|
|
instant immediately before ``started_at`` and that binding belongs to the
|
|
meter which this declaration just closed. All other occupied or ambiguous
|
|
cases retain the normal fail-closed overlap behaviour.
|
|
|
|
This function deliberately does not commit. The caller must keep the meter
|
|
declaration, binding handoff, and any billing recompute in one transaction.
|
|
"""
|
|
new_meter = _get_meter(session, new_meter_id)
|
|
channel = get_channel(session, channel_id)
|
|
expected_unit = COMMODITY_UNITS.get(new_meter.commodity)
|
|
if expected_unit is None or channel.unit != expected_unit:
|
|
raise BindingValidationError(
|
|
f"Meter commodity {new_meter.commodity!r} requires unit {expected_unit!r}, "
|
|
f"but channel has {channel.unit!r}."
|
|
)
|
|
|
|
boundary = _as_utc(started_at)
|
|
if _as_utc(new_meter.started_at) != boundary:
|
|
raise BindingValidationError(
|
|
"Meter-swap binding must start at the new meter's started_at boundary."
|
|
)
|
|
covering_bindings = [
|
|
binding
|
|
for binding in session.execute(
|
|
select(MeterSourceBinding).where(MeterSourceBinding.channel_id == channel_id)
|
|
).scalars()
|
|
if _as_utc(binding.started_at) < boundary
|
|
and (binding.ended_at is None or _as_utc(binding.ended_at) >= boundary)
|
|
]
|
|
|
|
if not covering_bindings:
|
|
return create_binding(
|
|
session,
|
|
meter_id=new_meter_id,
|
|
channel_id=channel_id,
|
|
started_at=started_at,
|
|
)
|
|
|
|
if old_meter_id is None or len(covering_bindings) != 1:
|
|
raise BindingOverlapError("Channel is occupied or has an ambiguous binding at meter swap.")
|
|
|
|
old_meter = _get_meter(session, old_meter_id)
|
|
old_binding = covering_bindings[0]
|
|
if (
|
|
old_meter.commodity != new_meter.commodity
|
|
or old_meter.ended_at is None
|
|
or _as_utc(old_meter.ended_at) != boundary
|
|
or old_binding.meter_id != old_meter.id
|
|
):
|
|
raise BindingOverlapError("Channel is occupied by a binding that cannot be handed off.")
|
|
|
|
update_binding(session, old_binding.id, ended_at=started_at)
|
|
return create_binding(
|
|
session,
|
|
meter_id=new_meter_id,
|
|
channel_id=channel_id,
|
|
started_at=started_at,
|
|
)
|
|
|
|
|
|
def update_binding(
|
|
session: Session,
|
|
binding_id: int,
|
|
*,
|
|
meter_id: int | None = None,
|
|
channel_id: int | None = None,
|
|
started_at: datetime | None = None,
|
|
ended_at: datetime | None | object = _UNSET,
|
|
) -> MeterSourceBinding:
|
|
"""Correct a binding while preserving half-open timeline constraints."""
|
|
binding = session.get(MeterSourceBinding, binding_id)
|
|
if binding is None:
|
|
raise BindingNotFoundError(f"Meter source binding {binding_id} was not found.")
|
|
new_meter_id = binding.meter_id if meter_id is None else meter_id
|
|
new_channel_id = binding.channel_id if channel_id is None else channel_id
|
|
new_started_at = binding.started_at if started_at is None else started_at
|
|
new_ended_at = binding.ended_at if ended_at is _UNSET else ended_at
|
|
now = _utc_now()
|
|
if _as_utc(new_started_at) > now or (new_ended_at is not None and _as_utc(new_ended_at) > now):
|
|
raise BindingValidationError("Binding boundaries must not be in the future.")
|
|
_validate_binding(
|
|
session,
|
|
meter_id=new_meter_id,
|
|
channel_id=new_channel_id,
|
|
started_at=new_started_at,
|
|
ended_at=new_ended_at,
|
|
excluding_ids={binding.id},
|
|
)
|
|
binding.meter_id = new_meter_id
|
|
binding.channel_id = new_channel_id
|
|
binding.started_at = new_started_at
|
|
binding.ended_at = new_ended_at
|
|
binding.updated_at = _utc_now()
|
|
return binding
|
|
|
|
|
|
def close_binding(session: Session, binding_id: int, *, ended_at: datetime) -> MeterSourceBinding:
|
|
"""Close an existing binding at its exclusive end boundary."""
|
|
return update_binding(session, binding_id, ended_at=ended_at)
|
|
|
|
|
|
def close_open_bindings_for_meter(session: Session, meter_id: int, *, ended_at: datetime) -> list[MeterSourceBinding]:
|
|
"""Close every open binding on a meter at one shared boundary."""
|
|
bindings = list(session.execute(
|
|
select(MeterSourceBinding).where(
|
|
MeterSourceBinding.meter_id == meter_id, MeterSourceBinding.ended_at.is_(None)
|
|
)
|
|
).scalars())
|
|
for binding in bindings:
|
|
update_binding(session, binding.id, ended_at=ended_at)
|
|
return bindings
|
|
|
|
|
|
def transfer_binding(
|
|
session: Session, *, target_meter_id: int, from_binding_id: int, to_channel_id: int,
|
|
effective_at: datetime,
|
|
) -> tuple[MeterSourceBinding, MeterSourceBinding]:
|
|
"""Atomically close a binding and open its replacement on the target meter."""
|
|
source = session.get(MeterSourceBinding, from_binding_id)
|
|
if source is None:
|
|
raise BindingNotFoundError(f"Meter source binding {from_binding_id} was not found.")
|
|
target = _get_meter(session, target_meter_id)
|
|
old_meter = _get_meter(session, source.meter_id)
|
|
effective_at = _as_utc(effective_at)
|
|
now = _utc_now()
|
|
if effective_at > now:
|
|
raise BindingValidationError("Binding transfer effective_at must not be in the future.")
|
|
if old_meter.commodity != target.commodity:
|
|
raise BindingValidationError("Binding transfer meters must have the same commodity.")
|
|
if source.ended_at is not None:
|
|
raise BindingValidationError("Only an open binding can be transferred.")
|
|
if old_meter.id == target.id:
|
|
close_at = effective_at
|
|
else:
|
|
# Recovery is deliberately narrow: the source meter must be the one
|
|
# and only most-recent closed predecessor in this commodity's timeline.
|
|
# A manually closed meter may leave an intentional epoch gap before the
|
|
# target is declared, so adjacency is not required.
|
|
if old_meter.ended_at is None:
|
|
raise BindingValidationError("Source binding must belong to a closed predecessor meter.")
|
|
timeline = list(session.execute(
|
|
select(Meter).where(Meter.commodity == target.commodity)
|
|
).scalars())
|
|
predecessors = [
|
|
meter for meter in timeline
|
|
if meter.id != target.id
|
|
and meter.ended_at is not None
|
|
and _as_utc(meter.ended_at) <= _as_utc(target.started_at)
|
|
]
|
|
if not predecessors:
|
|
raise BindingValidationError("Source meter is not the unique immediately preceding meter.")
|
|
latest_end = max(_as_utc(meter.ended_at) for meter in predecessors)
|
|
latest = [meter for meter in predecessors if _as_utc(meter.ended_at) == latest_end]
|
|
if len(latest) != 1 or latest[0].id != old_meter.id:
|
|
raise BindingValidationError("Source meter is not the unique immediately preceding meter.")
|
|
# Reject any overlapping epoch around either endpoint. A separate
|
|
# meter inside the gap is already excluded by the predecessor check;
|
|
# one extending into either endpoint is an ambiguous timeline too.
|
|
for meter in timeline:
|
|
if meter.id in {old_meter.id, target.id}:
|
|
continue
|
|
meter_end = _as_utc(meter.ended_at) if meter.ended_at is not None else None
|
|
if (
|
|
half_open_intervals_overlap(
|
|
_as_utc(old_meter.started_at), _as_utc(old_meter.ended_at),
|
|
_as_utc(meter.started_at), meter_end,
|
|
)
|
|
or half_open_intervals_overlap(
|
|
_as_utc(target.started_at),
|
|
_as_utc(target.ended_at) if target.ended_at is not None else None,
|
|
_as_utc(meter.started_at), meter_end,
|
|
)
|
|
):
|
|
raise BindingValidationError("Source meter has an ambiguous commodity timeline.")
|
|
close_at = _as_utc(old_meter.ended_at)
|
|
if effective_at < _as_utc(target.started_at):
|
|
raise BindingValidationError("Transfer effective_at must be within the target meter epoch.")
|
|
if effective_at < _as_utc(source.started_at):
|
|
raise BindingValidationError("Transfer effective_at precedes the source binding.")
|
|
# Validate the target before mutating the old row, then close/create in one session.
|
|
_validate_binding(session, meter_id=target.id, channel_id=to_channel_id,
|
|
started_at=effective_at, ended_at=None,
|
|
excluding_ids={source.id})
|
|
update_binding(session, source.id, ended_at=close_at)
|
|
created = create_binding(session, meter_id=target.id, channel_id=to_channel_id,
|
|
started_at=effective_at)
|
|
return source, created
|