import re from pathlib import Path from matplotlib.backend_bases import MouseButton from matplotlib.backends.backend_qtagg import FigureCanvasQTAgg, NavigationToolbar2QT from matplotlib.figure import Figure from PySide6.QtCore import Qt, Signal from PySide6.QtGui import QCloseEvent from PySide6.QtWidgets import ( QFileDialog, QInputDialog, QMainWindow, QMenu, QMessageBox, QTreeWidgetItem, QVBoxLayout, QWidget, ) from bedit.core.simulation.openmodelica import SimulationMessage, SimulationProgress from bedit.core.simulation.results import ( SimulationExecutionResult, SimulationGraph, SimulationResults, SimulationTrace, load_simulation_results, save_simulation_results, ) from bedit.gui.generated.ui_simulation_window import Ui_SimulationWindow from bedit.gui.preferences import application_settings class SimulationWindow(QMainWindow): """Persistent viewer for live and previously saved simulation 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.settings = application_settings() self._running = False self._run_generation = 0 self._file_path: Path | None = None self._rebuilding_graph_tabs = False self._updating_signal_checks = False self.results = SimulationResults() self._connect_actions() self._populate_view_menu() self.splitDockWidget( self.ui.statusDock, self.ui.logDock, Qt.Orientation.Vertical ) self.ui.statusDock.setFixedHeight(self.ui.statusDock.sizeHint().height()) self._restore_window_layout() 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 _connect_actions(self) -> None: self.ui.actionOpen.triggered.connect(self.open_results) self.ui.actionSave.triggered.connect(self.save_results) self.ui.actionSaveAs.triggered.connect(self.save_results_as) self.ui.actionClear.triggered.connect(self.clear) self.ui.actionExit.triggered.connect(self.close) self.ui.actionAbout.triggered.connect(self.show_about) self.ui.actionAboutQt.triggered.connect( lambda: QMessageBox.aboutQt(self, "About Qt Framework") ) self.ui.actionToggleResults.toggled.connect(self.ui.centralWidget.setVisible) self.ui.actionAddGraph.triggered.connect(self.add_graph_tab) self.ui.actionRemoveGraph.triggered.connect(self.remove_current_graph_tab) self.ui.graphTabs.tabBarDoubleClicked.connect(self.rename_graph_tab) self.ui.graphTabs.tabBar().tabMoved.connect(self._move_graph_tab) self.ui.graphTabs.currentChanged.connect(self._current_graph_changed) self.ui.signalsTree.itemChanged.connect(self._signal_check_changed) self.ui.signalsTree.customContextMenuRequested.connect( self.show_signal_context_menu ) def _populate_view_menu(self) -> None: self.ui.menuPanels.addAction(self.ui.actionToggleResults) for panel in (self.ui.statusDock, self.ui.logDock, self.ui.signalsDock): self.ui.menuPanels.addAction(panel.toggleViewAction()) self.ui.menuToolbars.addAction(self.ui.fileToolbar.toggleViewAction()) self.ui.menuToolbars.addAction(self.ui.workspaceToolbar.toggleViewAction()) def _restore_window_layout(self) -> None: geometry = self.settings.value("simulationWindow/geometry") if geometry is not None: self.restoreGeometry(geometry) state = self.settings.value("simulationWindow/state") if state is not None: self.restoreState(state) results_visible = self.settings.value( "simulationWindow/resultsVisible", True, type=bool ) self.ui.actionToggleResults.setChecked(results_visible) def _save_window_layout(self) -> None: self.settings.setValue("simulationWindow/geometry", self.saveGeometry()) self.settings.setValue("simulationWindow/state", self.saveState()) self.settings.setValue( "simulationWindow/resultsVisible", self.ui.actionToggleResults.isChecked(), ) self.settings.sync() def begin_run(self) -> tuple: """Reset transient run state while retaining the current workspace.""" self._run_generation += 1 generation = self._run_generation self._running = True self._file_path = None self.results.status = {} self.results.messages.clear() self.results.metadata.clear() self.ui.progressBar.setValue(0) self.ui.timeLabel.setText("Time: 0 s") self.ui.messageList.clear() self.ui.statusLabel.setText("Preparing simulation…") 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 prepare_run_model(self, model_name: str | None) -> None: """Retain graph configuration only when rerunning the same model.""" if not model_name: return if self.results.model_name != model_name: self.results = SimulationResults(model_name=model_name) self.clear_result_views() self.load_result_views() else: self.results.model_name = model_name self.ui.statusLabel.setText("Preparing simulation…") self._update_title() def clear(self, checked: bool = False, *, model_name: str = "") -> None: """Discard the displayed run and prepare an empty results document.""" del checked self._run_generation += 1 self._running = False self._file_path = None self.results = SimulationResults(model_name=model_name) self.ui.statusLabel.setText("No simulation has been run yet.") self.ui.progressBar.setValue(0) self.ui.timeLabel.setText("Time: 0 s") self.ui.messageList.clear() self.clear_result_views() self.load_result_views() self._update_title() def clear_result_views(self) -> None: """Clear custom plots before a run or loaded document is displayed. Future graph widgets should be placed in the Designer-owned ``resultsLayout`` and reset here. """ self._rebuilding_graph_tabs = True self._clear_graph_tab_widgets() self._rebuilding_graph_tabs = False self.ui.signalsTree.clear() def load_result_views(self) -> None: """Populate custom plots from ``self.results.data`` and traces. This is the intended integration point for a future plotting widget. """ self._rebuild_graph_tabs() self._rebuild_signal_tree() def _rebuild_graph_tabs(self) -> None: current_page = self.ui.graphTabs.currentWidget() current_graph_id = ( current_page.graph_id if isinstance(current_page, GraphWorkspacePage) else None ) self._rebuilding_graph_tabs = True self._clear_graph_tab_widgets() for graph in self.results.graphs: self.ui.graphTabs.addTab( GraphWorkspacePage(graph, self.results), graph.title ) if current_graph_id is not None: for index, graph in enumerate(self.results.graphs): if graph.id == current_graph_id: self.ui.graphTabs.setCurrentIndex(index) break self._rebuilding_graph_tabs = False self._update_graph_actions() self._sync_signal_checks() def _clear_graph_tab_widgets(self) -> None: while self.ui.graphTabs.count(): page = self.ui.graphTabs.widget(0) self.ui.graphTabs.removeTab(0) page.deleteLater() def add_graph_tab(self) -> None: used_titles = {graph.title for graph in self.results.graphs} number = 1 while f"Graph {number}" in used_titles: number += 1 graph = SimulationGraph(title=f"Graph {number}") self.results.graphs.append(graph) index = self.ui.graphTabs.addTab( GraphWorkspacePage(graph, self.results), graph.title ) self.ui.graphTabs.setCurrentIndex(index) self._update_graph_actions() def remove_current_graph_tab(self) -> None: index = self.ui.graphTabs.currentIndex() if index < 0 or index >= len(self.results.graphs): return self.results.graphs.pop(index) page = self.ui.graphTabs.widget(index) self.ui.graphTabs.removeTab(index) page.deleteLater() self._update_graph_actions() self._sync_signal_checks() def rename_graph_tab(self, index: int) -> None: if index < 0 or index >= len(self.results.graphs): return graph = self.results.graphs[index] title, accepted = QInputDialog.getText( self, "Rename Graph", "Title:", text=graph.title ) title = title.strip() if accepted and title: graph.title = title self.ui.graphTabs.setTabText(index, title) page = self.ui.graphTabs.widget(index) if isinstance(page, GraphWorkspacePage): page.refresh_chart() def _move_graph_tab(self, old_index: int, new_index: int) -> None: if self._rebuilding_graph_tabs or old_index == new_index: return graph = self.results.graphs.pop(old_index) self.results.graphs.insert(new_index, graph) self._sync_signal_checks() def _update_graph_actions(self) -> None: self.ui.actionRemoveGraph.setEnabled(bool(self.results.graphs)) def _rebuild_signal_tree(self) -> None: """Build a hierarchy while retaining each leaf's complete signal name.""" tree = self.ui.signalsTree self._updating_signal_checks = True try: tree.clear() items: dict[tuple[str, ...], QTreeWidgetItem] = {} for signal_name in sorted(self.results.data, key=str.casefold): parts = _signal_tree_parts(signal_name) if not parts: continue parent = tree.invisibleRootItem() for depth, part in enumerate(parts, start=1): path = parts[:depth] item = items.get(path) if item is None: item = QTreeWidgetItem(parent, [part]) items[path] = item parent = item parent.setData(0, Qt.ItemDataRole.UserRole, signal_name) metadata = self.results.signal_metadata.get(signal_name, {}) unit = metadata.get("unit", "") if unit: parent.setText(0, f"{parts[-1]} [{unit}]") tooltip = signal_name details = [ metadata.get("quantity", ""), f"Unit: {unit}" if unit else "", metadata.get("description", ""), ] details = [detail for detail in details if detail] if details: tooltip += "\n" + "\n".join(details) parent.setToolTip(0, tooltip) parent.setFlags(parent.flags() | Qt.ItemFlag.ItemIsUserCheckable) parent.setCheckState(0, Qt.CheckState.Unchecked) tree.expandToDepth(0) finally: self._updating_signal_checks = False self._sync_signal_checks() def _current_graph_changed(self, _index: int) -> None: if not self._rebuilding_graph_tabs: self._sync_signal_checks() def _current_graph(self) -> SimulationGraph | None: index = self.ui.graphTabs.currentIndex() if 0 <= index < len(self.results.graphs): return self.results.graphs[index] return None def _sync_signal_checks(self) -> None: graph = self._current_graph() enabled = {trace.name for trace in graph.traces} if graph else set() self._updating_signal_checks = True try: root = self.ui.signalsTree.invisibleRootItem() pending = [root.child(index) for index in range(root.childCount())] while pending: item = pending.pop() pending.extend( item.child(index) for index in range(item.childCount()) ) signal_name = item.data(0, Qt.ItemDataRole.UserRole) if isinstance(signal_name, str): item.setCheckState( 0, Qt.CheckState.Checked if signal_name in enabled else Qt.CheckState.Unchecked, ) finally: self._updating_signal_checks = False self.ui.signalsTree.setEnabled(graph is not None) def _signal_check_changed(self, item: QTreeWidgetItem, _column: int) -> None: if self._updating_signal_checks: return signal_name = item.data(0, Qt.ItemDataRole.UserRole) graph = self._current_graph() if not isinstance(signal_name, str) or graph is None: return enabled = item.checkState(0) == Qt.CheckState.Checked existing = next( (trace for trace in graph.traces if trace.name == signal_name), None ) if enabled and existing is None: metadata = self.results.signal_metadata.get(signal_name, {}) graph.traces.append( SimulationTrace( name=signal_name, y_label=metadata.get("quantity", ""), unit=metadata.get("unit", ""), ) ) elif not enabled and existing is not None: graph.traces.remove(existing) page = self.ui.graphTabs.currentWidget() if isinstance(page, GraphWorkspacePage): page.refresh_chart() def selected_signal_names(self) -> list[str]: """Return full column names selected for future plotting actions.""" names = [] for item in self.ui.signalsTree.selectedItems(): name = item.data(0, Qt.ItemDataRole.UserRole) if isinstance(name, str): names.append(name) return names def show_signal_context_menu(self, position) -> None: item = self.ui.signalsTree.itemAt(position) signal_name = ( item.data(0, Qt.ItemDataRole.UserRole) if item is not None else None ) graph = self._current_graph() if not isinstance(signal_name, str) or graph is None: return menu = QMenu(self) use_as_x_action = menu.addAction("Use as X Axis") use_as_x_action.setCheckable(True) use_as_x_action.setChecked(graph.x_axis == signal_name) selected = menu.exec( self.ui.signalsTree.viewport().mapToGlobal(position) ) if selected is use_as_x_action: self.set_x_axis_signal(signal_name) def set_x_axis_signal(self, signal_name: str) -> None: """Set the current graph's persisted horizontal data column.""" graph = self._current_graph() if graph is None or signal_name not in self.results.data: return graph.x_axis = signal_name page = self.ui.graphTabs.currentWidget() if isinstance(page, GraphWorkspacePage): page.refresh_chart() def open_results(self) -> None: file_name, _selected_filter = QFileDialog.getOpenFileName( self, "Open Simulation Results", "", "BEdit Binary Simulation Results (*.ber);;JSON Simulation Results (*.json)", ) if not file_name: return try: results = load_simulation_results(file_name) except ValueError as error: QMessageBox.warning(self, "Cannot Open Simulation Results", str(error)) return self._run_generation += 1 self._running = False self._file_path = Path(file_name) self.results = results self._display_results() def save_results(self) -> None: file_path = self._file_path if file_path is None: self.save_results_as() return self._save_results_to(file_path) def save_results_as(self) -> None: default_name = f"{_safe_file_stem(self.results.model_name)}-results.ber" file_name, selected_filter = QFileDialog.getSaveFileName( self, "Save Simulation Results As", default_name, "BEdit Binary Simulation Results (*.ber);;JSON Simulation Results (*.json)", ) if not file_name: return file_path = Path(file_name) if file_path.suffix.lower() not in {".ber", ".json"}: suffix = ".json" if selected_filter.startswith("JSON") else ".ber" file_path = file_path.with_suffix(suffix) self._save_results_to(file_path) def _save_results_to(self, file_path: Path) -> None: try: save_simulation_results(file_path, self.results) except OSError as error: QMessageBox.warning(self, "Cannot Save Simulation Results", str(error)) return self._file_path = file_path self._update_title() def _display_results(self) -> None: status = self.results.status self.ui.statusLabel.setText(str(status.get("phase", "Loaded results"))) self.ui.progressBar.setValue(int(status.get("progress", 0))) self.ui.timeLabel.setText(f"Time: {float(status.get('time', 0)):g} s") self.ui.messageList.clear() for message in self.results.messages: prefix = message.get("stream") or message.get("type") or "OpenModelica" self.ui.messageList.addItem(f"{prefix}: {message.get('text', '')}") self.clear_result_views() self.load_result_views() self._update_title() def _update_title(self) -> None: name = self.results.model_name or "Simulation" self.setWindowTitle(f"{name} — Simulation") 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: self.simulationFailed.emit(str(error)) def _show_progress(self, progress: SimulationProgress) -> None: self.results.status = { "phase": progress.phase, "currentStepSize": progress.current_step_size, "time": progress.time, "progress": progress.progress, } 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: self.results.messages.append( {"stream": message.stream, "type": message.type, "text": message.text} ) 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.results.status.update(phase="Simulation finished", progress=10000) if isinstance(result, SimulationExecutionResult): self.results.data = result.data self.results.signal_metadata = result.signal_metadata available_signals = set(result.data) for graph in self.results.graphs: graph.traces = [ trace for trace in graph.traces if trace.name in available_signals ] for trace in graph.traces: metadata = result.signal_metadata.get(trace.name, {}) trace.unit = metadata.get("unit", "") trace.y_label = metadata.get("quantity", "") if graph.x_axis not in available_signals: graph.x_axis = ( "time" if "time" in available_signals else next(iter(result.data), "") ) self.results.metadata["processReturnCode"] = result.return_code self.results.metadata["sourceResultFile"] = Path(result.result_file).name else: self.results.metadata["processResult"] = result self.ui.progressBar.setValue(10000) self.ui.statusLabel.setText("Simulation finished") self.load_result_views() def _show_error(self, message: str) -> None: self._running = False self.results.status["phase"] = "Simulation failed" self.results.messages.append( {"stream": "BEdit", "type": "error", "text": message} ) self.ui.statusLabel.setText("Simulation failed") self.ui.messageList.addItem(f"Error: {message}") def show_about(self) -> None: QMessageBox.about( self, "About BEdit Simulation", "

BEdit Simulation

" "

View live progress and open or save simulation results.

", ) def closeEvent(self, event: QCloseEvent) -> None: # noqa: N802 (Qt API name) self._save_window_layout() super().closeEvent(event) @property def is_running(self) -> bool: return self._running class GraphWorkspacePage(QWidget): """Matplotlib view of one persisted simulation graph definition.""" def __init__( self, graph: SimulationGraph, results: SimulationResults, parent=None ) -> None: super().__init__(parent) self.graph_id = graph.id self.graph = graph self.results = results self.plot_layout = QVBoxLayout(self) self.figure = Figure(layout="constrained") self.canvas = InteractiveFigureCanvas(self.figure) self.axes = self.figure.add_subplot(111) self.canvas.axes = self.axes self.navigation_toolbar = GraphNavigationToolbar(self.canvas, self) self.canvas.navigation_toolbar = self.navigation_toolbar self.plot_layout.addWidget(self.navigation_toolbar) self.plot_layout.addWidget(self.canvas) self.refresh_chart() def refresh_chart(self) -> None: self.axes.clear() x_values = self.results.data.get(self.graph.x_axis) for trace in self.graph.traces: y_values = self.results.data.get(trace.name) if y_values is None: continue horizontal = x_values if x_values is not None else range(len(y_values)) sample_count = min(len(horizontal), len(y_values)) color = trace.properties.get("color") unit = trace.unit or self.results.signal_metadata.get( trace.name, {} ).get("unit", "") label = f"{trace.name} [{unit}]" if unit else trace.name self.axes.plot( list(horizontal)[:sample_count], y_values[:sample_count], label=label, color=color if isinstance(color, str) and color else None, ) self.axes.set_title(self.graph.title) if x_values is not None: x_unit = self.results.signal_metadata.get( self.graph.x_axis, {} ).get("unit", "") x_label = ( f"{self.graph.x_axis} [{x_unit}]" if x_unit else self.graph.x_axis ) else: x_label = "sample" self.axes.set_xlabel(x_label) units = { trace.unit or self.results.signal_metadata.get(trace.name, {}).get("unit", "") for trace in self.graph.traces if self.results.data.get(trace.name) is not None } units.discard("") if len(units) == 1: self.axes.set_ylabel(next(iter(units))) self.axes.grid(True, alpha=0.25) if self.axes.lines: self.axes.legend() self.axes.relim() self.axes.autoscale_view() self.canvas.set_home_view() self.canvas.draw_idle() class GraphNavigationToolbar(NavigationToolbar2QT): """Navigation toolbar whose Home action includes direct canvas navigation.""" def home(self, *args) -> None: del args self.canvas.reset_home_view() class InteractiveFigureCanvas(FigureCanvasQTAgg): """Matplotlib canvas with always-available wheel zoom and drag pan.""" def __init__(self, figure: Figure) -> None: super().__init__(figure) self.axes = None self.navigation_toolbar = None self._pan_start = None self._home_view = None self.mpl_connect("scroll_event", self._zoom_at_cursor) self.mpl_connect("button_press_event", self._start_pan) self.mpl_connect("motion_notify_event", self._pan) self.mpl_connect("button_release_event", self._finish_pan) def _toolbar_is_active(self) -> bool: return bool( self.navigation_toolbar is not None and self.navigation_toolbar.mode ) def set_home_view(self) -> None: if self.axes is not None: self._home_view = (self.axes.get_xlim(), self.axes.get_ylim()) def reset_home_view(self) -> None: if self.axes is None or self._home_view is None: return x_limits, y_limits = self._home_view self.axes.set_xlim(x_limits) self.axes.set_ylim(y_limits) self.draw_idle() def _zoom_at_cursor(self, event) -> None: if ( self.axes is None or event.inaxes is not self.axes or event.xdata is None or event.ydata is None or self._toolbar_is_active() ): return scale = 0.8 if event.button == "up" else 1.25 left, right = self.axes.get_xlim() bottom, top = self.axes.get_ylim() self.axes.set_xlim( event.xdata - (event.xdata - left) * scale, event.xdata + (right - event.xdata) * scale, ) self.axes.set_ylim( event.ydata - (event.ydata - bottom) * scale, event.ydata + (top - event.ydata) * scale, ) self.draw_idle() def _start_pan(self, event) -> None: if ( self.axes is None or event.inaxes is not self.axes or event.button != MouseButton.LEFT or self._toolbar_is_active() ): return self._pan_start = ( event.x, event.y, self.axes.get_xlim(), self.axes.get_ylim(), ) def _pan(self, event) -> None: if ( self.axes is None or self._pan_start is None or event.x is None or event.y is None ): return start_x, start_y, x_limits, y_limits = self._pan_start width = max(self.axes.bbox.width, 1.0) height = max(self.axes.bbox.height, 1.0) delta_x = (event.x - start_x) * (x_limits[1] - x_limits[0]) / width delta_y = (event.y - start_y) * (y_limits[1] - y_limits[0]) / height self.axes.set_xlim(x_limits[0] - delta_x, x_limits[1] - delta_x) self.axes.set_ylim(y_limits[0] - delta_y, y_limits[1] - delta_y) self.draw_idle() def _finish_pan(self, _event) -> None: self._pan_start = None def _safe_file_stem(model_name: str) -> str: stem = re.sub(r"[^A-Za-z0-9_.-]+", "_", model_name).strip("._") return stem or "simulation" def _signal_tree_parts(signal_name: str) -> tuple[str, ...]: """Split a result name into readable component, variable, and index levels. OpenModelica emits state derivatives as names such as ``der(block.x)``. Keep those traces available, but display them below a ``Derivatives`` group on their owning component instead of creating a misleading top-level ``der(block`` branch. """ derivative = re.fullmatch(r"der\((.+)\)", signal_name) if derivative is not None: inner_name = derivative.group(1) inner_parts = _signal_tree_parts(inner_name) if inner_parts: index_count = len(re.findall(r"\[[^\]]+\]", inner_name.rsplit(".", 1)[-1])) variable_index = max(0, len(inner_parts) - index_count - 1) return ( *inner_parts[:variable_index], "Derivatives", *inner_parts[variable_index:], ) return ("Derivatives", signal_name) parts: list[str] = [] for segment in signal_name.split("."): if not segment: continue match = re.fullmatch(r"([^\[]+)((?:\[[^\]]+\])+)", segment) if match is None: parts.append(segment) continue parts.append(match.group(1)) parts.extend(re.findall(r"\[([^\]]+)\]", match.group(2))) return tuple(parts)