From 0a0f01c32136c6244274d5408ec15098ef0a9162 Mon Sep 17 00:00:00 2001
From: Hector van der Aa <hector@h3cx.dev>
Date: Tue, 29 Sep 2026 14:14:27 +0200
Subject: [PATCH] Added final details for offline processing

Added merge buffer methods and a (very inefficient) nearest neighbor
system.
---
 src/dynalab_core/buffer.py | 31 ++++++++++++++++++++++++
 src/dynalab_core/derive.py | 48 ++++++++++++++++++++++++++++++++++----
 2 files changed, 75 insertions(+), 4 deletions(-)

diff --git a/src/dynalab_core/buffer.py b/src/dynalab_core/buffer.py
index 762bc88..20c55c7 100644
--- a/src/dynalab_core/buffer.py
+++ b/src/dynalab_core/buffer.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 bisect import bisect_left
 from collections import defaultdict
 import csv
@@ -20,6 +21,10 @@ class ValueBuffer:
         with self._lock:
             self._samples.append((timestamp, uuid, value))
 
+    def merge(self, buffer: ValueBuffer) -> None:
+        with self._lock:
+            self._samples += buffer._samples
+
     def clear(self) -> None:
         with self._lock:
             self._samples.clear()
@@ -32,6 +37,32 @@ class ValueBuffer:
         with self._lock:
             return list({signal_id for _, signal_id, _ in self._samples})
 
+    def get_nearest_neighbor_no_lock(self, uuid: UUID, timestamp: int) -> float | None:
+        previous: tuple[int, UUID, float] | None = None
+
+        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 get_nearest_neighbor(self, uuid: UUID, timestamp: int) -> float | None:
         previous: tuple[int, UUID, float] | None = None
 
diff --git a/src/dynalab_core/derive.py b/src/dynalab_core/derive.py
index 6f2898a..cffd76d 100644
--- a/src/dynalab_core/derive.py
+++ b/src/dynalab_core/derive.py
@@ -3,6 +3,8 @@
 # SPDX-License-Identifier: GPL-3.0-or-later
 
 from collections.abc import Callable
+from concurrent.futures import ThreadPoolExecutor
+from copy import copy
 from datetime import datetime, timezone
 from enum import IntEnum
 import inspect
@@ -200,9 +202,12 @@ class DeriveUnit:
     def is_running(self) -> bool:
         return not self._stopped_event.is_set()
 
-    def process_offline(self, buffer: ValueBuffer) -> ValueBuffer:
+    def _process_offline_worker(
+        self, buffer: ValueBuffer, start: int, end: int
+    ) -> ValueBuffer:
+        samples = buffer._samples.copy()
         output_buffer: ValueBuffer = ValueBuffer()
-        for sample in buffer._samples:
+        for sample in samples[start:end]:
             if sample[1] in [signal.id for signal in self._input_signals]:
                 input_args: list[ValueDescriptor] = []
                 for signal in self._input_signals:
@@ -214,12 +219,47 @@ class DeriveUnit:
                         input_args.append(
                             ValueDescriptor(
                                 signal.id,
-                                buffer.get_nearest_neighbor(signal.id, sample[0]),
+                                buffer.get_nearest_neighbor_no_lock(
+                                    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)
+                output_buffer.append(
+                    result.timestamp, self._return_signal.id, result.value
+                )
+        return output_buffer
+
+    def process_offline(self, buffer: ValueBuffer) -> ValueBuffer:
+        num_samples = len(buffer._samples)
+        num_chunks = 16
+        chunk_size = num_samples // num_chunks
+
+        output_buffer: ValueBuffer = ValueBuffer()
+        with ThreadPoolExecutor(max_workers=num_chunks) as executor:
+            ranges: list[tuple[int, int]] = []
+
+            for i in range(num_chunks):
+                start = i * chunk_size
+
+                if i == num_chunks - 1:
+                    end = num_samples
+                else:
+                    end = (i + 1) * chunk_size
+
+                ranges.append((start, end))
+
+            futures = [
+                executor.submit(self._process_offline_worker, buffer, *range)
+                for range in ranges
+            ]
+
+            results = [future.result() for future in futures]
+
+            for result in results:
+                output_buffer.merge(result)
+
         return output_buffer