From 442b9569a7ba4173c5525603e450bcfe3082ab6e Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 09:09:31 +0000 Subject: [PATCH] feat(bec_status_box): side-by-side QWidget and QML demo with simulated services Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_015TeMnCZ9FjLpwiDkPu6K9j --- .../services/bec_status_box/compare_demo.py | 382 ++++++++++++++++++ 1 file changed, 382 insertions(+) create mode 100644 bec_widgets/widgets/services/bec_status_box/compare_demo.py diff --git a/bec_widgets/widgets/services/bec_status_box/compare_demo.py b/bec_widgets/widgets/services/bec_status_box/compare_demo.py new file mode 100644 index 00000000..91523729 --- /dev/null +++ b/bec_widgets/widgets/services/bec_status_box/compare_demo.py @@ -0,0 +1,382 @@ +"""Side-by-side demo of the QWidget and QML status box views, fed with simulated BEC services. + +No BEC server or Redis is needed. Run with: + + python -m bec_widgets.widgets.services.bec_status_box.compare_demo + +Both views sit in a BEC dock area (pass ``--plain`` for a simple splitter instead) and receive the +same simulated updates once per second, like the real widget. The control panel on the left sets +the state of every core service, adds or removes clients, and starts scripted scenarios. +""" + +from __future__ import annotations + +import argparse +import random +import sys +import tempfile +import time +from unittest import mock + +from bec_lib.messages import ( + BECStatus, + ServiceInfo, + ServiceMetricMessage, + ServiceVersions, + StatusMessage, +) +from qtpy.QtCore import QObject, Qt, QTimer, Signal +from qtpy.QtWidgets import ( + QApplication, + QCheckBox, + QComboBox, + QFormLayout, + QGroupBox, + QHBoxLayout, + QLabel, + QMainWindow, + QPushButton, + QScrollArea, + QSpinBox, + QSplitter, + QVBoxLayout, + QWidget, +) + +from bec_widgets.utils.colors import apply_theme, get_theme_name +from bec_widgets.widgets.services.bec_status_box.bec_status_box import BECStatusBox + +STATES = ["Running", "Busy", "Idle", "Error", "Not connected"] +STATE_TO_BEC = { + "Running": BECStatus.RUNNING, + "Busy": BECStatus.BUSY, + "Idle": BECStatus.IDLE, + "Error": BECStatus.ERROR, +} +CURRENT_VERSION = "3.39.5" +OLD_VERSION = "3.38.1" + + +class SimulatedServices(QObject): + """Stands in for BECServiceStatusMixin and emits simulated service updates.""" + + services_update = Signal(dict, dict) + + def __init__(self, parent=None): + super().__init__(parent) + self.core_states = {name: "Running" for name in BECStatusBox.CORE_SERVICES} + self.clients = 2 + self.gui_clients = 1 + self.dap_server = True + self.old_client = False + self.live_metrics = True + self.start_times: dict[str, float] = {} + self._tick = 0 + self.timer = QTimer(self) + self.timer.timeout.connect(self.emit_update) + self.timer.start(1000) + + def _info(self, user: str, host: str, version: str) -> ServiceInfo: + versions = ServiceVersions( + bec_lib=version, + bec_server=version, + bec_ipython_client=version, + bec_widgets=CURRENT_VERSION, + ophyd_devices="1.51.0", + ) + return ServiceInfo(user=user, hostname=host, versions=versions) + + def _metrics(self, name: str, user: str, host: str, index: int) -> dict: + now = time.time() + started = self.start_times.setdefault(name, now - random.uniform(600, 3 * 86400)) + jitter = random.uniform(-2, 2) if self.live_metrics else 0 + return { + "process_name": "python", + "username": user, + "hostname": host, + "cpu_percent": max(0.0, 3.0 + index + jitter), + "cpu_times": 12.0, + "cmdline": ["python", name], + "num_threads": 10 + index, + "pid": 4100 + index, + "memory_in_mb": 160.0 + 12 * index + jitter * 3, + "memory_used_percent": 1.0, + "create_time": started, + } + + def snapshot(self) -> tuple[dict, dict]: + """Build the services_info and services_metric dicts BEC would report now.""" + infos, metrics = {}, {} + services = [(name, "x12sa", "x12sa-bec-01") for name in BECStatusBox.CORE_SERVICES] + if self.dap_server: + services.append(("DAPServer/9f3c2a1b", "x12sa", "x12sa-bec-01")) + services += [ + (f"CLIBECClient/{i:02x}7a9e5c{i:04x}", "e12345", "x12sa-console-2") + for i in range(self.clients) + ] + services += [ + (f"BECGuiClient/{i:02x}c0ffee{i:04x}", "e12345", "x12sa-console-2") + for i in range(self.gui_clients) + ] + for index, (name, user, host) in enumerate(services): + state = self.core_states.get(name, "Running") + if state == "Not connected": + continue + version = ( + OLD_VERSION + if self.old_client and name.startswith("CLIBECClient/00") + else (CURRENT_VERSION) + ) + infos[name] = StatusMessage( + name=name, status=STATE_TO_BEC[state], info=self._info(user, host, version) + ) + metrics[name] = ServiceMetricMessage( + name=name, metrics=self._metrics(name, user, host, index) + ) + return infos, metrics + + def emit_update(self) -> None: + """Emit one update, like the real mixin does every second.""" + self._tick += 1 + self.services_update.emit(*self.snapshot()) + + def cleanup(self): + """Same interface as BECServiceStatusMixin.""" + + +class TimedStatusBox(BECStatusBox): + """BECStatusBox that records how long each update takes.""" + + timing = Signal(float) + + def update_service_status(self, services_info, services_metric): + start = time.perf_counter() + super().update_service_status(services_info, services_metric) + self.timing.emit((time.perf_counter() - start) * 1000) + + +class ControlPanel(QWidget): + """Controls for every state and interaction of the simulated services.""" + + def __init__(self, sim: SimulatedServices, boxes: dict[str, BECStatusBox], parent=None): + super().__init__(parent) + self.sim = sim + self.boxes = boxes + self.combos: dict[str, QComboBox] = {} + layout = QVBoxLayout(self) + + core = QGroupBox("Core services") + form = QFormLayout(core) + for name in BECStatusBox.CORE_SERVICES: + combo = QComboBox() + combo.addItems(STATES) + combo.currentTextChanged.connect(lambda state, n=name: self._set_state(n, state)) + self.combos[name] = combo + form.addRow(name, combo) + layout.addWidget(core) + + others = QGroupBox("Clients and other services") + form = QFormLayout(others) + self.clients = QSpinBox() + self.clients.setRange(0, 60) + self.clients.setValue(sim.clients) + self.clients.valueChanged.connect(lambda v: self._set("clients", v)) + form.addRow("IPython clients", self.clients) + self.gui_clients = QSpinBox() + self.gui_clients.setRange(0, 10) + self.gui_clients.setValue(sim.gui_clients) + self.gui_clients.valueChanged.connect(lambda v: self._set("gui_clients", v)) + form.addRow("GUI clients", self.gui_clients) + self.dap = QCheckBox("Data processing server") + self.dap.setChecked(True) + self.dap.toggled.connect(lambda v: self._set("dap_server", v)) + form.addRow(self.dap) + self.old = QCheckBox("First client runs an old bec_lib") + self.old.toggled.connect(lambda v: self._set("old_client", v)) + form.addRow(self.old) + layout.addWidget(others) + + scenarios = QGroupBox("Scenarios") + grid = QVBoxLayout(scenarios) + for label, slot in ( + ("All healthy", self.all_healthy), + ("Scan running (Scan server busy)", self.scan_running), + ("One core service down", self.one_down), + ("BEC stopped (all core down)", self.all_down), + ("Services starting (all idle)", self.all_idle), + ("Random incident", self.random_incident), + ("Crowded beamline (40 clients)", self.crowded), + ): + button = QPushButton(label) + button.clicked.connect(slot) + grid.addWidget(button) + self.flapping = QCheckBox("Flapping: random changes every 2 s") + self.flap_timer = QTimer(self) + self.flap_timer.timeout.connect(self.random_incident) + self.flapping.toggled.connect( + lambda on: self.flap_timer.start(2000) if on else self.flap_timer.stop() + ) + grid.addWidget(self.flapping) + layout.addWidget(scenarios) + + view = QGroupBox("Look") + form = QFormLayout(view) + self.live = QCheckBox("Live CPU and memory numbers") + self.live.setChecked(True) + self.live.toggled.connect(lambda v: self._set("live_metrics", v)) + form.addRow(self.live) + self.paused = QCheckBox("Pause updates") + self.paused.toggled.connect(lambda on: sim.timer.stop() if on else sim.timer.start(1000)) + form.addRow(self.paused) + theme = QPushButton("Toggle dark / light theme") + theme.clicked.connect( + lambda: apply_theme("light" if get_theme_name() == "dark" else "dark") + ) + form.addRow(theme) + self.compact = QCheckBox("Compact mode (LED + popup button)") + self.compact.toggled.connect(self._set_compact) + form.addRow(self.compact) + layout.addWidget(view) + + timing = QGroupBox("Last update time") + form = QFormLayout(timing) + for kind, box in boxes.items(): + label = QLabel("–") + box.timing.connect(lambda ms, lab=label: lab.setText(f"{ms:.2f} ms")) + form.addRow(kind, label) + layout.addWidget(timing) + layout.addStretch() + + def _push(self): + self.sim.emit_update() + + def _set(self, attr: str, value): + setattr(self.sim, attr, value) + self._push() + + def _set_state(self, name: str, state: str): + self.sim.core_states[name] = state + self._push() + + def _apply_states(self, states: dict[str, str]): + for name, state in states.items(): + combo = self.combos[name] + combo.blockSignals(True) + combo.setCurrentText(state) + combo.blockSignals(False) + self.sim.core_states[name] = state + self._push() + + def _set_compact(self, compact: bool): + for box in self.boxes.values(): + box.compact_view = compact + + def all_healthy(self): + self._apply_states({name: "Running" for name in BECStatusBox.CORE_SERVICES}) + + def scan_running(self): + states = {name: "Running" for name in BECStatusBox.CORE_SERVICES} + states["ScanServer"] = "Busy" + self._apply_states(states) + + def one_down(self): + states = {name: "Running" for name in BECStatusBox.CORE_SERVICES} + states["ScanBundler"] = "Not connected" + self._apply_states(states) + + def all_down(self): + self._apply_states({name: "Not connected" for name in BECStatusBox.CORE_SERVICES}) + + def all_idle(self): + self._apply_states({name: "Idle" for name in BECStatusBox.CORE_SERVICES}) + + def random_incident(self): + states = {name: "Running" for name in BECStatusBox.CORE_SERVICES} + for name in random.sample(BECStatusBox.CORE_SERVICES, random.randint(0, 2)): + states[name] = random.choice(STATES[1:]) + self._apply_states(states) + + def crowded(self): + self.clients.setValue(40) + + +class CompareWindow(QMainWindow): + """Main window with the control panel and both views.""" + + def __init__(self, plain: bool = False): + super().__init__() + self.setWindowTitle("BEC status box: QWidget vs QML") + self.sim = SimulatedServices(self) + client = mock.MagicMock() + self.boxes = { + "QWidget": TimedStatusBox( + client=client, bec_service_status_mixin=self.sim, view="widgets" + ), + "QML": TimedStatusBox(client=client, bec_service_status_mixin=self.sim, view="qml"), + } + controls = ControlPanel(self.sim, self.boxes) + scroll = QScrollArea() + scroll.setWidgetResizable(True) + scroll.setWidget(controls) + scroll.setMinimumWidth(320) + + splitter = QSplitter(Qt.Orientation.Horizontal) + splitter.addWidget(scroll) + if plain: + for kind, box in self.boxes.items(): + column = QWidget() + col_layout = QVBoxLayout(column) + title = QLabel(kind) + title.setStyleSheet("font-weight: bold;") + col_layout.addWidget(title) + col_layout.addWidget(box, 1) + splitter.addWidget(column) + else: + # pylint: disable=import-outside-toplevel + from bec_widgets.widgets.containers.dock_area import profile_utils + from bec_widgets.widgets.containers.dock_area.dock_area import BECDockArea + + # without a running BEC there is no client config for the profile folder + profiles = tempfile.mkdtemp(prefix="status_box_demo_profiles_") + profile_utils._settings_profiles_root = lambda: profiles # pylint: disable=W0212 + + self.dock_area = BECDockArea( + mode="utils", + profile_namespace="status_box_compare_demo", + auto_save_upon_exit=False, + enable_profile_management=False, + ) + for box in self.boxes.values(): + self.dock_area.new(box, where="right") + for dock in self.dock_area.dock_list(): + for kind, box in self.boxes.items(): + if dock.widget() is box: + dock.setWindowTitle(f"{kind} view") + splitter.addWidget(self.dock_area) + splitter.setStretchFactor(1, 1) + self.setCentralWidget(splitter) + self.resize(1250, 780) + self.sim.emit_update() + + def closeEvent(self, event): + self.sim.timer.stop() + for box in self.boxes.values(): + box.cleanup() + super().closeEvent(event) + + +def main(): # pragma: no cover + """Start the demo.""" + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--plain", action="store_true", help="splitter instead of a dock area") + parser.add_argument("--theme", choices=["dark", "light"], default="dark") + args = parser.parse_args() + app = QApplication.instance() or QApplication(sys.argv) + apply_theme(args.theme) + window = CompareWindow(plain=args.plain) + window.show() + sys.exit(app.exec()) + + +if __name__ == "__main__": # pragma: no cover + main()