diff --git a/docs/API.md b/docs/API.md index a076a7a..917d04a 100644 --- a/docs/API.md +++ b/docs/API.md @@ -43,7 +43,7 @@ again. | `host` | `str` | `"127.0.0.1"` | TCP bind address. | | `port` | `int` | `58763` | TCP bind port; validated from 0 through 65535. | | `thread_debug` | `bool` | `False` | Enables periodic per-thread CPU debug logging. | -| `playback_mode` | `"reprocess" \| "read"` | `"reprocess"` | Whether playback recalculates bound derived signals or reads their recorded values. | +| `playback_mode` | `"reprocess" \| "read"` | `"reprocess"` | Whether playback recalculates bound derived signals or reads their recorded values. Offline `reprocess()` requires `"read"`. | `CoreConfig.bind_str() -> str` returns `"host:port"`. @@ -106,6 +106,27 @@ Available only in `"offline"` mode (`CoreError(code="invalid_mode")`). Pass a a processing buffer and create its manifest. Omitting `input` before a recording is available raises `CoreError(code="no_playback_data")`. +### `reprocess() -> None` + +Recalculates derived signals in the current processing buffer using the bound +derive units. Call it in `"offline"` mode after `load_processing_buffer()`, with +`CoreConfig(playback_mode="read")`. Units with source inputs run first; units +that depend on their derived outputs run in dependency order. The operation +removes previously recorded derived samples and descriptors, then merges the +new outputs into the processing buffer. Calling it again replaces those +outputs. It does not write an archive or start playback; use `write()` to save +the result, or switch to `"playback"` mode, call `load_playback_engine()`, and +then `play()`. + +The mode, configuration, and buffer checks raise `CoreError` with codes +`"invalid_mode"`, `"invalid_config"`, and `"empty_buffer"`, respectively. +Unresolved derive dependencies raise `DeriveError(code="unresolved_dependencies")`; +a unit that produces no values raises `DeriveError(code="no_values")`. Errors +from a derive function propagate to the caller. Reprocessing changes the +in-memory processing buffer before all units finish, so a failed call may +leave its old derived data removed. Reload an unchanged archive before retrying +if that data is needed. + ## Signals and live values `SignalDescriptor` is the lightweight internal signal type in diff --git a/src/dynalab_core/__init__.py b/src/dynalab_core/__init__.py index f5107a0..30d2511 100644 --- a/src/dynalab_core/__init__.py +++ b/src/dynalab_core/__init__.py @@ -68,7 +68,10 @@ class Core: ) self._derive_registry = DeriveRegistry( - self._data_input_queue, self._get_live_value_descriptor + self._data_input_queue, + self._mode, + self._get_live_value_descriptor, + self.get_all_signal_descriptors, ) self._connector_registry = ConnectorRegistry(self._data_input_queue, self._mode) self._json_server = JsonServer(self._core_config, self._connector_registry) @@ -275,7 +278,79 @@ class Core: loaded_data = DLPak().read(input) self._processing_buffer = loaded_data - # def unload_playback_engine + def _reprocess_worker( + self, + result: dict, + ) -> None: + try: + self._processing_buffer.remove_derived() + + processed = self._derive_registry.process_offline(self._processing_buffer) + + self._processing_buffer.merge(processed) + + result["processed"] = processed + + except Exception as err: + result["error"] = err + + def reprocess(self) -> None: + if self._mode.value != "offline": + raise CoreError( + "Cannot reprocess unless the core is in 'offline' mode", + code="invalid_mode", + ) + + if self._core_config.playback_mode != "read": + raise CoreError( + "Core not configured to reprocess, config.playback_mode must be set to 'read'", + code="invalid_config", + ) + + if self._processing_buffer is None: + raise CoreError( + "Cannot reprocess an empty buffer", + code="empty_buffer", + ) + + log.info( + "Offline reprocessing started", + extra={ + "event": "core.reprocess_started", + "sample_count": len(self._processing_buffer._data), + }, + ) + result: dict[str, DLPak | Exception] = {} + + reprocess_thread = Thread( + target=self._reprocess_worker, + name="core_reprocess_thread", + args=(result,), + ) + reprocess_thread.start() + reprocess_thread.join() + + if error := result.get("error"): + log.exception( + "Offline reprocessing failed", + extra={ + "event": "core.reprocess_failed", + "exception_type": type(error).__name__, + }, + ) + raise error + + processed = result["processed"] + + log.info( + "Offline reprocessing completed with %d derived sample(s)", + len(processed._data), + extra={ + "event": "core.reprocess_completed", + "derived_sample_count": len(processed._data), + "sample_count": len(self._processing_buffer._data), + }, + ) def start_recording(self) -> None: if self._mode.value != "realtime": diff --git a/src/dynalab_core/buffer.py b/src/dynalab_core/buffer.py index 8eb89e8..762bc88 100644 --- a/src/dynalab_core/buffer.py +++ b/src/dynalab_core/buffer.py @@ -2,7 +2,7 @@ # Copyright (C) 2026 Association Exergie # SPDX-License-Identifier: GPL-3.0-or-later -from bisect import bisect_right +from bisect import bisect_left from collections import defaultdict import csv import io @@ -24,10 +24,41 @@ class ValueBuffer: with self._lock: self._samples.clear() + def clear_uuid(self, uuid: UUID) -> None: + with self._lock: + self._samples = [sample for sample in self._samples if sample[1] != uuid] + def get_signal_ids(self) -> list[UUID]: with self._lock: return list({signal_id for _, signal_id, _ in self._samples}) + def get_nearest_neighbor(self, uuid: UUID, timestamp: int) -> float | None: + previous: tuple[int, UUID, float] | None = None + + with self._lock: + for sample in self._samples: + if sample[1] != uuid: + continue + + if sample[0] >= timestamp: + if previous is None: + return sample[2] + + d1 = abs(sample[0] - timestamp) + d2 = abs(previous[0] - timestamp) + + if d1 < d2: + return sample[2] + else: + return previous[2] + + previous = sample + + if previous is not None: + return previous[2] + + return None + def __len__(self) -> int: with self._lock: return len(self._samples) @@ -141,7 +172,7 @@ class ValueBuffer: def get_samples_from(self, timestamp: int) -> list[tuple[int, UUID, float]]: with self._lock: - index = bisect_right( + index = bisect_left( self._samples, timestamp, key=lambda sample: sample[0], diff --git a/src/dynalab_core/derive.py b/src/dynalab_core/derive.py index 769b03d..6f2898a 100644 --- a/src/dynalab_core/derive.py +++ b/src/dynalab_core/derive.py @@ -3,6 +3,7 @@ # SPDX-License-Identifier: GPL-3.0-or-later from collections.abc import Callable +from datetime import datetime, timezone from enum import IntEnum import inspect import logging @@ -12,10 +13,13 @@ import threading import types from uuid import UUID, uuid4 -from dynalab_core.dlunit import DLUnit +from dynalab_core.buffer import ValueBuffer +from dynalab_core.dlpak import DLPak +from dynalab_core.dlunit import DLUnit, DeriveManifest from dynalab_core.errors import DeriveError from dynalab_core.protocols.packets.data import ValueDescriptor from dynalab_core.protocols.packets.handshake import SignalDescriptor +from dynalab_core.types import CoreModeState log = logging.getLogger(__name__) @@ -95,6 +99,17 @@ class DeriveUnit: ) self._worker_thread.start() + def get_manifest(self) -> DeriveManifest: + + json_input_signals = [signal.to_json() for signal in self._input_signals] + json_return_signal = self._return_signal.to_json() + + return DeriveManifest( + unit_uuid=self._id, + input_signals=json_input_signals, + return_signal=json_return_signal, + ) + def _worker_function(self) -> None: derive_uuid = str(self._id) log.debug( @@ -157,15 +172,6 @@ class DeriveUnit: if not self._stopped_event.is_set(): self._input_queue.put(input) - def process_offline(self, input_args: list[ValueDescriptor]) -> ValueDescriptor: - if len(input_args) == self._num_input_args: - return self._process_function(*input_args, self._return_signal) - raise DeriveError( - f"Derive unit expected {self._num_input_args} input value(s), " - f"received {len(input_args)}", - code="argument_mismatch", - ) - def get_input_signals(self) -> list[SignalDescriptor]: return self._input_signals @@ -194,17 +200,42 @@ class DeriveUnit: def is_running(self) -> bool: return not self._stopped_event.is_set() + def process_offline(self, buffer: ValueBuffer) -> ValueBuffer: + output_buffer: ValueBuffer = ValueBuffer() + for sample in buffer._samples: + if sample[1] in [signal.id for signal in self._input_signals]: + input_args: list[ValueDescriptor] = [] + for signal in self._input_signals: + if signal.id == sample[1]: + input_args.append( + ValueDescriptor(sample[1], sample[2], sample[0]) + ) + else: + input_args.append( + ValueDescriptor( + signal.id, + buffer.get_nearest_neighbor(signal.id, sample[0]), + sample[0], + ) + ) + result = self._process_function(*input_args, self._return_signal) + output_buffer.append(result.timestamp, self._return_signal.id, result.value) + return output_buffer + class DeriveRegistry: def __init__( self, core_input_queue: Queue, + core_mode: CoreModeState, get_live_value: Callable[[UUID], ValueDescriptor | None], + get_all_core_signal_descriptors: Callable[[], list[SignalDescriptor]], ) -> None: self._lock = RLock() self._units: dict[UUID, DeriveUnit] = {} self._core_input_queue = core_input_queue self._get_live_value = get_live_value + self._get_all_core_signal_descriptors = get_all_core_signal_descriptors self._routing_queue: Queue[ValueDescriptor] = Queue() self._routing_stop_event = threading.Event() self._routing_stopped_event = threading.Event() @@ -214,6 +245,8 @@ class DeriveRegistry: daemon=True, ) self._routing_worker_thread.start() + self._dependency_layers: list[list[UUID]] = [] + self._core_mode = core_mode def register(self, dlunit: DLUnit) -> DeriveUnit: @@ -343,7 +376,10 @@ class DeriveRegistry: signals: list[SignalDescriptor] = [] for unit in units: - if unit.is_running(): + if self._core_mode.value == "realtime": + if unit.is_running(): + signals.append(unit.return_signal()) + else: signals.append(unit.return_signal()) return signals @@ -393,6 +429,68 @@ class DeriveRegistry: ) return stopped + def _build_dependency_layers(self) -> None: + all_signals = self._get_all_core_signal_descriptors() + + source_signals: list[SignalDescriptor] = [ + signal for signal in all_signals if signal.origin == "source" + ] + + units = [unit.get_manifest() for unit in self._units.values()] + + dependent_units = [ + unit + for unit in units + if any(signal.origin == "derived" for signal in unit.input_signals) + ] + + L0_units = [ + unit + for unit in units + if not any(signal.origin == "derived" for signal in unit.input_signals) + ] + + unit_layers: list[list[DeriveManifest]] = [] + + unit_layers.append(L0_units) + + while len(dependent_units) > 0: + available_signal_ids = {signal.id for signal in source_signals} + available_signal_ids.update( + unit.return_signal.id for layer in unit_layers for unit in layer + ) + Lx_units = [ + unit + for unit in dependent_units + if all(sig.id in available_signal_ids for sig in unit.input_signals) + ] + + if not Lx_units: + unresolved = ", ".join(str(unit.unit_uuid) for unit in dependent_units) + raise DeriveError( + f"Unresolved derive dependencies for units: {unresolved}", + code="unresolved_dependencies", + ) + + ready_ids = {unit.unit_uuid for unit in Lx_units} + dependent_units = [ + unit for unit in dependent_units if unit.unit_uuid not in ready_ids + ] + unit_layers.append(Lx_units) + + self._dependency_layers = [ + [manifest.unit_uuid for manifest in layer] for layer in unit_layers + ] + log.debug( + "Built %d offline derive dependency layer(s)", + len(self._dependency_layers), + extra={ + "event": "derive.offline_layers_built", + "layer_count": len(self._dependency_layers), + "unit_count": len(units), + }, + ) + def _routing_worker(self) -> None: log.debug( "Derive routing worker started", @@ -450,3 +548,61 @@ class DeriveRegistry: args.append(value_descriptor) else: unit.put_data(args) + + def process_offline(self, buffer: DLPak) -> DLPak: + self._build_dependency_layers() + working_buffer = ValueBuffer() + for timestamp, signal_id, value in buffer._data._samples: + working_buffer.append(timestamp, signal_id, value) + processed_buffer = ValueBuffer() + + for layer_index, layer in enumerate(self._dependency_layers): + layer_units = [self._units[uuid] for uuid in layer] + for unit in layer_units: + returned_buffer = unit.process_offline(working_buffer) + if not returned_buffer._samples: + log.warning( + "Offline derive unit %s returned no values", + unit.uuid(), + extra={ + "event": "derive.offline_no_values", + "derive_uuid": str(unit.uuid()), + "layer_index": layer_index, + }, + ) + raise DeriveError( + "Offline processing returned no values", code="no_values" + ) + log.debug( + "Offline derive unit %s produced %d sample(s)", + unit.uuid(), + len(returned_buffer), + extra={ + "event": "derive.offline_unit_processed", + "derive_uuid": str(unit.uuid()), + "layer_index": layer_index, + "sample_count": len(returned_buffer), + }, + ) + for sample in returned_buffer._samples: + processed_buffer.append(sample[0], sample[1], sample[2]) + working_buffer.append(sample[0], sample[1], sample[2]) + working_buffer.sort() + log.debug( + "Offline derive layer %d processed", + layer_index, + extra={ + "event": "derive.offline_layer_processed", + "layer_index": layer_index, + "unit_count": len(layer_units), + }, + ) + + processed_buffer.sort() + + processed_pak: DLPak = DLPak() + processed_pak.set_data(processed_buffer) + processed_pak.set_manifest( + datetime.now(timezone.utc), self.get_all_signal_descriptors() + ) + return processed_pak diff --git a/src/dynalab_core/dlpak.py b/src/dynalab_core/dlpak.py index 0744872..f634305 100644 --- a/src/dynalab_core/dlpak.py +++ b/src/dynalab_core/dlpak.py @@ -2,6 +2,7 @@ # Copyright (C) 2026 Association Exergie # SPDX-License-Identifier: GPL-3.0-or-later +from __future__ import annotations from copy import copy from datetime import datetime, timezone import logging @@ -59,6 +60,51 @@ class DLPak: timestamp=timestamp, signals=manifest_signals, comments="" ) + def merge(self, dlpak: DLPak) -> None: + for sample in dlpak._data._samples: + self._data.append(sample[0], sample[1], sample[2]) + self._data.sort() + + for signal in dlpak._manifest.signals: + self._manifest.signals.append(signal) + log.debug( + "Merged %d sample(s) and %d signal(s) into DLPak", + len(dlpak._data), + len(dlpak._manifest.signals), + extra={ + "event": "dlpak.merged", + "added_sample_count": len(dlpak._data), + "added_signal_count": len(dlpak._manifest.signals), + }, + ) + + def remove_derived(self) -> None: + if self._manifest is not None: + original_sample_count = len(self._data) + remove_uuids = [ + signal.id + for signal in self._manifest.signals + if signal.origin == "derived" + ] + for remove_uuid in remove_uuids: + self._data.clear_uuid(remove_uuid) + + self._manifest.signals = [ + signal + for signal in self._manifest.signals + if signal.origin != "derived" + ] + log.debug( + "Removed %d derived signal(s) and %d sample(s) from DLPak", + len(remove_uuids), + original_sample_count - len(self._data), + extra={ + "event": "dlpak.derived_removed", + "removed_signal_count": len(remove_uuids), + "removed_sample_count": original_sample_count - len(self._data), + }, + ) + def get_all_signal_descriptors(self) -> list[SignalDescriptor]: return copy([signal.to_pure() for signal in self._manifest.signals]) diff --git a/test/manual/core_tui.py b/test/manual/core_tui.py index 0e62ef7..51061dc 100644 --- a/test/manual/core_tui.py +++ b/test/manual/core_tui.py @@ -120,6 +120,8 @@ class Dashboard: self.open_mode_selection() elif key.lower() == "l": self.load_current_stage() + elif key.lower() == "p": + self.reprocess_offline() elif key == " ": self.toggle_playback() except DynaLabError as error: @@ -132,7 +134,9 @@ class Dashboard: if self.core.get_recording_state(): self.core.stop_recording() - self.set_message("Recording stopped; select Offline and press L to load it", "green") + self.set_message( + "Recording stopped; select Offline and press L to load it", "green" + ) else: self.core.start_recording() self.set_message("Recording started", "bold red") @@ -154,7 +158,9 @@ class Dashboard: mode = modes.get(key) if mode is None: - self.set_message("Choose 1 Realtime, 2 Offline, 3 Playback, or Esc", "yellow") + self.set_message( + "Choose 1 Realtime, 2 Offline, 3 Playback, or Esc", "yellow" + ) return if self.core.get_playback_state() and mode != "playback": @@ -163,7 +169,9 @@ class Dashboard: self.mode_selection_open = False if mode == "offline": - self.set_message("Offline mode selected; press L to load the recording", "green") + self.set_message( + "Offline mode selected; press L to load the recording", "green" + ) elif mode == "playback": self.set_message( "Playback mode selected; press L to load the offline buffer", "green" @@ -191,6 +199,14 @@ class Dashboard: self.set_message("Select Offline or Playback before loading", "yellow") + def reprocess_offline(self) -> None: + if self.core.get_mode() != "offline": + self.set_message("Reprocessing is only available in offline mode", "yellow") + return + + self.core.reprocess() + self.set_message("Offline reprocessing requested", "green") + def toggle_playback(self) -> None: if self.core.get_mode() != "playback": self.set_message("Switch to playback mode before pressing Space", "yellow") @@ -252,7 +268,8 @@ class Dashboard: controls = Text.from_markup( "[bold]R[/bold] Record [bold]M[/bold] Choose mode " - "[bold]L[/bold] Load stage [bold]Space[/bold] Play/Pause " + "[bold]L[/bold] Load stage [bold]P[/bold] Reprocess offline " + "[bold]Space[/bold] Play/Pause " "[bold]Q[/bold] Quit" ) log_table = Table.grid(expand=True, padding=(0, 1)) @@ -288,7 +305,9 @@ class Dashboard: mode_picker.add_column(width=3, style="bold cyan") mode_picker.add_column(ratio=1) mode_picker.add_row("1", "Realtime — receive connector data and record") - mode_picker.add_row("2", "Offline — load a completed recording into a processing buffer") + mode_picker.add_row( + "2", "Offline — load a completed recording into a processing buffer" + ) mode_picker.add_row("3", "Playback — load the offline buffer and replay it") mode_picker.add_row("Esc", "Cancel") panels.append( @@ -329,6 +348,9 @@ def main() -> None: processing_source = ( Path(__file__).with_name("process.py").read_text(encoding="utf-8") ) + processing_source_2 = ( + Path(__file__).with_name("process2.py").read_text(encoding="utf-8") + ) dashboard = Dashboard(core, config, log_handler) derive_unit = DLUnit() @@ -357,7 +379,30 @@ def main() -> None: ), ) derive_unit.set_data(processing_source) - core.bind_derive_unit(dlunit=derive_unit, file=None) + + derive_unit_2 = DLUnit() + derive_unit_2.set_manifest( + unit_uuid=uuid4(), + input_signals=[ + SignalDescriptor( + id=derive_unit._manifest.return_signal.id, + name="Input signal", + type="number", + timeout_ms=5000, + origin="derived", + ) + ], + return_signal=SignalDescriptor( + id=uuid4(), + name="Sq derive", + type="number", + timeout_ms=5000, + origin="derived", + ), + ) + derive_unit_2.set_data(processing_source_2) + core.bind_derive_unit(dlunit=derive_unit) + core.bind_derive_unit(dlunit=derive_unit_2) try: core.start() diff --git a/test/manual/process2.py b/test/manual/process2.py new file mode 100644 index 0000000..b2f3d26 --- /dev/null +++ b/test/manual/process2.py @@ -0,0 +1,9 @@ +from dynalab_core.protocols.packets.data import ValueDescriptor +from dynalab_core.protocols.packets.handshake import SignalDescriptor + + +def process2(a: ValueDescriptor, r: SignalDescriptor) -> ValueDescriptor: + val = ValueDescriptor( + signal_id=r.id, value=a.value * a.value, timestamp=a.timestamp + ) + return val