Simulate json or beb model from the command line

pyproject.toml

@@ -9,12 +9,14 @@ description = "A Bondgraph and block scheme simulator"
 readme = "README.md"
 requires-python = ">=3.10"
 dependencies = [
+    "matplotlib>=3.8,<4",
     "msgpack>=1.0,<2",
     "PySide6>=6.7,<7",
 ]
 
 [project.scripts]
 bedit-graphviz = "bedit_util.graphviz:main"
+bedit-simulate = "bedit_util.simulate:main"
 
 [project.optional-dependencies]
 dev = [

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",
+]

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]

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

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
+            ),
+            "",
+        ]
+    )

src/bedit_util/__init__.py

@@ -2,4 +2,7 @@
 
 from .graphviz import document_to_dot, render_document
 
-__all__ = ["document_to_dot", "render_document"]
+__all__ = [
+    "document_to_dot",
+    "render_document",
+]

src/bedit_util/component_path.py

@@ -0,0 +1,41 @@
+"""Resolve components by their dotted name path."""
+
+from __future__ import annotations
+
+from bedit_core.models import Component, ComponentID, Document, GraphImplementation
+
+
+def component_paths(
+    root: Component,
+    prefix: tuple[str, ...] = (),
+) -> list[tuple[Component, tuple[str, ...]]]:
+    """Return components paired with their dotted-name path parts."""
+    result: list[tuple[Component, tuple[str, ...]]] = []
+
+    def visit(component: Component, path: tuple[str, ...]) -> None:
+        current_path = path + (component.name,)
+        result.append((component, current_path))
+        if isinstance(component.implementation, GraphImplementation):
+            for child in component.implementation.graph.components.values():
+                visit(child, current_path)
+
+    visit(root, prefix)
+    return result
+
+
+def find_component(
+    document: Document,
+    requested_path: str,
+) -> tuple[ComponentID, Component, tuple[str, ...]]:
+    """Find exactly one component by its dotted name path."""
+    matches: list[tuple[ComponentID, Component, tuple[str, ...]]] = []
+    for root_id, root in document.root.items():
+        for component, path in component_paths(root):
+            if ".".join(path) == requested_path:
+                matches.append((root_id, component, path))
+
+    if len(matches) == 1:
+        return matches[0]
+    if not matches:
+        raise ValueError(f"component path {requested_path!r} was not found")
+    raise ValueError(f"component path {requested_path!r} is ambiguous")

src/bedit_util/graphviz.py

@@ -11,12 +11,12 @@ from bedit_core.bondgraph import BondGraphNetwork, flatten_bondgraph
 from bedit_core.models import (
     BondCausality,
     Component,
-    ComponentID,
     Document,
-    GraphImplementation,
 )
 from bedit_core.serialization import load
 
+from .component_path import component_paths, find_component
+
 
 def document_to_dot(document: Document, component_path: str) -> str:
     """Return DOT for the hierarchy starting at ``component_path``."""
@@ -27,9 +27,9 @@ def document_to_dot(document: Document, component_path: str) -> str:
         "  edge [fontname=\"sans-serif\", arrowsize=0.8];",
     ]
 
-    root_id, component, path = _find_component(document, component_path)
+    root_id, component, path = find_component(document, component_path)
     network = flatten_bondgraph(component)
-    paths = _component_paths(component, path[:-1])
+    paths = component_paths(component, path[:-1])
     node_ids = {
         id(item): f"{root_id}:{'/'.join(item_path)}"
         for item, item_path in paths
@@ -90,42 +90,6 @@ def main(argv: list[str] | None = None) -> int:
     return 0
 
 
-def _component_paths(
-    root: Component,
-    prefix: tuple[str, ...] = (),
-) -> list[tuple[Component, tuple[str, ...]]]:
-    """Return components paired with dotted-name path parts."""
-    result: list[tuple[Component, tuple[str, ...]]] = []
-
-    def visit(component: Component, path: tuple[str, ...]) -> None:
-        current_path = path + (component.name,)
-        result.append((component, current_path))
-        if isinstance(component.implementation, GraphImplementation):
-            for child in component.implementation.graph.components.values():
-                visit(child, current_path)
-
-    visit(root, prefix)
-    return result
-
-
-def _find_component(
-    document: Document,
-    requested_path: str,
-) -> tuple[ComponentID, Component, tuple[str, ...]]:
-    """Find a component by its dotted name path."""
-    matches: list[tuple[ComponentID, Component, tuple[str, ...]]] = []
-    for root_id, root in document.root.items():
-        for component, path in _component_paths(root):
-            if ".".join(path) == requested_path:
-                matches.append((root_id, component, path))
-
-    if len(matches) == 1:
-        return matches[0]
-    if not matches:
-        raise ValueError(f"component path {requested_path!r} was not found")
-    raise ValueError(f"component path {requested_path!r} is ambiguous")
-
-
 def _nodes(
     network: BondGraphNetwork,
     node_ids: dict[int, str],

src/bedit_util/simulate.py

@@ -0,0 +1,248 @@
+"""Command-line simulation and interactive result plotting."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import sys
+import tempfile
+from collections.abc import Callable
+from pathlib import Path
+
+from bedit_core.serialization import load
+from bedit_simulation import (
+    OpenModelicaError,
+    OpenModelicaRunner,
+    Simulation,
+    SimulationOptions,
+    SimulationResult,
+)
+
+from .component_path import find_component
+
+
+async def run_simulation(
+    model: Path,
+    component_path: str,
+    options: SimulationOptions,
+    *,
+    omc_command: str = "omc",
+    working_directory: Path | None = None,
+    timeout: float | None = None,
+    progress_callback: Callable[[int], None] | None = None,
+) -> SimulationResult:
+    """Load and simulate one component selected by dotted path."""
+    document = load(model)
+    _root_id, component, _path = find_component(document, component_path)
+    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 _run_with_progress(
+            simulation,
+            options,
+            progress_callback,
+        )
+    with tempfile.TemporaryDirectory(prefix="bedit-simulation-") as temporary:
+        await simulation.build(temporary)
+        result = await _run_with_progress(
+            simulation,
+            options,
+            progress_callback,
+        )
+        return SimulationResult(
+            model_name=result.model_name,
+            data=result.data,
+            process_output=result.process_output,
+            process_errors=result.process_errors,
+        )
+
+
+async def _run_with_progress(
+    simulation: Simulation,
+    options: SimulationOptions,
+    callback: Callable[[int], None] | None,
+) -> SimulationResult:
+    task = asyncio.create_task(simulation.run(options))
+    last_progress = -1
+    while not task.done():
+        progress = simulation.get_progress()
+        if callback is not None and progress != last_progress:
+            callback(progress)
+            last_progress = progress
+        await asyncio.sleep(0.1)
+    result = await task
+    if callback is not None and last_progress != 100:
+        callback(100)
+    return result
+
+
+class _ProgressBar:
+    """Small dependency-free terminal progress bar."""
+
+    def __init__(self, width: int = 30) -> None:
+        self.width = width
+        self._shown = False
+
+    def update(self, percentage: int) -> None:
+        percentage = max(0, min(100, percentage))
+        completed = self.width * percentage // 100
+        bar = "#" * completed + "-" * (self.width - completed)
+        print(
+            f"\rSimulating [{bar}] {percentage:3d}%",
+            end="",
+            file=sys.stderr,
+            flush=True,
+        )
+        self._shown = True
+
+    def close(self) -> None:
+        if self._shown:
+            print(file=sys.stderr)
+            self._shown = False
+
+
+def create_results_figure(
+    result: SimulationResult,
+    *,
+    title: str | None = None,
+):
+    """Create a Matplotlib figure with controls for trace visibility."""
+    import matplotlib.pyplot as plt
+    from matplotlib.widgets import Button, CheckButtons
+
+    trace_names = [name for name in result.data if name != "time"]
+    if not trace_names:
+        raise ValueError("simulation results contain no plottable traces")
+
+    figure, plot = plt.subplots(figsize=(12, 7))
+    figure.subplots_adjust(right=0.72)
+    plot.set_title(title or result.model_name)
+    plot.set_xlabel("time" if "time" in result.data else "sample")
+    plot.grid(True, alpha=0.3)
+
+    time = result.data.get("time")
+    lines = {}
+    for name in trace_names:
+        values = result.data[name]
+        x_values = time if time is not None and len(time) == len(values) else range(len(values))
+        line, = plot.plot(x_values, values, label=name)
+        lines[name] = line
+
+    checks_axis = figure.add_axes((0.75, 0.17, 0.23, 0.76))
+    checks = CheckButtons(
+        checks_axis,
+        trace_names,
+        [True] * len(trace_names),
+    )
+    for label in checks.labels:
+        label.set_fontsize(8)
+    checks_axis.set_title("Traces", fontsize=10)
+
+    def toggle(label: str) -> None:
+        line = lines[label]
+        line.set_visible(not line.get_visible())
+        figure.canvas.draw_idle()
+
+    checks.on_clicked(toggle)
+
+    def set_all(visible: bool) -> None:
+        for index, active in enumerate(checks.get_status()):
+            if active != visible:
+                checks.set_active(index)
+
+    all_button = Button(
+        figure.add_axes((0.75, 0.07, 0.10, 0.05)),
+        "All",
+    )
+    none_button = Button(
+        figure.add_axes((0.88, 0.07, 0.10, 0.05)),
+        "None",
+    )
+    all_button.on_clicked(lambda _event: set_all(True))
+    none_button.on_clicked(lambda _event: set_all(False))
+
+    # Keep widget objects alive for as long as the figure exists.
+    figure._bedit_widgets = (checks, all_button, none_button)  # type: ignore[attr-defined]
+
+    set_all(False)
+    return figure
+
+
+def show_results(
+    result: SimulationResult,
+    *,
+    title: str | None = None,
+) -> None:
+    """Open the interactive Matplotlib result window."""
+    import matplotlib.pyplot as plt
+
+    create_results_figure(result, title=title)
+    plt.show()
+
+
+def _parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        prog="bedit-simulate",
+        description="Simulate a component from a .json or .beb model.",
+    )
+    parser.add_argument("model", type=Path, help="input .json or .beb model")
+    parser.add_argument(
+        "component_path",
+        help="dotted component path, for example Root.Subsystem",
+    )
+    parser.add_argument("--start-time", type=float, default=0.0)
+    parser.add_argument("--stop-time", type=float, default=1.0)
+    parser.add_argument("--intervals", type=int, default=500)
+    parser.add_argument("--tolerance", type=float, default=1e-6)
+    parser.add_argument("--method", help="OpenModelica solver method")
+    parser.add_argument(
+        "--omc-command",
+        default="omc",
+        help="OMC executable, wrapper script, or command prefix",
+    )
+    parser.add_argument(
+        "--work-dir",
+        type=Path,
+        help="retain generated Modelica and result artifacts here",
+    )
+    parser.add_argument("--timeout", type=float, help="OMC timeout in seconds")
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    """Run a simulation and show its results in Matplotlib."""
+    parser = _parser()
+    args = parser.parse_args(argv)
+    progress = _ProgressBar()
+    try:
+        options = SimulationOptions(
+            start_time=args.start_time,
+            stop_time=args.stop_time,
+            number_of_intervals=args.intervals,
+            tolerance=args.tolerance,
+            method=args.method,
+        )
+        result = asyncio.run(
+            run_simulation(
+                args.model,
+                args.component_path,
+                options,
+                omc_command=args.omc_command,
+                working_directory=args.work_dir,
+                timeout=args.timeout,
+                progress_callback=progress.update,
+            )
+        )
+    except (OSError, ValueError, OpenModelicaError) as error:
+        progress.close()
+        parser.error(str(error))
+    progress.close()
+
+    show_results(result, title=args.component_path)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/manual_test.py

@@ -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])

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'<status phase="integration" currentStepSize="0.1" '
+                b'time="1" progress="50"/>\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())

tests/unit/test_simulation_util.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+import matplotlib
+import pytest
+
+from bedit_simulation import SimulationResult
+from bedit_util.simulate import create_results_figure
+
+matplotlib.use("Agg")
+
+
+@pytest.mark.unit
+def test_creates_result_figure_with_trace_controls() -> None:
+    import matplotlib.pyplot as plt
+
+    result = SimulationResult(
+        model_name="Example",
+        data={
+            "time": [0.0, 1.0],
+            "x": [1.0, 2.0],
+            "y": [3.0, 4.0],
+        },
+    )
+
+    figure = create_results_figure(result)
+
+    assert {line.get_label() for line in figure.axes[0].lines} == {"x", "y"}
+    assert len(figure._bedit_widgets) == 3  # type: ignore[attr-defined]
+    plt.close(figure)