diff --git a/app/api/routes/api/meter_sources.py b/app/api/routes/api/meter_sources.py index 746cd81..574ed27 100644 --- a/app/api/routes/api/meter_sources.py +++ b/app/api/routes/api/meter_sources.py @@ -12,10 +12,12 @@ 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.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel, WarmteLinkReading from app.schemas.meter_source import ( - BindingCreate, BindingListResponse, BindingPatch, BindingResponse, ChannelReadingResponse, + BindingCreate, BindingListResponse, BindingPatch, BindingResponse, ChannelBindingSummaryResponse, + ChannelReadingResponse, ChannelReadingsResponse, CommoditiesResponse, CommodityResponse, DiscoverResponse, + DiscoverChannelResponse, MeterSourceChannelListResponse, MeterSourceChannelResponse, MeterSourceCreate, MeterSourceListResponse, MeterSourcePatch, MeterSourceResponse, SourceConfigFieldResponse, SourceProfileResponse, SourceProfilesResponse, @@ -181,8 +183,25 @@ 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.") + if not source.enabled: + return DiscoverResponse( + requested=False, supported=True, status="error", + detail="The WarmteLink source is disabled.", channels=_discover_channels(db, source), + ) + # This merely schedules lifecycle convergence. It never opens a serial + # descriptor or waits for a frame in the request thread; the one managed + # worker remains the sole owner of serial I/O and can keep reconnecting. + request = warmtelink_worker_manager.request_discovery(source.id) + if request.completed.is_set(): + # A worker may have accepted a frame during the bounded wait. + # Refresh only durable accepted metadata, never candidates/raw data. + db.expire_all() + source = _source_or_404(db, source_uuid) + return DiscoverResponse( + requested=request.status != "error", supported=True, status=request.status, + request_id=request.request_id or None, detail=request.detail, + channels=_discover_channels(db, source), + ) return DiscoverResponse(requested=False, supported=True, status="managed_by_runtime", detail="This source is discovered by its runtime subscription; no connection was opened.") @@ -195,32 +214,61 @@ def source_channels(source_uuid: str, db: Session = Depends(get_db), items = [] for channel in channels: bindings = list_bindings(db, channel_id=channel.id) + meter_ids = [binding.meter_id for binding in bindings] 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], + bound_meter_ids=meter_ids, + binding_summary=ChannelBindingSummaryResponse(count=len(bindings), meter_ids=meter_ids), )) - return MeterSourceChannelListResponse(items=items, total=len(items)) + return MeterSourceChannelListResponse(items=items, total=len(items), source_status=source.status) @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), + from_: datetime | None = Query(default=None, alias="from"), + to: datetime | None = Query(default=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)) + channel = _channel_or_404(db, source, channel_uuid) + if from_ is not None and to is not None and _as_utc(from_) >= _as_utc(to): + raise HTTPException(status_code=422, detail="'from' must be earlier than 'to'.") + if source.kind == "warmtelink_serial": + statement = select(WarmteLinkReading).where(WarmteLinkReading.channel_id == channel.id) + model = WarmteLinkReading + else: + # DSMR remains a source-level protocol history. Its channel is the + # public electricity identity, while payload/telegram diagnostics stay + # private to ingestion and the legacy latest endpoint. + statement = select(DsmrReading).where(DsmrReading.meter_source_id == source.id) + model = DsmrReading + if from_ is not None: + statement = statement.where(model.recorded_at >= _as_utc(from_)) + if to is not None: + statement = statement.where(model.recorded_at < _as_utc(to)) + rows = list(db.execute(statement.order_by(model.recorded_at.asc()).limit(limit)).scalars()) + return ChannelReadingsResponse( + items=[ChannelReadingResponse( + recorded_at=row.recorded_at, + value=getattr(row, "value", None), quality=getattr(row, "quality", None), + ) for row in rows], + total=len(rows), + ) + + +def _discover_channels(db: Session, source: MeterSource) -> list[DiscoverChannelResponse]: + """Return only public, accepted channel metadata for discover responses.""" + return [ + DiscoverChannelResponse( + uuid=channel.uuid, label=channel.label, unit=channel.unit, + latest_value=channel.latest_value, latest_at=channel.latest_at, + latest_quality=channel.latest_quality, + ) + for channel in db.execute( + select(MeterSourceChannel).where(MeterSourceChannel.source_id == source.id) + ).scalars() + ] @router.get("/meters/{meter_id}/bindings", response_model=BindingListResponse) diff --git a/app/schemas/meter_source.py b/app/schemas/meter_source.py index 9426d2a..9e47780 100644 --- a/app/schemas/meter_source.py +++ b/app/schemas/meter_source.py @@ -64,7 +64,23 @@ class DiscoverResponse(BaseModel): requested: bool supported: bool status: str + request_id: int | None = None detail: str | None = None + channels: list["DiscoverChannelResponse"] = Field(default_factory=list) + + +class DiscoverChannelResponse(BaseModel): + uuid: str + label: str + unit: str + latest_value: Decimal | None + latest_at: datetime | None + latest_quality: str | None + + +class ChannelBindingSummaryResponse(BaseModel): + count: int + meter_ids: list[int] class CommodityResponse(BaseModel): @@ -88,11 +104,13 @@ class MeterSourceChannelResponse(BaseModel): latest_quality: str | None binding_count: int bound_meter_ids: list[int] + binding_summary: ChannelBindingSummaryResponse class MeterSourceChannelListResponse(BaseModel): items: list[MeterSourceChannelResponse] total: int + source_status: str class ChannelReadingResponse(BaseModel): diff --git a/app/services/warmtelink_worker.py b/app/services/warmtelink_worker.py index 62cb488..04434fa 100644 --- a/app/services/warmtelink_worker.py +++ b/app/services/warmtelink_worker.py @@ -8,8 +8,8 @@ from __future__ import annotations from collections.abc import Callable from contextlib import suppress -from dataclasses import dataclass -from datetime import UTC, datetime +from dataclasses import dataclass, field +from datetime import UTC, datetime, timedelta import logging import threading from typing import Protocol @@ -27,6 +27,9 @@ logger = logging.getLogger(__name__) _BACKOFF_SECONDS = (1, 2, 4, 8, 16, 32, 60) _JOIN_TIMEOUT_SECONDS = 5 +_DISCOVERY_LOCK_TIMEOUT_SECONDS = 0.05 +_DISCOVERY_WAIT_SECONDS = 0.1 +_DISCOVERY_TIMEOUT_SECONDS = 5 class ReadOnlySerial(Protocol): @@ -69,6 +72,18 @@ class _WorkerConfig: config: dict +@dataclass +class DiscoveryRequest: + """One source-scoped request, completed only by its serial owner.""" + + request_id: int + source_id: int + deadline: datetime + status: str = "pending" + detail: str | None = None + completed: threading.Event = field(default_factory=threading.Event) + + class WarmteLinkWorker: """One interruptible, read-only serial loop for one meter source.""" @@ -88,6 +103,8 @@ class WarmteLinkWorker: self._clock = clock or _EventClock() self._serial: ReadOnlySerial | None = None self._serial_lock = threading.Lock() + self._discovery_lock = threading.Lock() + self._discoveries: list[DiscoveryRequest] = [] # Never inherit a daemon flag from a caller's background thread: a serial # descriptor and its orderly shutdown must remain visible to the process. self._thread = threading.Thread( @@ -105,6 +122,35 @@ class WarmteLinkWorker: self._stop_event.set() self._close_serial() + def request_discovery(self, request: DiscoveryRequest) -> None: + """Queue a read request; this worker remains the sole serial owner.""" + with self._discovery_lock: + self._discoveries.append(request) + + def _finish_discoveries(self, status: str, detail: str | None = None) -> None: + now = datetime.now(UTC) + with self._discovery_lock: + pending, self._discoveries = self._discoveries, [] + for request in pending: + if request.completed.is_set(): + continue + if request.deadline <= now and status == "completed": + request.status, request.detail = "error", "Discovery timed out." + else: + request.status, request.detail = status, detail + request.completed.set() + + def _expire_discoveries(self) -> None: + now = datetime.now(UTC) + with self._discovery_lock: + expired = [request for request in self._discoveries if request.deadline <= now] + self._discoveries = [request for request in self._discoveries if request.deadline > now] + for request in expired: + if request.completed.is_set(): + continue + request.status, request.detail = "error", "Discovery timed out." + request.completed.set() + def join(self, timeout: float = _JOIN_TIMEOUT_SECONDS) -> bool: self._thread.join(timeout) return not self._thread.is_alive() @@ -142,6 +188,7 @@ class WarmteLinkWorker: return self._serial = device while not self._stop_event.is_set(): + self._expire_discoveries() chunk = device.read(1024) if not chunk: # ``timeout`` reads are normal, but still yield so a bad @@ -152,13 +199,16 @@ class WarmteLinkWorker: for frame in frames: if self._stop_event.is_set(): break - self._ingestor.handle_frame( + admitted = self._ingestor.handle_frame( self.source_id, frame, session_factory=self._session_factory ) + if admitted: + self._finish_discoveries("completed") # A complete frame proves transport recovery even if its # contents are rejected by the privacy/admission layer. backoff_index = 0 except Exception: + self._finish_discoveries("error", "WarmteLink discovery failed.") self._record_error("WarmteLink serial connection failed") delay = _BACKOFF_SECONDS[min(backoff_index, len(_BACKOFF_SECONDS) - 1)] backoff_index += 1 @@ -181,6 +231,7 @@ class WarmteLinkWorkerManager: self._workers: dict[int, tuple[_WorkerConfig, WarmteLinkWorker]] = {} self._lock = threading.Lock() self._reapers: set[int] = set() + self._next_discovery_id = 0 self._shutting_down = False @property @@ -203,6 +254,48 @@ class WarmteLinkWorkerManager: self._shutting_down = False self.reconcile() + def request_discovery(self, source_id: int) -> DiscoveryRequest: + """Ask the current source worker for one bounded read/discovery attempt. + + This intentionally does not reconcile or open a descriptor. Lifecycle + convergence remains separate; a request can neither replace nor stop a + worker when an HTTP client times out or disconnects. + """ + now = datetime.now(UTC) + request = DiscoveryRequest(0, source_id, now) + if not self._lock.acquire(timeout=_DISCOVERY_LOCK_TIMEOUT_SECONDS): + request.status, request.detail = "error", "Discovery queue is busy." + request.completed.set() + return request + try: + self._next_discovery_id += 1 + request.request_id = self._next_discovery_id + request.deadline = now + timedelta(seconds=_DISCOVERY_TIMEOUT_SECONDS) + if self._shutting_down: + request.status, request.detail = "error", "WarmteLink manager is stopped." + request.completed.set() + elif (entry := self._workers.get(source_id)) is None: + request.status, request.detail = "error", "WarmteLink worker is not running." + request.completed.set() + else: + entry[1].request_discovery(request) + timer = threading.Timer(_DISCOVERY_TIMEOUT_SECONDS, self._timeout_discovery, args=(request,)) + timer.daemon = True + timer.start() + finally: + self._lock.release() + # A tiny bounded wait makes an immediately available frame observable, + # without turning an HTTP call into serial I/O or an unbounded wait. + request.completed.wait(_DISCOVERY_WAIT_SECONDS) + return request + + @staticmethod + def _timeout_discovery(request: DiscoveryRequest) -> None: + """Resolve a stale HTTP request without touching its healthy worker.""" + if not request.completed.is_set(): + request.status, request.detail = "error", "Discovery timed out." + request.completed.set() + def _read_desired(self) -> dict[int, _WorkerConfig]: with self._session_factory() as session: return { diff --git a/docs/design/m8-warmtelink-energy.md b/docs/design/m8-warmtelink-energy.md index a4046fc..3d475f1 100644 --- a/docs/design/m8-warmtelink-energy.md +++ b/docs/design/m8-warmtelink-energy.md @@ -689,7 +689,7 @@ T01～T06 先把现有 DSMR 安全迁到统一 source/binding；T07～T11 再接 ### M8-T11 — WarmteLink discover、latest 与 history API -- **Status**: `todo` +- **Status**: `done` - **Depends**: M8-T10 - **Context**: worker 链路稳定后，把一次发现、状态和规范化历史接到已建立的 source API。 diff --git a/frontend/src/api/schema.d.ts b/frontend/src/api/schema.d.ts index c3b057c..85f9ae9 100644 --- a/frontend/src/api/schema.d.ts +++ b/frontend/src/api/schema.d.ts @@ -1453,6 +1453,13 @@ export interface components { /** Enabled */ enabled: boolean; }; + /** ChannelBindingSummaryResponse */ + ChannelBindingSummaryResponse: { + /** Count */ + count: number; + /** Meter Ids */ + meter_ids: number[]; + }; /** ChannelReadingResponse */ ChannelReadingResponse: { /** @@ -1753,6 +1760,21 @@ export interface components { /** Name */ name: string; }; + /** DiscoverChannelResponse */ + DiscoverChannelResponse: { + /** Uuid */ + uuid: string; + /** Label */ + label: string; + /** Unit */ + unit: string; + /** Latest Value */ + latest_value: string | null; + /** Latest At */ + latest_at: string | null; + /** Latest Quality */ + latest_quality: string | null; + }; /** DiscoverResponse */ DiscoverResponse: { /** Requested */ @@ -1761,8 +1783,12 @@ export interface components { supported: boolean; /** Status */ status: string; + /** Request Id */ + request_id?: number | null; /** Detail */ detail?: string | null; + /** Channels */ + channels?: components["schemas"]["DiscoverChannelResponse"][]; }; /** * DsmrLatestResponse @@ -2051,6 +2077,8 @@ export interface components { items: components["schemas"]["MeterSourceChannelResponse"][]; /** Total */ total: number; + /** Source Status */ + source_status: string; }; /** MeterSourceChannelResponse */ MeterSourceChannelResponse: { @@ -2074,6 +2102,7 @@ export interface components { binding_count: number; /** Bound Meter Ids */ bound_meter_ids: number[]; + binding_summary: components["schemas"]["ChannelBindingSummaryResponse"]; }; /** MeterSourceCreate */ MeterSourceCreate: { @@ -3977,8 +4006,8 @@ export interface operations { parameters: { query?: { limit?: number; - start?: string | null; - end?: string | null; + from?: string | null; + to?: string | null; }; header?: never; path: { diff --git a/openapi/openapi.json b/openapi/openapi.json index 7f722d5..fb47f40 100644 --- a/openapi/openapi.json +++ b/openapi/openapi.json @@ -1945,7 +1945,7 @@ } }, { - "name": "start", + "name": "from", "in": "query", "required": false, "schema": { @@ -1958,11 +1958,11 @@ "type": "null" } ], - "title": "Start" + "title": "From" } }, { - "name": "end", + "name": "to", "in": "query", "required": false, "schema": { @@ -1975,7 +1975,7 @@ "type": "null" } ], - "title": "End" + "title": "To" } } ], @@ -3462,6 +3462,27 @@ "title": "CatalogEntrySchema", "description": "An entity from the catalog with its current toggle state." }, + "ChannelBindingSummaryResponse": { + "properties": { + "count": { + "type": "integer", + "title": "Count" + }, + "meter_ids": { + "items": { + "type": "integer" + }, + "type": "array", + "title": "Meter Ids" + } + }, + "type": "object", + "required": [ + "count", + "meter_ids" + ], + "title": "ChannelBindingSummaryResponse" + }, "ChannelReadingResponse": { "properties": { "recorded_at": { @@ -4055,6 +4076,67 @@ "title": "DeviceInfoSchema", "description": "HA device grouping info for an exposable entity." }, + "DiscoverChannelResponse": { + "properties": { + "uuid": { + "type": "string", + "title": "Uuid" + }, + "label": { + "type": "string", + "title": "Label" + }, + "unit": { + "type": "string", + "title": "Unit" + }, + "latest_value": { + "anyOf": [ + { + "type": "string", + "pattern": "^(?!^[-+.]*$)[+-]?0*\\d*\\.?\\d*$" + }, + { + "type": "null" + } + ], + "title": "Latest Value" + }, + "latest_at": { + "anyOf": [ + { + "type": "string", + "format": "date-time" + }, + { + "type": "null" + } + ], + "title": "Latest At" + }, + "latest_quality": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Latest Quality" + } + }, + "type": "object", + "required": [ + "uuid", + "label", + "unit", + "latest_value", + "latest_at", + "latest_quality" + ], + "title": "DiscoverChannelResponse" + }, "DiscoverResponse": { "properties": { "requested": { @@ -4069,6 +4151,17 @@ "type": "string", "title": "Status" }, + "request_id": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Request Id" + }, "detail": { "anyOf": [ { @@ -4079,6 +4172,13 @@ } ], "title": "Detail" + }, + "channels": { + "items": { + "$ref": "#/components/schemas/DiscoverChannelResponse" + }, + "type": "array", + "title": "Channels" } }, "type": "object", @@ -4691,12 +4791,17 @@ "total": { "type": "integer", "title": "Total" + }, + "source_status": { + "type": "string", + "title": "Source Status" } }, "type": "object", "required": [ "items", - "total" + "total", + "source_status" ], "title": "MeterSourceChannelListResponse" }, @@ -4781,6 +4886,9 @@ }, "type": "array", "title": "Bound Meter Ids" + }, + "binding_summary": { + "$ref": "#/components/schemas/ChannelBindingSummaryResponse" } }, "type": "object", @@ -4794,7 +4902,8 @@ "latest_at", "latest_quality", "binding_count", - "bound_meter_ids" + "bound_meter_ids", + "binding_summary" ], "title": "MeterSourceChannelResponse" }, diff --git a/openapi/openapi.yaml b/openapi/openapi.yaml index aec1e8e..e85d463 100644 --- a/openapi/openapi.yaml +++ b/openapi/openapi.yaml @@ -1467,7 +1467,7 @@ paths: minimum: 1 default: 100 title: Limit - - name: start + - name: from in: query required: false schema: @@ -1475,8 +1475,8 @@ paths: - type: string format: date-time - type: 'null' - title: Start - - name: end + title: From + - name: to in: query required: false schema: @@ -1484,7 +1484,7 @@ paths: - type: string format: date-time - type: 'null' - title: End + title: To responses: '200': description: Successful Response @@ -2597,6 +2597,21 @@ components: - enabled title: CatalogEntrySchema description: An entity from the catalog with its current toggle state. + ChannelBindingSummaryResponse: + properties: + count: + type: integer + title: Count + meter_ids: + items: + type: integer + type: array + title: Meter Ids + type: object + required: + - count + - meter_ids + title: ChannelBindingSummaryResponse ChannelReadingResponse: properties: recorded_at: @@ -3050,6 +3065,43 @@ components: - name title: DeviceInfoSchema description: HA device grouping info for an exposable entity. + DiscoverChannelResponse: + properties: + uuid: + type: string + title: Uuid + label: + type: string + title: Label + unit: + type: string + title: Unit + latest_value: + anyOf: + - type: string + pattern: ^(?!^[-+.]*$)[+-]?0*\d*\.?\d*$ + - type: 'null' + title: Latest Value + latest_at: + anyOf: + - type: string + format: date-time + - type: 'null' + title: Latest At + latest_quality: + anyOf: + - type: string + - type: 'null' + title: Latest Quality + type: object + required: + - uuid + - label + - unit + - latest_value + - latest_at + - latest_quality + title: DiscoverChannelResponse DiscoverResponse: properties: requested: @@ -3061,11 +3113,21 @@ components: status: type: string title: Status + request_id: + anyOf: + - type: integer + - type: 'null' + title: Request Id detail: anyOf: - type: string - type: 'null' title: Detail + channels: + items: + $ref: '#/components/schemas/DiscoverChannelResponse' + type: array + title: Channels type: object required: - requested @@ -3542,10 +3604,14 @@ components: total: type: integer title: Total + source_status: + type: string + title: Source Status type: object required: - items - total + - source_status title: MeterSourceChannelListResponse MeterSourceChannelResponse: properties: @@ -3593,6 +3659,8 @@ components: type: integer type: array title: Bound Meter Ids + binding_summary: + $ref: '#/components/schemas/ChannelBindingSummaryResponse' type: object required: - uuid @@ -3605,6 +3673,7 @@ components: - latest_quality - binding_count - bound_meter_ids + - binding_summary title: MeterSourceChannelResponse MeterSourceCreate: properties: diff --git a/tests/test_meter_source_api.py b/tests/test_meter_source_api.py index d1159d0..6b0a541 100644 --- a/tests/test_meter_source_api.py +++ b/tests/test_meter_source_api.py @@ -2,7 +2,12 @@ from __future__ import annotations -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta +from decimal import Decimal +from queue import Empty, Queue +from types import SimpleNamespace +import threading +import time from unittest.mock import patch from fastapi.testclient import TestClient @@ -10,7 +15,7 @@ from sqlalchemy import create_engine, select from sqlalchemy.orm import Session from app.models.energy import DsmrReading, Meter -from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel +from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel, WarmteLinkReading _CSRF = "test-csrf-token" @@ -212,6 +217,7 @@ def test_source_and_binding_error_contracts_csrf_timezone_and_dsmr_compatibility readings = client.get(f"/api/energy/sources/{source_uuid}/channels/{channel_uuid}/readings") assert readings.status_code == 200 assert readings.json()["total"] == 1 + assert readings.json()["items"] == [{"recorded_at": now.isoformat().replace("+00:00", ""), "value": None, "quality": None}] assert "telegram_id" not in readings.text latest = client.get("/api/energy/dsmr/latest") assert latest.status_code == 200 @@ -314,7 +320,9 @@ def test_source_channel_binding_response_contract_and_discover_capabilities(auth assert discovered.status_code == 200 assert discovered.json() == { "requested": False, "supported": True, "status": "managed_by_runtime", + "request_id": None, "detail": "This source is discovered by its runtime subscription; no connection was opened.", + "channels": [], } channels = client.get(f"/api/energy/sources/{source['uuid']}/channels") assert channels.status_code == 200 @@ -322,8 +330,9 @@ def test_source_channel_binding_response_contract_and_discover_capabilities(auth assert channel["uuid"] == channel_uuid assert set(channel) == { "uuid", "label", "suggested_commodity", "unit", "device_type", "latest_value", - "latest_at", "latest_quality", "binding_count", "bound_meter_ids", + "latest_at", "latest_quality", "binding_count", "bound_meter_ids", "binding_summary", } + assert channels.json()["source_status"] == "unknown" meter = client.post("/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, json={ "label": "Contract meter", "started_at": "2030-01-01T00:00:00Z", "reason": "initial", @@ -381,3 +390,261 @@ def test_binding_patch_omitted_null_and_adjacent_half_open_boundaries(auth_datab }) assert adjacent.status_code == 201 engine.dispose() + + +def test_warmtelink_discover_and_minute_history_are_bounded_and_private(auth_database, monkeypatch): + """Discover delegates to the manager; readings expose accepted minute samples only.""" + from app.api.routes.api import meter_sources + + requested: list[int] = [] + monkeypatch.setattr( + meter_sources.warmtelink_worker_manager, "request_discovery", + lambda source_id: requested.append(source_id) or SimpleNamespace( + status="completed", request_id=1, detail=None, completed=SimpleNamespace(is_set=lambda: False), + ), + ) + monkeypatch.setattr(meter_sources.warmtelink_worker_manager, "reconcile", lambda: None) + client, engine = _client(auth_database) + with client: + _login(client) + created = client.post("/api/energy/sources", headers={"X-CSRF-Token": _CSRF}, json={ + "name": "WarmteLink", "kind": "warmtelink_serial", "config": {"path": "/dev/fake"}, + }) + assert created.status_code == 201 + source_uuid = created.json()["uuid"] + now = datetime(2030, 1, 1, 12, 0, 30, tzinfo=UTC) + with Session(engine) as session: + source = session.execute(select(MeterSource).where(MeterSource.uuid == source_uuid)).scalar_one() + source.status = "online" + channel = MeterSourceChannel( + source_id=source.id, channel_key="heating", label="Heating", unit="GJ", + latest_value=Decimal("7.002"), latest_at=now, latest_quality="unverifiable", + created_at=now, updated_at=now, + ) + session.add(channel) + session.flush() + session.add_all([ + WarmteLinkReading( + channel_id=channel.id, recorded_at=now - timedelta(minutes=1), received_at=now, + value=Decimal("7.001"), unit="GJ", quality="unverifiable", equipment_fingerprint="masked", + ), + WarmteLinkReading( + channel_id=channel.id, recorded_at=now, received_at=now, + value=Decimal("7.002"), unit="GJ", quality="unverifiable", equipment_fingerprint="masked", + ), + ]) + session.commit() + channel_uuid = channel.uuid + + discover = client.post(f"/api/energy/sources/{source_uuid}/discover", headers={"X-CSRF-Token": _CSRF}) + assert discover.status_code == 200 + assert discover.json()["status"] == "completed" + assert requested and "fingerprint" not in discover.text and "channel_key" not in discover.text + assert discover.json()["channels"][0]["latest_quality"] == "unverifiable" + + history = client.get( + f"/api/energy/sources/{source_uuid}/channels/{channel_uuid}/readings", + params={"from": "2030-01-01T11:59:00Z", "to": "2030-01-01T12:01:00Z", "limit": 1}, + ) + assert history.status_code == 200 + assert history.json()["total"] == 1 + assert history.json()["items"] == [{ + "recorded_at": "2030-01-01T11:59:30", "value": "7.001", "quality": "unverifiable", + }] + assert client.get( + f"/api/energy/sources/{source_uuid}/channels/{channel_uuid}/readings", + params={"from": "2030-01-01T12:01:00Z", "to": "2030-01-01T12:00:00Z"}, + ).status_code == 422 + assert client.get( + f"/api/energy/sources/{source_uuid}/channels/{channel_uuid}/readings", params={"limit": 0} + ).status_code == 422 + engine.dispose() + + +def test_warmtelink_discover_auth_csrf_disabled_and_source_ownership(auth_database, monkeypatch): + from app.api.routes.api import meter_sources + + monkeypatch.setattr( + meter_sources.warmtelink_worker_manager, "request_discovery", + lambda _source_id: SimpleNamespace( + status="pending", request_id=1, detail=None, completed=SimpleNamespace(is_set=lambda: False), + ), + ) + monkeypatch.setattr(meter_sources.warmtelink_worker_manager, "reconcile", lambda: None) + client, engine = _client(auth_database) + with client: + serial = client.post("/api/energy/sources", headers={"X-CSRF-Token": _CSRF}, json={ + "name": "Serial", "kind": "warmtelink_serial", "enabled": False, "config": {"path": "/dev/fake"}, + }) + assert serial.status_code == 401 # no session yet + _login(client) + serial = client.post("/api/energy/sources", headers={"X-CSRF-Token": _CSRF}, json={ + "name": "Serial", "kind": "warmtelink_serial", "enabled": False, "config": {"path": "/dev/fake"}, + }) + other = _create_source(client) + channel_uuid = _add_channel(engine, other["uuid"]) + assert client.post(f"/api/energy/sources/{serial.json()['uuid']}/discover").status_code == 403 + disabled = client.post( + f"/api/energy/sources/{serial.json()['uuid']}/discover", headers={"X-CSRF-Token": _CSRF} + ) + assert disabled.status_code == 200 and disabled.json()["status"] == "error" + assert client.get( + f"/api/energy/sources/{serial.json()['uuid']}/channels/{channel_uuid}/readings" + ).status_code == 404 + engine.dispose() + + +def test_discovery_manager_is_source_scoped_and_never_replaces_a_worker(auth_database): + """The real manager queues requests on one fake read-only serial owner.""" + from app.services.warmtelink_worker import WarmteLinkWorkerManager + + engine = create_engine(auth_database["app_url"], connect_args={"check_same_thread": False}) + with Session(engine) as session: + now = datetime.now(UTC) + source = MeterSource( + name="Serial", kind="warmtelink_serial", enabled=True, config={"path": "/dev/fake"}, + created_at=now, updated_at=now, + ) + session.add(source) + session.commit() + source_id = source.id + + class FakeReadOnlyWorker: + instances: list["FakeReadOnlyWorker"] = [] + + def __init__(self, _source_id, _config, **_kwargs): + self.requests = [] + self.thread = SimpleNamespace(is_alive=lambda: True) + self.__class__.instances.append(self) + + def start(self): + return None + + def stop(self): + return None + + def join(self, timeout=5): + return True + + def request_discovery(self, request): + self.requests.append(request) + + manager = WarmteLinkWorkerManager( + session_factory=lambda: Session(engine), worker_factory=FakeReadOnlyWorker, + ) + manager.reconcile() + assert manager.worker_count == 1 + first = manager.request_discovery(source_id) + assert first.status == "pending" + worker = FakeReadOnlyWorker.instances[0] + assert len(worker.requests) == 1 + # A client timing out/cancelling leaves the queued request and worker alone; + # completing it later cannot open another descriptor or create bindings. + worker.requests[0].status = "completed" + worker.requests[0].completed.set() + results = [] + threads = [threading.Thread(target=lambda: results.append(manager.request_discovery(source_id))) for _ in range(2)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + assert manager.worker_count == 1 + assert len(FakeReadOnlyWorker.instances) == 1 + assert len(worker.requests) == 3 + with Session(engine) as session: + assert session.query(Meter).count() == 0 + assert session.query(MeterSourceBinding).count() == 0 + manager.shutdown() + engine.dispose() + + +def test_real_warmtelink_discovery_waits_for_admission_and_keeps_its_serial_owner( + auth_database, monkeypatch, +): + """A rejected candidate is neither a discovery success nor exposed metadata.""" + from app.integrations.p1 import dsmr_crc16 + from app.services import warmtelink_worker + from app.services.warmtelink_worker import WarmteLinkWorkerManager + + engine = create_engine(auth_database["app_url"], connect_args={"check_same_thread": False}) + with Session(engine) as session: + now = datetime.now(UTC) + source = MeterSource( + name="Serial", kind="warmtelink_serial", enabled=True, config={"path": "/dev/fake"}, + created_at=now, updated_at=now, + ) + session.add(source) + session.commit() + source_id = source.id + + def frame(second: int, *, crc: bool = False) -> bytes: + body = ( + b"/WARMTE\r\n" + + f"0-0:1.0.0(2608221200{second:02d}S)\r\n".encode() + + b"0-0:96.1.1(REDACTED)\r\n" + + b"0-1:24.1.0(006)\r\n" + + b"0-1:96.1.0(REDACTED)\r\n" + + f"0-1:24.2.1(2608221200{second:02d}S)(5.900*m3)\r\n".encode() + + b"0-2:24.1.0(012)\r\n" + + b"0-2:96.1.0(REDACTED)\r\n" + + f"0-2:24.2.1(2608221200{second:02d}S)(0.017*GJ)\r\n".encode() + ) + payload = body + b"!" + return payload + (f"{dsmr_crc16(payload):04X}".encode() if crc else b"") + b"\r\n" + + class FakeReadOnlySerial: + instances: list["FakeReadOnlySerial"] = [] + + def __init__(self): + self.frames: Queue[bytes] = Queue() + self.closed = False + self.__class__.instances.append(self) + + def read(self, _size: int = 1) -> bytes: + try: + return self.frames.get(timeout=0.01) + except Empty: + return b"" + + def close(self) -> None: + self.closed = True + + monkeypatch.setattr(warmtelink_worker, "_DISCOVERY_TIMEOUT_SECONDS", 0.15) + monkeypatch.setattr(warmtelink_worker, "_DISCOVERY_WAIT_SECONDS", 0.02) + manager = WarmteLinkWorkerManager( + session_factory=lambda: Session(engine), serial_factory=lambda _config: FakeReadOnlySerial(), + ) + manager.reconcile() + serial = FakeReadOnlySerial.instances[0] + + request = manager.request_discovery(source_id) + assert request.status == "pending" + serial.frames.put(frame(0)) # First unverifiable candidate is not admitted. + time.sleep(0.04) + assert not request.completed.is_set() + with Session(engine) as session: + assert session.query(MeterSourceChannel).filter_by(source_id=source_id).count() == 0 + assert session.query(MeterSourceBinding).count() == 0 + + serial.frames.put(frame(10)) # Strictly continuous successor admits both channels. + assert request.completed.wait(1) + assert request.status == "completed" + with Session(engine) as session: + assert session.query(MeterSourceChannel).filter_by(source_id=source_id).count() == 2 + assert session.query(MeterSourceBinding).count() == 0 + + rejected = manager.request_discovery(source_id) + assert rejected.status == "pending" + serial.frames.put(b"/malformed!\r\n") + assert rejected.completed.wait(1) + assert rejected.status == "error" and rejected.detail == "Discovery timed out." + assert manager.worker_count == 1 and len(FakeReadOnlySerial.instances) == 1 + + recovered = manager.request_discovery(source_id) + serial.frames.put(frame(20, crc=True)) + assert recovered.completed.wait(1) + assert recovered.status == "completed" + assert manager.worker_count == 1 and len(FakeReadOnlySerial.instances) == 1 + manager.shutdown() + assert serial.closed + engine.dispose()