import csv import json import zlib from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Any from uuid import uuid4 import msgpack RESULTS_FORMAT = "bedit-simulation-results" RESULTS_VERSION = 1 @dataclass class SimulationTrace: """One plottable series; samples can be filled by a future result importer.""" name: str x_values: list[float] = field(default_factory=list) y_values: list[float] = field(default_factory=list) x_label: str = "time" y_label: str = "" unit: str = "" properties: dict[str, Any] = field(default_factory=dict) @dataclass class SimulationGraph: """One graph workspace tab and its configured traces.""" id: str = field(default_factory=lambda: str(uuid4())) title: str = "Graph 1" x_axis: str = "time" traces: list[SimulationTrace] = field(default_factory=list) @dataclass class SimulationResults: """Serializable state displayed by the standalone simulation window.""" model_name: str = "" status: dict[str, Any] = field(default_factory=dict) messages: list[dict[str, str]] = field(default_factory=list) data: dict[str, list[float]] = field(default_factory=dict) graphs: list[SimulationGraph] = field( default_factory=lambda: [SimulationGraph()] ) metadata: dict[str, Any] = field(default_factory=dict) def to_dict(self) -> dict[str, Any]: return { "format": RESULTS_FORMAT, "version": RESULTS_VERSION, "modelName": self.model_name, "status": dict(self.status), "messages": [dict(message) for message in self.messages], "data": {name: list(values) for name, values in self.data.items()}, "graphs": [asdict(graph) for graph in self.graphs], "metadata": dict(self.metadata), } @classmethod def from_dict(cls, data: dict[str, Any]) -> "SimulationResults": if data.get("format") != RESULTS_FORMAT: raise ValueError("Not a BEdit simulation-results file") if data.get("version") != RESULTS_VERSION: raise ValueError(f"Unsupported simulation-results version: {data.get('version')!r}") try: graphs = [ SimulationGraph( id=str(graph["id"]), title=str(graph["title"]), x_axis=str(graph.get("x_axis", "time")), traces=[ SimulationTrace(**trace) for trace in graph.get("traces", []) ], ) for graph in data.get("graphs", []) ] return cls( model_name=str(data.get("modelName", "")), status=dict(data.get("status", {})), messages=[dict(message) for message in data.get("messages", [])], data={ str(name): [float(value) for value in values] for name, values in dict(data.get("data", {})).items() }, graphs=graphs, metadata=dict(data.get("metadata", {})), ) except (KeyError, TypeError, ValueError) as error: raise ValueError("Malformed simulation-results data") from error @dataclass(frozen=True) class SimulationExecutionResult: """Completed process information delivered before its temp files disappear.""" return_code: int result_file: str data: dict[str, list[float]] def load_openmodelica_csv(path: str | Path) -> dict[str, list[float]]: """Read an OpenModelica CSV result as one numeric array per column.""" source = Path(path) try: with source.open(newline="", encoding="utf-8") as file: reader = csv.reader(file) headers = next(reader) if not headers or any(not header for header in headers): raise ValueError("The result CSV has an invalid header") if len(set(headers)) != len(headers): raise ValueError("The result CSV contains duplicate column names") columns = {header: [] for header in headers} for row_number, row in enumerate(reader, start=2): if len(row) != len(headers): raise ValueError( f"Result CSV row {row_number} has {len(row)} values; " f"expected {len(headers)}" ) for header, value in zip(headers, row, strict=True): columns[header].append(float(value)) except OSError as error: raise ValueError(f"Could not read OpenModelica results: {error}") from error except StopIteration as error: raise ValueError("The OpenModelica result CSV is empty") from error except ValueError as error: if str(error).startswith(("The result CSV", "Result CSV")): raise raise ValueError(f"The OpenModelica result CSV is not numeric: {error}") from error return columns class JsonSimulationResultsSerializer: @staticmethod def load(path: Path) -> SimulationResults: try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as error: raise ValueError(f"Could not read simulation results: {error}") from error if not isinstance(data, dict): raise ValueError("Simulation-results root must be an object") return SimulationResults.from_dict(data) @staticmethod def save(results: SimulationResults, path: Path) -> None: temporary_path = path.with_suffix(path.suffix + ".tmp") temporary_path.write_text( json.dumps(results.to_dict(), indent=2, ensure_ascii=False) + "\n", encoding="utf-8", ) temporary_path.replace(path) class BerSimulationResultsSerializer: """Compressed MessagePack serializer for binary simulation results.""" MAGIC = b"BER\x00" VERSION = 1 @classmethod def load(cls, path: Path) -> SimulationResults: try: payload = path.read_bytes() except OSError as error: raise ValueError(f"Could not read simulation results: {error}") from error header = cls.MAGIC + bytes([cls.VERSION]) if not payload.startswith(header): raise ValueError("This is not a supported BEdit binary results file") try: data = msgpack.unpackb( zlib.decompress(payload[len(header) :]), raw=False ) except (ValueError, zlib.error, msgpack.exceptions.MsgpackException) as error: raise ValueError("The BEdit binary results file is damaged") from error if not isinstance(data, dict): raise ValueError("The BEdit binary results file has an invalid root value") return SimulationResults.from_dict(data) @classmethod def save(cls, results: SimulationResults, path: Path) -> None: packed = msgpack.packb(results.to_dict(), use_bin_type=True) payload = cls.MAGIC + bytes([cls.VERSION]) + zlib.compress(packed, level=9) temporary_path = path.with_suffix(path.suffix + ".tmp") temporary_path.write_bytes(payload) temporary_path.replace(path) class SimulationResultsSerializer: """Select JSON or compressed MessagePack based on the file extension.""" @staticmethod def load(path: str | Path) -> SimulationResults: target = Path(path) serializer = ( BerSimulationResultsSerializer if target.suffix.lower() == ".ber" else JsonSimulationResultsSerializer ) return serializer.load(target) @staticmethod def save(results: SimulationResults, path: str | Path) -> None: target = Path(path) serializer = ( BerSimulationResultsSerializer if target.suffix.lower() == ".ber" else JsonSimulationResultsSerializer ) serializer.save(results, target) def save_simulation_results(path: str | Path, results: SimulationResults) -> None: SimulationResultsSerializer.save(results, path) def load_simulation_results(path: str | Path) -> SimulationResults: return SimulationResultsSerializer.load(path)