M8-T06: add meter source management API

2026-08-23 21:22:05 +02:00
parent 1ea2f659e0
commit 5855fff451
10 changed files with 3857 additions and 8 deletions
@@ -0,0 +1,259 @@
+"""Authenticated HTTP contract for meter sources, channels, and bindings."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.api.routes.api.deps import require_csrf, require_session
+from app.dependencies import get_db
+from app.integrations.meter_sources import SourceProfileError, list_source_profiles, sanitize_source_config
+from app.models.energy import DsmrReading, Meter
+from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel
+from app.schemas.meter_source import (
+    BindingCreate, BindingListResponse, BindingPatch, BindingResponse, ChannelReadingResponse,
+    ChannelReadingsResponse, CommoditiesResponse, CommodityResponse, DiscoverResponse,
+    MeterSourceChannelListResponse, MeterSourceChannelResponse, MeterSourceCreate,
+    MeterSourceListResponse, MeterSourcePatch, MeterSourceResponse, SourceConfigFieldResponse,
+    SourceProfileResponse, SourceProfilesResponse,
+)
+from app.services.auth import AuthenticatedSession
+from app.services.meter_sources import (
+    BindingNotFoundError, ChannelNotFoundError, MeterNotFoundError,
+    MeterSourceError, SourceDeleteRestrictedError, SourceNotFoundError, create_binding,
+    create_source, delete_source, list_bindings, list_sources, update_binding, update_source,
+)
+from app.services import timezone as _tz_mod
+
+router = APIRouter(prefix="/api/energy", tags=["api-energy-meter-sources"])
+
+
+def _as_utc(value: datetime) -> datetime:
+    if value.tzinfo is None:
+        return value.replace(tzinfo=_tz_mod.local_tz()).astimezone(UTC)
+    return value.astimezone(UTC)
+
+
+def _source_or_404(db: Session, uuid: str) -> MeterSource:
+    source = db.execute(select(MeterSource).where(MeterSource.uuid == uuid)).scalar_one_or_none()
+    if source is None:
+        raise HTTPException(status_code=404, detail="Meter source not found.")
+    return source
+
+
+def _channel_or_404(db: Session, source: MeterSource, uuid: str) -> MeterSourceChannel:
+    channel = db.execute(
+        select(MeterSourceChannel).where(
+            MeterSourceChannel.uuid == uuid, MeterSourceChannel.source_id == source.id
+        )
+    ).scalar_one_or_none()
+    if channel is None:
+        raise HTTPException(status_code=404, detail="Meter source channel not found.")
+    return channel
+
+
+def _source_response(source: MeterSource) -> MeterSourceResponse:
+    return MeterSourceResponse(
+        uuid=source.uuid, name=source.name, kind=source.kind, enabled=source.enabled,
+        config=sanitize_source_config(source.kind, source.config), status=source.status,
+        last_seen_at=source.last_seen_at, last_error=source.last_error,
+        created_at=source.created_at, updated_at=source.updated_at,
+    )
+
+
+def binding_response(binding: MeterSourceBinding) -> BindingResponse:
+    return BindingResponse(
+        uuid=binding.uuid, meter_id=binding.meter_id, source_channel_uuid=binding.channel.uuid,
+        source_uuid=binding.channel.source.uuid, started_at=binding.started_at, ended_at=binding.ended_at,
+        created_at=binding.created_at, updated_at=binding.updated_at,
+    )
+
+
+def _binding_error(exc: MeterSourceError) -> HTTPException:
+    if isinstance(exc, (SourceNotFoundError, ChannelNotFoundError, MeterNotFoundError, BindingNotFoundError)):
+        return HTTPException(status_code=404, detail=str(exc))
+    return HTTPException(status_code=422, detail=str(exc))
+
+
+@router.get("/source-profiles", response_model=SourceProfilesResponse)
+def source_profiles(_auth: AuthenticatedSession = Depends(require_session)) -> SourceProfilesResponse:
+    """Return profile metadata; default secrets are never populated with stored values."""
+    profiles = []
+    for profile in list_source_profiles():
+        fields = [
+            SourceConfigFieldResponse(name=f.name, value_type=f.value_type.__name__, default=f.default,
+                                      required=f.required, secret=f.secret)
+            for f in profile.fields
+        ]
+        profiles.append(SourceProfileResponse(
+            kind=profile.kind, fields=fields,
+            defaults={f.name: f.default for f in profile.fields if not f.required},
+            capabilities=sorted(profile.capabilities), allowed_units=sorted(profile.allowed_units),
+        ))
+    return SourceProfilesResponse(items=profiles)
+
+
+@router.get("/commodities", response_model=CommoditiesResponse)
+def commodities(_auth: AuthenticatedSession = Depends(require_session)) -> CommoditiesResponse:
+    return CommoditiesResponse(items=[
+        CommodityResponse(key="electricity", unit="kWh", capabilities=["meter", "binding", "cost"]),
+        CommodityResponse(key="heating", unit="GJ", capabilities=["meter", "binding"]),
+        CommodityResponse(key="hot_water", unit="m³", capabilities=["meter", "binding"]),
+    ])
+
+
+@router.get("/sources", response_model=MeterSourceListResponse)
+def get_sources(db: Session = Depends(get_db), _auth: AuthenticatedSession = Depends(require_session)) -> MeterSourceListResponse:
+    items = [_source_response(source) for source in list_sources(db)]
+    return MeterSourceListResponse(items=items, total=len(items))
+
+
+@router.post("/sources", response_model=MeterSourceResponse, status_code=status.HTTP_201_CREATED)
+def post_source(body: MeterSourceCreate, db: Session = Depends(get_db),
+                _auth: AuthenticatedSession = Depends(require_session), _csrf: None = Depends(require_csrf)) -> MeterSourceResponse:
+    try:
+        source = create_source(db, name=body.name, kind=body.kind, config=body.config, enabled=body.enabled)
+        db.commit()
+        db.refresh(source)
+        return _source_response(source)
+    except (SourceProfileError, MeterSourceError) as exc:
+        db.rollback()
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+
+
+@router.get("/sources/{source_uuid}", response_model=MeterSourceResponse)
+def get_source_detail(source_uuid: str, db: Session = Depends(get_db),
+                      _auth: AuthenticatedSession = Depends(require_session)) -> MeterSourceResponse:
+    return _source_response(_source_or_404(db, source_uuid))
+
+
+@router.patch("/sources/{source_uuid}", response_model=MeterSourceResponse)
+def patch_source(source_uuid: str, body: MeterSourcePatch, db: Session = Depends(get_db),
+                 _auth: AuthenticatedSession = Depends(require_session), _csrf: None = Depends(require_csrf)) -> MeterSourceResponse:
+    source = _source_or_404(db, source_uuid)
+    try:
+        updated = update_source(db, source.id, name=body.name, enabled=body.enabled, config_patch=body.config)
+        db.commit()
+        db.refresh(updated)
+        return _source_response(updated)
+    except (SourceProfileError, MeterSourceError) as exc:
+        db.rollback()
+        raise _binding_error(exc) from exc
+
+
+@router.delete(
+    "/sources/{source_uuid}", status_code=status.HTTP_204_NO_CONTENT, response_model=None
+)
+def remove_source(source_uuid: str, db: Session = Depends(get_db),
+                  _auth: AuthenticatedSession = Depends(require_session), _csrf: None = Depends(require_csrf)) -> None:
+    source = _source_or_404(db, source_uuid)
+    # DSMR readings are not a relationship on MeterSource to avoid loading a large history.
+    if db.execute(select(DsmrReading.id).where(DsmrReading.meter_source_id == source.id).limit(1)).scalar() is not None:
+        raise HTTPException(status_code=409, detail="Meter source has dependent readings.")
+    try:
+        delete_source(db, source.id)
+        db.commit()
+        return Response(status_code=status.HTTP_204_NO_CONTENT)
+    except SourceDeleteRestrictedError as exc:
+        db.rollback()
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
+
+
+@router.post("/sources/{source_uuid}/discover", response_model=DiscoverResponse)
+def discover_source(source_uuid: str, db: Session = Depends(get_db),
+                    _auth: AuthenticatedSession = Depends(require_session), _csrf: None = Depends(require_csrf)) -> DiscoverResponse:
+    source = _source_or_404(db, source_uuid)
+    if source.kind == "warmtelink_serial":
+        return DiscoverResponse(requested=False, supported=False, status="not_implemented",
+                                detail="Serial discovery is available after the WarmteLink worker is installed.")
+    return DiscoverResponse(requested=False, supported=True, status="managed_by_runtime",
+                            detail="This source is discovered by its runtime subscription; no connection was opened.")
+
+
+@router.get("/sources/{source_uuid}/channels", response_model=MeterSourceChannelListResponse)
+def source_channels(source_uuid: str, db: Session = Depends(get_db),
+                    _auth: AuthenticatedSession = Depends(require_session)) -> MeterSourceChannelListResponse:
+    source = _source_or_404(db, source_uuid)
+    channels = db.execute(select(MeterSourceChannel).where(MeterSourceChannel.source_id == source.id)).scalars().all()
+    items = []
+    for channel in channels:
+        bindings = list_bindings(db, channel_id=channel.id)
+        items.append(MeterSourceChannelResponse(
+            uuid=channel.uuid, label=channel.label, suggested_commodity=channel.suggested_commodity,
+            unit=channel.unit, device_type=channel.device_type, latest_value=channel.latest_value,
+            latest_at=channel.latest_at, latest_quality=channel.latest_quality, binding_count=len(bindings),
+            bound_meter_ids=[binding.meter_id for binding in bindings],
+        ))
+    return MeterSourceChannelListResponse(items=items, total=len(items))
+
+
+@router.get("/sources/{source_uuid}/channels/{channel_uuid}/readings", response_model=ChannelReadingsResponse)
+def channel_readings(source_uuid: str, channel_uuid: str, limit: int = Query(default=100, ge=1, le=1000),
+                     start: datetime | None = None, end: datetime | None = None, db: Session = Depends(get_db),
+                     _auth: AuthenticatedSession = Depends(require_session)) -> ChannelReadingsResponse:
+    source = _source_or_404(db, source_uuid)
+    _channel_or_404(db, source, channel_uuid)
+    if source.kind != "dsmr_mqtt":
+        return ChannelReadingsResponse(items=[], total=0)
+    statement = select(DsmrReading).where(DsmrReading.meter_source_id == source.id)
+    if start is not None:
+        statement = statement.where(DsmrReading.recorded_at >= _as_utc(start))
+    if end is not None:
+        statement = statement.where(DsmrReading.recorded_at < _as_utc(end))
+    rows = list(db.execute(statement.order_by(DsmrReading.recorded_at.desc()).limit(limit)).scalars())
+    # The DSMR payload remains available only from its legacy compatibility endpoint;
+    # this generic endpoint intentionally exposes no telegram/equipment identifiers.
+    return ChannelReadingsResponse(items=[ChannelReadingResponse(recorded_at=row.recorded_at) for row in rows], total=len(rows))
+
+
+@router.get("/meters/{meter_id}/bindings", response_model=BindingListResponse)
+def meter_bindings(meter_id: int, db: Session = Depends(get_db),
+                   _auth: AuthenticatedSession = Depends(require_session)) -> BindingListResponse:
+    if db.get(Meter, meter_id) is None:
+        raise HTTPException(status_code=404, detail="Meter not found.")
+    items = [binding_response(binding) for binding in list_bindings(db, meter_id=meter_id)]
+    return BindingListResponse(items=items, total=len(items))
+
+
+@router.post("/meters/{meter_id}/bindings", response_model=BindingResponse, status_code=status.HTTP_201_CREATED)
+def post_meter_binding(meter_id: int, body: BindingCreate, db: Session = Depends(get_db),
+                       _auth: AuthenticatedSession = Depends(require_session), _csrf: None = Depends(require_csrf)) -> BindingResponse:
+    channel = db.execute(select(MeterSourceChannel).where(MeterSourceChannel.uuid == body.source_channel_uuid)).scalar_one_or_none()
+    if channel is None:
+        raise HTTPException(status_code=404, detail="Meter source channel not found.")
+    try:
+        binding = create_binding(db, meter_id=meter_id, channel_id=channel.id, started_at=_as_utc(body.started_at),
+                                 ended_at=_as_utc(body.ended_at) if body.ended_at else None)
+        db.commit()
+        db.refresh(binding)
+        return binding_response(binding)
+    except MeterSourceError as exc:
+        db.rollback()
+        raise _binding_error(exc) from exc
+
+
+@router.patch("/bindings/{binding_uuid}", response_model=BindingResponse)
+def patch_binding(binding_uuid: str, body: BindingPatch, db: Session = Depends(get_db),
+                  _auth: AuthenticatedSession = Depends(require_session), _csrf: None = Depends(require_csrf)) -> BindingResponse:
+    binding = db.execute(select(MeterSourceBinding).where(MeterSourceBinding.uuid == binding_uuid)).scalar_one_or_none()
+    if binding is None:
+        raise HTTPException(status_code=404, detail="Meter source binding not found.")
+    try:
+        # ``ended_at`` has three meaningful states in the service layer: omitted
+        # keeps the existing boundary, null reopens the interval, and a datetime
+        # changes the exclusive end.  Do not collapse omitted into null here.
+        changes: dict[str, datetime | None] = {}
+        if "started_at" in body.model_fields_set:
+            changes["started_at"] = _as_utc(body.started_at) if body.started_at is not None else None
+        if "ended_at" in body.model_fields_set:
+            changes["ended_at"] = _as_utc(body.ended_at) if body.ended_at is not None else None
+        updated = update_binding(db, binding.id, **changes)
+        db.commit()
+        db.refresh(updated)
+        return binding_response(updated)
+    except MeterSourceError as exc:
+        db.rollback()
+        raise _binding_error(exc) from exc
@@ -48,16 +48,20 @@ from typing import Optional

 from fastapi import APIRouter, Depends, HTTPException, status
 from sqlalchemy.orm import Session
+from sqlalchemy import select

 from app.api.routes.api.deps import require_csrf, require_session
 from app.dependencies import get_db
 from app.models.energy import Meter
+from app.models.meter_source import MeterSourceChannel
 from app.schemas.meter import (
    MeterDeclareRequest,
+    MeterBindingSummary,
    MeterListResponse,
    MeterPatchRequest,
    MeterResponse,
 )
+from app.services.meter_sources import ChannelNotFoundError, MeterSourceError, create_binding
 from app.services import timezone as _tz_mod
 from app.services.auth import AuthenticatedSession
 from app.services.energy_cost import recompute_range
@@ -153,6 +157,22 @@ def _trigger_recompute(db: Session, start: datetime, label: str) -> int:
    return n


+def _meter_response(meter: Meter) -> MeterResponse:
+    """Serialize meter plus binding summaries without exposing source config."""
+    response = MeterResponse.model_validate(meter)
+    response.bindings = [
+        MeterBindingSummary(
+            uuid=binding.uuid,
+            source_channel_uuid=binding.channel.uuid,
+            source_uuid=binding.channel.source.uuid,
+            started_at=binding.started_at,
+            ended_at=binding.ended_at,
+        )
+        for binding in meter.source_bindings
+    ]
+    return response
+
+
 # ---------------------------------------------------------------------------
 # GET /api/energy/meters
 # ---------------------------------------------------------------------------
@@ -170,7 +190,7 @@ def list_energy_meters(
    has the latest ``started_at``.
    """
    meters = list_meters(db)
-    items = [MeterResponse.model_validate(m) for m in meters]
+    items = [_meter_response(m) for m in meters]
    return MeterListResponse(items=items, total=len(items))


@@ -219,14 +239,29 @@ def declare_energy_meter(
            commodity=body.commodity,
            note=body.note,
        )
-    except MeterOverlapError as exc:
+        db.flush()  # assign PK before an optional binding and recompute
+        if body.source_channel_uuid is not None:
+            channel = db.execute(
+                select(MeterSourceChannel).where(MeterSourceChannel.uuid == body.source_channel_uuid)
+            ).scalar_one_or_none()
+            if channel is None:
+                raise ChannelNotFoundError("Meter source channel was not found.")
+            create_binding(
+                db,
+                meter_id=new_meter.id,
+                channel_id=channel.id,
+                started_at=started_at_utc,
+            )
+    except (MeterOverlapError, MeterSourceError) as exc:
+        # declare_meter may already have closed the previous epoch.  Rolling back
+        # here makes Meter + binding declaration genuinely atomic.
+        db.rollback()
        raise HTTPException(
-            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            status_code=(status.HTTP_404_NOT_FOUND if isinstance(exc, ChannelNotFoundError)
+                         else status.HTTP_422_UNPROCESSABLE_ENTITY),
            detail=str(exc),
        )

-    db.flush()  # assign PK before recompute (recompute uses session, needs meter in DB)
-
    # Retroactive recompute: re-judge attribution from the new boundary onward.
    now = datetime.now(UTC)
    if started_at_utc < now:
@@ -247,7 +282,7 @@ def declare_energy_meter(
        new_meter.label,
        started_at_utc.isoformat(),
    )
-    return MeterResponse.model_validate(new_meter)
+    return _meter_response(new_meter)


 # ---------------------------------------------------------------------------
@@ -328,4 +363,4 @@ def patch_energy_meter(
        meter.label,
        meter.started_at,
    )
-    return MeterResponse.model_validate(meter)
+    return _meter_response(meter)