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_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/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)