diff --git a/bec_widgets/applications/launch_window.py b/bec_widgets/applications/launch_window.py index cc8274fd..46b282dc 100644 --- a/bec_widgets/applications/launch_window.py +++ b/bec_widgets/applications/launch_window.py @@ -24,13 +24,11 @@ from bec_widgets.utils.bec_plugin_helper import get_all_plugin_widgets from bec_widgets.utils.container_utils import WidgetContainerUtils from bec_widgets.utils.error_popups import SafeSlot from bec_widgets.utils.name_utils import pascal_to_snake -from bec_widgets.utils.plugin_utils import get_plugin_auto_updates from bec_widgets.utils.round_frame import RoundedFrame from bec_widgets.utils.rpc_register import RPCRegister from bec_widgets.utils.screen_utils import apply_window_geometry, centered_geometry_for_app from bec_widgets.utils.toolbars.toolbar import ModularToolBar from bec_widgets.utils.ui_loader import UILoader -from bec_widgets.widgets.containers.auto_update.auto_updates import AutoUpdates from bec_widgets.widgets.containers.dock_area.dock_area import BECDockArea from bec_widgets.widgets.containers.dock_area.profile_utils import get_last_profile, list_profiles from bec_widgets.widgets.containers.main_window.main_window import BECMainWindow, BECMainWindowNoRPC @@ -237,20 +235,6 @@ class LaunchWindow(BECMainWindow): ) self._refresh_dock_area_profiles(preserve_selection=False) - self.available_auto_updates: dict[str, type[AutoUpdates]] = ( - self._update_available_auto_updates() - ) - self.register_tile( - name="auto_update", - icon_path=os.path.join(MODULE_PATH, "assets", "app_icons", "auto_update.png"), - top_label="Get automated", - main_label="BEC Auto Update Dock Area", - description="Dock area with auto update functionality for BEC widgets plotting.", - action_button=self._open_auto_update, - show_selector=True, - selector_items=list(self.available_auto_updates.keys()) + ["Default"], - ) - self.register_tile( name="custom_ui_file", icon_path=os.path.join(MODULE_PATH, "assets", "app_icons", "ui_loader_tile.png"), @@ -462,10 +446,6 @@ class LaunchWindow(BECMainWindow): return None return self._launch_custom_ui_file(ui_file) - if launch_script == "auto_update": - auto_update = kwargs.pop("auto_update", None) - return self._launch_auto_update(auto_update, geometry=geometry) - if launch_script == "widget": widget = kwargs.pop("widget", None) if widget is None: @@ -538,23 +518,6 @@ class LaunchWindow(BECMainWindow): logger.info(f"Launched custom UI: {filename}, type: {type(window).__name__}") return window - def _launch_auto_update( - self, auto_update: str, geometry: tuple[int, int, int, int] | None = None - ) -> AutoUpdates: - if auto_update in self.available_auto_updates: - auto_update_cls = self.available_auto_updates[auto_update] - window = auto_update_cls() - else: - - auto_update = "auto_updates" - window = AutoUpdates() - - window.resize(window.minimumSizeHint()) - window.setWindowTitle(f"BEC - {window.objectName()}") - apply_window_geometry(window, geometry) - window.show() - return window - def _launch_widget( self, widget: type[BECWidget], geometry: tuple[int, int, int, int] | None = None ) -> QWidget: @@ -583,18 +546,6 @@ class LaunchWindow(BECMainWindow): super().apply_theme(theme) - def _open_auto_update(self): - """ - Open the auto update window. - """ - if self.tiles["auto_update"].selector is None: - auto_update = None - else: - auto_update = self.tiles["auto_update"].selector.currentText() - if auto_update == "Default": - auto_update = None - return self.launch("auto_update", auto_update=auto_update) - def _open_dock_area(self): """ Open Advanced Dock Area using the selected profile. @@ -635,19 +586,6 @@ class LaunchWindow(BECMainWindow): ) self.launch("custom_ui_file", ui_file=ui_file) - @staticmethod - def _update_available_auto_updates() -> dict[str, type[AutoUpdates]]: - """ - Load all available auto updates from the plugin repository. - """ - try: - auto_updates = get_plugin_auto_updates() - logger.info(f"Available auto updates: {auto_updates.keys()}") - except Exception as exc: - logger.error(f"Failed to load auto updates: {exc}") - return {} - return auto_updates - def show_launcher(self): """ Show the launcher window. diff --git a/bec_widgets/assets/app_icons/auto_update.png b/bec_widgets/assets/app_icons/auto_update.png deleted file mode 100644 index aa66cebe..00000000 Binary files a/bec_widgets/assets/app_icons/auto_update.png and /dev/null differ diff --git a/bec_widgets/cli/client.py b/bec_widgets/cli/client.py index 56b0f853..e424fcad 100644 --- a/bec_widgets/cli/client.py +++ b/bec_widgets/cli/client.py @@ -92,44 +92,6 @@ class AdminView(RPCBase): """ -class AutoUpdates(RPCBase): - _IMPORT_MODULE = "bec_widgets.widgets.containers.auto_update.auto_updates" - - @property - @rpc_call - def enabled(self) -> "bool": - """ - Get the enabled status of the auto updates. - """ - - @enabled.setter - @rpc_call - def enabled(self) -> "bool": - """ - Get the enabled status of the auto updates. - """ - - @property - @rpc_call - def selected_device(self) -> "str | None": - """ - Get the selected device from the auto update config. - - Returns: - str: The selected device. If no device is selected, None is returned. - """ - - @selected_device.setter - @rpc_call - def selected_device(self) -> "str | None": - """ - Get the selected device from the auto update config. - - Returns: - str: The selected device. If no device is selected, None is returned. - """ - - class AvailableDeviceResources(RPCBase): _IMPORT_MODULE = "bec_widgets.widgets.control.device_manager.components.available_device_resources.available_device_resources" diff --git a/bec_widgets/utils/plugin_utils.py b/bec_widgets/utils/plugin_utils.py index 0e4e082f..0041c138 100644 --- a/bec_widgets/utils/plugin_utils.py +++ b/bec_widgets/utils/plugin_utils.py @@ -2,7 +2,6 @@ from __future__ import annotations import ast import importlib -import inspect import os from dataclasses import dataclass from functools import lru_cache @@ -14,43 +13,6 @@ if TYPE_CHECKING: # pragma: no cover from bec_widgets.utils.bec_connector import BECConnector from bec_widgets.utils.bec_widget import BECWidget - from bec_widgets.widgets.containers.auto_update.auto_updates import AutoUpdates - - -def get_plugin_auto_updates() -> dict[str, type[AutoUpdates]]: - """ - Get all available auto update classes from the plugin directory. AutoUpdates must inherit from AutoUpdate and be - placed in the plugin repository's bec_widgets/auto_updates directory. The entry point for the auto updates is - specified in the respective pyproject.toml file using the following key: - [project.entry-points."bec.widgets.auto_updates"] - plugin_widgets_update = ".bec_widgets.auto_updates" - - e.g. - [project.entry-points."bec.widgets.auto_updates"] - plugin_widgets_update = "pxiii_bec.bec_widgets.auto_updates" - - Returns: - dict[str, AutoUpdates]: A dictionary of widget names and their respective classes. - """ - from bec_lib.plugin_helper import _get_available_plugins - - modules = _get_available_plugins("bec.widgets.auto_updates") - loaded_plugins = {} - for module in modules: - mods = inspect.getmembers(module, predicate=_filter_auto_updates) - for name, mod_cls in mods: - if name in loaded_plugins: - print(f"Duplicated auto update {name}.") - loaded_plugins[name] = mod_cls - return loaded_plugins - - -def _filter_auto_updates(obj): - from bec_widgets.widgets.containers.auto_update.auto_updates import AutoUpdates - - return ( - inspect.isclass(obj) and issubclass(obj, AutoUpdates) and not obj.__name__ == "AutoUpdates" - ) @dataclass diff --git a/bec_widgets/utils/rpc_server.py b/bec_widgets/utils/rpc_server.py index eff7798d..94b09820 100644 --- a/bec_widgets/utils/rpc_server.py +++ b/bec_widgets/utils/rpc_server.py @@ -515,14 +515,10 @@ class RPCServer: if wait and not self.rpc_register.object_is_registered(connector): raise RegistryNotReadyError(f"Connector {connector} not registered yet") - widget_class = getattr(connector, "rpc_widget_class", None) - if not widget_class: - widget_class = connector.__class__.__name__ - return { "gui_id": connector.gui_id, "object_name": connector.object_name or connector.__class__.__name__, - "widget_class": widget_class, + "widget_class": connector.__class__.__name__, "config": config_dict, "container_proxy": container_proxy, "__rpc__": getattr(connector, "rpc_exposed", True), diff --git a/bec_widgets/widgets/containers/auto_update/__init__.py b/bec_widgets/widgets/containers/auto_update/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/bec_widgets/widgets/containers/auto_update/auto_updates.py b/bec_widgets/widgets/containers/auto_update/auto_updates.py deleted file mode 100644 index 95424c16..00000000 --- a/bec_widgets/widgets/containers/auto_update/auto_updates.py +++ /dev/null @@ -1,379 +0,0 @@ -from __future__ import annotations - -from typing import TYPE_CHECKING, Literal, overload - -from bec_lib.endpoints import MessageEndpoints -from bec_lib.logger import bec_logger -from bec_lib.messages import ScanStatusMessage - -from bec_widgets.utils.error_popups import SafeSlot -from bec_widgets.widgets.containers.dock_area.dock_area import BECDockArea -from bec_widgets.widgets.containers.main_window.main_window import BECMainWindow -from bec_widgets.widgets.containers.qt_ads import CDockWidget - -if TYPE_CHECKING: # pragma: no cover - from bec_widgets.utils.bec_widget import BECWidget - from bec_widgets.widgets.plots.image.image import Image - from bec_widgets.widgets.plots.motor_map.motor_map import MotorMap - from bec_widgets.widgets.plots.multi_waveform.multi_waveform import MultiWaveform - from bec_widgets.widgets.plots.scatter_waveform.scatter_waveform import ScatterWaveform - from bec_widgets.widgets.plots.waveform.waveform import Waveform - - -logger = bec_logger.logger - - -class AutoUpdates(BECMainWindow): - _default_dock: CDockWidget | None - USER_ACCESS = ["enabled", "enabled.setter", "selected_device", "selected_device.setter"] - RPC = True - PLUGIN = False - - # enforce that subclasses have the same rpc widget class - rpc_widget_class = "AutoUpdates" - - def __init__( - self, parent=None, gui_id: str = None, window_title="Auto Update", *args, **kwargs - ): - super().__init__(parent=parent, gui_id=gui_id, window_title=window_title, **kwargs) - - self.dock_area = BECDockArea( - parent=self, - object_name="dock_area", - enable_profile_management=False, - startup_profile="skip", - ) - self.setCentralWidget(self.dock_area) - self._auto_update_selected_device: str | None = None - - self._default_dock = None # type: ignore - self.current_widget: BECWidget | None = None - self.dock_name = None - self._enabled = True - self.start_auto_update() - - def start_auto_update(self): - """ - Establish all connections for the auto updates. - """ - self.bec_dispatcher.connect_slot(self._on_scan_status, MessageEndpoints.scan_status()) - - def stop_auto_update(self): - """ - Disconnect all connections for the auto updates. - """ - self.bec_dispatcher.disconnect_slot( - self._on_scan_status, MessageEndpoints.scan_status() # type: ignore - ) - - @property - def selected_device(self) -> str | None: - """ - Get the selected device from the auto update config. - - Returns: - str: The selected device. If no device is selected, None is returned. - """ - return self._auto_update_selected_device - - @selected_device.setter - def selected_device(self, value: str | None) -> None: - """ - Set the selected device in the auto update config. - - Args: - value(str): The selected device. - """ - self._auto_update_selected_device = value - - @SafeSlot() - def _on_scan_status(self, content: dict, metadata: dict) -> None: - """ - Callback for scan status messages. - """ - msg = ScanStatusMessage(**content, metadata=metadata) - if not self.enabled: - return - - self.enable_gui_highlights(True) - - match msg.status: - case "open": - self.on_scan_open(msg) - case "closed": - self.on_scan_closed(msg) - case ["aborted", "halted"]: - self.on_scan_abort(msg) - case _: - pass - - def start_default_dock(self): - """ - Create a default dock for the auto updates. - """ - self.dock_area.delete_all() - self.dock_name = "update_dock" - self.current_widget = self.dock_area.new("Waveform") - docks = self.dock_area.dock_list() - self._default_dock = docks[0] if docks else None - - @overload - def set_dock_to_widget(self, widget: Literal["Waveform"]) -> Waveform: ... - - @overload - def set_dock_to_widget(self, widget: Literal["Image"]) -> Image: ... - - @overload - def set_dock_to_widget(self, widget: Literal["ScatterWaveform"]) -> ScatterWaveform: ... - - @overload - def set_dock_to_widget(self, widget: Literal["MotorMap"]) -> MotorMap: ... - - @overload - def set_dock_to_widget(self, widget: Literal["MultiWaveform"]) -> MultiWaveform: ... - - def set_dock_to_widget( - self, - widget: Literal["Waveform", "Image", "ScatterWaveform", "MotorMap", "MultiWaveForm"] | str, - ) -> BECWidget: - """ - Set the dock to the widget. - - Args: - widget (str): The widget to set the dock to. Must be the name of a valid widget class. - - Returns: - BECWidget: The widget that was set. - """ - if self.current_widget is None: - logger.warning( - f"Auto Updates: No default dock found. Creating a new one with name {self.dock_name}" - ) - self.start_default_dock() - assert self.current_widget is not None - - if self.current_widget.__class__.__name__ != widget: - self.dock_area.delete_all() - self.current_widget = self.dock_area.new(widget) - docks = self.dock_area.dock_list() - self._default_dock = docks[0] if docks else None - return self.current_widget - - def get_selected_device( - self, monitored_devices, selected_device: str | None = None - ) -> str | None: - """ - Get the selected device for the plot. If no device is selected, the first - device in the monitored devices list is selected. - """ - - if selected_device is None: - selected_device = self.selected_device - if selected_device: - return selected_device - if len(monitored_devices) > 0: - sel_device = monitored_devices[0] - return sel_device - return None - - def enable_gui_highlights(self, enable: bool) -> None: - """ - Enable or disable GUI highlights. - - Args: - enable (bool): Whether to enable or disable the highlights. - """ - if enable: - title = self.dock_area.window().windowTitle() - if " [Auto Updates]" in title: - return - self.dock_area.window().setWindowTitle(f"{title} [Auto Updates]") - else: - title = self.dock_area.window().windowTitle() - self.dock_area.window().setWindowTitle(title.replace(" [Auto Updates]", "")) - - @property - def enabled(self) -> bool: - """ - Get the enabled status of the auto updates. - """ - return self._enabled - - @enabled.setter - def enabled(self, value: bool) -> None: - """ - Set the enabled status of the auto updates. - """ - if self._enabled == value: - return - self._enabled = value - - if value: - self.start_auto_update() - self.enable_gui_highlights(True) - self.on_start() - else: - self.stop_auto_update() - self.enable_gui_highlights(False) - self.on_stop() - - def cleanup(self) -> None: - """ - Cleanup procedure to run when the auto updates are disabled. - """ - self.enabled = False - self.stop_auto_update() - self.dock_area.close() - self.dock_area.deleteLater() - self.dock_area = None - super().cleanup() - - ######################################################################## - ################# Update Functions ##################################### - ######################################################################## - - def simple_line_scan(self, info: ScanStatusMessage) -> None: - """ - Simple line scan. - - Args: - info (ScanStatusMessage): The scan status message. - """ - - # Set the dock to the waveform widget - wf = self.set_dock_to_widget("Waveform") - - # Get the scan report devices reported by the scan - dev_x = info.scan_report_devices[0] # type: ignore - - # For the y axis, get the selected device - dev_y = self.get_selected_device(info.readout_priority["monitored"]) # type: ignore - if not dev_y: - return - - # Clear the waveform widget and plot the data - # with the scan number and device names - # as the label and title - wf.clear_all() - wf.plot( - device_x=dev_x, - device_y=dev_y, - label=f"Scan {info.scan_number} - {dev_y}", - title=f"Scan {info.scan_number}", - x_label=dev_x, - y_label=dev_y, - ) - - logger.info( - f"Auto Update [simple_line_scan]: Started plot with: device_x={dev_x}, device_y={dev_y}" - ) - - def simple_grid_scan(self, info: ScanStatusMessage) -> None: - """ - Simple grid scan. - - Args: - info (ScanStatusMessage): The scan status message. - """ - # Set the dock to the scatter waveform widget - scatter = self.set_dock_to_widget("ScatterWaveform") - - # Get the scan report devices reported by the scan - dev_x, dev_y = info.scan_report_devices[0], info.scan_report_devices[1] # type: ignore - dev_z = self.get_selected_device(info.readout_priority["monitored"]) # type: ignore - - if None in (dev_x, dev_y, dev_z): - return - - # Clear the scatter waveform widget and plot the data - scatter.clear_all() - scatter.plot( - device_x=dev_x, - device_y=dev_y, - device_z=dev_z, - label=f"Scan {info.scan_number} - {dev_z}", - ) - - logger.info( - f"Auto Update [simple_grid_scan]: Started plot with: device_x={dev_x}, device_y={dev_y}, device_z={dev_z}" - ) - - def best_effort(self, info: ScanStatusMessage) -> None: - """ - Best effort scan. - - Args: - info (ScanStatusMessage): The scan status message. - """ - - # If the scan report devices are empty, there is nothing we can do - if not info.scan_report_devices: - return - dev_x = info.scan_report_devices[0] # type: ignore - dev_y = self.get_selected_device(info.readout_priority["monitored"]) # type: ignore - if not dev_y: - return - - # Set the dock to the waveform widget - wf = self.set_dock_to_widget("Waveform") - - # Clear the waveform widget and plot the data - wf.clear_all() - wf.plot( - device_x=dev_x, - device_y=dev_y, - label=f"Scan {info.scan_number} - {dev_y}", - title=f"Scan {info.scan_number}", - x_label=dev_x, - y_label=dev_y, - ) - - logger.info( - f"Auto Update [best_effort]: Started plot with: device_x={dev_x}, device_y={dev_y}" - ) - - ####################################################################### - ################# GUI Callbacks ####################################### - ####################################################################### - - def on_start(self) -> None: - """ - Procedure to run when the auto updates are enabled. - """ - self.start_default_dock() - - def on_stop(self) -> None: - """ - Procedure to run when the auto updates are disabled. - """ - - def on_scan_open(self, msg: ScanStatusMessage) -> None: - """ - Procedure to run when a scan starts. - - Args: - msg (ScanStatusMessage): The scan status message. - """ - if msg.scan_name == "line_scan" and msg.scan_report_devices: - return self.simple_line_scan(msg) - if msg.scan_name == "grid_scan" and msg.scan_report_devices: - return self.simple_grid_scan(msg) - if msg.scan_report_devices: - return self.best_effort(msg) - return None - - def on_scan_closed(self, msg: ScanStatusMessage) -> None: - """ - Procedure to run when a scan ends. - - Args: - msg (ScanStatusMessage): The scan status message. - """ - - def on_scan_abort(self, msg: ScanStatusMessage) -> None: - """ - Procedure to run when a scan is aborted. - - Args: - msg (ScanStatusMessage): The scan status message. - """ diff --git a/tests/unit_tests/test_launch_window.py b/tests/unit_tests/test_launch_window.py index 2836cb3d..d5571027 100644 --- a/tests/unit_tests/test_launch_window.py +++ b/tests/unit_tests/test_launch_window.py @@ -11,7 +11,6 @@ from qtpy.QtWidgets import QWidget import bec_widgets from bec_widgets.applications.launch_window import START_EMPTY_PROFILE_OPTION, LaunchWindow from bec_widgets.tests.client_mocks import mocked_client -from bec_widgets.widgets.containers.auto_update.auto_updates import AutoUpdates from bec_widgets.widgets.containers.main_window.main_window import BECMainWindow base_path = os.path.dirname(bec_widgets.__file__) @@ -73,40 +72,6 @@ def test_launch_window_launch_ui_file(bec_launch_window): res.deleteLater() -def test_launch_window_launch_default_auto_update(bec_launch_window): - # Mock the auto update selection - bec_launch_window.tiles["auto_update"].selector.setCurrentText("Default") - - # Call the method to launch the auto update - res = bec_launch_window._open_auto_update() - - assert isinstance(res, AutoUpdates) - assert res.windowTitle() == "BEC - AutoUpdates" - - # We need to manually close the launched window as it is not registered with qtbot. - # In real usage, the GUIServer would handle this in the sigint handler in case of a ctrl-c initiated shutdown. - res.close() - res.deleteLater() - - -def test_launch_window_launch_plugin_auto_update(bec_launch_window): - class PluginAutoUpdate(AutoUpdates): ... - - bec_launch_window.available_auto_updates = {"PluginAutoUpdate": PluginAutoUpdate} - bec_launch_window.tiles["auto_update"].selector.clear() - bec_launch_window.tiles["auto_update"].selector.addItems( - list(bec_launch_window.available_auto_updates.keys()) + ["Default"] - ) - bec_launch_window.tiles["auto_update"].selector.setCurrentText("PluginAutoUpdate") - res = bec_launch_window._open_auto_update() - assert isinstance(res, PluginAutoUpdate) - assert res.windowTitle() == "BEC - PluginAutoUpdate" - # We need to manually close the launched window as it is not registered with qtbot. - # In real usage, the GUIServer would handle this in the sigint handler in case of a ctrl-c initiated shutdown. - res.close() - res.deleteLater() - - def test_launch_window_dock_area_selector_has_start_empty_option(bec_launch_window): selector = bec_launch_window.tiles["dock_area"].selector assert selector is not None