Implemented initial DLPak design

Implemented DLPak design with zip storage, DLPak consists of two files,
a manifest.json which provides details on the signals, time of recording
and currently blank comment field and data.csv which stores all the
signals as csv with the nanosecond timestamp and indexed against the
signal id
This commit is contained in:
h3cx committed 2026-08-06 15:29:02 +01:00
1 parent ffcea13c0b
commit 1cd6bb695c
4 files changed
+113 -17

No files matched your search

+67
View File
@@ -0,0 +1,67 @@
# 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 datetime import datetime, timezone
from pathlib import Path
from typing import Literal
from zipfile import ZIP_DEFLATED, ZipFile
from pydantic import BaseModel, Field
from dynalab_core.buffer import ValueBuffer
from dynalab_core.errors import DLPakNoDataError
from dynalab_core.protocols.packets.handshake import SignalDescriptor
class RecordManifest(BaseModel):
timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
signals: list[SignalDescriptor] = Field(default_factory=list)
comments: str = ""
class DLPak:
def __init__(self) -> None:
self._dir: Path | None = None
self._filename: str | None = None
self._savetype: Literal["zip", "folder"] = "zip"
self._data: ValueBuffer | None = None
self._manifest: RecordManifest | None = None
def set_data(self, data: ValueBuffer) -> None:
self._data = data
def set_manifest(
self, timestamp: datetime, all_signals: list[SignalDescriptor]
) -> None:
if self._data is None:
raise DLPakNoDataError
data_signal_ids = self._data.get_signal_ids()
manifest_signals: list[SignalDescriptor] = []
for signal in all_signals:
if signal.id in data_signal_ids:
manifest_signals.append(signal)
self._manifest = RecordManifest(
timestamp=timestamp, signals=manifest_signals, comments=""
)
def write(self, dir: str | Path, filename: str) -> None:
output_dir = Path(dir)
if output_dir.is_file():
output_dir = output_dir.parent
file_path = Path(filename)
if file_path.suffix.lower() != ".dlpak":
file_path = file_path.with_suffix(".dlpak")
output_path = output_dir / file_path
with ZipFile(output_path, mode="w", compression=ZIP_DEFLATED) as archive:
archive.writestr("manifest.json", self._manifest.model_dump_json(indent=2))
archive.writestr("data.csv", self._data.export_csv())