Implemented live updates using the growing MAT file.

This commit is contained in:
Joppe Blondel
2026-08-19 11:50:21 +02:00
parent be9e05732f
commit eef77aed09
5 changed files with 45872 additions and 7679 deletions
+2 -1
View File
@@ -6,7 +6,7 @@ from .openmodelica import (
ProcessResult,
SimulationCancelledError,
)
from .results import SimulationResult, load_openmodelica_csv, load_openmodelica_mat
from .results import SimulationResult, load_openmodelica_csv, load_openmodelica_mat, load_openmodelica_mat_snapshot
from .compile import compile_component, compile_component_sync
from .simulation import (
ModelBuildResult,
@@ -39,5 +39,6 @@ __all__ = [
"compile_component_sync",
"load_openmodelica_csv",
"load_openmodelica_mat",
"load_openmodelica_mat_snapshot",
"simulate",
]
+55 -2
View File
@@ -3,6 +3,8 @@
from __future__ import annotations
import csv
import io
import struct
from dataclasses import dataclass
from pathlib import Path
@@ -54,6 +56,34 @@ def load_openmodelica_mat(path: str | Path) -> dict[str, list[float]]:
raise ValueError("reading OpenModelica MAT results requires DyMat") from None
try:
result = DyMatFile(str(Path(path)))
columns = _dymat_columns(result)
except (IndexError, KeyError, OSError, TypeError, ValueError) as error:
raise ValueError(f"could not read OpenModelica MAT results: {error}") from error
return columns
def load_openmodelica_mat_snapshot(path: str | Path) -> dict[str, list[float]]:
"""Read the complete time-step columns currently available in a growing MAT file."""
try:
from DyMat import DyMatFile
except ImportError:
raise ValueError("reading OpenModelica MAT results requires DyMat") from None
try:
snapshot = bytearray(Path(path).read_bytes())
header_offset, byte_order, rows, columns, element_size, data_offset = _mat_v4_matrix(snapshot, "data_2")
available_columns = (len(snapshot) - data_offset) // (rows * element_size)
complete_columns = min(columns, available_columns) if columns > 0 else available_columns
if complete_columns < 2:
raise ValueError("the MAT result does not contain enough complete samples yet")
struct.pack_into(f"{byte_order}i", snapshot, header_offset + 8, complete_columns)
del snapshot[data_offset + complete_columns * rows * element_size:]
columns_data = _dymat_columns(DyMatFile(io.BytesIO(snapshot)))
except (IndexError, KeyError, OSError, TypeError, ValueError, struct.error) as error:
raise ValueError(f"could not read growing OpenModelica MAT results: {error}") from error
return columns_data
def _dymat_columns(result) -> dict[str, list[float]]:
blocks = result.blocks()
time_block = 2 if 2 in blocks else blocks[-1]
times = result.abscissa(time_block, valuesOnly=True).astype(float).tolist()
@@ -63,6 +93,29 @@ def load_openmodelica_mat(path: str | Path) -> dict[str, list[float]]:
if len(values) != len(times):
values = [values[0]] * len(times)
columns[name] = values
except (IndexError, KeyError, OSError, TypeError, ValueError) as error:
raise ValueError(f"could not read OpenModelica MAT results: {error}") from error
return columns
def _mat_v4_matrix(content: bytearray, target: str) -> tuple[int, str, int, int, int, int]:
offset = 0
while offset + 20 <= len(content):
byte_order = _mat_v4_byte_order(content, offset)
matrix_type, rows, columns, imaginary, name_length = struct.unpack_from(f"{byte_order}5i", content, offset)
if rows <= 0 or columns < 0 or imaginary not in (0, 1) or name_length <= 0:
break
name_offset = offset + 20
data_offset = name_offset + name_length
name = bytes(content[name_offset:data_offset]).split(b"\x00", 1)[0].decode("ascii", errors="replace")
precision = (matrix_type // 10) % 10
element_size = {0: 8, 1: 4, 2: 4, 3: 2, 4: 2, 5: 1}.get(precision)
if element_size is None:
break
if name == target:
return offset, byte_order, rows, columns, element_size, data_offset
offset = data_offset + rows * columns * element_size * (2 if imaginary else 1)
raise ValueError(f"MAT v4 matrix {target!r} was not found")
def _mat_v4_byte_order(content: bytearray, offset: int) -> str:
little_type = struct.unpack_from("<i", content, offset)[0]
return "<" if little_type // 1000 in (0, 1) and (little_type // 10) % 10 <= 5 else ">"
+17 -31
View File
@@ -9,9 +9,6 @@ from .openmodelica import OpenModelicaRunner
from .results import SimulationResult
from .simulation import Simulation, SimulationOptions, SimulationStateError
_SEGMENTS_PER_RUN = 10
@dataclass(frozen=True)
class SimulationRunSettings:
start_time: float = 0.0
@@ -42,63 +39,54 @@ class SimulationSession:
self.results = list(results)
self._simulation = Simulation(runner)
self._simulation.load_compiled(model_name, executable)
self._segments_completed = 0
self._segments_total = 1
@property
def is_running(self) -> bool:
return self._simulation.is_running
def get_progress(self) -> int:
return min(100, int((self._segments_completed + self._simulation.get_progress() / 100) * 100 / self._segments_total))
return self._simulation.get_progress()
async def run_next(self) -> SimulationResult:
return await self.run_next_segmented()
async def run_next_segmented(self, segment_completed: Callable[[SimulationResult], None] | None = None) -> SimulationResult:
"""Advance one configured duration in restartable, plot-ready segments."""
"""Advance one configured duration in one process with live result snapshots."""
previous_file = self._restart_file()
if previous_file is None and self.current_end_time != self.settings.start_time:
self.current_end_time = self.settings.start_time
self.results.clear()
run_start_time = self.current_end_time
run_stop_time = run_start_time + self.settings.duration
segment_duration = self.settings.duration / _SEGMENTS_PER_RUN
intervals = self._segment_intervals(segment_duration)
self._segments_completed = 0
self._segments_total = _SEGMENTS_PER_RUN
result: SimulationResult | None = None
for segment in range(_SEGMENTS_PER_RUN):
start_time = self.current_end_time
stop_time = run_stop_time if segment == _SEGMENTS_PER_RUN - 1 else run_start_time + (segment + 1) * segment_duration
result_file = self._simulation.last_build.executable.parent / f"{self._simulation.last_build.executable.stem}_segment_{len(self.results):06d}.mat"
stop_time = start_time + self.settings.duration
completed_results = list(self.results)
result_file = self._simulation.last_build.executable.parent / f"{self._simulation.last_build.executable.stem}_segment_{len(completed_results):06d}.mat"
def publish(result: SimulationResult) -> None:
self.results[:] = [*completed_results, result]
if segment_completed is not None:
segment_completed(result)
options = SimulationOptions(
start_time=start_time,
stop_time=stop_time,
number_of_intervals=intervals,
number_of_intervals=self._run_intervals(),
tolerance=self.settings.tolerance,
method=self.settings.method,
result_file=result_file,
initialization_file=previous_file,
initialization_time=start_time if previous_file is not None else None,
skip_initialization=previous_file is not None,
intermediate_result=publish,
)
result = await self._simulation.run(options)
self.current_end_time = stop_time
self.results.append(result)
previous_file = result.result_file
self._segments_completed += 1
if segment_completed is not None:
segment_completed(result)
assert result is not None
self._segments_total = 1
self._segments_completed = 1
self.results[:] = [*completed_results, result]
return result
def _segment_intervals(self, segment_duration: float) -> int:
def _run_intervals(self) -> int:
if self.settings.use_timed_steps:
return max(1, math.ceil(segment_duration / self.settings.step_size))
return max(1, math.ceil(self.settings.number_of_steps / _SEGMENTS_PER_RUN))
return max(1, math.ceil(self.settings.duration / self.settings.step_size))
return self.settings.number_of_steps
def _restart_file(self) -> Path | None:
for result in reversed(self.results):
@@ -114,5 +102,3 @@ class SimulationSession:
raise SimulationStateError("cannot reset while a simulation is running")
self.current_end_time = self.settings.start_time
self.results.clear()
self._segments_completed = 0
self._segments_total = 1
+24 -3
View File
@@ -10,7 +10,7 @@ import time
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from pathlib import Path
from collections.abc import Sequence
from collections.abc import Callable, Sequence
from threading import Event, Lock, Thread
from bedit_core.modelica import CompositionResult, compose_modelica_model
@@ -22,7 +22,7 @@ from .openmodelica import (
ProcessResult,
_run_on_worker,
)
from .results import SimulationResult, load_openmodelica_csv, load_openmodelica_mat
from .results import SimulationResult, load_openmodelica_csv, load_openmodelica_mat, load_openmodelica_mat_snapshot
_MODEL_FILE = "model.mo"
_RUN_SCRIPT_FILE = "run.mos"
@@ -43,6 +43,7 @@ class SimulationOptions:
initialization_file: str | Path | None = None
initialization_time: float | None = None
skip_initialization: bool = False
intermediate_result: Callable[[SimulationResult], None] | None = None
def __post_init__(self) -> None:
if self.stop_time <= self.start_time:
@@ -518,7 +519,7 @@ class Simulation:
reader = Thread(
target=self._read_progress,
args=(server, command_finished, reader_errors),
args=(server, command_finished, reader_errors, options.result_file, self._model_name, options.intermediate_result),
name="bedit-simulation-progress",
daemon=True,
)
@@ -568,6 +569,9 @@ class Simulation:
server: socket.socket,
command_finished: Event,
errors: list[BaseException],
result_file: str | Path | None,
model_name: str | None,
intermediate_result: Callable[[SimulationResult], None] | None,
) -> None:
try:
connection: socket.socket | None = None
@@ -586,6 +590,7 @@ class Simulation:
raise OpenModelicaError(
"simulation did not connect to the progress server"
)
next_snapshot = 10
with connection, connection.makefile("r", encoding="utf-8") as stream:
for line in stream:
try:
@@ -595,9 +600,25 @@ class Simulation:
# messages. Keep consuming the stream so the simulator
# does not receive SIGPIPE.
continue
percentage = self.get_progress()
if intermediate_result is not None and percentage >= next_snapshot:
self._publish_intermediate_result(result_file, model_name, intermediate_result)
next_snapshot = (percentage // 10 + 1) * 10
except (OSError, OpenModelicaError) as error:
errors.append(error)
def _publish_intermediate_result(self, result_file: str | Path | None, model_name: str | None, callback: Callable[[SimulationResult], None]) -> None:
if result_file is None or model_name is None:
return
path = Path(result_file)
if not path.is_file() or path.suffix.lower() != ".mat":
return
try:
data = load_openmodelica_mat_snapshot(path)
except Exception: # noqa: BLE001 - the MAT file may be incomplete while the simulator writes it
return
callback(SimulationResult(model_name=model_name, data=data, result_file=path))
def _parse_progress_line(self, line: str) -> None:
element = ET.fromstring(line.strip())
if element.tag != "status":
+45756 -7624
View File
File diff suppressed because it is too large Load Diff