mirror of
https://github.com/bec-project/bec_widgets.git
synced 2025-07-13 19:21:50 +02:00
refactor(auto_update): auto_update changed to be BECMainWindow; removed auto update logic from BECDockArea
This commit is contained in:
@ -1,13 +1,13 @@
|
||||
from bec_widgets.cli.auto_updates import AutoUpdates
|
||||
from bec_widgets.widgets.containers.auto_update.auto_updates import AutoUpdates
|
||||
from bec_widgets.widgets.containers.dock.dock_area import BECDockArea
|
||||
|
||||
|
||||
def dock_area(object_name: str | None = None):
|
||||
def dock_area(object_name: str | None = None) -> BECDockArea:
|
||||
_dock_area = BECDockArea(object_name=object_name)
|
||||
return _dock_area
|
||||
|
||||
|
||||
def auto_update_dock_area(object_name: str | None = None) -> BECDockArea:
|
||||
def auto_update_dock_area(object_name: str | None = None) -> AutoUpdates:
|
||||
"""
|
||||
Create a dock area with auto update enabled.
|
||||
|
||||
@ -17,7 +17,5 @@ def auto_update_dock_area(object_name: str | None = None) -> BECDockArea:
|
||||
Returns:
|
||||
BECDockArea: The created dock area.
|
||||
"""
|
||||
_dock_area = BECDockArea(object_name=object_name)
|
||||
_dock_area.set_auto_update(AutoUpdates)
|
||||
_dock_area.auto_update.enabled = True # type:ignore
|
||||
return _dock_area
|
||||
_auto_update = AutoUpdates(object_name=object_name)
|
||||
return _auto_update
|
||||
|
@ -72,6 +72,42 @@ for plugin_name, plugin_class in inspect.getmembers(plugin_client, inspect.iscla
|
||||
globals()[plugin_name] = plugin_class
|
||||
|
||||
|
||||
class AutoUpdates(RPCBase):
|
||||
@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 BECDock(RPCBase):
|
||||
@property
|
||||
@rpc_call
|
||||
@ -341,26 +377,6 @@ class BECDockArea(RPCBase):
|
||||
dict: The state of the dock area.
|
||||
"""
|
||||
|
||||
@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.
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def restore_state(
|
||||
self, state: "dict" = None, missing: "Literal['ignore', 'error']" = "ignore", extra="bottom"
|
||||
|
@ -7,11 +7,12 @@ 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.dock_area import BECDockArea
|
||||
from bec_widgets.widgets.containers.main_window.main_window import BECMainWindow
|
||||
|
||||
if TYPE_CHECKING: # pragma: no cover
|
||||
from bec_widgets.utils.bec_widget import BECWidget
|
||||
from bec_widgets.widgets.containers.dock.dock import BECDock
|
||||
from bec_widgets.widgets.containers.dock.dock_area import BECDockArea
|
||||
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
|
||||
@ -22,24 +23,33 @@ if TYPE_CHECKING: # pragma: no cover
|
||||
logger = bec_logger.logger
|
||||
|
||||
|
||||
class AutoUpdates:
|
||||
class AutoUpdates(BECMainWindow):
|
||||
_default_dock: BECDock
|
||||
USER_ACCESS = ["enabled", "enabled.setter", "selected_device", "selected_device.setter"]
|
||||
RPC = True
|
||||
|
||||
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")
|
||||
self.setCentralWidget(self.dock_area)
|
||||
self._auto_update_selected_device: str | None = None
|
||||
|
||||
def __init__(self, dock_area: BECDockArea):
|
||||
self.dock_area = dock_area
|
||||
self.bec_dispatcher = dock_area.bec_dispatcher
|
||||
self._default_dock = None # type:ignore
|
||||
self.current_widget: BECWidget | None = None
|
||||
self.dock_name = None
|
||||
self._enabled = False
|
||||
self._enabled = True
|
||||
self.start_auto_update()
|
||||
|
||||
def connect(self):
|
||||
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 disconnect(self):
|
||||
def stop_auto_update(self):
|
||||
"""
|
||||
Disconnect all connections for the auto updates.
|
||||
"""
|
||||
@ -47,6 +57,26 @@ class AutoUpdates:
|
||||
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:
|
||||
"""
|
||||
@ -125,7 +155,7 @@ class AutoUpdates:
|
||||
"""
|
||||
|
||||
if selected_device is None:
|
||||
selected_device = self.dock_area.selected_device
|
||||
selected_device = self.selected_device
|
||||
if selected_device:
|
||||
return selected_device
|
||||
if len(monitored_devices) > 0:
|
||||
@ -166,11 +196,11 @@ class AutoUpdates:
|
||||
self._enabled = value
|
||||
|
||||
if value:
|
||||
self.connect()
|
||||
self.start_auto_update()
|
||||
self.enable_gui_highlights(True)
|
||||
self.on_start()
|
||||
else:
|
||||
self.disconnect()
|
||||
self.stop_auto_update()
|
||||
self.enable_gui_highlights(False)
|
||||
self.on_stop()
|
||||
|
||||
@ -228,6 +258,9 @@ class AutoUpdates:
|
||||
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 dev_x or dev_y or dev_z is None:
|
||||
return
|
||||
|
||||
# Clear the scatter waveform widget and plot the data
|
||||
scatter.clear_all()
|
||||
scatter.plot(
|
||||
@ -315,3 +348,14 @@ class AutoUpdates:
|
||||
Args:
|
||||
msg (ScanStatusMessage): The scan status message.
|
||||
"""
|
||||
|
||||
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()
|
@ -1,6 +1,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING, Literal, Optional
|
||||
from typing import Literal, Optional
|
||||
from weakref import WeakValueDictionary
|
||||
|
||||
from bec_lib.logger import bec_logger
|
||||
@ -37,9 +37,6 @@ from bec_widgets.widgets.services.bec_status_box.bec_status_box import BECStatus
|
||||
from bec_widgets.widgets.utility.logpanel.logpanel import LogPanel
|
||||
from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import DarkModeButton
|
||||
|
||||
if TYPE_CHECKING: # pragma: no cover
|
||||
from bec_widgets.cli.auto_updates import AutoUpdates
|
||||
|
||||
logger = bec_logger.logger
|
||||
|
||||
|
||||
@ -67,8 +64,6 @@ class BECDockArea(BECWidget, QWidget):
|
||||
"detach_dock",
|
||||
"attach_all",
|
||||
"save_state",
|
||||
"selected_device",
|
||||
"selected_device.setter",
|
||||
"restore_state",
|
||||
]
|
||||
|
||||
@ -100,8 +95,6 @@ class BECDockArea(BECWidget, QWidget):
|
||||
self.layout.setSpacing(5)
|
||||
self.layout.setContentsMargins(0, 0, 0, 0)
|
||||
|
||||
self.auto_update: AutoUpdates | None = None
|
||||
self._auto_update_selected_device: str | None = None
|
||||
self._instructions_visible = True
|
||||
|
||||
self.dark_mode_button = DarkModeButton(parent=self, parent_id=self.gui_id, toolbar=True)
|
||||
@ -258,35 +251,6 @@ class BECDockArea(BECWidget, QWidget):
|
||||
"Add docks using 'new' method from CLI\n or \n Add widget docks using the toolbar",
|
||||
)
|
||||
|
||||
@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
|
||||
|
||||
def set_auto_update(self, auto_update_cls: type[AutoUpdates]) -> None:
|
||||
"""
|
||||
Set the auto update object for the dock area.
|
||||
|
||||
Args:
|
||||
auto_update(AutoUpdates): The auto update object.
|
||||
"""
|
||||
self.auto_update = auto_update_cls(self)
|
||||
|
||||
@property
|
||||
def panels(self) -> dict[str, BECDock]:
|
||||
"""
|
||||
@ -471,8 +435,6 @@ class BECDockArea(BECWidget, QWidget):
|
||||
"""
|
||||
Cleanup the dock area.
|
||||
"""
|
||||
if self.auto_update:
|
||||
self.auto_update.enabled = False
|
||||
self.delete_all()
|
||||
self.toolbar.close()
|
||||
self.toolbar.deleteLater()
|
||||
|
Reference in New Issue
Block a user