From ffcea13c0b85aebf9da0fc357b3c9758bc39652f Mon Sep 17 00:00:00 2001 From: Hector van der Aa Date: Thu, 6 Aug 2026 12:57:59 +0100 Subject: [PATCH] Added ValueBuffer implementation and simple recording ValueBuffer is now the default buffer object for storing data for recording and processing purposes, a simple recording implementation has been done, core.py example starts recording as soon as the core has started then on exit it saves the output as a csv file through the ValueBuffer export_csv method --- .gitignore | 4 +++ src/dynalab_core/__init__.py | 10 ++++-- src/dynalab_core/buffer.py | 60 ++++++++++++++++++++++++++++++++++++ test/manual/core.py | 2 ++ test/manual/peer.py | 24 ++++----------- 5 files changed, 80 insertions(+), 20 deletions(-) create mode 100644 src/dynalab_core/buffer.py diff --git a/.gitignore b/.gitignore index 505a3b1..775747e 100644 --- a/.gitignore +++ b/.gitignore @@ -8,3 +8,7 @@ wheels/ # Virtual environments .venv + +# Output files +*.csv +*.dlpak diff --git a/src/dynalab_core/__init__.py b/src/dynalab_core/__init__.py index 5e581d5..c97349d 100644 --- a/src/dynalab_core/__init__.py +++ b/src/dynalab_core/__init__.py @@ -6,11 +6,11 @@ import logging from queue import Empty, Queue import threading from threading import Lock, Thread -from time import monotonic, monotonic_ns, sleep from typing import Literal from uuid import UUID +from dynalab_core.buffer import ValueBuffer from dynalab_core.config import CoreConfig from dynalab_core.constants import CORE_VERSION from dynalab_core.errors import CoreStateMismatchError @@ -33,10 +33,10 @@ class Core: "unintid" ) self._recording = threading.Event() + self._recording_buffer = ValueBuffer() self._core_version: VersionDescriptor = CORE_VERSION self._stop_event: threading.Event = threading.Event() self._core_config: CoreConfig = config - # TODO: Implement live values dict self._live_values: dict[UUID, Value] = {} self._live_values_lock = Lock() @@ -105,10 +105,12 @@ class Core: ) def start_recording(self) -> None: + self._recording_buffer.clear() self._recording.set() def stop_recording(self) -> None: self._recording.clear() + self._recording_buffer.export_csv("output.csv") def get_signal_descriptor(self, signal_id: UUID) -> SignalDescriptor | None: return self._connector_registry.get_signal_descriptor(signal_id) @@ -155,3 +157,7 @@ class Core: self._live_values[message.signal_id] = live_value live_value.update(message.value, message.timestamp) + if self._recording.is_set(): + self._recording_buffer.append( + message.timestamp, message.signal_id, message.value + ) diff --git a/src/dynalab_core/buffer.py b/src/dynalab_core/buffer.py new file mode 100644 index 0000000..6f5d511 --- /dev/null +++ b/src/dynalab_core/buffer.py @@ -0,0 +1,60 @@ +# Copyright (C) 2026 Hector van der Aa +# Copyright (C) 2026 Association Exergie +# SPDX-License-Identifier: GPL-3.0-or-later + +from collections import defaultdict +import csv +from pathlib import Path +from threading import Lock +from uuid import UUID + + +class ValueBuffer: + def __init__(self) -> None: + self._samples: list[tuple[int, UUID, float]] = [] + self._lock = Lock() + + def append(self, timestamp: int, uuid: UUID, value: float) -> None: + with self._lock: + self._samples.append((timestamp, uuid, value)) + + def clear(self) -> None: + with self._lock: + self._samples.clear() + + def __len__(self) -> int: + with self._lock: + return len(self._samples) + + def export_csv(self, path: str | Path) -> None: + with self._lock: + samples = list(self._samples) + + rows: dict[int, dict[UUID, float]] = defaultdict(dict) + signal_ids: set[UUID] = set() + + for timestamp_ns, signal_id, value in samples: + rows[timestamp_ns][signal_id] = value + signal_ids.add(signal_id) + + ordered_signal_ids = sorted(signal_ids, key=str) + + with Path(path).open(mode="w", newline="", encoding="utf-8") as file: + writer = csv.writer(file) + + writer.writerow( + ["timestamp_ns", *(str(signal_id) for signal_id in ordered_signal_ids)] + ) + + for timestamp_ns in sorted(rows): + timestamp_values = rows[timestamp_ns] + + writer.writerow( + [ + timestamp_ns, + *( + timestamp_values.get(signal_id, "") + for signal_id in ordered_signal_ids + ), + ] + ) diff --git a/test/manual/core.py b/test/manual/core.py index 8db3e19..47e758f 100644 --- a/test/manual/core.py +++ b/test/manual/core.py @@ -19,6 +19,7 @@ config = CoreConfig(port=8765) log.info("Configured manual core on %s", config.bind_str()) dl_core = Core(config) dl_core.start() +dl_core.start_recording() try: while True: @@ -27,4 +28,5 @@ try: log.debug(f"Core values: {values}") except KeyboardInterrupt: log.info("Received keyboard interrupt") + dl_core.stop_recording() dl_core.stop() diff --git a/test/manual/peer.py b/test/manual/peer.py index 3f0c754..cea0da0 100644 --- a/test/manual/peer.py +++ b/test/manual/peer.py @@ -116,24 +116,12 @@ def value_sender_thread( next_send_time += period - value = ValueDescriptor( - signal_id=signal_id, value=2.0, timestamp=monotonic_ns() - ) - - message = ValueBatch( - values=[ - value, - value, - value, - value, - value, - value, - value, - value, - value, - value, - ] - ) + message = ValueBatch(values=[]) + for i in range(10): + value = ValueDescriptor( + signal_id=signal_id, value=2.0, timestamp=monotonic_ns() + ) + message.values.append(value) future = asyncio.run_coroutine_threadsafe( send_message(