joppe/BondGraph
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Start of a sim window

Browse Source
This commit is contained in:
Joppe Blondel
2026-07-21 13:54:57 +02:00
parent 6fb2478589
commit cdfc891980
10 changed files with 646 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

215
BEdit/src/bedit/core/simulation/openmodelica.py
View File

@@ -1,12 +1,16 @@
import json
import os
import shlex
import shutil
import socket
import tempfile
import time
import xml.etree.ElementTree as ET
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path
from queue import Queue
from threading import Lock, Thread, current_thread
from threading import Event, Lock, Thread, current_thread
from typing import Any
from bedit.core.application_log import get_logger
@@ -19,17 +23,32 @@ ErrorCallback = Callable[[Exception], None]
@dataclass(frozen=True)
class _Request:
expression: str
parsed: bool
description: str
operation: Callable[[Any, Path], Any]
callback: ResultCallback | None
error_callback: ErrorCallback | None
@dataclass(frozen=True)
class SimulationProgress:
phase: str
current_step_size: float
time: float
progress: int
@dataclass(frozen=True)
class SimulationMessage:
stream: str
type: str
text: str
class OpenModelicaInterface:
"""Asynchronous, persistent interface to one OpenModelica session.
The worker and OMC session are created lazily for the first request. Callback
functions execute on the worker thread and must not manipulate Qt widgets.
The worker and OMC session are created lazily for the first request. Callbacks
execute on background threads and must not manipulate Qt widgets directly.
"""
def __init__(self, executable_path: str = "") -> None:
@@ -59,32 +78,45 @@ class OpenModelicaInterface:
"""Request the OpenModelica version without blocking the caller."""
self.send_expression("getVersion()", callback, error_callback)
def load_model(
def build_model(
self,
model: str,
callback: ResultCallback | None = None,
error_callback: ErrorCallback | None = None,
) -> None:
"""Load a composed model into OpenModelica"""
self.send_expression(f"loadString({json.dumps(model)})", callback, error_callback)
def compile_model(
self,
model_name: str,
callback: ResultCallback | None = None,
error_callback: ErrorCallback | None = None,
) -> None:
"""Compile a loaded model in OpenModelica"""
self.send_expression(f"buildModel({model_name})", callback, error_callback)
def exec_system(
"""Load and build one composed model as an ordered worker operation."""
def operation(omc, _temp_dir: Path):
loaded = omc.sendExpression(f"loadString({json.dumps(model)})")
if loaded is not True:
raise RuntimeError("OpenModelica could not load the composed model")
return omc.sendExpression(f"buildModel({model_name})")
self._submit("build model", operation, callback, error_callback)
def run_model(
self,
cmd: str,
executable: str,
arguments: list[str],
progress_callback: Callable[[SimulationProgress], None] | None = None,
message_callback: Callable[[SimulationMessage], None] | None = None,
callback: ResultCallback | None = None,
error_callback: ErrorCallback | None = None,
) -> None:
"""Execute system command from OpenModelica"""
self.send_expression(f"system(\"{cmd}\", \"system_out.txt\")", callback, error_callback)
"""Run a built model and consume its XML/TCP status stream."""
def operation(omc, temp_dir: Path):
return _run_model_with_tcp(
omc,
executable,
arguments,
temp_dir,
progress_callback,
message_callback,
)
self._submit("run simulation", operation, callback, error_callback)
def send_expression(
self,
@@ -96,7 +128,19 @@ class OpenModelicaInterface:
) -> None:
"""Queue an OMC expression for ordered execution on the worker thread."""
request = _Request(expression, parsed, callback, error_callback)
def operation(omc, _temp_dir: Path):
return omc.sendExpression(expression, parsed=parsed)
self._submit(expression, operation, callback, error_callback)
def _submit(
self,
description: str,
operation: Callable[[Any, Path], Any],
callback: ResultCallback | None,
error_callback: ErrorCallback | None,
) -> None:
request = _Request(description, operation, callback, error_callback)
self._ensure_worker().put(request)
def shutdown(self, *, wait: bool = True) -> None:
@@ -158,13 +202,11 @@ class OpenModelicaInterface:
raise RuntimeError(
f"OpenModelica could not use {str(temp_dir)!r}"
)
result = omc.sendExpression(
request.expression, parsed=request.parsed
)
result = request.operation(omc, temp_dir)
except Exception as error:
if request.error_callback is None:
log.exception(
"OpenModelica request failed: %s", request.expression
"OpenModelica request failed: %s", request.description
)
else:
_deliver_callback(request.error_callback, error)
@@ -214,6 +256,127 @@ def _create_session(executable_path: str):
return OMCSessionZMQ(omhome=_openmodelica_home(executable_path))
def _run_model_with_tcp(
omc,
executable: str,
arguments: list[str],
temp_dir: Path,
progress_callback: Callable[[SimulationProgress], None] | None,
message_callback: Callable[[SimulationMessage], None] | None,
) -> int:
executable_path = Path(executable)
executable_command = executable
if not executable_path.is_absolute() and executable_path.parent == Path("."):
executable_command = f"./{executable}"
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
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 = shlex.join([
executable_command,
*arguments,
f"-port={port}",
"-logFormat=xmltcp",
])
output_path = temp_dir / "simulation-output.txt"
command_finished = Event()
reader_finished = Event()
reader_errors: list[Exception] = []
def read_progress() -> None:
try:
connection = _accept_simulation_connection(server, command_finished)
with connection, connection.makefile(
"r", encoding="utf-8"
) as stream:
for line in stream:
_handle_simulation_xml(
line, progress_callback, message_callback
)
except Exception as error:
reader_errors.append(error)
finally:
reader_finished.set()
reader = Thread(
target=read_progress,
name="bedit-simulation-progress",
daemon=True,
)
reader.start()
log.info("Starting simulation through OpenModelica: %s", command)
try:
return_code = omc.sendExpression(
f"system({json.dumps(command)}, {json.dumps(str(output_path))})"
)
finally:
command_finished.set()
if not reader_finished.wait(20.0):
raise TimeoutError("Simulation progress connection did not close")
if reader_errors:
raise reader_errors[0]
if return_code != 0:
output = output_path.read_text(errors="replace") if output_path.exists() else ""
detail = f": {output.strip()}" if output.strip() else ""
raise RuntimeError(
f"Simulation process exited with status {return_code}{detail}"
)
return int(return_code)
def _accept_simulation_connection(
server: socket.socket, command_finished: Event
) -> socket.socket:
deadline = time.monotonic() + 15.0
while time.monotonic() < deadline:
try:
connection, _address = server.accept()
return connection
except socket.timeout:
if command_finished.is_set():
raise RuntimeError(
"Simulation command finished before opening its progress connection"
)
raise TimeoutError("Simulation did not connect to the progress server")
def _handle_simulation_xml(
line: str,
progress_callback: Callable[[SimulationProgress], None] | None,
message_callback: Callable[[SimulationMessage], None] | None,
) -> None:
text = line.strip()
if not text:
return
try:
element = ET.fromstring(text)
except ET.ParseError:
log.warning("Invalid simulation status XML: %s", text)
return
if element.tag == "status" and progress_callback is not None:
_deliver_callback(
progress_callback,
SimulationProgress(
phase=element.get("phase", ""),
current_step_size=float(element.get("currentStepSize", 0)),
time=float(element.get("time", 0)),
progress=int(float(element.get("progress", 0))),
),
)
elif element.tag == "message":
message = SimulationMessage(
stream=element.get("stream", ""),
type=element.get("type", ""),
text=element.get("text", ""),
)
log.info("OpenModelica %s: %s", message.stream, message.text)
if message_callback is not None:
_deliver_callback(message_callback, message)
def _ompython_transport_files(omc) -> set[Path]:
"""Return only the log and port files owned by this OMPython session."""

129
BEdit/src/bedit/core/simulation/service.py
View File

@@ -1,3 +1,4 @@
from collections.abc import Callable
from typing import Any
from bedit.core.application_log import get_logger
@@ -6,6 +7,8 @@ from bedit.core.simulation.openmodelica import (
ErrorCallback,
OpenModelicaInterface,
ResultCallback,
SimulationMessage,
SimulationProgress,
)
@@ -23,7 +26,8 @@ class Simulation:
self.model_name: str | None = None
self._openmodelica_path = openmodelica_path
self.openmodelica = OpenModelicaInterface(openmodelica_path)
self.model_path : str | None = None
self.model_path: str | None = None
self.simulation_progress: SimulationProgress | None = None
@property
def openmodelica_path(self) -> str:
@@ -34,7 +38,12 @@ class Simulation:
self._openmodelica_path = value
self.openmodelica.configure(value)
def compose(self, graph: dict[str, Any]) -> None:
def compose(
self,
graph: dict[str, Any],
callback: Callable[[str], None] | None = None,
error_callback: ErrorCallback | None = None,
) -> None:
"""Compose and retain the active graph's Modelica representation."""
self.model_path = None
@@ -45,58 +54,98 @@ class Simulation:
self.id_list = result.objects_by_id
self.last_composition_output = result.modelica
self.model_name = result.model_name
log.info("Composed OpenModelica model:\n%s", result.modelica)
# Compile openmodelica model
self.openmodelica.load_model(
self.last_composition_output,
lambda result: log.info("Loading OK: %s", result),
lambda error: log.info("Loading ERROR: %s", repr(error)),
)
def _model_compiled(result):
log.info("Compiling OK: %s", result)
self.model_path = result[0]
try:
self.model_path = str(result[0])
except (IndexError, TypeError) as error:
failure = RuntimeError(
f"OpenModelica returned an invalid build result: {result!r}"
)
failure.__cause__ = error
if error_callback is not None:
error_callback(failure)
else:
log.error("%s", failure)
return
if callback is not None:
callback(self.model_path)
self.openmodelica.compile_model(
self.openmodelica.build_model(
self.last_composition_output,
self.model_name,
_model_compiled,
lambda error: log.info("Compiling ERROR: %s", repr(error)),
error_callback,
)
def run_simulation(self, graph: dict[str, Any]) -> None:
"""Reserved for a future simulation request."""
def run_simulation(
self,
graph: dict[str, Any],
progress_callback: Callable[[SimulationProgress], None] | None = None,
message_callback: Callable[[SimulationMessage], None] | None = None,
callback: ResultCallback | None = None,
error_callback: ErrorCallback | None = None,
) -> None:
"""Compose, build, and asynchronously run the current graph."""
if self.model_path is None:
self.compose(graph)
self.simulation_progress = None
self.openmodelica.get_version(lambda val: log.info(val))
log.info(f"Running model binary {self.model_path}")
def report_progress(progress: SimulationProgress) -> None:
self.simulation_progress = progress
if progress_callback is not None:
progress_callback(progress)
self.openmodelica.exec_system(
f'./{self.model_name} {self.build_simulation_arguments()}',
lambda result: log.info("Running OK: %s", result),
lambda error: log.info("Running ERROR: %s", repr(error)),
)
def run_model(model_path: str) -> None:
try:
arguments = self.build_simulation_arguments()
except Exception as error:
if error_callback is not None:
error_callback(error)
else:
log.exception("Could not prepare simulation arguments")
return
self.openmodelica.run_model(
model_path,
arguments,
report_progress,
message_callback,
callback,
error_callback,
)
self.compose(graph, run_model, error_callback)
def get_progress(self) -> SimulationProgress | None:
"""Return the most recently received simulation status."""
return self.simulation_progress
def shutdown(self, *, wait: bool = True) -> None:
"""Close OpenModelica and clean its generated working directory."""
self.openmodelica.shutdown(wait=wait)
self.model_path = None
def build_simulation_arguments(self) -> str:
opts = self.last_composition_input['implementation']['graph']['simulation']
args = "-port=40696 -outputFormat=csv -logFormat=xmltcp "
args += f"-startTime={opts['startTime']} -stopTime={opts['stopTime']} "
step_size = opts['intervalTime']
if opts['intervalMode'] == 'numberOfIntervals':
step_size = (opts['stopTime']-opts['startTime'])/opts['numberOfIntervals']
args += f"-stepSize={step_size}"
log.info("Running model with: %s", args)
return args
def build_simulation_arguments(self) -> list[str]:
opts = self.last_composition_input["implementation"]["graph"].get(
"simulation", {}
)
start_time = float(opts.get("startTime", 0.0))
stop_time = float(opts.get("stopTime", 1.0))
interval_mode = opts.get("intervalMode", "numberOfIntervals")
interval_time = float(opts.get("intervalTime", 0.002))
if interval_mode == "numberOfIntervals":
intervals = int(opts.get("numberOfIntervals", 500))
if intervals <= 0:
raise ValueError("Number of simulation intervals must be positive")
interval_time = (stop_time - start_time) / intervals
if interval_time <= 0:
raise ValueError("Simulation interval must be positive")
arguments = [
"-outputFormat=csv",
f"-startTime={start_time}",
f"-stopTime={stop_time}",
f"-stepSize={interval_time}",
]
log.info("Running model with: %s", arguments)
return arguments

16
BEdit/src/bedit/gui/controllers/document.py
View File

@@ -427,11 +427,23 @@ class DocumentController(QObject):
raise ValueError("Open a graph component before composing")
self.simulation.compose(component.to_dict())
def run_simulation(self) -> None:
def run_simulation(
self,
progress_callback=None,
message_callback=None,
callback=None,
error_callback=None,
) -> None:
component = self.active_component
if component is None or component.implementation_kind != "graph":
raise ValueError("Open a graph component before running a simulation")
self.simulation.run_simulation(component.to_dict())
self.simulation.run_simulation(
component.to_dict(),
progress_callback,
message_callback,
callback,
error_callback,
)
def set_route_waypoints(self, item_kind: str, item_id: str, waypoints: list[QPointF]) -> None:
item = (

94
BEdit/src/bedit/gui/generated/ui_simulation_window.py Normal file
View File

@@ -0,0 +1,94 @@
# -*- coding: utf-8 -*-
################################################################################
## Form generated from reading UI file 'simulation_window.ui'
##
## Created by: Qt User Interface Compiler version 6.11.1
##
## WARNING! All changes made in this file will be lost when recompiling UI file!
################################################################################
from PySide6.QtCore import (QCoreApplication, QDate, QDateTime, QLocale,
QMetaObject, QObject, QPoint, QRect,
QSize, QTime, QUrl, Qt)
from PySide6.QtGui import (QBrush, QColor, QConicalGradient, QCursor,
QFont, QFontDatabase, QGradient, QIcon,
QImage, QKeySequence, QLinearGradient, QPainter,
QPalette, QPixmap, QRadialGradient, QTransform)
from PySide6.QtWidgets import (QApplication, QLabel, QListWidget, QListWidgetItem,
QMainWindow, QProgressBar, QSizePolicy, QTabWidget,
QVBoxLayout, QWidget)
class Ui_SimulationWindow(object):
def setupUi(self, SimulationWindow):
if not SimulationWindow.objectName():
SimulationWindow.setObjectName(u"SimulationWindow")
SimulationWindow.resize(720, 480)
self.centralWidget = QWidget(SimulationWindow)
self.centralWidget.setObjectName(u"centralWidget")
self.windowLayout = QVBoxLayout(self.centralWidget)
self.windowLayout.setObjectName(u"windowLayout")
self.statusLabel = QLabel(self.centralWidget)
self.statusLabel.setObjectName(u"statusLabel")
self.windowLayout.addWidget(self.statusLabel)
self.progressBar = QProgressBar(self.centralWidget)
self.progressBar.setObjectName(u"progressBar")
self.progressBar.setMaximum(10000)
self.progressBar.setValue(0)
self.windowLayout.addWidget(self.progressBar)
self.timeLabel = QLabel(self.centralWidget)
self.timeLabel.setObjectName(u"timeLabel")
self.windowLayout.addWidget(self.timeLabel)
self.resultsTabs = QTabWidget(self.centralWidget)
self.resultsTabs.setObjectName(u"resultsTabs")
self.logTab = QWidget()
self.logTab.setObjectName(u"logTab")
self.logLayout = QVBoxLayout(self.logTab)
self.logLayout.setObjectName(u"logLayout")
self.messageList = QListWidget(self.logTab)
self.messageList.setObjectName(u"messageList")
self.messageList.setAlternatingRowColors(True)
self.logLayout.addWidget(self.messageList)
self.resultsTabs.addTab(self.logTab, "")
self.resultsTab = QWidget()
self.resultsTab.setObjectName(u"resultsTab")
self.resultsLayout = QVBoxLayout(self.resultsTab)
self.resultsLayout.setObjectName(u"resultsLayout")
self.resultsPlaceholder = QLabel(self.resultsTab)
self.resultsPlaceholder.setObjectName(u"resultsPlaceholder")
self.resultsPlaceholder.setAlignment(Qt.AlignmentFlag.AlignCenter)
self.resultsLayout.addWidget(self.resultsPlaceholder)
self.resultsTabs.addTab(self.resultsTab, "")
self.windowLayout.addWidget(self.resultsTabs)
SimulationWindow.setCentralWidget(self.centralWidget)
self.retranslateUi(SimulationWindow)
self.resultsTabs.setCurrentIndex(0)
QMetaObject.connectSlotsByName(SimulationWindow)
# setupUi
def retranslateUi(self, SimulationWindow):
SimulationWindow.setWindowTitle(QCoreApplication.translate("SimulationWindow", u"Simulation", None))
self.statusLabel.setText(QCoreApplication.translate("SimulationWindow", u"No simulation has been run yet.", None))
self.progressBar.setFormat(QCoreApplication.translate("SimulationWindow", u"%p%", None))
self.timeLabel.setText(QCoreApplication.translate("SimulationWindow", u"Time: 0 s", None))
self.resultsTabs.setTabText(self.resultsTabs.indexOf(self.logTab), QCoreApplication.translate("SimulationWindow", u"Simulation Log", None))
self.resultsPlaceholder.setText(QCoreApplication.translate("SimulationWindow", u"Simulation graphs and result controls can be added here.", None))
self.resultsTabs.setTabText(self.resultsTabs.indexOf(self.resultsTab), QCoreApplication.translate("SimulationWindow", u"Results", None))
# retranslateUi

11
BEdit/src/bedit/gui/main_window.py
View File

@@ -32,6 +32,7 @@ from bedit.gui.dialogs.settings import SettingsDialog
from bedit.gui.dialogs.port_options import PortOptionsDialog
from bedit.gui.dialogs.parameter_options import ParameterOptionsDialog
from bedit.gui.dialogs.simulation_settings import SimulationSettingsDialog
from bedit.gui.simulation_window import SimulationWindow
from bedit.gui.preferences import application_settings
from bedit.gui.simulation_reload import reload_simulation
from bedit.gui.generated.ui_main_window import Ui_MainWindow
@@ -54,6 +55,7 @@ class MainWindow(QMainWindow):
self.log.info("BEdit started")
self.settings = application_settings()
self._applying_text_definition = False
self._simulation_window = SimulationWindow(self)
self.libraries = LibraryRepository(self)
self.simulation = Simulation(
@@ -248,11 +250,16 @@ class MainWindow(QMainWindow):
@Slot()
def run_simulation(self) -> None:
window = self._simulation_window
callbacks = window.begin_run()
window.show()
window.raise_()
window.activateWindow()
try:
self.document_controller.run_simulation()
self.document_controller.run_simulation(*callbacks)
except Exception as error:
self.log.exception("Simulation run failed")
QMessageBox.warning(self, "Cannot run simulation", str(error))
window.report_start_error(error)
def _restore_window_geometry(self) -> None:
geometry = self.settings.value("window/geometry")

102
BEdit/src/bedit/gui/simulation_window.py Normal file
View File

@@ -0,0 +1,102 @@
from PySide6.QtCore import Signal
from PySide6.QtWidgets import QMainWindow
from bedit.core.simulation.openmodelica import SimulationMessage, SimulationProgress
from bedit.gui.generated.ui_simulation_window import Ui_SimulationWindow
class SimulationWindow(QMainWindow):
"""Persistent, reusable view of simulation progress and results."""
progressReceived = Signal(object)
messageReceived = Signal(object)
simulationFinished = Signal(object)
simulationFailed = Signal(str)
def __init__(self, parent=None) -> None:
super().__init__(parent)
self.ui = Ui_SimulationWindow()
self.ui.setupUi(self)
self._running = False
self._run_generation = 0
self.progressReceived.connect(self._show_progress)
self.messageReceived.connect(self._show_message)
self.simulationFinished.connect(self._show_finished)
self.simulationFailed.connect(self._show_error)
def begin_run(self) -> tuple:
"""Reset the view and return callbacks bound to this particular run."""
self._run_generation += 1
generation = self._run_generation
self._running = True
self.ui.statusLabel.setText("Preparing simulation…")
self.ui.progressBar.setValue(0)
self.ui.timeLabel.setText("Time: 0 s")
self.ui.messageList.clear()
self.clear_results()
return (
lambda progress: self._report_progress(generation, progress),
lambda message: self._report_message(generation, message),
lambda result: self._report_finished(generation, result),
lambda error: self._report_error(generation, error),
)
def clear_results(self) -> None:
"""Clear future plots and result models before a new run.
Add graph widgets and their clearing logic here when simulation result
loading and plotting are implemented. The Designer-owned `resultsLayout`
is the intended container for those widgets.
"""
def _report_progress(
self, generation: int, progress: SimulationProgress
) -> None:
if generation == self._run_generation:
self.progressReceived.emit(progress)
def _report_message(self, generation: int, message: SimulationMessage) -> None:
if generation == self._run_generation:
self.messageReceived.emit(message)
def _report_finished(self, generation: int, result) -> None:
if generation == self._run_generation:
self.simulationFinished.emit(result)
def _report_error(self, generation: int, error: Exception) -> None:
if generation == self._run_generation:
self.simulationFailed.emit(str(error))
def report_start_error(self, error: Exception) -> None:
"""Report an error raised before asynchronous callbacks were installed."""
self.simulationFailed.emit(str(error))
def _show_progress(self, progress: SimulationProgress) -> None:
self.ui.statusLabel.setText(progress.phase or "Running")
self.ui.timeLabel.setText(f"Time: {progress.time:g} s")
self.ui.progressBar.setValue(max(0, min(10000, progress.progress)))
def _show_message(self, message: SimulationMessage) -> None:
prefix = message.stream or message.type or "OpenModelica"
self.ui.messageList.addItem(f"{prefix}: {message.text}")
self.ui.messageList.scrollToBottom()
def _show_finished(self, _result) -> None:
self._running = False
self.ui.progressBar.setValue(10000)
self.ui.statusLabel.setText("Simulation finished")
self.load_results()
def _show_error(self, message: str) -> None:
self._running = False
self.ui.statusLabel.setText("Simulation failed")
self.ui.messageList.addItem(f"Error: {message}")
def load_results(self) -> None:
"""Populate future plots and result controls after a successful run."""
@property
def is_running(self) -> bool:
return self._running