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
This commit is contained in:
1 parent
1690032f5b
commit
ffcea13c0b
5 files changed
+80
-20
No files matched your search
@@ -0,0 +1,60 @@
|
||||
# Copyright (C) 2026 Hector van der Aa <hector@h3cx.dev>
|
||||
# Copyright (C) 2026 Association Exergie <association.exergie@gmail.com>
|
||||
# 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
|
||||
),
|
||||
]
|
||||
)
|
||||
Reference in new issue
Block a user