diff --git a/app/services/warmtelink_worker.py b/app/services/warmtelink_worker.py index 04434fa..260b64e 100644 --- a/app/services/warmtelink_worker.py +++ b/app/services/warmtelink_worker.py @@ -177,7 +177,6 @@ class WarmteLinkWorker: def _run(self) -> None: backoff_index = 0 - framer = TelegramFramer() while not self._stop_event.is_set(): try: device = self._serial_factory(self.config) @@ -187,6 +186,11 @@ class WarmteLinkWorker: device.close() return self._serial = device + # A disconnect makes any bytes buffered from the previous + # descriptor untrustworthy. In particular, never let a + # trailing partial telegram be completed by a newly opened + # device. + framer = TelegramFramer() while not self._stop_event.is_set(): self._expire_discoveries() chunk = device.read(1024) diff --git a/frontend/src/energy/SourceManager.test.tsx b/frontend/src/energy/SourceManager.test.tsx index ed3968f..ea6fbf7 100644 --- a/frontend/src/energy/SourceManager.test.tsx +++ b/frontend/src/energy/SourceManager.test.tsx @@ -8,7 +8,8 @@ vi.mock('../api/client', () => ({ default: { GET: (...a: unknown[]) => mockGet(. const source = { uuid: 's1', name: 'WarmteLink', kind: 'warmtelink_serial', enabled: true, status: 'online', config: {} } describe('SourceManager API states', () => { beforeEach(() => vi.clearAllMocks()) - it('renders source list, detail and a completed discovery result', async () => { const user = userEvent.setup(); mockGet.mockImplementation((path: string) => Promise.resolve({ data: path === '/api/energy/sources' ? { items: [source] } : path.includes('channels') ? { items: [] } : source })); mockPost.mockResolvedValue({ data: { status: 'completed' } }); renderWithProviders(); await user.click(await screen.findByText('WarmteLink')); expect(await screen.findByText(/No channels discovered yet/)).toBeInTheDocument(); await user.click(screen.getByRole('button', { name: /Discover channels/ })); expect(await screen.findByText(/Discovery completed/)).toBeInTheDocument() }) + it('renders source list, detail and a completed bounded discovery refresh', async () => { const user = userEvent.setup(); mockGet.mockImplementation((path: string) => Promise.resolve({ data: path === '/api/energy/sources' ? { items: [source] } : path.includes('channels') ? { items: [] } : source })); mockPost.mockResolvedValue({ data: { status: 'completed' } }); renderWithProviders(); await user.click(await screen.findByText('WarmteLink')); expect(await screen.findByText(/No channels discovered yet/)).toBeInTheDocument(); await user.click(screen.getByRole('button', { name: /Refresh discovered channels/ })); expect(await screen.findByText(/Discovery completed/)).toBeInTheDocument() }) + it('explains that an enabled serial source reconnects automatically', async () => { const user = userEvent.setup(); mockGet.mockImplementation((path: string) => Promise.resolve({ data: path === '/api/energy/sources' ? { items: [{ ...source, status: 'error', last_error: 'serial unavailable' }] } : path.includes('channels') ? { items: [] } : { ...source, status: 'error', last_error: 'serial unavailable' } })); renderWithProviders(); await user.click(await screen.findByText('WarmteLink')); expect(await screen.findAllByText(/reconnecting automatically/i)).not.toHaveLength(0); expect(screen.getByRole('button', { name: /Refresh discovered channels/ })).toBeInTheDocument() }) it('renders source list empty and error states', async () => { mockGet.mockResolvedValueOnce({ data: { items: [] } }); const { unmount } = renderWithProviders(); expect(await screen.findByText(/No sources configured/)).toBeInTheDocument(); unmount(); mockGet.mockRejectedValueOnce(new Error('offline')); renderWithProviders(); expect(await screen.findByText(/Failed to load sources/)).toBeInTheDocument() }) it('contains the wide four-column source table in a scroll area', async () => { mockGet.mockResolvedValue({ data: { items: [source] } }); renderWithProviders(); expect(await screen.findByTestId('sources-table-scrollarea')).toBeInTheDocument(); expect(screen.getByTestId('sources-table')).toHaveStyle({ minWidth: '640px' }) }) it('safely deletes an unreferenced source and clears its selection', async () => { const user = userEvent.setup(); let items = [source]; mockGet.mockImplementation((path: string) => Promise.resolve({ data: path === '/api/energy/sources' ? { items } : path.includes('channels') ? { items: [] } : source })); mockDelete.mockImplementation(async () => { items = []; return { data: undefined } }); renderWithProviders(); await user.click(await screen.findByText('WarmteLink')); await user.click(screen.getByRole('button', { name: 'Delete source' })); await waitFor(() => expect(mockDelete).toHaveBeenCalledWith('/api/energy/sources/{source_uuid}', { params: { path: { source_uuid: 's1' } } })); expect(await screen.findByText(/No sources configured/)).toBeInTheDocument(); expect(screen.queryByRole('button', { name: 'Delete source' })).not.toBeInTheDocument() }) diff --git a/frontend/src/energy/SourceManager.tsx b/frontend/src/energy/SourceManager.tsx index 11071f9..aad2ff4 100644 --- a/frontend/src/energy/SourceManager.tsx +++ b/frontend/src/energy/SourceManager.tsx @@ -12,7 +12,7 @@ export function SourceManager() { if (sources.isLoading) return
if (sources.isError || !sources.data) return Failed to load sources. return Sources - {sources.data.items.length === 0 ? No sources configured yet. : NameTypeStatusLast seen{sources.data.items.map((source) => {source.kind}{source.enabled ? source.status : 'disabled'}{source.last_error && {source.last_error}}{source.last_seen_at ? formatLocalDateTime(source.last_seen_at) : '—'})}
} + {sources.data.items.length === 0 ? No sources configured yet. : NameTypeStatusLast seen{sources.data.items.map((source) => {source.kind}{source.enabled ? source.status : 'disabled'}{source.enabled && source.status === 'error' && Reconnecting automatically.}{source.last_error && {source.last_error}}{source.last_seen_at ? formatLocalDateTime(source.last_seen_at) : '—'})}
} {selected && setSelected(null)} />}{form !== null && setForm(null)} />} } @@ -22,8 +22,9 @@ function SourceDetail({ uuid, onEdit, onDeleted }: { uuid: string; onEdit: (sour if (source.isLoading) return ; if (source.isError || !source.data) return Failed to load source. const detail = source.data! const result = discover.data?.data - return {detail.name} + return {detail.name} {deleteError && {deleteError}} + {detail.enabled && detail.status === 'error' && The worker is reconnecting automatically. Refresh only requests a bounded status update; it does not start ingestion.} {discover.isPending && Discovery pending…}{discover.isError && Discovery request failed. Check the source and try again.} {result && Discovery {result.status}: {result.detail ?? (result.status === 'completed' ? 'Channels refreshed.' : 'Waiting for discovery.')}} {detail.kind.includes('dsmr') && } diff --git a/frontend/src/energy/hooks.ts b/frontend/src/energy/hooks.ts index 818e57c..b1d61f3 100644 --- a/frontend/src/energy/hooks.ts +++ b/frontend/src/energy/hooks.ts @@ -523,12 +523,12 @@ export function useSourceProfiles() { export function useSources() { return useQuery({ queryKey: ['energy-sources'], queryFn: async () => { const res = await apiClient.GET('/api/energy/sources'); return res.data - } }) + }, refetchInterval: 5_000 }) } export function useSource(uuid: string | null) { return useQuery({ queryKey: ['energy-source', uuid], enabled: !!uuid, queryFn: async () => { const res = await apiClient.GET('/api/energy/sources/{source_uuid}', { params: { path: { source_uuid: uuid! } } }); return res.data - } }) + }, refetchInterval: 3_000 }) } function invalidateSourceQueries(qc: ReturnType) { void qc.invalidateQueries({ queryKey: ['energy-sources'] }); void qc.invalidateQueries({ queryKey: ['energy-source'] }); @@ -540,8 +540,8 @@ export function useCreateSource() { const qc = useQueryClient(); return useMutat export function useUpdateSource() { const qc = useQueryClient(); return useMutation({ mutationFn: ({ uuid, body }: { uuid: string; body: MeterSourcePatch }) => apiClient.PATCH('/api/energy/sources/{source_uuid}', { params: { path: { source_uuid: uuid } }, body }), onSuccess: () => invalidateSourceQueries(qc) }) } export function useDeleteSource() { const qc = useQueryClient(); return useMutation({ mutationFn: (uuid: string) => apiClient.DELETE('/api/energy/sources/{source_uuid}', { params: { path: { source_uuid: uuid } } }), onSuccess: () => invalidateSourceQueries(qc) }) } export function useDiscoverSource() { const qc = useQueryClient(); return useMutation({ mutationFn: (uuid: string) => apiClient.POST('/api/energy/sources/{source_uuid}/discover', { params: { path: { source_uuid: uuid } } }), onSuccess: () => invalidateSourceQueries(qc) }) } -export function useSourceChannels(uuid: string | null) { return useQuery({ queryKey: ['energy-source-channels', uuid], enabled: !!uuid, queryFn: async () => { const res = await apiClient.GET('/api/energy/sources/{source_uuid}/channels', { params: { path: { source_uuid: uuid! } } }); return res.data } }) } -export function useChannelReadings(sourceUuid: string | null, channelUuid: string | null) { return useQuery({ queryKey: ['energy-channel-readings', sourceUuid, channelUuid], enabled: !!sourceUuid && !!channelUuid, queryFn: async () => { const res = await apiClient.GET('/api/energy/sources/{source_uuid}/channels/{channel_uuid}/readings', { params: { path: { source_uuid: sourceUuid!, channel_uuid: channelUuid! }, query: { limit: 60 } } }); return res.data } }) } +export function useSourceChannels(uuid: string | null) { return useQuery({ queryKey: ['energy-source-channels', uuid], enabled: !!uuid, queryFn: async () => { const res = await apiClient.GET('/api/energy/sources/{source_uuid}/channels', { params: { path: { source_uuid: uuid! } } }); return res.data }, refetchInterval: 3_000 }) } +export function useChannelReadings(sourceUuid: string | null, channelUuid: string | null) { return useQuery({ queryKey: ['energy-channel-readings', sourceUuid, channelUuid], enabled: !!sourceUuid && !!channelUuid, queryFn: async () => { const res = await apiClient.GET('/api/energy/sources/{source_uuid}/channels/{channel_uuid}/readings', { params: { path: { source_uuid: sourceUuid!, channel_uuid: channelUuid! }, query: { limit: 60 } } }); return res.data }, refetchInterval: 5_000 }) } export function useMeterBindings(id: number | null) { return useQuery({ queryKey: ['energy-meter-bindings', id], enabled: id != null, queryFn: async () => { const res = await apiClient.GET('/api/energy/meters/{meter_id}/bindings', { params: { path: { meter_id: id! } } }); return res.data } }) } export function useCreateBinding() { const qc = useQueryClient(); return useMutation({ mutationFn: ({ id, body }: { id: number; body: BindingCreate }) => apiClient.POST('/api/energy/meters/{meter_id}/bindings', { params: { path: { meter_id: id } }, body }), onSuccess: () => invalidateSourceQueries(qc) }) } export function useCloseBinding() { const qc = useQueryClient(); return useMutation({ mutationFn: ({ uuid, ended_at }: { uuid: string; ended_at: string }) => apiClient.PATCH('/api/energy/bindings/{binding_uuid}', { params: { path: { binding_uuid: uuid } }, body: { ended_at } }), onSuccess: () => invalidateSourceQueries(qc) }) } diff --git a/tests/test_warmtelink_worker.py b/tests/test_warmtelink_worker.py index cef860a..120332f 100644 --- a/tests/test_warmtelink_worker.py +++ b/tests/test_warmtelink_worker.py @@ -5,7 +5,7 @@ from __future__ import annotations import time import threading from datetime import UTC, datetime -from queue import Queue +from queue import Empty, Queue import anyio from fastapi import FastAPI @@ -14,6 +14,7 @@ from sqlalchemy.orm import sessionmaker from app.db import Base from app.models.meter_source import MeterSource +from app.services.warmtelink_ingest import WarmteLinkIngestor from app.services.warmtelink_worker import WarmteLinkWorker, WarmteLinkWorkerManager, open_warmtelink_serial @@ -156,6 +157,128 @@ def test_worker_resets_backoff_after_failures_then_a_complete_frame(): assert clock.delays[:3] == [1, 2, 1] +def test_worker_discards_partial_frame_when_a_new_serial_descriptor_reconnects(): + """A new device must not complete bytes buffered from its predecessor.""" + class _PartialThenDisconnect: + def __init__(self): + self.closed = False + self.calls = 0 + + def read(self, _size=1): + self.calls += 1 + if self.calls == 1: + return b"/stale-partial" + raise OSError("disconnected") + + def close(self): + self.closed = True + + class _FreshDevice: + def __init__(self): + self.closed = False + self.calls = 0 + + def read(self, _size=1): + self.calls += 1 + if self.calls == 1: + return b"/fresh\r\n!\r\n" + raise OSError("disconnected") + + def close(self): + self.closed = True + + stale, fresh = _PartialThenDisconnect(), _FreshDevice() + clock, ingestor = _Clock(), _Ingestor() + devices = iter([stale, fresh, _BrokenSerial()]) + worker = WarmteLinkWorker( + 3, {"path": "/dev/private"}, session_factory=lambda: None, + serial_factory=lambda _config: next(devices), ingestor=ingestor, clock=clock, + ) + worker.start() + deadline = time.monotonic() + 1 + while not ingestor.frames and time.monotonic() < deadline: + time.sleep(0.005) + worker.stop() + assert worker.join(1) + assert ingestor.frames == [(3, b"/fresh\r\n!\r\n")] + assert stale.closed and fresh.closed + assert clock.delays[:2] == [1, 1] + + +def test_worker_recovers_from_initial_open_failure_and_ingests_without_discovery(tmp_path): + """Startup retries independently of a manual discovery request.""" + from app.integrations.p1 import dsmr_crc16 + from app.models.meter_source import MeterSourceChannel + + factory, engine = _session_factory(tmp_path) + received_at = datetime(2026, 8, 22, 10, 0, 30, tzinfo=UTC) + with factory() as session: + source = session.query(MeterSource).filter_by(name="one").one() + source_id = source.id + session.query(MeterSource).filter_by(name="two").one().enabled = False + session.commit() + + class _QueueSerial: + def __init__(self): + self.closed = False + self.frames: Queue[bytes] = Queue() + + def read(self, _size=1): + try: + return self.frames.get(timeout=0.01) + except Empty: + return b"" + + def close(self): + self.closed = True + + def frame() -> bytes: + body = ( + b"/WARMTE\r\n" + b"0-0:1.0.0(260822120000S)\r\n" + 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" + b"0-1:24.2.1(260822120000S)(5.900*m3)\r\n" + b"0-2:24.1.0(012)\r\n" + b"0-2:96.1.0(REDACTED)\r\n" + b"0-2:24.2.1(260822120000S)(0.017*GJ)\r\n" + ) + payload = body + b"!" + return payload + f"{dsmr_crc16(payload):04X}".encode() + b"\r\n" + + serial, clock = _QueueSerial(), _Clock() + attempts = 0 + + def open_after_one_failure(_config): + nonlocal attempts + attempts += 1 + if attempts == 1: + raise OSError("unavailable") + return serial + + worker = WarmteLinkWorker( + source_id, {"path": "/dev/fake"}, session_factory=factory, + serial_factory=open_after_one_failure, ingestor=WarmteLinkIngestor(clock=lambda: received_at), clock=clock, + ) + worker.start() + serial.frames.put(frame()) + deadline = time.monotonic() + 1 + while time.monotonic() < deadline: + with factory() as session: + source = session.get(MeterSource, source_id) + if source is not None and source.status == "online": + assert session.query(MeterSourceChannel).filter_by(source_id=source_id).count() == 2 + break + time.sleep(0.005) + else: + raise AssertionError("worker did not recover and ingest its startup frame") + worker.stop() + assert worker.join(1) + assert attempts >= 2 and clock.delays[0] == 1 and serial.closed + engine.dispose() + + def test_worker_is_non_daemon_even_when_constructed_by_a_daemon_parent(): result: Queue[bool] = Queue()