Initial offline process implementation

Implemented the initial offline processing implementation, with
dependency layer building and ordered processing

Processing is still single threaded and unoptimized so it can take some
time to complete, the next step is to move from a single threaded design
to a ThreadPoolExecutor that splits up the buffer into a set number of
chunks for processing

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":

src/dynalab_core/buffer.py

@@ -2,7 +2,7 @@
 # Copyright (C) 2026 Association Exergie <association.exergie@gmail.com>
 # 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],

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

src/dynalab_core/dlpak.py

@@ -2,6 +2,7 @@
 # Copyright (C) 2026 Association Exergie <association.exergie@gmail.com>
 # 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])