mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-03-04 16:02:51 +01:00
feat(generate_cli): RPC API from content widget can be merged with the RPC API of the container widget statically
This commit is contained in:
@@ -9,6 +9,8 @@ class DockAreaView(ViewBase):
|
||||
Modular dock area view for arranging and managing multiple dockable widgets.
|
||||
"""
|
||||
|
||||
RPC_CONTENT_CLASS = BECDockArea
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
parent: QWidget | None = None,
|
||||
@@ -20,6 +22,10 @@ class DockAreaView(ViewBase):
|
||||
):
|
||||
super().__init__(parent=parent, content=content, view_id=view_id, title=title, **kwargs)
|
||||
self.dock_area = BECDockArea(
|
||||
self, profile_namespace="bec", auto_profile_namespace=False, object_name="DockArea"
|
||||
self,
|
||||
profile_namespace="bec",
|
||||
auto_profile_namespace=False,
|
||||
object_name="DockArea",
|
||||
rpc_exposed=False,
|
||||
)
|
||||
self.set_content(self.dock_area)
|
||||
|
||||
@@ -51,6 +51,8 @@ class ViewBase(BECWidget, QWidget):
|
||||
RPC = True
|
||||
PLUGIN = False
|
||||
USER_ACCESS = ["activate"]
|
||||
RPC_CONTENT_CLASS: type[QWidget] | None = None
|
||||
RPC_CONTENT_ATTR = "content"
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
|
||||
@@ -1005,6 +1005,16 @@ class DapComboBox(RPCBase):
|
||||
"""
|
||||
|
||||
|
||||
class DeveloperView(RPCBase):
|
||||
"""A view for users to write scripts and macros and execute them within the application."""
|
||||
|
||||
@rpc_call
|
||||
def activate(self) -> "None":
|
||||
"""
|
||||
Switch the parent application to this view.
|
||||
"""
|
||||
|
||||
|
||||
class DeviceBrowser(RPCBase):
|
||||
"""DeviceBrowser is a widget that displays all available devices in the current BEC session."""
|
||||
|
||||
@@ -1090,6 +1100,239 @@ class DockAreaView(RPCBase):
|
||||
Switch the parent application to this view.
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def new(
|
||||
self,
|
||||
widget: "QWidget | str",
|
||||
*,
|
||||
closable: "bool" = True,
|
||||
floatable: "bool" = True,
|
||||
movable: "bool" = True,
|
||||
start_floating: "bool" = False,
|
||||
where: "Literal['left', 'right', 'top', 'bottom'] | None" = None,
|
||||
tab_with: "CDockWidget | QWidget | str | None" = None,
|
||||
relative_to: "CDockWidget | QWidget | str | None" = None,
|
||||
show_title_bar: "bool | None" = None,
|
||||
title_buttons: "Mapping[str, bool] | Sequence[str] | str | None" = None,
|
||||
show_settings_action: "bool | None" = None,
|
||||
promote_central: "bool" = False,
|
||||
object_name: "str | None" = None,
|
||||
**widget_kwargs,
|
||||
) -> "QWidget | BECWidget":
|
||||
"""
|
||||
Create a new widget (or reuse an instance) and add it as a dock.
|
||||
|
||||
Args:
|
||||
widget(QWidget | str): Instance or registered widget type string.
|
||||
closable(bool): Whether the dock is closable.
|
||||
floatable(bool): Whether the dock is floatable.
|
||||
movable(bool): Whether the dock is movable.
|
||||
start_floating(bool): Whether to start the dock floating.
|
||||
where(Literal["left", "right", "top", "bottom"] | None): Dock placement hint relative to the dock area (ignored when
|
||||
``relative_to`` is provided without an explicit value).
|
||||
tab_with(CDockWidget | QWidget | str | None): Existing dock (or widget/name) to tab the new dock alongside.
|
||||
relative_to(CDockWidget | QWidget | str | None): Existing dock (or widget/name) used as the positional anchor.
|
||||
When supplied and ``where`` is ``None``, the new dock inherits the
|
||||
anchor's current dock area.
|
||||
show_title_bar(bool | None): Explicitly show or hide the dock area's title bar.
|
||||
title_buttons(Mapping[str, bool] | Sequence[str] | str | None): Mapping or iterable describing which title bar buttons should
|
||||
remain visible. Provide a mapping of button names (``"float"``,
|
||||
``"close"``, ``"menu"``, ``"auto_hide"``, ``"minimize"``) to booleans,
|
||||
or a sequence of button names to hide.
|
||||
show_settings_action(bool | None): Control whether a dock settings/property action should
|
||||
be installed. Defaults to ``False`` for the basic dock area; subclasses
|
||||
such as `AdvancedDockArea` override the default to ``True``.
|
||||
promote_central(bool): When True, promote the created dock to be the dock manager's
|
||||
central widget (useful for editor stacks or other root content).
|
||||
object_name(str | None): Optional object name to assign to the created widget.
|
||||
**widget_kwargs: Additional keyword arguments passed to the widget constructor
|
||||
when creating by type name.
|
||||
|
||||
Returns:
|
||||
BECWidget: The created or reused widget instance.
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def widget_map(self, bec_widgets_only: "bool" = True) -> "dict[str, QWidget]":
|
||||
"""
|
||||
Return a dictionary mapping widget names to their corresponding widgets.
|
||||
|
||||
Args:
|
||||
bec_widgets_only(bool): If True, only include widgets that are BECConnector instances.
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def widget_list(self, bec_widgets_only: "bool" = True) -> "list[QWidget]":
|
||||
"""
|
||||
Return a list of widgets contained in the dock area.
|
||||
|
||||
Args:
|
||||
bec_widgets_only(bool): If True, only include widgets that are BECConnector instances.
|
||||
"""
|
||||
|
||||
@property
|
||||
@rpc_call
|
||||
def workspace_is_locked(self) -> "bool":
|
||||
"""
|
||||
Get or set the lock state of the workspace.
|
||||
|
||||
Returns:
|
||||
bool: True if the workspace is locked, False otherwise.
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def attach_all(self):
|
||||
"""
|
||||
Re-attach floating docks back into the dock manager.
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def delete_all(self):
|
||||
"""
|
||||
Delete all docks and their associated widgets.
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def delete(self, object_name: "str") -> "bool":
|
||||
"""
|
||||
Remove a widget from the dock area by its object name.
|
||||
|
||||
Args:
|
||||
object_name: The object name of the widget to remove.
|
||||
|
||||
Returns:
|
||||
bool: True if the widget was found and removed, False otherwise.
|
||||
|
||||
Raises:
|
||||
ValueError: If no widget with the given object name is found.
|
||||
|
||||
Example:
|
||||
>>> dock_area.delete("my_widget")
|
||||
True
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def set_layout_ratios(
|
||||
self,
|
||||
*,
|
||||
horizontal: "Sequence[float] | Mapping[int | str, float] | None" = None,
|
||||
vertical: "Sequence[float] | Mapping[int | str, float] | None" = None,
|
||||
splitter_overrides: "Mapping[int | str | Sequence[int], Sequence[float] | Mapping[int | str, float]] | None" = None,
|
||||
) -> "None":
|
||||
"""
|
||||
Adjust splitter ratios in the dock layout.
|
||||
|
||||
Args:
|
||||
horizontal: Weights applied to every horizontal splitter encountered.
|
||||
vertical: Weights applied to every vertical splitter encountered.
|
||||
splitter_overrides: Optional overrides targeting specific splitters identified
|
||||
by their index path (e.g. ``{0: [1, 2], (1, 0): [3, 5]}``). Paths are zero-based
|
||||
indices following the splitter hierarchy, starting from the root splitter.
|
||||
|
||||
Example:
|
||||
To build three columns with custom per-column ratios::
|
||||
|
||||
area.set_layout_ratios(
|
||||
horizontal=[1, 2, 1], # column widths
|
||||
splitter_overrides={
|
||||
0: [1, 2], # column 0 (two rows)
|
||||
1: [3, 2, 1], # column 1 (three rows)
|
||||
2: [1], # column 2 (single row)
|
||||
},
|
||||
)
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def describe_layout(self) -> "list[dict[str, Any]]":
|
||||
"""
|
||||
Return metadata describing splitter paths, orientations, and contained docks.
|
||||
|
||||
Useful for determining the keys to use in `set_layout_ratios(splitter_overrides=...)`.
|
||||
"""
|
||||
|
||||
@property
|
||||
@rpc_call
|
||||
def mode(self) -> "str":
|
||||
"""
|
||||
None
|
||||
"""
|
||||
|
||||
@mode.setter
|
||||
@rpc_call
|
||||
def mode(self) -> "str":
|
||||
"""
|
||||
None
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def list_profiles(self) -> "list[str]":
|
||||
"""
|
||||
List available workspace profiles in the current namespace.
|
||||
|
||||
Returns:
|
||||
list[str]: List of profile names.
|
||||
"""
|
||||
|
||||
@rpc_timeout(None)
|
||||
@rpc_call
|
||||
def save_profile(
|
||||
self,
|
||||
name: "str | None" = None,
|
||||
*,
|
||||
show_dialog: "bool" = False,
|
||||
quick_select: "bool | None" = None,
|
||||
):
|
||||
"""
|
||||
Save the current workspace profile.
|
||||
|
||||
On first save of a given name:
|
||||
- writes a default copy to states/default/<name>.ini with tag=default and created_at
|
||||
- writes a user copy to states/user/<name>.ini with tag=user and created_at
|
||||
On subsequent saves of user-owned profiles:
|
||||
- updates both the default and user copies so restore uses the latest snapshot.
|
||||
Read-only bundled profiles cannot be overwritten.
|
||||
|
||||
Args:
|
||||
name (str | None): The name of the profile to save. If None and show_dialog is True,
|
||||
prompts the user.
|
||||
show_dialog (bool): If True, shows the SaveProfileDialog for user interaction.
|
||||
If False (default), saves directly without user interaction (useful for CLI usage).
|
||||
quick_select (bool | None): Whether to include the profile in quick selection.
|
||||
If None (default), uses the existing value or True for new profiles.
|
||||
Only used when show_dialog is False; otherwise the dialog provides the value.
|
||||
"""
|
||||
|
||||
@rpc_timeout(None)
|
||||
@rpc_call
|
||||
def load_profile(self, name: "str | None" = None):
|
||||
"""
|
||||
Load a workspace profile.
|
||||
|
||||
Before switching, persist the current profile to the user copy.
|
||||
Prefer loading the user copy; fall back to the default copy.
|
||||
|
||||
Args:
|
||||
name (str | None): The name of the profile to load. If None, prompts the user.
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def delete_profile(self, name: "str | None" = None, show_dialog: "bool" = False) -> "bool":
|
||||
"""
|
||||
Delete a workspace profile.
|
||||
|
||||
Args:
|
||||
name: The name of the profile to delete. If None, uses the currently
|
||||
selected profile from the toolbar combo box (for UI usage).
|
||||
show_dialog: If True, show confirmation dialog before deletion.
|
||||
Defaults to False for CLI/programmatic usage.
|
||||
|
||||
Returns:
|
||||
bool: True if the profile was deleted, False otherwise.
|
||||
|
||||
Raises:
|
||||
ValueError: If the profile is read-only or doesn't exist (when show_dialog=False).
|
||||
"""
|
||||
|
||||
|
||||
class DockAreaWidget(RPCBase):
|
||||
"""Lightweight dock area that exposes the core Qt ADS docking helpers without any"""
|
||||
|
||||
@@ -164,17 +164,13 @@ class {class_name}(RPCBase):"""
|
||||
self.content += f"""
|
||||
\"\"\"{class_docs}\"\"\"
|
||||
"""
|
||||
if not cls.USER_ACCESS:
|
||||
user_access_entries = self._get_user_access_entries(cls)
|
||||
if not user_access_entries:
|
||||
self.content += """...
|
||||
"""
|
||||
|
||||
for method in cls.USER_ACCESS:
|
||||
is_property_setter = False
|
||||
obj = getattr(cls, method, None)
|
||||
if obj is None:
|
||||
obj = getattr(cls, method.split(".setter")[0], None)
|
||||
is_property_setter = True
|
||||
method = method.split(".setter")[0]
|
||||
for method_entry in user_access_entries:
|
||||
method, obj, is_property_setter = self._resolve_method_object(cls, method_entry)
|
||||
if obj is None:
|
||||
raise AttributeError(
|
||||
f"Method {method} not found in class {cls.__name__}. "
|
||||
@@ -216,6 +212,34 @@ class {class_name}(RPCBase):"""
|
||||
{doc}
|
||||
\"\"\""""
|
||||
|
||||
@staticmethod
|
||||
def _get_user_access_entries(cls) -> list[str]:
|
||||
entries = list(getattr(cls, "USER_ACCESS", []))
|
||||
content_cls = getattr(cls, "RPC_CONTENT_CLASS", None)
|
||||
if content_cls is not None:
|
||||
entries.extend(getattr(content_cls, "USER_ACCESS", []))
|
||||
return list(dict.fromkeys(entries))
|
||||
|
||||
@staticmethod
|
||||
def _resolve_method_object(cls, method_entry: str):
|
||||
method_name = method_entry
|
||||
is_property_setter = False
|
||||
|
||||
if method_entry.endswith(".setter"):
|
||||
is_property_setter = True
|
||||
method_name = method_entry.split(".setter")[0]
|
||||
|
||||
candidate_classes = [cls]
|
||||
content_cls = getattr(cls, "RPC_CONTENT_CLASS", None)
|
||||
if content_cls is not None:
|
||||
candidate_classes.append(content_cls)
|
||||
|
||||
for candidate_cls in candidate_classes:
|
||||
obj = getattr(candidate_cls, method_name, None)
|
||||
if obj is not None:
|
||||
return method_name, obj, is_property_setter
|
||||
return method_name, None, is_property_setter
|
||||
|
||||
def _rpc_call(self, timeout_info: dict[str, float | None]):
|
||||
"""
|
||||
Decorator to mark a method as an RPC call.
|
||||
|
||||
@@ -181,18 +181,58 @@ class RPCServer:
|
||||
obj.show()
|
||||
res = None
|
||||
else:
|
||||
method_obj = getattr(obj, method)
|
||||
target_obj, method_obj = self._resolve_rpc_target(obj, method)
|
||||
# check if the method accepts args and kwargs
|
||||
if not callable(method_obj):
|
||||
if not args:
|
||||
res = method_obj
|
||||
else:
|
||||
setattr(obj, method, args[0])
|
||||
setattr(target_obj, method, args[0])
|
||||
res = None
|
||||
else:
|
||||
res = method_obj(*args, **kwargs)
|
||||
return res
|
||||
|
||||
def _resolve_rpc_target(self, obj, method: str) -> tuple[object, object]:
|
||||
"""
|
||||
Resolve a method/property access target for RPC execution.
|
||||
|
||||
Primary target is the object itself. If not found there and the class defines
|
||||
``RPC_CONTENT_CLASS``, unresolved method names can be delegated to the content
|
||||
widget referenced by ``RPC_CONTENT_ATTR`` (default ``content``), but only when
|
||||
the method is explicitly listed in the content class ``USER_ACCESS``.
|
||||
"""
|
||||
if hasattr(obj, method):
|
||||
return obj, getattr(obj, method)
|
||||
|
||||
content_cls = getattr(type(obj), "RPC_CONTENT_CLASS", None)
|
||||
if content_cls is None:
|
||||
raise AttributeError(f"{type(obj).__name__} has no attribute '{method}'")
|
||||
|
||||
content_user_access = set()
|
||||
for entry in getattr(content_cls, "USER_ACCESS", []):
|
||||
if entry.endswith(".setter"):
|
||||
content_user_access.add(entry.split(".setter")[0])
|
||||
else:
|
||||
content_user_access.add(entry)
|
||||
|
||||
if method not in content_user_access:
|
||||
raise AttributeError(f"{type(obj).__name__} has no attribute '{method}'")
|
||||
|
||||
content_attr = getattr(type(obj), "RPC_CONTENT_ATTR", "content")
|
||||
target_obj = getattr(obj, content_attr, None)
|
||||
if target_obj is None:
|
||||
raise AttributeError(
|
||||
f"{type(obj).__name__} has no content target '{content_attr}' for RPC delegation"
|
||||
)
|
||||
if not isinstance(target_obj, content_cls):
|
||||
raise AttributeError(
|
||||
f"{type(obj).__name__}.{content_attr} is not instance of {content_cls.__name__}"
|
||||
)
|
||||
if not hasattr(target_obj, method):
|
||||
raise AttributeError(f"{content_cls.__name__} has no attribute '{method}'")
|
||||
return target_obj, getattr(target_obj, method)
|
||||
|
||||
def run_system_rpc(self, method: str, args: list, kwargs: dict):
|
||||
if method == "system.launch_dock_area":
|
||||
return self._launch_dock_area(*args, **kwargs)
|
||||
|
||||
@@ -34,6 +34,31 @@ class MockBECFigure:
|
||||
"""Remove a plot from the figure."""
|
||||
|
||||
|
||||
class MockContentWidget:
|
||||
USER_ACCESS = ["list_profiles", "mode", "mode.setter"]
|
||||
|
||||
def list_profiles(self) -> list[str]:
|
||||
"""List profiles."""
|
||||
return []
|
||||
|
||||
@property
|
||||
def mode(self) -> str:
|
||||
"""Current mode."""
|
||||
return "creator"
|
||||
|
||||
@mode.setter
|
||||
def mode(self, value: str) -> None:
|
||||
_ = value
|
||||
|
||||
|
||||
class MockViewWithContent:
|
||||
USER_ACCESS = ["activate"]
|
||||
RPC_CONTENT_CLASS = MockContentWidget
|
||||
|
||||
def activate(self):
|
||||
"""Activate view."""
|
||||
|
||||
|
||||
def test_client_generator_with_black_formatting():
|
||||
generator = ClientGenerator(base=True)
|
||||
container = BECClassContainer()
|
||||
@@ -228,6 +253,16 @@ def test_generate_content_for_class():
|
||||
assert "Test method" in generator.content
|
||||
|
||||
|
||||
def test_generate_content_for_class_uses_rpc_content_class_user_access():
|
||||
generator = ClientGenerator()
|
||||
generator.generate_content_for_class(MockViewWithContent)
|
||||
|
||||
assert "def activate(self):" in generator.content
|
||||
assert "def list_profiles(self) -> list[str]:" in generator.content
|
||||
assert "def mode(self) -> str:" in generator.content
|
||||
assert "@mode.setter" in generator.content
|
||||
|
||||
|
||||
def test_write_is_black_formatted(tmp_path):
|
||||
"""
|
||||
Test the write method of the ClientGenerator class.
|
||||
|
||||
@@ -4,6 +4,7 @@ from qtpy.QtWidgets import QWidget
|
||||
|
||||
from bec_widgets.applications.main_app import BECMainApp
|
||||
from bec_widgets.applications.views.view import ViewBase
|
||||
from bec_widgets.utils.bec_widget import BECWidget
|
||||
|
||||
from .client_mocks import mocked_client
|
||||
|
||||
@@ -133,7 +134,28 @@ def test_view_switch_method_switches_to_target(app_with_spies, qtbot):
|
||||
def test_view_content_widget_is_hidden_from_namespace(app_with_spies):
|
||||
app, _, _, _ = app_with_spies
|
||||
assert app.dock_area.content is app.dock_area.dock_area
|
||||
assert app.dock_area.content.skip_rpc_namespace is True
|
||||
|
||||
|
||||
def test_developer_plotting_area_parent_id_uses_view_namespace(app_with_spies):
|
||||
app, _, _, _ = app_with_spies
|
||||
plotting_area = app.developer_view.developer_widget.plotting_ads
|
||||
|
||||
assert plotting_area.parent_id == app.developer_view.gui_id
|
||||
|
||||
|
||||
def test_parent_id_ignores_plain_qwidget_between_connectors(qtbot, mocked_client):
|
||||
class RootConnector(BECWidget, QWidget):
|
||||
RPC = True
|
||||
|
||||
class ChildConnector(BECWidget, QWidget):
|
||||
RPC = True
|
||||
|
||||
root = RootConnector(client=mocked_client)
|
||||
qtbot.addWidget(root)
|
||||
spacer = QWidget(root)
|
||||
child = ChildConnector(parent=spacer, client=mocked_client)
|
||||
|
||||
assert child.parent_id == root.gui_id
|
||||
|
||||
|
||||
def test_guided_tour_is_initialized(app_with_spies):
|
||||
|
||||
@@ -109,6 +109,14 @@ def test_show_launcher_creates_launcher_when_missing(bec_main_window):
|
||||
assert bec_main_window._launcher_window is launcher
|
||||
|
||||
|
||||
def test_hidden_scan_progress_parent_blocks_children_namespace(bec_main_window):
|
||||
hidden_progress = bec_main_window._scan_progress_bar_full
|
||||
nested_progress = hidden_progress.progressbar
|
||||
|
||||
assert hidden_progress.rpc_exposed is False
|
||||
assert nested_progress.parent_id == hidden_progress.gui_id
|
||||
|
||||
|
||||
#################################################################
|
||||
# Tests for BECMainWindow Addons
|
||||
#################################################################
|
||||
|
||||
@@ -138,3 +138,36 @@ def test_serialize_result_and_send_max_delay_exceeded(rpc_server, qtbot, dummy_w
|
||||
assert args[1] is False # accepted=False
|
||||
assert "error" in args[2]
|
||||
assert "Max delay exceeded" in args[2]["error"]
|
||||
|
||||
|
||||
def test_run_rpc_delegates_to_rpc_content_class(rpc_server):
|
||||
class Content:
|
||||
USER_ACCESS = ["foo", "mode", "mode.setter"]
|
||||
|
||||
def __init__(self):
|
||||
self._mode = "initial"
|
||||
|
||||
def foo(self):
|
||||
return "ok"
|
||||
|
||||
@property
|
||||
def mode(self):
|
||||
return self._mode
|
||||
|
||||
@mode.setter
|
||||
def mode(self, value):
|
||||
self._mode = value
|
||||
|
||||
class View:
|
||||
RPC_CONTENT_CLASS = Content
|
||||
RPC_CONTENT_ATTR = "content"
|
||||
|
||||
def __init__(self):
|
||||
self.content = Content()
|
||||
|
||||
view = View()
|
||||
|
||||
assert rpc_server.run_rpc(view, "foo", [], {}) == "ok"
|
||||
assert rpc_server.run_rpc(view, "mode", [], {}) == "initial"
|
||||
assert rpc_server.run_rpc(view, "mode", ["creator"], {}) is None
|
||||
assert view.content.mode == "creator"
|
||||
|
||||
Reference in New Issue
Block a user