h3cx
142646396e
API has been update to return copies when returning immutable objects to avoid passing internal references to the user that could then modify them, breaking the internal state. Even if the copy operation is obsolete in some cases, it is kept in as a safety should the API functions be modified which would cause a return reference issue, the concerned API methods are not to be requested at high frequency either way so the performance hit is estimated to be minimal, the only API method that may be queried at a high freqency is fetching live values which are returned as floats which are immutable and therefore not concerned
149 lines
4.5 KiB
Python
149 lines
4.5 KiB
Python
# 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
|
|
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Literal, Self
|
|
from uuid import UUID
|
|
from zipfile import ZIP_DEFLATED, ZipFile
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
from dynalab_core.errors import DLPakError, DLUnitError
|
|
from dynalab_core.protocols.packets.handshake import (
|
|
JsonSignalDescriptor,
|
|
SignalDescriptor,
|
|
)
|
|
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class DeriveManifest(BaseModel):
|
|
unit_uuid: UUID
|
|
input_signals: list[JsonSignalDescriptor] = Field(default_factory=list)
|
|
return_signal: JsonSignalDescriptor
|
|
comments: str = ""
|
|
|
|
|
|
class DLUnit:
|
|
def __init__(self) -> None:
|
|
self._dir: Path | None = None
|
|
self._filename: str | None = None
|
|
self._savetype: Literal["zip", "folder"] = "zip"
|
|
self._data: str | None = None
|
|
self._manifest: DeriveManifest | None = None
|
|
|
|
def set_data(self, data: str) -> None:
|
|
self._data = data
|
|
|
|
def set_manifest(
|
|
self,
|
|
unit_uuid: UUID,
|
|
input_signals: list[SignalDescriptor],
|
|
return_signal: SignalDescriptor,
|
|
) -> None:
|
|
|
|
json_input_signals = [signal.to_json() for signal in input_signals]
|
|
json_return_signal = return_signal.to_json()
|
|
|
|
self._manifest = DeriveManifest(
|
|
unit_uuid=unit_uuid,
|
|
input_signals=json_input_signals,
|
|
return_signal=json_return_signal,
|
|
)
|
|
|
|
def read(self, file: str | Path) -> Self:
|
|
input_path = Path(file)
|
|
|
|
try:
|
|
with ZipFile(input_path, mode="r") as archive:
|
|
manifest = DeriveManifest.model_validate_json(
|
|
archive.read("manifest.json")
|
|
)
|
|
data: str = archive.read("unit.py").decode("utf-8")
|
|
except Exception as error:
|
|
log.exception(
|
|
"Failed to read DLUnit archive from %s",
|
|
input_path.name,
|
|
extra={
|
|
"event": "dlunit.read_failed",
|
|
"input_filename": input_path.name,
|
|
"exception_type": type(error).__name__,
|
|
},
|
|
)
|
|
raise DLUnitError(
|
|
f"Could not read DLUnit archive {input_path}: {error}",
|
|
code="read_failed",
|
|
) from error
|
|
|
|
self._dir = input_path.parent
|
|
self._filename = input_path.name
|
|
self._data = data
|
|
self._manifest = manifest
|
|
|
|
log.info(
|
|
"Read DLUnit archive from %s",
|
|
input_path.name,
|
|
extra={
|
|
"event": "dlunit.read",
|
|
"input_filename": input_path.name,
|
|
},
|
|
)
|
|
return self
|
|
|
|
def write(self, dir: str | Path, filename: str) -> None:
|
|
output_dir = Path(dir)
|
|
|
|
if self._data is None:
|
|
raise DLUnitError(
|
|
"Cannot write a DLUnit archive before setting its data",
|
|
code="no_data",
|
|
)
|
|
if self._manifest is None:
|
|
raise DLUnitError(
|
|
"Cannot write a DLUnit archive before creating its manifest",
|
|
code="no_manifest",
|
|
)
|
|
|
|
if output_dir.is_file():
|
|
output_dir = output_dir.parent
|
|
|
|
file_path = Path(filename)
|
|
|
|
if file_path.suffix.lower() != ".dlunit":
|
|
file_path = file_path.with_suffix(".dlunit")
|
|
|
|
output_path = output_dir / file_path
|
|
|
|
try:
|
|
with ZipFile(output_path, mode="w", compression=ZIP_DEFLATED) as archive:
|
|
archive.writestr(
|
|
"manifest.json", self._manifest.model_dump_json(indent=2)
|
|
)
|
|
archive.writestr("unit.py", self._data.export_csv())
|
|
except Exception as error:
|
|
log.exception(
|
|
"Failed to write DLUnit archive to %s",
|
|
output_path.name,
|
|
extra={
|
|
"event": "dlunit.write_failed",
|
|
"output_filename": output_path.name,
|
|
"exception_type": type(error).__name__,
|
|
},
|
|
)
|
|
raise DLUnitError(
|
|
f"Could not write DLUnit archive {output_path}: {error}",
|
|
code="write_failed",
|
|
) from error
|
|
|
|
log.info(
|
|
"Wrote DLUnit archive to %s",
|
|
output_path.name,
|
|
extra={
|
|
"event": "dlunit.written",
|
|
"output_filename": output_path.name,
|
|
},
|
|
)
|