1
0
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:
2026-03-03 13:26:29 +01:00
committed by Jan Wyzula
parent a7ab8f55f2
commit 59408d6fab
9 changed files with 425 additions and 12 deletions

View File

@@ -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)

View File

@@ -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,

View File

@@ -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"""

View File

@@ -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.

View File

@@ -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)

View File

@@ -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.

View File

@@ -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):

View File

@@ -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
#################################################################

View File

@@ -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"