From 46b8dd826327d96280bc601d06c1daa716dae2fc Mon Sep 17 00:00:00 2001 From: Joppe Blondel Date: Fri, 24 Jul 2026 16:30:31 +0200 Subject: [PATCH] Simulate json or beb model from the command line --- pyproject.toml | 2 + src/bedit_util/__init__.py | 5 +- src/bedit_util/component_path.py | 41 +++++ src/bedit_util/graphviz.py | 44 +---- src/bedit_util/simulate.py | 248 +++++++++++++++++++++++++++++ tests/unit/test_simulation_util.py | 29 ++++ 6 files changed, 328 insertions(+), 41 deletions(-) create mode 100644 src/bedit_util/component_path.py create mode 100644 src/bedit_util/simulate.py create mode 100644 tests/unit/test_simulation_util.py diff --git a/pyproject.toml b/pyproject.toml index 720fb7a..94c8282 100644 --- a/pyproject.toml +++ b/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 = [ diff --git a/src/bedit_util/__init__.py b/src/bedit_util/__init__.py index 1faaf9c..927e3d5 100644 --- a/src/bedit_util/__init__.py +++ b/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", +] diff --git a/src/bedit_util/component_path.py b/src/bedit_util/component_path.py new file mode 100644 index 0000000..acd2d1a --- /dev/null +++ b/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") diff --git a/src/bedit_util/graphviz.py b/src/bedit_util/graphviz.py index 48680b6..01e126f 100644 --- a/src/bedit_util/graphviz.py +++ b/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], diff --git a/src/bedit_util/simulate.py b/src/bedit_util/simulate.py new file mode 100644 index 0000000..9e146d7 --- /dev/null +++ b/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()) diff --git a/tests/unit/test_simulation_util.py b/tests/unit/test_simulation_util.py new file mode 100644 index 0000000..3da1e93 --- /dev/null +++ b/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)