From b09caa34756194383dd8b3b1a5e3b3405cf34c8e Mon Sep 17 00:00:00 2001 From: Joppe Blondel Date: Fri, 24 Jul 2026 16:30:14 +0200 Subject: [PATCH] Simulate with OMC --- src/bedit_simulation/__init__.py | 34 ++ src/bedit_simulation/openmodelica.py | 153 ++++++ src/bedit_simulation/results.py | 46 ++ src/bedit_simulation/simulation.py | 691 +++++++++++++++++++++++++++ tests/manual_test.py | 15 - tests/unit/test_simulation.py | 167 +++++++ 6 files changed, 1091 insertions(+), 15 deletions(-) create mode 100644 src/bedit_simulation/__init__.py create mode 100644 src/bedit_simulation/openmodelica.py create mode 100644 src/bedit_simulation/results.py create mode 100644 src/bedit_simulation/simulation.py delete mode 100644 tests/manual_test.py create mode 100644 tests/unit/test_simulation.py diff --git a/src/bedit_simulation/__init__.py b/src/bedit_simulation/__init__.py new file mode 100644 index 0000000..cc4f7da --- /dev/null +++ b/src/bedit_simulation/__init__.py @@ -0,0 +1,34 @@ +"""Simple, application-independent OpenModelica simulation API.""" + +from .openmodelica import ( + OpenModelicaError, + OpenModelicaRunner, + ProcessResult, +) +from .results import SimulationResult, load_openmodelica_csv +from .simulation import ( + ModelBuildResult, + ModelCheckResult, + ModelInfo, + Simulation, + SimulationOptions, + SimulationProgress, + SimulationStateError, + simulate, +) + +__all__ = [ + "OpenModelicaError", + "OpenModelicaRunner", + "ProcessResult", + "ModelBuildResult", + "ModelCheckResult", + "ModelInfo", + "Simulation", + "SimulationOptions", + "SimulationProgress", + "SimulationStateError", + "SimulationResult", + "load_openmodelica_csv", + "simulate", +] diff --git a/src/bedit_simulation/openmodelica.py b/src/bedit_simulation/openmodelica.py new file mode 100644 index 0000000..da59f0c --- /dev/null +++ b/src/bedit_simulation/openmodelica.py @@ -0,0 +1,153 @@ +"""Run OMC directly or through a user-provided wrapper command.""" + +from __future__ import annotations + +import asyncio +import os +import shlex +import subprocess +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass +from pathlib import Path +from threading import Event, Thread +from typing import Any, TypeVar + + +@dataclass(frozen=True) +class ProcessResult: + """Captured output from one OMC invocation.""" + + command: tuple[str, ...] + return_code: int + stdout: str + stderr: str + + +class OpenModelicaError(RuntimeError): + """Raised when OMC cannot start or reports a failure.""" + + +ProcessExecutor = Callable[ + [Sequence[str], Path, float | None, Mapping[str, str] | None], + ProcessResult, +] +_Result = TypeVar("_Result") + + +class OpenModelicaRunner: + """Execute ``.mos`` scripts using an OMC command or wrapper script. + + ``command`` may be ``"omc"``, a path to a wrapper script, or a shell-like + command prefix such as ``"docker run ... omc"``. It is split with + :func:`shlex.split` and is never passed through a shell. + """ + + def __init__( + self, + command: str | Sequence[str] = "omc", + *, + timeout: float | None = None, + environment: Mapping[str, str] | None = None, + executor: ProcessExecutor | None = None, + ) -> None: + self.command = _command_parts(command) + self.timeout = timeout + self.environment = environment + self._executor = executor or _execute_process + + def _run(self, script: Path, working_directory: Path) -> ProcessResult: + """Run ``script`` in ``working_directory`` and return captured output.""" + command = (*self.command, str(script.resolve())) + try: + result = self._executor( + command, + working_directory, + self.timeout, + self.environment, + ) + except FileNotFoundError: + raise OpenModelicaError( + f"OpenModelica command {self.command[0]!r} was not found" + ) from None + except subprocess.TimeoutExpired as error: + raise OpenModelicaError( + f"OpenModelica timed out after {error.timeout} seconds" + ) from error + if result.return_code != 0: + details = result.stderr.strip() or result.stdout.strip() + suffix = f": {details}" if details else "" + raise OpenModelicaError( + f"OpenModelica exited with status {result.return_code}{suffix}" + ) + return result + + async def run( + self, + script: Path, + working_directory: Path, + ) -> ProcessResult: + """Run OMC on an asyncio worker thread.""" + return await _run_on_worker(self._run, script, working_directory) + + +def _command_parts(command: str | Sequence[str]) -> tuple[str, ...]: + if isinstance(command, str): + expanded = os.path.expandvars(os.path.expanduser(command)) + parts = tuple(shlex.split(expanded)) + else: + parts = tuple(map(str, command)) + if not parts: + raise ValueError("OpenModelica command cannot be empty") + return parts + + +def _execute_process( + command: Sequence[str], + working_directory: Path, + timeout: float | None, + environment: Mapping[str, str] | None, +) -> ProcessResult: + process_environment = None + if environment is not None: + process_environment = {**os.environ, **environment} + completed = subprocess.run( + list(command), + cwd=working_directory, + env=process_environment, + timeout=timeout, + text=True, + capture_output=True, + check=False, + ) + return ProcessResult( + command=tuple(command), + return_code=completed.returncode, + stdout=completed.stdout, + stderr=completed.stderr, + ) + + +async def _run_on_worker( + operation: Callable[..., _Result], + *args: Any, + **kwargs: Any, +) -> _Result: + """Run a blocking callable on a thread and await it without Qt.""" + finished = Event() + results: list[_Result] = [] + errors: list[BaseException] = [] + + def invoke() -> None: + try: + results.append(operation(*args, **kwargs)) + except BaseException as error: + errors.append(error) + finally: + finished.set() + + Thread(target=invoke, name="bedit-openmodelica", daemon=True).start() + while not finished.is_set(): + await asyncio.sleep(0.01) + if errors: + raise errors[0] + return results[0] diff --git a/src/bedit_simulation/results.py b/src/bedit_simulation/results.py new file mode 100644 index 0000000..b599f0e --- /dev/null +++ b/src/bedit_simulation/results.py @@ -0,0 +1,46 @@ +"""Simulation result types and OpenModelica CSV loading.""" + +from __future__ import annotations + +import csv +from dataclasses import dataclass +from pathlib import Path + + +@dataclass(frozen=True) +class SimulationResult: + """Numeric results and process information from a completed simulation.""" + + model_name: str + data: dict[str, list[float]] + process_output: str = "" + process_errors: str = "" + result_file: Path | None = None + + +def load_openmodelica_csv(path: str | Path) -> dict[str, list[float]]: + """Load an OpenModelica CSV result into one list per variable.""" + source = Path(path) + try: + with source.open(newline="", encoding="utf-8") as stream: + reader = csv.reader(stream) + headers = next(reader) + if not headers or len(headers) != len(set(headers)): + raise ValueError("invalid or duplicate CSV headers") + columns: dict[str, list[float]] = { + header: [] + for header in headers + } + for row_number, row in enumerate(reader, start=2): + if len(row) != len(headers): + raise ValueError( + f"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 simulation results: {error}") from error + except StopIteration as error: + raise ValueError("simulation result CSV is empty") from error + return columns diff --git a/src/bedit_simulation/simulation.py b/src/bedit_simulation/simulation.py new file mode 100644 index 0000000..d6f3d6c --- /dev/null +++ b/src/bedit_simulation/simulation.py @@ -0,0 +1,691 @@ +"""High-level Modelica composition and simulation service.""" + +from __future__ import annotations + +import json +import shlex +import socket +import tempfile +import time +import xml.etree.ElementTree as ET +from dataclasses import dataclass +from pathlib import Path +from collections.abc import Sequence +from threading import Event, Lock, Thread + +from bedit_core.modelica import CompositionResult, compose_modelica_model +from bedit_core.models import Component + +from .openmodelica import ( + OpenModelicaError, + OpenModelicaRunner, + ProcessResult, + _run_on_worker, +) +from .results import SimulationResult, load_openmodelica_csv + +_MODEL_FILE = "model.mo" +_RUN_SCRIPT_FILE = "run.mos" +_RUN_OUTPUT_FILE = "simulation-output.txt" + + +@dataclass(frozen=True) +class SimulationOptions: + """Time range and numerical options passed to OpenModelica.""" + + start_time: float = 0.0 + stop_time: float = 1.0 + number_of_intervals: int = 500 + tolerance: float = 1e-6 + method: str | None = None + + def __post_init__(self) -> None: + if self.stop_time <= self.start_time: + raise ValueError("stop_time must be greater than start_time") + if self.number_of_intervals <= 0: + raise ValueError("number_of_intervals must be positive") + if self.tolerance <= 0: + raise ValueError("tolerance must be positive") + + +@dataclass(frozen=True) +class ModelCheckResult: + """Result of loading and checking one Modelica model.""" + + model_name: str + successful: bool + output: str + errors: str + + +@dataclass(frozen=True) +class ModelInfo: + """Raw OpenModelica introspection output for a loaded model.""" + + model_name: str + output: str + errors: str + + +@dataclass(frozen=True) +class ModelBuildResult: + """Compiled model executable and the OMC process output.""" + + model_name: str + executable: Path + output: str + errors: str + + +@dataclass(frozen=True) +class SimulationProgress: + """Latest progress status reported by the simulation executable.""" + + percentage: int = 0 + phase: str = "" + time: float = 0.0 + current_step_size: float = 0.0 + + +class SimulationStateError(RuntimeError): + """Raised when an operation requires a model that has not been loaded.""" + + +class Simulation: + """Stateful asynchronous OpenModelica session facade.""" + + def __init__( + self, + runner: OpenModelicaRunner | None = None, + ) -> None: + self.runner = runner or OpenModelicaRunner() + self._modelica: str | None = None + self._model_name: str | None = None + self.last_check: ModelCheckResult | None = None + self.last_info: ModelInfo | None = None + self.last_build: ModelBuildResult | None = None + self.last_result: SimulationResult | None = None + self._progress = SimulationProgress() + self._progress_lock = Lock() + + @staticmethod + def _compose(component: Component) -> CompositionResult: + return compose_modelica_model(component) + + @property + def is_loaded(self) -> bool: + """Return whether this instance has an active model.""" + return self._modelica is not None + + @property + def model_name(self) -> str | None: + """Return the active Modelica model name.""" + return self._model_name + + @property + def modelica(self) -> str | None: + """Return the active Modelica source.""" + return self._modelica + + async def load(self, component: Component) -> CompositionResult: + """Compose ``component`` and make it the active model.""" + composition = await _run_on_worker(self._compose, component) + self._set_model(composition.modelica, composition.model_name) + return composition + + async def load_modelica( + self, + modelica: str, + model_name: str, + ) -> CompositionResult: + """Make existing Modelica source the active model.""" + if not modelica.strip(): + raise ValueError("Modelica source cannot be empty") + if not model_name.strip(): + raise ValueError("model_name cannot be empty") + self._set_model(modelica, model_name) + return CompositionResult(modelica, model_name) + + async def clear(self) -> None: + """Forget the active model.""" + self._modelica = None + self._model_name = None + self._reset_results() + + def _set_model(self, modelica: str, model_name: str) -> None: + self._modelica = modelica + self._model_name = model_name + self._reset_results() + + def _reset_results(self) -> None: + self.last_check = None + self.last_info = None + self.last_build = None + self.last_result = None + self._set_progress(SimulationProgress()) + + def get_progress(self) -> int: + """Return the most recently reported simulation progress, from 0 to 100.""" + with self._progress_lock: + return self._progress.percentage + + @property + def progress(self) -> SimulationProgress: + """Return the complete latest progress report.""" + with self._progress_lock: + return self._progress + + def _set_progress(self, progress: SimulationProgress) -> None: + with self._progress_lock: + self._progress = progress + + def _active_model(self) -> tuple[str, str]: + if self._modelica is None or self._model_name is None: + raise SimulationStateError( + "no model is loaded; call load() or load_modelica() first" + ) + return self._modelica, self._model_name + + def _get_version(self) -> str: + """Return the version reported by the configured OMC command.""" + result = self._execute("getVersion()") + for line in reversed(result.stdout.splitlines()): + try: + value = json.loads(line) + except json.JSONDecodeError: + continue + if isinstance(value, str): + return value + return result.stdout.strip() + + async def get_version(self) -> str: + """Return the OMC version without blocking the event-loop thread.""" + return await _run_on_worker(self._get_version) + + def _execute( + self, + commands: str | Sequence[str], + *, + working_directory: str | Path | None = None, + ) -> ProcessResult: + """Execute arbitrary OMC scripting commands and return raw output.""" + if isinstance(commands, str): + script = commands + else: + script = "\n".join( + command if command.rstrip().endswith(";") else f"{command};" + for command in commands + ) + if not script.strip(): + raise ValueError("OpenModelica commands cannot be empty") + if not script.endswith("\n"): + script += "\n" + + if working_directory is not None: + directory = Path(working_directory) + directory.mkdir(parents=True, exist_ok=True) + return self._execute_script(script, directory, "commands.mos") + with tempfile.TemporaryDirectory(prefix="bedit-omc-") as temporary: + return self._execute_script( + script, + Path(temporary), + "commands.mos", + ) + + async def execute( + self, + commands: str | Sequence[str], + *, + working_directory: str | Path | None = None, + ) -> ProcessResult: + """Execute arbitrary OMC commands on a worker thread.""" + return await _run_on_worker( + self._execute, + commands, + working_directory=working_directory, + ) + + async def check( + self, + *, + working_directory: str | Path | None = None, + ) -> ModelCheckResult: + """Check the active model on a worker thread.""" + modelica, model_name = self._active_model() + result = await _run_on_worker( + self._check_modelica, + modelica, + model_name, + working_directory=working_directory, + ) + self.last_check = result + return result + + def _check_modelica( + self, + modelica: str, + model_name: str, + *, + working_directory: str | Path | None = None, + ) -> ModelCheckResult: + """Load and run ``checkModel`` on existing Modelica source.""" + process = self._run_model_commands( + modelica, + [ + f"checkModel({model_name})", + "getErrorString()", + ], + working_directory, + ) + return ModelCheckResult( + model_name=model_name, + successful="completed successfully" in process.stdout.lower(), + output=process.stdout, + errors=process.stderr, + ) + + async def get_model_info( + self, + *, + working_directory: str | Path | None = None, + ) -> ModelInfo: + """Inspect the active model on a worker thread.""" + modelica, model_name = self._active_model() + result = await _run_on_worker( + self._get_modelica_info, + modelica, + model_name, + working_directory=working_directory, + ) + self.last_info = result + return result + + def _get_modelica_info( + self, + modelica: str, + model_name: str, + *, + working_directory: str | Path | None = None, + ) -> ModelInfo: + """Return raw class, component, equation, and connection information.""" + process = self._run_model_commands( + modelica, + [ + f"getClassInformation({model_name})", + f"getComponents({model_name})", + f"getEquationCount({model_name})", + f"getConnectionCount({model_name})", + "getErrorString()", + ], + working_directory, + ) + return ModelInfo( + model_name=model_name, + output=process.stdout, + errors=process.stderr, + ) + + async def build( + self, + working_directory: str | Path, + ) -> ModelBuildResult: + """Build the active model and retain its artifacts.""" + modelica, model_name = self._active_model() + result = await _run_on_worker( + self._build_modelica, + modelica, + model_name, + working_directory, + ) + self.last_build = result + return result + + def _build_modelica( + self, + modelica: str, + model_name: str, + working_directory: str | Path, + ) -> ModelBuildResult: + """Compile Modelica source and retain its artifacts in a directory.""" + directory = Path(working_directory) + directory.mkdir(parents=True, exist_ok=True) + process = self._run_model_commands( + modelica, + [ + f'buildModel({model_name}, outputFormat="csv")', + "getErrorString()", + ], + directory, + ) + executable = directory / model_name + if not executable.is_file(): + windows_executable = executable.with_suffix(".exe") + if windows_executable.is_file(): + executable = windows_executable + else: + details = process.stderr.strip() or process.stdout.strip() + suffix = f": {details}" if details else "" + raise OpenModelicaError( + f"OpenModelica did not build {model_name!r}{suffix}" + ) + return ModelBuildResult( + model_name=model_name, + executable=executable, + output=process.stdout, + errors=process.stderr, + ) + + async def compile( + self, + working_directory: str | Path, + ) -> ModelBuildResult: + """Alias for :meth:`build` using compiler-oriented terminology.""" + return await self.build(working_directory) + + async def run( + self, + options: SimulationOptions | None = None, + *, + working_directory: str | Path | None = None, + ) -> SimulationResult: + """Run the previously built executable. + + :meth:`build` must have completed successfully first. + """ + _modelica, model_name = self._active_model() + build = self.last_build + if ( + build is None + or build.model_name != model_name + or not build.executable.is_file() + ): + raise SimulationStateError( + "the active model has not been built; call build() first" + ) + directory = ( + Path(working_directory) + if working_directory is not None + else build.executable.parent + ) + if directory.resolve() != build.executable.parent.resolve(): + raise SimulationStateError( + "working_directory must be the directory used by build()" + ) + self._set_progress(SimulationProgress()) + result = await _run_on_worker( + self._run_built_model, + model_name, + build.executable, + options, + directory, + ) + self.last_result = result + return result + + def _run_built_model( + self, + model_name: str, + executable: Path, + options: SimulationOptions | None = None, + working_directory: Path | None = None, + ) -> SimulationResult: + """Run an existing compiled Modelica executable.""" + simulation_options = options or SimulationOptions() + directory = working_directory or executable.parent + return self._run_in_directory( + model_name, + executable, + simulation_options, + directory, + ) + + def _run_in_directory( + self, + model_name: str, + executable: Path, + options: SimulationOptions, + directory: Path, + ) -> SimulationResult: + result_path = directory / f"{executable.stem}_res.csv" + process = self._run_executable(executable, options, directory) + if not result_path.is_file(): + details = process.stderr.strip() or process.stdout.strip() + suffix = f": {details}" if details else "" + raise OpenModelicaError( + f"simulation did not create {result_path.name!r}{suffix}" + ) + self._set_progress( + SimulationProgress( + percentage=100, + phase=self.progress.phase, + time=options.stop_time, + current_step_size=self.progress.current_step_size, + ) + ) + return SimulationResult( + model_name=model_name, + data=load_openmodelica_csv(result_path), + process_output=process.stdout, + process_errors=process.stderr, + result_file=result_path, + ) + + def _run_executable( + self, + executable: Path, + options: SimulationOptions, + directory: Path, + ) -> ProcessResult: + server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + server.bind(("127.0.0.1", 0)) + server.listen(1) + server.settimeout(0.25) + port = server.getsockname()[1] + command_finished = Event() + reader_errors: list[BaseException] = [] + + reader = Thread( + target=self._read_progress, + args=(server, command_finished, reader_errors), + name="bedit-simulation-progress", + daemon=True, + ) + reader.start() + + step_size = ( + options.stop_time - options.start_time + ) / options.number_of_intervals + arguments = [ + str(executable.resolve()), + f"-startTime={options.start_time}", + f"-stopTime={options.stop_time}", + f"-stepSize={step_size}", + f"-tolerance={options.tolerance}", + "-outputFormat=csv", + f"-port={port}", + "-logFormat=xmltcp", + ] + if options.method: + arguments.append(f"-s={options.method}") + script_path = directory / _RUN_SCRIPT_FILE + script_path.write_text( + _executable_script(arguments, directory), + encoding="utf-8", + ) + try: + process = self.runner._run(script_path, directory) + finally: + command_finished.set() + reader.join(timeout=20) + server.close() + if reader.is_alive(): + raise OpenModelicaError("simulation progress connection did not close") + if reader_errors: + raise OpenModelicaError( + f"could not read simulation progress: {reader_errors[0]}" + ) from reader_errors[0] + return process + + def _read_progress( + self, + server: socket.socket, + command_finished: Event, + errors: list[BaseException], + ) -> None: + try: + connection: socket.socket | None = None + deadline = time.monotonic() + 15.0 + while connection is None and time.monotonic() < deadline: + try: + connection, _ = server.accept() + except TimeoutError: + if command_finished.is_set(): + raise OpenModelicaError( + "simulation finished before opening its progress " + "connection" + ) + continue + if connection is None: + raise OpenModelicaError( + "simulation did not connect to the progress server" + ) + with connection, connection.makefile("r", encoding="utf-8") as stream: + for line in stream: + try: + self._parse_progress_line(line) + except (ET.ParseError, ValueError): + # OpenModelica occasionally emits incomplete XML + # messages. Keep consuming the stream so the simulator + # does not receive SIGPIPE. + continue + except (OSError, OpenModelicaError) as error: + errors.append(error) + + def _parse_progress_line(self, line: str) -> None: + element = ET.fromstring(line.strip()) + if element.tag != "status": + return + raw_progress = float(element.attrib.get("progress", 0)) + percentage = int(raw_progress) + self._set_progress( + SimulationProgress( + percentage=max(0, min(100, percentage)), + phase=element.attrib.get("phase", ""), + time=float(element.attrib.get("time", 0)), + current_step_size=float( + element.attrib.get("currentStepSize", 0) + ), + ) + ) + + def _run_model_commands( + self, + modelica: str, + commands: Sequence[str], + working_directory: str | Path | None, + ) -> ProcessResult: + if not modelica.strip(): + raise ValueError("Modelica source cannot be empty") + if working_directory is not None: + directory = Path(working_directory) + directory.mkdir(parents=True, exist_ok=True) + return self._run_model_commands_in_directory( + modelica, + commands, + directory, + ) + with tempfile.TemporaryDirectory(prefix="bedit-omc-") as temporary: + return self._run_model_commands_in_directory( + modelica, + commands, + Path(temporary), + ) + + def _run_model_commands_in_directory( + self, + modelica: str, + commands: Sequence[str], + directory: Path, + ) -> ProcessResult: + (directory / _MODEL_FILE).write_text(modelica, encoding="utf-8") + script = _load_model_script(directory, commands) + return self._execute_script(script, directory, "model-command.mos") + + def _execute_script( + self, + script: str, + directory: Path, + filename: str, + ) -> ProcessResult: + script_path = directory / filename + script_path.write_text(script, encoding="utf-8") + return self.runner._run(script_path, directory) + + +async def simulate( + component: Component, + options: SimulationOptions | None = None, + *, + omc_command: str = "omc", + working_directory: str | Path | None = None, + timeout: float | None = None, +) -> SimulationResult: + """Convenience API for simulating a component on a worker thread.""" + runner = OpenModelicaRunner(omc_command, timeout=timeout) + simulation = Simulation(runner) + await simulation.load(component) + if working_directory is not None: + await simulation.build(working_directory) + return await simulation.run(options) + with tempfile.TemporaryDirectory(prefix="bedit-simulation-") as temporary: + await simulation.build(temporary) + result = await simulation.run(options) + return SimulationResult( + model_name=result.model_name, + data=result.data, + process_output=result.process_output, + process_errors=result.process_errors, + ) + + +def _executable_script( + arguments: Sequence[str], + working_directory: Path, +) -> str: + """Create an OMC script that starts a compiled simulation binary.""" + command = shlex.join(arguments) + return "\n".join( + [ + f"cd({json.dumps(str(working_directory.resolve()))});", + f"status := system({json.dumps(command)}, " + f"{json.dumps(_RUN_OUTPUT_FILE)});", + "if status <> 0 then", + f" print(readFile({json.dumps(_RUN_OUTPUT_FILE)}));", + " exit(1);", + "end if;", + "", + ] + ) + + +def _load_model_script( + working_directory: Path, + commands: Sequence[str], +) -> str: + """Create a script that loads ``model.mo`` before custom commands.""" + return "\n".join( + [ + f"cd({json.dumps(str(working_directory.resolve()))});", + f"loaded := loadFile({json.dumps(_MODEL_FILE)});", + "if not loaded then", + " print(getErrorString());", + " exit(1);", + "end if;", + *( + command if command.rstrip().endswith(";") else f"{command};" + for command in commands + ), + "", + ] + ) diff --git a/tests/manual_test.py b/tests/manual_test.py deleted file mode 100644 index d72b855..0000000 --- a/tests/manual_test.py +++ /dev/null @@ -1,15 +0,0 @@ -import sys -from bedit_core.serialization import load, save -from bedit_core.bondgraph import causality_inference -from bedit_core.modelica import compose_modelica_model - -def main(path: str): - doc = load(path) - for id, root in doc.root.items(): - doc.root[id] = causality_inference(doc.root[id]) - composition = compose_modelica_model(doc.root[id]) - print(composition.modelica) - save(doc, path) - -if __name__ == "__main__": - main(sys.argv[1]) \ No newline at end of file diff --git a/tests/unit/test_simulation.py b/tests/unit/test_simulation.py new file mode 100644 index 0000000..c1f39a1 --- /dev/null +++ b/tests/unit/test_simulation.py @@ -0,0 +1,167 @@ +from __future__ import annotations + +import asyncio +import re +import socket +import threading +from pathlib import Path +from typing import Mapping, Sequence + +import pytest + +from bedit_simulation import ( + OpenModelicaRunner, + ProcessResult, + Simulation, + SimulationOptions, + SimulationStateError, +) + + +def _fake_omc( + command: Sequence[str], + working_directory: Path, + timeout: float | None, + environment: Mapping[str, str] | None, +) -> ProcessResult: + del timeout, environment + script = Path(command[-1]).read_text(encoding="utf-8") + if "buildModel(Example" in script: + (working_directory / "Example").touch() + if "system(" in script: + port_match = re.search(r"-port=(\d+)", script) + assert port_match is not None + with socket.create_connection( + ("127.0.0.1", int(port_match.group(1))) + ) as connection: + connection.sendall( + b'\n' + ) + (working_directory / "Example_res.csv").write_text( + '"time","x"\n0,1\n1,2\n', + encoding="utf-8", + ) + return ProcessResult( + tuple(command), + 0, + '"Check of Example completed successfully."\n', + "", + ) + + +@pytest.mark.unit +def test_runs_omc_through_configured_command(tmp_path: Path) -> None: + runner = OpenModelicaRunner( + ["./run-omc-in-docker"], + executor=_fake_omc, + ) + + async def run() -> object: + simulation = Simulation(runner) + await simulation.load_modelica( + "model Example\nend Example;\n", + "Example", + ) + await simulation.build(tmp_path) + return await simulation.run( + SimulationOptions(stop_time=2, number_of_intervals=20), + ) + + result = asyncio.run(run()) + + assert result.data == {"time": [0.0, 1.0], "x": [1.0, 2.0]} + assert result.result_file == tmp_path / "Example_res.csv" + script = (tmp_path / "run.mos").read_text(encoding="utf-8") + assert f'cd("{tmp_path}")' in script + assert "stepSize=0.1" in script + assert "-logFormat=xmltcp" in script + assert "simulate(Example" not in script + + +@pytest.mark.unit +def test_rejects_invalid_time_range() -> None: + with pytest.raises(ValueError, match="stop_time"): + SimulationOptions(start_time=1, stop_time=1) + + +@pytest.mark.unit +def test_run_requires_a_built_model() -> None: + simulation = Simulation(OpenModelicaRunner("omc", executor=_fake_omc)) + + async def run() -> None: + await simulation.load_modelica( + "model Example\nend Example;\n", + "Example", + ) + with pytest.raises(SimulationStateError, match=r"call build\(\) first"): + await simulation.run() + + asyncio.run(run()) + + +@pytest.mark.unit +def test_checks_existing_modelica_source(tmp_path: Path) -> None: + simulation = Simulation(OpenModelicaRunner("omc", executor=_fake_omc)) + + async def check() -> object: + await simulation.load_modelica( + "model Example\nend Example;\n", + "Example", + ) + return await simulation.check(working_directory=tmp_path) + + result = asyncio.run(check()) + + assert result.successful + script = (tmp_path / "model-command.mos").read_text(encoding="utf-8") + assert "checkModel(Example);" in script + + +@pytest.mark.unit +def test_async_execution_uses_a_worker_thread(tmp_path: Path) -> None: + caller_thread = threading.get_ident() + execution_threads: list[int] = [] + + def executor( + command: Sequence[str], + working_directory: Path, + timeout: float | None, + environment: Mapping[str, str] | None, + ) -> ProcessResult: + del working_directory, timeout, environment + execution_threads.append(threading.get_ident()) + return ProcessResult(tuple(command), 0, '"OpenModelica test"', "") + + simulation = Simulation(OpenModelicaRunner("omc", executor=executor)) + + result = asyncio.run( + simulation.execute( + "getVersion()", + working_directory=tmp_path, + ) + ) + + assert result.return_code == 0 + assert len(execution_threads) == 1 + assert execution_threads[0] != caller_thread + + +@pytest.mark.unit +def test_requires_and_retains_a_loaded_model(tmp_path: Path) -> None: + simulation = Simulation(OpenModelicaRunner("omc", executor=_fake_omc)) + + async def use_session() -> None: + with pytest.raises(SimulationStateError, match="no model is loaded"): + await simulation.check() + + await simulation.load_modelica( + "model Example\nend Example;\n", + "Example", + ) + check = await simulation.check(working_directory=tmp_path) + + assert simulation.model_name == "Example" + assert simulation.last_check is check + + asyncio.run(use_session())