bec/bec_widgets
1
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-06-08 14:18:51 +02:00
Code Activity

feat: add broadcast and introduce _name for bec_connector, same as dock/widget names on client side

Browse Source
This commit is contained in:
appel_c
2025-03-04 15:52:10 +01:00
committed by wyzula-jan
parent b51bbe5ea9
commit a7020553f9
14 changed files with 602 additions and 369 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bec_widgets/cli/auto_updates.py
+3 -3
View File
@@ -35,9 +35,9 @@ class AutoUpdates:
Create a default dock for the auto updates.
"""
self.dock_name = "default_figure"
self._default_dock = self.gui.add_dock(self.dock_name)
self._default_dock.add_widget("BECFigure")
self._default_fig = self._default_dock.widget_list[0]
self._default_dock = self.gui.new(self.dock_name)
self._default_dock.new("BECFigure")
self._default_fig = self._default_dock.elements_list[0]
@staticmethod
def get_scan_info(msg) -> ScanInfo:
bec_widgets/cli/client.py
+128 -135
View File
@@ -230,14 +230,7 @@ class BECDock(RPCBase):
@property
@rpc_call
def _rpc_id(self) -> "str":
"""
Get the RPC ID of the widget.
"""
@property
@rpc_call
def widget_list(self) -> "list[BECWidget]":
def element_list(self) -> "list[BECWidget]":
"""
Get the widgets in the dock.
@@ -245,12 +238,66 @@ class BECDock(RPCBase):
widgets(list): The widgets in the dock.
"""
@property
@rpc_call
def elements(self) -> "dict[str, BECWidget]":
"""
Get the widgets in the dock.
Returns:
widgets(dict): The widgets in the dock.
"""
@rpc_call
def new(
self,
widget: "BECWidget | str",
name: "str | None" = None,
row: "int | None" = None,
col: "int" = 0,
rowspan: "int" = 1,
colspan: "int" = 1,
shift: "Literal['down', 'up', 'left', 'right']" = "down",
) -> "BECWidget":
"""
Add a widget to the dock.
Args:
widget(QWidget): The widget to add.
row(int): The row to add the widget to. If None, the widget will be added to the next available row.
col(int): The column to add the widget to.
rowspan(int): The number of rows the widget should span.
colspan(int): The number of columns the widget should span.
shift(Literal["down", "up", "left", "right"]): The direction to shift the widgets if the position is occupied.
"""
@rpc_call
def show(self):
"""
Show the dock.
"""
@rpc_call
def hide(self):
"""
Hide the dock.
"""
@rpc_call
def show_title_bar(self):
"""
Hide the title bar of the dock.
"""
@rpc_call
def set_title(self, title: "str"):
"""
Set the title of the dock.
Args:
title(str): The title of the dock.
"""
@rpc_call
def hide_title_bar(self):
"""
@@ -267,38 +314,7 @@ class BECDock(RPCBase):
"""
@rpc_call
def set_title(self, title: "str"):
"""
Set the title of the dock.
Args:
title(str): The title of the dock.
"""
@rpc_call
def add_widget(
self,
widget: "BECWidget | str",
row=None,
col=0,
rowspan=1,
colspan=1,
shift: "Literal['down', 'up', 'left', 'right']" = "down",
) -> "BECWidget":
"""
Add a widget to the dock.
Args:
widget(QWidget): The widget to add.
row(int): The row to add the widget to. If None, the widget will be added to the next available row.
col(int): The column to add the widget to.
rowspan(int): The number of rows the widget should span.
colspan(int): The number of columns the widget should span.
shift(Literal["down", "up", "left", "right"]): The direction to shift the widgets if the position is occupied.
"""
@rpc_call
def list_eligible_widgets(self) -> "list":
def available_widgets(self) -> "list":
"""
List all widgets that can be added to the dock.
@@ -318,18 +334,18 @@ class BECDock(RPCBase):
"""
@rpc_call
def remove_widget(self, widget_rpc_id: "str"):
def delete(self, widget_name: "str") -> "None":
"""
Remove a widget from the dock.
Args:
widget_rpc_id(str): The ID of the widget to remove.
widget_name(str): Delete the widget with the given name.
"""
@rpc_call
def remove(self):
def delete_all(self):
"""
Remove the dock from the parent dock area.
Remove all widgets from the dock.
"""
@rpc_call
@@ -346,21 +362,50 @@ class BECDock(RPCBase):
class BECDockArea(RPCBase):
@property
@rpc_call
def _config_dict(self) -> "dict":
def new(
self,
name: "str | None" = None,
widget: "str | QWidget | None" = None,
widget_name: "str | None" = None,
position: "Literal['bottom', 'top', 'left', 'right', 'above', 'below']" = "bottom",
relative_to: "BECDock | None" = None,
closable: "bool" = True,
floating: "bool" = False,
row: "int | None" = None,
col: "int" = 0,
rowspan: "int" = 1,
colspan: "int" = 1,
) -> "BECDock":
"""
Get the configuration of the widget.
Add a dock to the dock area. Dock has QGridLayout as layout manager by default.
Args:
name(str): The name of the dock to be displayed and for further references. Has to be unique.
widget(str|QWidget|None): The widget to be added to the dock. While using RPC, only BEC RPC widgets from RPCWidgetHandler are allowed.
position(Literal["bottom", "top", "left", "right", "above", "below"]): The position of the dock.
relative_to(BECDock): The dock to which the new dock should be added relative to.
closable(bool): Whether the dock is closable.
floating(bool): Whether the dock is detached after creating.
row(int): The row of the added widget.
col(int): The column of the added widget.
rowspan(int): The rowspan of the added widget.
colspan(int): The colspan of the added widget.
Returns:
dict: The configuration of the widget.
BECDock: The created dock.
"""
@property
@rpc_call
def selected_device(self) -> "str":
def show(self):
"""
None
Show all windows including floating docks.
"""
@rpc_call
def hide(self):
"""
Hide all windows including floating docks.
"""
@property
@@ -372,76 +417,29 @@ class BECDockArea(RPCBase):
dock_dict(dict): The docks in the dock area.
"""
@property
@rpc_call
def save_state(self) -> "dict":
def panel_list(self) -> "list[BECDock]":
"""
Save the state of the dock area.
Get the docks in the dock area.
Returns:
dict: The state of the dock area.
list: The docks in the dock area.
"""
@rpc_call
def remove_dock(self, name: "str"):
def delete(self, dock_name: "str"):
"""
Remove a dock by name and ensure it is properly closed and cleaned up.
Delete a dock by name.
Args:
name(str): The name of the dock to remove.
dock_name(str): The name of the dock to delete.
"""
@rpc_call
def restore_state(
self, state: "dict" = None, missing: "Literal['ignore', 'error']" = "ignore", extra="bottom"
):
def delete_all(self) -> "None":
"""
Restore the state of the dock area. If no state is provided, the last state is restored.
Args:
state(dict): The state to restore.
missing(Literal['ignore','error']): What to do if a dock is missing.
extra(str): Extra docks that are in the dockarea but that are not mentioned in state will be added to the bottom of the dockarea, unless otherwise specified by the extra argument.
"""
@rpc_call
def add_dock(
self,
name: "str" = None,
position: "Literal['bottom', 'top', 'left', 'right', 'above', 'below']" = None,
relative_to: "BECDock | None" = None,
closable: "bool" = True,
floating: "bool" = False,
prefix: "str" = "dock",
widget: "str | QWidget | None" = None,
row: "int" = None,
col: "int" = 0,
rowspan: "int" = 1,
colspan: "int" = 1,
) -> "BECDock":
"""
Add a dock to the dock area. Dock has QGridLayout as layout manager by default.
Args:
name(str): The name of the dock to be displayed and for further references. Has to be unique.
position(Literal["bottom", "top", "left", "right", "above", "below"]): The position of the dock.
relative_to(BECDock): The dock to which the new dock should be added relative to.
closable(bool): Whether the dock is closable.
floating(bool): Whether the dock is detached after creating.
prefix(str): The prefix for the dock name if no name is provided.
widget(str|QWidget|None): The widget to be added to the dock. While using RPC, only BEC RPC widgets from RPCWidgetHandler are allowed.
row(int): The row of the added widget.
col(int): The column of the added widget.
rowspan(int): The rowspan of the added widget.
colspan(int): The colspan of the added widget.
Returns:
BECDock: The created dock.
"""
@rpc_call
def clear_all(self):
"""
Close all docks and remove all temp areas.
Delete all docks.
"""
@rpc_call
@@ -462,40 +460,35 @@ class BECDockArea(RPCBase):
Return all floating docks to the dock area.
"""
@rpc_call
def _get_all_rpc(self) -> "dict":
"""
Get all registered RPC objects.
"""
@property
@rpc_call
def temp_areas(self) -> "list":
"""
Get the temporary areas in the dock area.
Returns:
list: The temporary areas in the dock area.
"""
@rpc_call
def show(self):
"""
Show all windows including floating docks.
"""
@rpc_call
def hide(self):
"""
Hide all windows including floating docks.
"""
@rpc_call
def delete(self):
def selected_device(self) -> "str":
"""
None
"""
@rpc_call
def save_state(self) -> "dict":
"""
Save the state of the dock area.
Returns:
dict: The state of the dock area.
"""
@rpc_call
def restore_state(
self, state: "dict" = None, missing: "Literal['ignore', 'error']" = "ignore", extra="bottom"
):
"""
Restore the state of the dock area. If no state is provided, the last state is restored.
Args:
state(dict): The state to restore.
missing(Literal['ignore','error']): What to do if a dock is missing.
extra(str): Extra docks that are in the dockarea but that are not mentioned in state will be added to the bottom of the dockarea, unless otherwise specified by the extra argument.
"""
class BECFigure(RPCBase):
@property
bec_widgets/cli/client_utils.py
+56 -28
View File
@@ -7,6 +7,7 @@ import os
import select
import subprocess
import threading
import time
from contextlib import contextmanager
from typing import TYPE_CHECKING
@@ -22,13 +23,12 @@ if TYPE_CHECKING:
from bec_lib import messages
from bec_lib.connector import MessageObject
from bec_lib.device import DeviceBase
from bec_widgets.utils.bec_dispatcher import BECDispatcher
from bec_lib.redis_connector import StreamMessage
else:
messages = lazy_import("bec_lib.messages")
# from bec_lib.connector import MessageObject
MessageObject = lazy_import_from("bec_lib.connector", ("MessageObject",))
BECDispatcher = lazy_import_from("bec_widgets.utils.bec_dispatcher", ("BECDispatcher",))
StreamMessage = lazy_import_from("bec_lib.redis_connector", ("StreamMessage",))
logger = bec_logger.logger
@@ -156,10 +156,11 @@ def wait_for_server(client):
### in the generated client module. So, here a class with the same name
### is created, and client module is patched.
class BECDockArea(client.BECDockArea):
# FIXME fix the delete method
def delete(self):
if self is BECGuiClient._top_level["bec"]:
raise RuntimeError("Cannot delete bec window")
super().delete()
super().delete_all()
try:
del BECGuiClient._top_level[self._gui_id]
except KeyError:
@@ -185,6 +186,7 @@ class BECGuiClient(RPCBase):
self._process = None
self._process_output_processing_thread = None
self._exposed_widgets = []
self._registry_state = {}
@property
def windows(self):
@@ -243,7 +245,7 @@ class BECGuiClient(RPCBase):
def _start_update_script(self) -> None:
self._client.connector.register(MessageEndpoints.scan_status(), cb=self._handle_msg_update)
def _handle_msg_update(self, msg: MessageObject) -> None:
def _handle_msg_update(self, msg: StreamMessage) -> None:
if self.auto_updates is not None:
# pylint: disable=protected-access
return self._update_script_msg_parser(msg.value)
@@ -256,19 +258,28 @@ class BECGuiClient(RPCBase):
return self.auto_updates.do_update(msg)
def _gui_post_startup(self):
widget = BECDockArea(gui_id=self._default_dock_name, parent=self)
self._add_widget_to_top_level(self._default_dock_name, widget)
if self._auto_updates_enabled:
if self._auto_updates is None:
auto_updates = self._get_update_script()
if auto_updates is None:
AutoUpdates.create_default_dock = True
AutoUpdates.enabled = True
auto_updates = AutoUpdates(self._top_level[self._default_dock_name])
if auto_updates.create_default_dock:
auto_updates.start_default_dock()
self._start_update_script()
self._auto_updates = auto_updates
timeout = 10
while time.time() < time.time() + timeout:
if len(list(self._registry_state.keys())) == 0:
time.sleep(0.1)
else:
break
key = list(self._registry_state.keys())[0]
gui_id = self._registry_state[key]["gui_id"]
name = self._registry_state[key]["name"]
widget = BECDockArea(gui_id=gui_id, name=name, parent=self)
self._add_widget_to_top_level(name, widget)
# if self._auto_updates_enabled:
# if self._auto_updates is None:
# auto_updates = self._get_update_script()
# if auto_updates is None:
# AutoUpdates.create_default_dock = True
# AutoUpdates.enabled = True
# auto_updates = AutoUpdates(self._top_level[name])
# if auto_updates.create_default_dock:
# auto_updates.start_default_dock()
# self._start_update_script()
# self._auto_updates = auto_updates
self._do_show_all()
self._gui_started_event.set()
@@ -308,12 +319,18 @@ class BECGuiClient(RPCBase):
return rpc_client._run_rpc("_dump")
def _start(self):
self._client.connector.register(
MessageEndpoints.gui_registry_state(self._gui_id), cb=self._handle_registry_update
)
return self._start_server()
def start(self):
# FIXME keeping backwards compatibility for now
return self._start()
def _handle_registry_update(self, msg: StreamMessage) -> None:
self._registry_state = msg["data"].state
def _do_show_all(self):
rpc_client = RPCBase(gui_id=f"{self._gui_id}:window", parent=self)
rpc_client._run_rpc("show")
@@ -340,23 +357,31 @@ class BECGuiClient(RPCBase):
def hide(self):
return self._hide_all()
def new(self, title: str = None, wait: bool = True) -> BECDockArea:
"""Create a new top-level dock area"""
def new(self, name: str | None = None, wait: bool = True) -> BECDockArea:
"""Create a new top-level dock area.
Args:
name(str, optional): The name of the dock area. Defaults to None.
wait(bool, optional): Whether to wait for the server to start. Defaults to True.
Returns:
BECDockArea: The new dock area.
"""
if wait:
with wait_for_server(self):
rpc_client = RPCBase(gui_id=f"{self._gui_id}:window", parent=self)
widget = rpc_client._run_rpc("new_dock_area", title)
self._add_widget_to_top_level(widget._gui_id, widget)
widget = rpc_client._run_rpc(
"new_dock_area", name
) # pylint: disable=protected-access
self._add_widget_to_top_level(widget._name, widget)
return widget
widget = rpc_client._run_rpc("new_dock_area", name) # pylint: disable=protected-access
rpc_client = RPCBase(gui_id=f"{self._gui_id}:window", parent=self)
widget = rpc_client._run_rpc("new_dock_area", title)
self._add_widget_to_top_level(widget._gui_id, widget)
self._add_widget_to_top_level(widget._name, widget)
return widget
def _add_widget_to_top_level(self, widget_id: str, widget: BECDockArea) -> None:
self._top_level[widget_id] = widget
setattr(self, widget_id, widget)
self._exposed_widgets.append(widget_id)
self._update_top_level_widgets()
def _update_top_level_widgets(self):
for widget_id in self._exposed_widgets:
@@ -367,8 +392,11 @@ class BECGuiClient(RPCBase):
setattr(self, widget_id, widget)
self._exposed_widgets.append(widget_id)
def close(self) -> None:
# FIXME: keeping backwards compatibility for now
def close(self):
# Needed to shut down gui for IPythonClient, will be remove in future
self.kill()
def kill(self) -> None:
self._close()
def _close(self) -> None:
bec_widgets/cli/rpc/rpc_base.py
+12 -4
View File
@@ -3,7 +3,7 @@ from __future__ import annotations
import threading
import uuid
from functools import wraps
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Any, cast
from bec_lib.client import BECClient
from bec_lib.endpoints import MessageEndpoints
@@ -61,10 +61,17 @@ class RPCResponseTimeoutError(Exception):
class RPCBase:
def __init__(self, gui_id: str = None, config: dict = None, parent=None) -> None:
def __init__(
self,
gui_id: str | None = None,
config: dict | None = None,
name: str | None = None,
parent=None,
) -> None:
self._client = BECClient() # BECClient is a singleton; here, we simply get the instance
self._config = config if config is not None else {}
self._gui_id = gui_id if gui_id is not None else str(uuid.uuid4())[:5]
self._name = name if name is not None else str(uuid.uuid4())[:5]
self._parent = parent
self._msg_wait_event = threading.Event()
self._rpc_response = None
@@ -88,7 +95,7 @@ class RPCBase:
parent = parent._parent
return parent
def _run_rpc(self, method, *args, wait_for_rpc_response=True, timeout=3, **kwargs):
def _run_rpc(self, method, *args, wait_for_rpc_response=True, timeout=3, **kwargs) -> Any:
"""
Run the RPC call.
@@ -162,7 +169,8 @@ class RPCBase:
cls = getattr(client, cls)
# print(msg_result)
return cls(parent=self, **msg_result)
ret = cls(parent=self, **msg_result)
return ret
return msg_result
def _gui_is_alive(self):
bec_widgets/cli/rpc/rpc_register.py
+53 -13
View File
@@ -2,11 +2,20 @@ from __future__ import annotations
from functools import wraps
from threading import Lock
from typing import Callable
from typing import TYPE_CHECKING, Callable
from weakref import WeakValueDictionary
from bec_lib.logger import bec_logger
from qtpy.QtCore import QObject
if TYPE_CHECKING:
from bec_widgets.utils.bec_connector import BECConnector
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
logger = bec_logger.logger
def broadcast_update(func):
"""
@@ -68,7 +77,7 @@ class RPCRegister:
raise ValueError(f"RPC object {rpc} must have a 'gui_id' attribute.")
self._rpc_register.pop(rpc.gui_id, None)
def get_rpc_by_id(self, gui_id: str) -> QObject:
def get_rpc_by_id(self, gui_id: str) -> QObject | None:
"""
Get an RPC object by its ID.
@@ -76,11 +85,25 @@ class RPCRegister:
gui_id(str): The ID of the RPC object to be retrieved.
Returns:
QObject: The RPC object with the given ID.
QObject | None: The RPC object with the given ID or None
"""
rpc_object = self._rpc_register.get(gui_id, None)
return rpc_object
def get_rpc_by_name(self, name: str) -> QObject | None:
"""
Get an RPC object by its name.
Args:
name(str): The name of the RPC object to be retrieved.
Returns:
QObject | None: The RPC object with the given name.
"""
rpc_object = [rpc for rpc in self._rpc_register if rpc._name == name]
rpc_object = rpc_object[0] if len(rpc_object) > 0 else None
return rpc_object
def list_all_connections(self) -> dict:
"""
List all the registered RPC objects.
@@ -92,24 +115,41 @@ class RPCRegister:
connections = dict(self._rpc_register)
return connections
def get_rpc_by_type(self, type_name) -> list[str]:
"""
Get all RPC objects of a certain type.
def get_names_of_rpc_by_class_type(
self, cls: BECWidget | BECConnector | BECDock | BECDockArea
) -> list[str]:
"""Get all the names of the widgets.
Args:
type_name(str): The type of the RPC object to be retrieved.
Returns:
list: A list of RPC objects of the given type.
cls(BECWidget | BECConnector): The class of the RPC object to be retrieved.
"""
rpc_objects = [rpc for rpc in self._rpc_register if rpc.startswith(type_name)]
return rpc_objects
# This retrieves any rpc objects that are subclass of BECWidget,
# i.e. curve and image items are excluded
widgets = [rpc for rpc in self._rpc_register.values() if isinstance(rpc, cls)]
return [widget._name for widget in widgets]
# def get_names_by_class_name(self, class_name: str) -> list[str]:
# """
# Get all RPC objects of a class, i.e. BECDockArea, BECDock.
# Args:
# class_name(str): The type of the RPC object to be retrieved.
# Returns:
# list: A list of names of RPC objects of the given type.
# """
# rpc_objects = [
# rpc._name
# for name, rpc in self._rpc_register.items()
# if rpc.__class__.__name__ == class_name
# ]
# return rpc_objects
def broadcast(self):
"""
Broadcast the update to all the callbacks.
"""
print("Broadcasting")
# print("Broadcasting")
connections = self.list_all_connections()
for callback in self.callbacks:
callback(connections)
bec_widgets/cli/rpc/rpc_widget_handler.py
+10 -7
View File
@@ -1,6 +1,8 @@
from __future__ import annotations
from bec_widgets.utils import BECConnector
from typing import Any
from bec_widgets.utils.bec_widget import BECWidget
class RPCWidgetHandler:
@@ -10,7 +12,7 @@ class RPCWidgetHandler:
self._widget_classes = None
@property
def widget_classes(self):
def widget_classes(self) -> dict[str, Any]:
"""
Get the available widget classes.
@@ -19,7 +21,7 @@ class RPCWidgetHandler:
"""
if self._widget_classes is None:
self.update_available_widgets()
return self._widget_classes
return self._widget_classes # type: ignore
def update_available_widgets(self):
"""
@@ -33,22 +35,23 @@ class RPCWidgetHandler:
clss = get_custom_classes("bec_widgets")
self._widget_classes = {cls.__name__: cls for cls in clss.widgets}
def create_widget(self, widget_type, **kwargs) -> BECConnector:
def create_widget(self, widget_type, name: str, **kwargs) -> BECWidget:
"""
Create a widget from an RPC message.
Args:
widget_type(str): The type of the widget.
name (str): The name of the widget.
**kwargs: The keyword arguments for the widget.
Returns:
widget(BECConnector): The created widget.
widget(BECWidget): The created widget.
"""
if self._widget_classes is None:
self.update_available_widgets()
widget_class = self._widget_classes.get(widget_type)
widget_class = self._widget_classes.get(widget_type) # type: ignore
if widget_class:
return widget_class(**kwargs)
return widget_class(name=name, **kwargs)
raise ValueError(f"Unknown widget type: {widget_type}")
bec_widgets/cli/server.py
+19 -11
View File
@@ -15,6 +15,7 @@ from bec_lib.utils.import_utils import lazy_import
from qtpy.QtCore import Qt, QTimer
from redis.exceptions import RedisError
from bec_widgets.cli.rpc import rpc_register
from bec_widgets.cli.rpc.rpc_register import RPCRegister
from bec_widgets.qt_utils.error_popups import ErrorPopupUtility
from bec_widgets.utils import BECDispatcher
@@ -36,6 +37,8 @@ def rpc_exception_hook(err_func):
old_exception_hook = popup.custom_exception_hook
# install err_func, if it is a callable
# IMPORTANT, Keep self here, because this method is overwriting the custom_exception_hook
# of the ErrorPopupUtility (popup instance) class.
def custom_exception_hook(self, exc_type, value, tb, **kwargs):
err_func({"error": popup.get_error_message(exc_type, value, tb)})
@@ -64,10 +67,9 @@ class BECWidgetsCLIServer:
self.client = self.dispatcher.client if client is None else client
self.client.start()
self.gui_id = gui_id
self.gui = gui_class(gui_id=gui_class_id)
# register broadcast callback
self.rpc_register = RPCRegister()
self.rpc_register.add_rpc(self.gui)
self.rpc_register.add_callback(self.broadcast_registry_update)
self.dispatcher.connect_slot(
self.on_rpc_update, MessageEndpoints.gui_instructions(self.gui_id)
)
@@ -79,6 +81,8 @@ class BECWidgetsCLIServer:
self.status = messages.BECStatus.RUNNING
logger.success(f"Server started with gui_id: {self.gui_id}")
# Create initial object -> BECFigure or BECDockArea
self.gui = gui_class(parent=None, name=gui_class_id)
def on_rpc_update(self, msg: dict, metadata: dict):
request_id = metadata.get("request_id")
@@ -136,6 +140,9 @@ class BECWidgetsCLIServer:
if isinstance(obj, BECConnector):
return {
"gui_id": obj.gui_id,
"name": (
obj._name if hasattr(obj, "_name") else obj.__class__.__name__
), # pylint: disable=protected-access
"widget_class": obj.__class__.__name__,
"config": obj.config.model_dump(),
"__rpc__": True,
@@ -165,11 +172,14 @@ class BECWidgetsCLIServer:
if val.__class__.__name__ == "BECDockArea"
}
logger.info(f"Broadcasting registry update: {data}")
# self.client.connector.set(
# MessageEndpoints.gui_registry_update(self.gui_id),
# messages.RegistryUpdateMessage(connections=connections),
# expire=10,
# )
for key, val in data.items():
logger.info(f"DockArea: {key} - docks: {len(val['config']['docks'])}")
logger.warning(f"Broadcasting registry update: {data}")
self.client.connector.xadd(
MessageEndpoints.gui_registry_state(self.gui_id),
msg_dict={"data": messages.GUIRegistryStateMessage(state=data)},
max_size=1, # only single message in stream
)
def shutdown(self): # TODO not sure if needed when cleanup is done at level of BECConnector
logger.info(f"Shutting down server with gui_id: {self.gui_id}")
@@ -288,7 +298,7 @@ def main():
# store gui id within QApplication object, to make it available to all widgets
app.gui_id = args.id
# args.id = "52e70"
# args.id = "abff6"
server = _start_server(args.id, gui_class, args.gui_class_id, args.config)
win = BECMainWindow(gui_id=f"{server.gui_id}:window")
@@ -296,8 +306,6 @@ def main():
win.setWindowTitle("BEC")
RPCRegister().add_rpc(win)
RPCRegister().add_callback(server.broadcast_registry_update)
gui = server.gui
win.setCentralWidget(gui)
if not args.hide:
bec_widgets/examples/jupyter_console/jupyter_console_window.py
+8 -4
View File
@@ -198,14 +198,18 @@ class JupyterConsoleWindow(QWidget): # pragma: no cover:
def _init_dock(self):
self.d0 = self.dock.add_dock(name="dock_0")
self.mm = self.d0.add_widget("BECMotorMapWidget")
self.d0 = self.dock.new(name="dock_0")
self.mm = self.d0.new("BECMotorMapWidget")
self.mm.change_motors("samx", "samy")
self.d1 = self.dock.add_dock(name="dock_1", position="right")
self.im = self.d1.add_widget("BECImageWidget")
self.d1 = self.dock.new(name="dock_1", position="right")
self.im = self.d1.new("BECImageWidget")
self.im.image("waveform", "1d")
self.d2 = self.dock.new(name="dock_2", position="bottom")
self.wf = self.d2.new("BECFigure", row=0, col=0)
self.mw = self.wf.multi_waveform(monitor="waveform") # , config=config)
self.mw = None # self.wf.multi_waveform(monitor="waveform") # , config=config)
self.dock.save_state()
bec_widgets/utils/bec_connector.py
+26 -9
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
import os
import time
import uuid
from datetime import datetime
from typing import TYPE_CHECKING, Optional
from bec_lib.logger import bec_logger
@@ -15,6 +16,7 @@ from qtpy.QtWidgets import QApplication
from bec_widgets.cli.rpc.rpc_register import RPCRegister
from bec_widgets.qt_utils.error_popups import ErrorPopupUtility
from bec_widgets.qt_utils.error_popups import SafeSlot as pyqtSlot
from bec_widgets.utils.container_utils import WidgetContainerUtils
from bec_widgets.utils.yaml_dialog import load_yaml, load_yaml_gui, save_yaml, save_yaml_gui
if TYPE_CHECKING:
@@ -39,8 +41,7 @@ class ConnectionConfig(BaseModel):
"""Generate a GUI ID if none is provided."""
if v is None:
widget_class = values.data["widget_class"]
v = f"{widget_class}_{str(time.time())}"
return v
v = f"{widget_class}_{datetime.now().strftime('%Y_%m_%d_%H_%M_%S_%f')}"
return v
@@ -75,7 +76,13 @@ class BECConnector:
USER_ACCESS = ["_config_dict", "_get_all_rpc", "_rpc_id"]
EXIT_HANDLERS = {}
def __init__(self, client=None, config: ConnectionConfig = None, gui_id: str = None):
def __init__(
self,
client=None,
config: ConnectionConfig | None = None,
gui_id: str | None = None,
name: str | None = None,
):
# BEC related connections
self.bec_dispatcher = BECDispatcher(client=client)
self.client = self.bec_dispatcher.client if client is None else client
@@ -103,15 +110,22 @@ class BECConnector:
)
self.config = ConnectionConfig(widget_class=self.__class__.__name__)
# I feel that we should not allow BECConnector to be created with a custom gui_id
# because this would break with the logic in the RPCRegister of retrieving widgets by type
# iterating over all widgets and checkinf if the register widget starts with the string that is passsed.
# If the gui_id is randomly generated, this would break since that widget would have a
# gui_id that is generated in a different way.
if gui_id:
self.config.gui_id = gui_id
self.gui_id = gui_id
self.gui_id: str = gui_id
else:
self.gui_id = self.config.gui_id
# register widget to rpc register
# be careful: when registering, and the object is not a BECWidget,
# cleanup has to be called manually since there is no 'closeEvent'
self.gui_id: str = self.config.gui_id # type: ignore
if name is None:
name = self.__class__.__name__
else:
if not WidgetContainerUtils.has_name_valid_chars(name):
raise ValueError(f"Name {name} contains invalid characters.")
self._name = name if name else self.__class__.__name__
self.rpc_register = RPCRegister()
self.rpc_register.add_rpc(self)
@@ -288,9 +302,12 @@ class BECConnector:
Args:
config (ConnectionConfig | dict): Configuration settings.
"""
gui_id = getattr(config, "gui_id", None)
if isinstance(config, dict):
config = ConnectionConfig(**config)
self.config = config
if gui_id and config.gui_id != gui_id: # Recreating config should not overwrite the gui_id
self.config.gui_id = gui_id
def get_config(self, dict_output: bool = True) -> dict | BaseModel:
"""
bec_widgets/utils/bec_widget.py
+11 -4
View File
@@ -7,6 +7,7 @@ from qtpy.QtWidgets import QApplication, QWidget
from bec_widgets.utils.bec_connector import BECConnector, ConnectionConfig
from bec_widgets.utils.colors import set_theme
from bec_widgets.utils.container_utils import WidgetContainerUtils
logger = bec_logger.logger
@@ -22,8 +23,9 @@ class BECWidget(BECConnector):
self,
client=None,
config: ConnectionConfig = None,
gui_id: str = None,
gui_id: str | None = None,
theme_update: bool = False,
name: str | None = None,
**kwargs,
):
"""
@@ -45,9 +47,14 @@ class BECWidget(BECConnector):
"""
if not isinstance(self, QWidget):
raise RuntimeError(f"{repr(self)} is not a subclass of QWidget")
super().__init__(client=client, config=config, gui_id=gui_id, **kwargs)
# Set the theme to auto if it is not set yet
# Create a default name if None is provided
if name is None:
name = "bec_widget_init_without_name"
# name = self.__class__.__name__
# Check for invalid chars in the name
if not WidgetContainerUtils.has_name_valid_chars(name):
raise ValueError(f"Name {name} contains invalid characters.")
super().__init__(client=client, config=config, gui_id=gui_id, name=name)
app = QApplication.instance()
if not hasattr(app, "theme"):
# DO NOT SET THE THEME TO AUTO! Otherwise, the qwebengineview will segfault
bec_widgets/utils/container_utils.py
+37 -12
View File
@@ -1,30 +1,55 @@
from __future__ import annotations
import itertools
from typing import Type
from typing import Literal, Type
from qtpy.QtWidgets import QWidget
from bec_widgets.cli.rpc.rpc_register import RPCRegister
class WidgetContainerUtils:
# We need one handler that checks if a WIDGET of a given name is already created for that DOCKAREA
# 1. If the name exists, then it depends whether the name was auto-generated -> add _1 to the name
# or alternatively raise an error that it can't be added again ( just raise an error)
# 2. Dock names in between docks should also be unique
@staticmethod
def generate_unique_widget_id(container: dict, prefix: str = "widget") -> str:
"""
Generate a unique widget ID.
def has_name_valid_chars(name: str) -> bool:
"""Check if the name is valid.
Args:
container(dict): The container of widgets.
prefix(str): The prefix of the widget ID.
name(str): The name to be checked.
Returns:
widget_id(str): The unique widget ID.
bool: True if the name is valid, False otherwise.
"""
existing_ids = set(container.keys())
for i in itertools.count(1):
widget_id = f"{prefix}_{i}"
if widget_id not in existing_ids:
return widget_id
if not name or len(name) > 256:
return False # Don't accept empty names or names longer than 256 characters
check_value = name.replace("_", "").replace("-", "")
if not check_value.isalnum() or not check_value.isascii():
return False
return True
@staticmethod
def generate_unique_name(name: str, list_of_names: list[str] | None = None) -> str:
"""Generate a unique ID.
Args:
name(str): The name of the widget.
Returns:
tuple (str): The unique name
"""
if list_of_names is None:
list_of_names = []
ii = 0
while ii < 1000: # 1000 is arbritrary!
name_candidate = f"{name}_{ii}"
if name_candidate not in list_of_names:
return name_candidate
ii += 1
raise ValueError("Could not generate a unique name after within 1000 attempts.")
@staticmethod
def find_first_widget_by_class(
bec_widgets/widgets/containers/dock/dock.py
+129 -47
View File
@@ -1,7 +1,9 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Any, Literal, Optional
from typing import TYPE_CHECKING, Any, Literal, Optional, cast
from weakref import WeakValueDictionary
from bec_lib.logger import bec_logger
from pydantic import Field
from pyqtgraph.dockarea import Dock, DockLabel
from qtpy import QtCore, QtGui
@@ -9,17 +11,22 @@ from qtpy import QtCore, QtGui
from bec_widgets.cli.rpc.rpc_widget_handler import widget_handler
from bec_widgets.utils import ConnectionConfig, GridLayoutManager
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.container_utils import WidgetContainerUtils
logger = bec_logger.logger
if TYPE_CHECKING:
from qtpy.QtWidgets import QWidget
from bec_widgets.widgets.containers.dock.dock_area import BECDockArea
class DockConfig(ConnectionConfig):
widgets: dict[str, Any] = Field({}, description="The widgets in the dock.")
position: Literal["bottom", "top", "left", "right", "above", "below"] = Field(
"bottom", description="The position of the dock."
)
parent_dock_area: Optional[str] = Field(
parent_dock_area: Optional[str] | None = Field(
None, description="The GUI ID of parent dock area of the dock."
)
@@ -103,17 +110,19 @@ class BECDock(BECWidget, Dock):
ICON_NAME = "widgets"
USER_ACCESS = [
"_config_dict",
"_rpc_id",
"widget_list",
"element_list",
"elements",
"new",
"show",
"hide",
"show_title_bar",
"set_title",
"hide_title_bar",
"get_widgets_positions",
"set_title",
"add_widget",
"list_eligible_widgets",
"available_widgets",
"move_widget",
"remove_widget",
"remove",
"delete",
"delete_all",
"attach",
"detach",
]
@@ -121,7 +130,7 @@ class BECDock(BECWidget, Dock):
def __init__(
self,
parent: QWidget | None = None,
parent_dock_area: QWidget | None = None,
parent_dock_area: BECDockArea | None = None,
config: DockConfig | None = None,
name: str | None = None,
client=None,
@@ -131,19 +140,20 @@ class BECDock(BECWidget, Dock):
) -> None:
if config is None:
config = DockConfig(
widget_class=self.__class__.__name__, parent_dock_area=parent_dock_area.gui_id
widget_class=self.__class__.__name__, parent_dock_area=parent_dock_area._name
)
else:
if isinstance(config, dict):
config = DockConfig(**config)
self.config = config
super().__init__(client=client, config=config, gui_id=gui_id)
super().__init__(
client=client, config=config, gui_id=gui_id, name=name
) # Name was checked and created in BEC Widget
label = CustomDockLabel(text=name, closable=closable)
Dock.__init__(self, name=name, label=label, **kwargs)
# Dock.__init__(self, name=name, **kwargs)
self.parent_dock_area = parent_dock_area
# Layout Manager
self.layout_manager = GridLayoutManager(self.layout)
@@ -173,7 +183,18 @@ class BECDock(BECWidget, Dock):
super().float()
@property
def widget_list(self) -> list[BECWidget]:
def elements(self) -> dict[str, BECWidget]:
"""
Get the widgets in the dock.
Returns:
widgets(dict): The widgets in the dock.
"""
# pylint: disable=protected-access
return dict((widget._name, widget) for widget in self.element_list)
@property
def element_list(self) -> list[BECWidget]:
"""
Get the widgets in the dock.
@@ -182,10 +203,6 @@ class BECDock(BECWidget, Dock):
"""
return self.widgets
@widget_list.setter
def widget_list(self, value: list[BECWidget]):
self.widgets = value
def hide_title_bar(self):
"""
Hide the title bar of the dock.
@@ -194,6 +211,20 @@ class BECDock(BECWidget, Dock):
self.label.hide()
self.labelHidden = True
def show(self):
"""
Show the dock.
"""
super().show()
self.show_title_bar()
def hide(self):
"""
Hide the dock.
"""
self.hide_title_bar()
super().hide()
def show_title_bar(self):
"""
Hide the title bar of the dock.
@@ -211,7 +242,6 @@ class BECDock(BECWidget, Dock):
"""
self.orig_area.docks[title] = self.orig_area.docks.pop(self.name())
self.setTitle(title)
self._name = title
def get_widgets_positions(self) -> dict:
"""
@@ -222,7 +252,7 @@ class BECDock(BECWidget, Dock):
"""
return self.layout_manager.get_widgets_positions()
def list_eligible_widgets(
def available_widgets(
self,
) -> list: # TODO can be moved to some util mixin like container class for rpc widgets
"""
@@ -233,13 +263,21 @@ class BECDock(BECWidget, Dock):
"""
return list(widget_handler.widget_classes.keys())
def add_widget(
def _get_list_of_widget_name_of_parent_dock_area(self):
docks = self.parent_dock_area.panel_list
widgets = []
for dock in docks:
widgets.extend(dock.elements.keys())
return widgets
def new(
self,
widget: BECWidget | str,
row=None,
col=0,
rowspan=1,
colspan=1,
name: str | None = None,
row: int | None = None,
col: int = 0,
rowspan: int = 1,
colspan: int = 1,
shift: Literal["down", "up", "left", "right"] = "down",
) -> BECWidget:
"""
@@ -254,21 +292,37 @@ class BECDock(BECWidget, Dock):
shift(Literal["down", "up", "left", "right"]): The direction to shift the widgets if the position is occupied.
"""
if row is None:
# row = cast(int, self.layout.rowCount()) # type:ignore
row = self.layout.rowCount()
# row = cast(int, row)
if self.layout_manager.is_position_occupied(row, col):
self.layout_manager.shift_widgets(shift, start_row=row)
existing_widgets_parent_dock = self._get_list_of_widget_name_of_parent_dock_area()
if name is not None: # Name is provided
if name in existing_widgets_parent_dock:
# pylint: disable=protected-access
raise ValueError(
f"Name {name} must be unique for widgets, but already exists in DockArea "
f"with name: {self.parent_dock_area._name} and id {self.parent_dock_area.gui_id}."
)
else: # Name is not provided
name = WidgetContainerUtils.generate_unique_name(
name=(
widget if isinstance(widget, str) else widget._name
), # pylint: disable=protected-access
list_of_names=existing_widgets_parent_dock,
)
if isinstance(widget, str):
widget = widget_handler.create_widget(widget)
widget = cast(BECWidget, widget_handler.create_widget(widget_type=widget, name=name))
else:
widget = widget
widget._name = name # pylint: disable=protected-access
self.addWidget(widget, row=row, col=col, rowspan=rowspan, colspan=colspan)
if hasattr(widget, "config"):
self.config.widgets[widget.gui_id] = widget.config
self.config.widgets[widget._name] = widget.config
return widget
def move_widget(self, widget: QWidget, new_row: int, new_col: int):
@@ -294,32 +348,51 @@ class BECDock(BECWidget, Dock):
"""
self.float()
def remove_widget(self, widget_rpc_id: str):
"""
Remove a widget from the dock.
Args:
widget_rpc_id(str): The ID of the widget to remove.
"""
widget = self.rpc_register.get_rpc_by_id(widget_rpc_id)
self.layout.removeWidget(widget)
self.config.widgets.pop(widget_rpc_id, None)
widget.close()
def remove(self):
"""
Remove the dock from the parent dock area.
"""
# self.cleanup()
self.parent_dock_area.remove_dock(self.name())
self.parent_dock_area.delete(self.gui_id)
def delete(self, widget_name: str) -> None:
"""
Remove a widget from the dock.
Args:
widget_name(str): Delete the widget with the given name.
"""
widget = [widget for widget in self.widgets if widget._name == widget_name]
if not widget:
logger.warning(
f"Widget with name {widget_name} not found in dock {self.name()}. "
f"Checking if gui_id was passed as widget_name."
)
# Try to find the widget in the RPC register
widget = self.rpc_register.get_rpc_by_id(widget_name)
if widget is None:
logger.warning(
f"Widget not found for name or gui_id: {widget_name} in dock {self.name()}"
)
return
widget = widget[0]
self.layout.removeWidget(widget)
self.config.widgets.pop(widget._name, None)
if widget in self.widgets:
self.widgets.remove(widget)
widget.close()
def delete_all(self):
"""
Remove all widgets from the dock.
"""
for widget in self.widgets:
self.delete(widget._name)
def cleanup(self):
"""
Clean up the dock, including all its widgets.
"""
for widget in self.widgets:
if hasattr(widget, "cleanup"):
widget.cleanup()
self.delete_all()
self.widgets.clear()
self.label.close()
self.label.deleteLater()
@@ -333,3 +406,12 @@ class BECDock(BECWidget, Dock):
self.cleanup()
super().close()
self.parent_dock_area.dock_area.docks.pop(self.name(), None)
if __name__ == "__main__":
from qtpy.QtWidgets import QApplication
app = QApplication([])
dock = BECDock(name="dock")
dock.show()
app.exec_()
bec_widgets/widgets/containers/dock/dock_area.py
+96 -81
View File
@@ -1,6 +1,7 @@
from __future__ import annotations
from typing import Literal, Optional
from unittest.mock import NonCallableMagicMock
from weakref import WeakValueDictionary
from bec_lib.endpoints import MessageEndpoints
@@ -10,6 +11,7 @@ from qtpy.QtCore import QSize, Qt
from qtpy.QtGui import QPainter, QPaintEvent
from qtpy.QtWidgets import QApplication, QSizePolicy, QVBoxLayout, QWidget
from bec_widgets.cli.rpc.rpc_register import RPCRegister
from bec_widgets.qt_utils.error_popups import SafeSlot
from bec_widgets.qt_utils.toolbar import (
ExpandableMenuAction,
@@ -44,21 +46,18 @@ class DockAreaConfig(ConnectionConfig):
class BECDockArea(BECWidget, QWidget):
PLUGIN = True
USER_ACCESS = [
"_config_dict",
"selected_device",
"panels",
"save_state",
"remove_dock",
"restore_state",
"add_dock",
"clear_all",
"detach_dock",
"attach_all",
"_get_all_rpc",
"temp_areas",
"new",
"show",
"hide",
"panels",
"panel_list",
"delete",
"delete_all",
"detach_dock",
"attach_all",
"selected_device",
"save_state",
"restore_state",
]
def __init__(
@@ -67,6 +66,8 @@ class BECDockArea(BECWidget, QWidget):
config: DockAreaConfig | None = None,
client=None,
gui_id: str = None,
name: str | None = None,
**kwargs,
) -> None:
if config is None:
config = DockAreaConfig(widget_class=self.__class__.__name__)
@@ -74,7 +75,7 @@ class BECDockArea(BECWidget, QWidget):
if isinstance(config, dict):
config = DockAreaConfig(**config)
self.config = config
super().__init__(client=client, config=config, gui_id=gui_id)
super().__init__(client=client, config=config, gui_id=gui_id, name=name, **kwargs)
QWidget.__init__(self, parent=parent)
self.layout = QVBoxLayout(self)
self.layout.setSpacing(5)
@@ -169,41 +170,41 @@ class BECDockArea(BECWidget, QWidget):
def _hook_toolbar(self):
# Menu Plot
self.toolbar.widgets["menu_plots"].widgets["waveform"].triggered.connect(
lambda: self.add_dock(widget="Waveform", prefix="waveform")
lambda: self._create_widget_from_toolbar(widget_name="Waveform")
)
self.toolbar.widgets["menu_plots"].widgets["multi_waveform"].triggered.connect(
lambda: self.add_dock(widget="BECMultiWaveformWidget", prefix="multi_waveform")
lambda: self._create_widget_from_toolbar(widget_name="BECMultiWaveformWidget")
)
self.toolbar.widgets["menu_plots"].widgets["image"].triggered.connect(
lambda: self.add_dock(widget="BECImageWidget", prefix="image")
lambda: self._create_widget_from_toolbar(widget_name="BECImageWidget")
)
self.toolbar.widgets["menu_plots"].widgets["motor_map"].triggered.connect(
lambda: self.add_dock(widget="BECMotorMapWidget", prefix="motor_map")
lambda: self._create_widget_from_toolbar(widget_name="BECMotorMapWidget")
)
# Menu Devices
self.toolbar.widgets["menu_devices"].widgets["scan_control"].triggered.connect(
lambda: self.add_dock(widget="ScanControl", prefix="scan_control")
lambda: self._create_widget_from_toolbar(widget_name="ScanControl")
)
self.toolbar.widgets["menu_devices"].widgets["positioner_box"].triggered.connect(
lambda: self.add_dock(widget="PositionerBox", prefix="positioner_box")
lambda: self._create_widget_from_toolbar(widget_name="PositionerBox")
)
# Menu Utils
self.toolbar.widgets["menu_utils"].widgets["queue"].triggered.connect(
lambda: self.add_dock(widget="BECQueue", prefix="queue")
lambda: self._create_widget_from_toolbar(widget_name="BECQueue")
)
self.toolbar.widgets["menu_utils"].widgets["status"].triggered.connect(
lambda: self.add_dock(widget="BECStatusBox", prefix="status")
lambda: self._create_widget_from_toolbar(widget_name="BECStatusBox")
)
self.toolbar.widgets["menu_utils"].widgets["vs_code"].triggered.connect(
lambda: self.add_dock(widget="VSCodeEditor", prefix="vs_code")
lambda: self._create_widget_from_toolbar(widget_name="VSCodeEditor")
)
self.toolbar.widgets["menu_utils"].widgets["progress_bar"].triggered.connect(
lambda: self.add_dock(widget="RingProgressBar", prefix="progress_bar")
lambda: self._create_widget_from_toolbar(widget_name="RingProgressBar")
)
self.toolbar.widgets["menu_utils"].widgets["log_panel"].triggered.connect(
lambda: self.add_dock(widget="LogPanel", prefix="log_panel")
lambda: self._create_widget_from_toolbar(widget_name="LogPanel")
)
# Icons
@@ -211,6 +212,11 @@ class BECDockArea(BECWidget, QWidget):
self.toolbar.widgets["save_state"].action.triggered.connect(self.save_state)
self.toolbar.widgets["restore_state"].action.triggered.connect(self.restore_state)
@SafeSlot()
def _create_widget_from_toolbar(self, widget_name: str) -> None:
dock_name = WidgetContainerUtils.generate_unique_name(widget_name, self.panels.keys())
dock: BECDock = self.new(name=dock_name, widget=widget_name)
def paintEvent(self, event: QPaintEvent): # TODO decide if we want any default instructions
super().paintEvent(event)
if self._instructions_visible:
@@ -218,7 +224,7 @@ class BECDockArea(BECWidget, QWidget):
painter.drawText(
self.rect(),
Qt.AlignCenter,
"Add docks using 'add_dock' method from CLI\n or \n Add widget docks using the toolbar",
"Add docks using 'new' method from CLI\n or \n Add widget docks using the toolbar",
)
@property
@@ -243,7 +249,17 @@ class BECDockArea(BECWidget, QWidget):
@panels.setter
def panels(self, value: dict[str, BECDock]):
self.dock_area.docks = WeakValueDictionary(value)
self.dock_area.docks = WeakValueDictionary(value) # This can not work can it?
@property
def panel_list(self) -> list[BECDock]:
"""
Get the docks in the dock area.
Returns:
list: The docks in the dock area.
"""
return list(self.dock_area.docks.values())
@property
def temp_areas(self) -> list:
@@ -287,36 +303,17 @@ class BECDockArea(BECWidget, QWidget):
self.config.docks_state = last_state
return last_state
def remove_dock(self, name: str):
"""
Remove a dock by name and ensure it is properly closed and cleaned up.
Args:
name(str): The name of the dock to remove.
"""
dock = self.dock_area.docks.pop(name, None)
self.config.docks.pop(name, None)
if dock:
dock.close()
dock.deleteLater()
if len(self.dock_area.docks) <= 1:
for dock in self.dock_area.docks.values():
dock.hide_title_bar()
else:
raise ValueError(f"Dock with name {name} does not exist.")
@SafeSlot(popup_error=True)
def add_dock(
def new(
self,
name: str = None,
position: Literal["bottom", "top", "left", "right", "above", "below"] = None,
name: str | None = None,
widget: str | QWidget | None = None,
widget_name: str | None = None,
position: Literal["bottom", "top", "left", "right", "above", "below"] = "bottom",
relative_to: BECDock | None = None,
closable: bool = True,
floating: bool = False,
prefix: str = "dock",
widget: str | QWidget | None = None,
row: int = None,
row: int | None = None,
col: int = 0,
rowspan: int = 1,
colspan: int = 1,
@@ -326,12 +323,11 @@ class BECDockArea(BECWidget, QWidget):
Args:
name(str): The name of the dock to be displayed and for further references. Has to be unique.
widget(str|QWidget|None): The widget to be added to the dock. While using RPC, only BEC RPC widgets from RPCWidgetHandler are allowed.
position(Literal["bottom", "top", "left", "right", "above", "below"]): The position of the dock.
relative_to(BECDock): The dock to which the new dock should be added relative to.
closable(bool): Whether the dock is closable.
floating(bool): Whether the dock is detached after creating.
prefix(str): The prefix for the dock name if no name is provided.
widget(str|QWidget|None): The widget to be added to the dock. While using RPC, only BEC RPC widgets from RPCWidgetHandler are allowed.
row(int): The row of the added widget.
col(int): The column of the added widget.
rowspan(int): The rowspan of the added widget.
@@ -340,21 +336,22 @@ class BECDockArea(BECWidget, QWidget):
Returns:
BECDock: The created dock.
"""
if name is None:
name = WidgetContainerUtils.generate_unique_widget_id(
container=self.dock_area.docks, prefix=prefix
dock_names = [dock._name for dock in self.panel_list] # pylint: disable=protected-access
if name is not None: # Name is provided
if name in dock_names:
raise ValueError(
f"Name {name} must be unique for docks, but already exists in DockArea "
f"with name: {self._name} and id {self.gui_id}."
)
else: # Name is not provided
name = WidgetContainerUtils.generate_unique_name(
name=self.__class__.__name__, list_of_names=dock_names
)
if name in set(self.dock_area.docks.keys()):
raise ValueError(f"Dock with name {name} already exists.")
if position is None:
position = "bottom"
dock = BECDock(name=name, parent_dock_area=self, closable=closable)
dock.config.position = position
self.config.docks[name] = dock.config
self.config.docks[dock.name()] = dock.config
# The dock.name is equal to the name passed to BECDock
self.dock_area.addDock(dock=dock, position=position, relativeTo=relative_to)
if len(self.dock_area.docks) <= 1:
@@ -363,10 +360,11 @@ class BECDockArea(BECWidget, QWidget):
for dock in self.dock_area.docks.values():
dock.show_title_bar()
if widget is not None and isinstance(widget, str):
dock.add_widget(widget=widget, row=row, col=col, rowspan=rowspan, colspan=colspan)
elif widget is not None and isinstance(widget, QWidget):
dock.addWidget(widget, row=row, col=col, rowspan=rowspan, colspan=colspan)
if widget is not None:
# Check if widget name exists.
dock.new(
widget=widget, name=widget_name, row=row, col=col, rowspan=rowspan, colspan=colspan
)
if (
self._instructions_visible
): # TODO still decide how initial instructions should be handled
@@ -408,20 +406,11 @@ class BECDockArea(BECWidget, QWidget):
area.window().close()
area.window().deleteLater()
def clear_all(self):
"""
Close all docks and remove all temp areas.
"""
self.attach_all()
for dock in dict(self.dock_area.docks).values():
dock.remove()
self.dock_area.docks.clear()
def cleanup(self):
"""
Cleanup the dock area.
"""
self.clear_all()
self.delete_all()
self.toolbar.close()
self.toolbar.deleteLater()
self.dock_area.close()
@@ -465,9 +454,31 @@ class BECDockArea(BECWidget, QWidget):
continue
docks.window().hide()
def delete(self):
self.hide()
self.deleteLater()
def delete_all(self) -> None:
"""
Delete all docks.
"""
self.attach_all()
for dock_name in self.panels.keys():
self.delete(dock_name)
def delete(self, dock_name: str):
"""
Delete a dock by name.
Args:
dock_name(str): The name of the dock to delete.
"""
dock = self.dock_area.docks.pop(dock_name, None)
self.config.docks.pop(dock_name, None)
if dock:
dock.close()
dock.deleteLater()
if len(self.dock_area.docks) <= 1:
for dock in self.dock_area.docks.values():
dock.hide_title_bar()
else:
raise ValueError(f"Dock with name {dock_name} does not exist.")
if __name__ == "__main__": # pragma: no cover
@@ -477,5 +488,9 @@ if __name__ == "__main__": # pragma: no cover
app = QApplication([])
set_theme("auto")
dock_area = BECDockArea()
dock_1 = dock_area.new(name="dock_0", widget="BECWaveformWidget")
dock_1 = dock_area.new(name="dock_0", widget="BECWaveformWidget")
dock_1.new(widget="BECWaveformWidget")
dock_area.show()
app.topLevelWidgets()
app.exec_()
bec_widgets/widgets/containers/main_window/main_window.py
+14 -11
View File
@@ -2,6 +2,7 @@ from qtpy.QtWidgets import QApplication, QMainWindow
from bec_widgets.cli.rpc.rpc_register import RPCRegister
from bec_widgets.utils import BECConnector
from bec_widgets.utils.container_utils import WidgetContainerUtils
from bec_widgets.widgets.containers.dock.dock_area import BECDockArea
@@ -34,18 +35,20 @@ class BECMainWindow(QMainWindow, BECConnector):
}
return info
def new_dock_area(self, name: str | None = None):
if name is None:
name = "BEC"
def new_dock_area(self, name: str | None = None) -> BECDockArea:
rpc_register = RPCRegister()
existing_dock_areas = rpc_register.get_names_of_rpc_by_class_type(BECDockArea)
if name is not None:
if name in existing_dock_areas:
raise ValueError(
f"Name {name} must be unique for dock areas, but already exists: {existing_dock_areas}."
)
else:
name = "BEC - " + name
self.rpc_register = RPCRegister()
gui_id = name.replace(" - ", "_").replace(" ", "_").lower()
existing_widgets = self.rpc_register.get_rpc_by_type(gui_id)
if existing_widgets:
name = f"{name} {len(existing_widgets) + 1}"
dock_area = BECDockArea(gui_id=name.replace(" - ", "_").replace(" ", "_").lower())
name = "dock_area"
name = WidgetContainerUtils.generate_unique_name(name, existing_dock_areas)
dock_area = BECDockArea(name=name)
dock_area.resize(dock_area.minimumSizeHint())
dock_area.window().setWindowTitle(name)
# TODO Should we simply use the specified name as title here?
dock_area.window().setWindowTitle(f"BEC - {name}")
dock_area.show()
return dock_area