feat: move to lazy widget import

bec_widgets/utils/bec_plugin_helper.py

@@ -4,6 +4,7 @@ import importlib.metadata
 import inspect
 import pkgutil
 import traceback
+from functools import lru_cache
 from importlib import util as importlib_util
 from importlib.machinery import FileFinder, ModuleSpec, SourceFileLoader
 from types import ModuleType
@@ -11,7 +12,11 @@ from typing import Generator
 
 from bec_lib.logger import bec_logger
 
-from bec_widgets.utils.plugin_utils import BECClassContainer, BECClassInfo
+from bec_widgets.utils.plugin_utils import (
+    BECClassContainer,
+    BECClassInfo,
+    rpc_widget_registry_from_source,
+)
 
 logger = bec_logger.logger
 
@@ -53,6 +58,14 @@ def _submodule_by_name(module: ModuleType, name: str):
     return None
 
 
+def _submodule_spec_by_name(module: ModuleType, name: str) -> ModuleSpec | None:
+    for module_info in pkgutil.iter_modules(module.__path__):
+        if module_info.name != name or not isinstance(module_info.module_finder, FileFinder):
+            continue
+        return module_info.module_finder.find_spec(module_info.name)
+    return None
+
+
 def _get_widgets_from_module(module: ModuleType) -> BECClassContainer:
     """Find any BECWidget subclasses in the given module and return them with their info."""
     from bec_widgets.utils.bec_widget import BECWidget  # avoid circular import
@@ -90,16 +103,55 @@ def get_plugin_client_module() -> ModuleType | None:
     return _submodule_by_name(plugin, "client") if (plugin := user_widget_plugin()) else None
 
 
+def get_plugin_designer_module() -> ModuleType | None:
+    """If there is a plugin repository installed, return the designer module."""
+    return (
+        _submodule_by_name(plugin, "designer_plugins") if (plugin := user_widget_plugin()) else None
+    )
+
+
+@lru_cache
+def get_plugin_rpc_widget_registry() -> dict[str, tuple[str, str]]:
+    """If there is a plugin repository installed, return the RPC widget registry."""
+    plugin = user_widget_plugin()
+    if plugin is None:
+        return {}
+
+    client_spec = _submodule_spec_by_name(plugin, "client")
+    if client_spec is not None and client_spec.origin:
+        try:
+            return rpc_widget_registry_from_source(client_spec.origin)
+        except (OSError, SyntaxError) as exc:
+            logger.warning(f"Could not parse plugin RPC widget registry: {exc}")
+
+    client_module = get_plugin_client_module()
+    if client_module is None:
+        return {}
+    registry = {}
+    for plugin_name, plugin_class in inspect.getmembers(client_module, inspect.isclass):
+        if hasattr(plugin_class, "_IMPORT_MODULE"):
+            registry[plugin_name] = (plugin_class._IMPORT_MODULE, plugin_class.__name__)
+    return registry
+
+
+@lru_cache
+def get_plugin_designer_registry() -> dict[str, tuple[str, str]]:
+    """If there is a plugin repository installed, return the designer plugin registry."""
+    designer_module = get_plugin_designer_module()
+    if designer_module and hasattr(designer_module, "designer_plugins"):
+        return designer_module.designer_plugins
+    return {}
+
+
 def get_all_plugin_widgets() -> BECClassContainer:
     """If there is a plugin repository installed, load all widgets from it."""
     if plugin := user_widget_plugin():
         return _all_widgets_from_all_submods(plugin)
-    else:
-        return BECClassContainer()
+    return BECClassContainer()
 
 
 if __name__ == "__main__":  # pragma: no cover
-
+    widgets = get_plugin_rpc_widget_registry()
     client = get_plugin_client_module()
     print(get_all_plugin_widgets())
     ...

bec_widgets/utils/generate_cli.py

@@ -14,7 +14,11 @@ import isort
 from bec_lib.logger import bec_logger
 from qtpy.QtCore import Property as QtProperty
 
-from bec_widgets.utils.generate_designer_plugin import DesignerPluginGenerator, plugin_filenames
+from bec_widgets.utils.generate_designer_plugin import (
+    DesignerPluginGenerator,
+    DesignerPluginInfo,
+    plugin_filenames,
+)
 from bec_widgets.utils.plugin_utils import BECClassContainer, get_custom_classes
 
 logger = bec_logger.logger
@@ -250,6 +254,50 @@ class {class_name}(RPCBase):\n"""
             file.write(formatted_content)
 
 
+def write_designer_plugins(plugin_infos: list[DesignerPluginInfo], file_name: str):
+    """
+    Write a registry of Qt widget classes with designer plugins.
+
+    Args:
+        plugin_infos(list[DesignerPluginInfo]): The designer plugin metadata to write.
+        file_name(str): The name of the file to write to.
+    """
+    plugin_infos = sorted(plugin_infos, key=lambda info: info.plugin_name_pascal)
+    content = """# This file was automatically generated by generate_cli.py
+# type: ignore
+from __future__ import annotations
+
+# pylint: skip-file
+
+designer_plugins = {
+"""
+    for info in plugin_infos:
+        widget_module = info.plugin_class.__module__
+        widget_class = info.plugin_name_pascal
+        content += f'    "{info.plugin_name_pascal}": ("{widget_module}", "{widget_class}"),\n'
+
+    content += """
+}
+"""
+
+    try:
+        formatted_content = black.format_str(content, mode=black.Mode(line_length=100))
+    except black.NothingChanged:
+        formatted_content = content
+
+    config = isort.Config(
+        profile="black",
+        line_length=100,
+        multi_line_output=3,
+        include_trailing_comma=False,
+        known_first_party=["bec_widgets"],
+    )
+    formatted_content = isort.code(formatted_content, config=config)
+
+    with open(file_name, "w", encoding="utf-8") as file:
+        file.write(formatted_content)
+
+
 def main():
     """
     Main entry point for the script, controlled by command line arguments.
@@ -303,6 +351,8 @@ def main():
     else:
         non_overwrite_classes = []
 
+    designer_plugin_infos = []
+
     for cls in rpc_classes.plugins:
         logger.info(f"Writing bec-designer plugin files for {cls.__name__}...")
 
@@ -310,21 +360,30 @@ def main():
             logger.error(
                 f"Not writing plugin files for {cls.__name__} because a built-in widget with that name exists"
             )
+            continue
 
         plugin = DesignerPluginGenerator(cls)
-        if not hasattr(plugin, "info"):
+        if not hasattr(plugin, "info") or plugin.excluded:
             continue
 
         def _exists(file: str):
             return os.path.exists(os.path.join(plugin.info.base_path, file))
 
         if any(_exists(file) for file in plugin_filenames(plugin.info.plugin_name_snake)):
+            if _exists(plugin.filenames.plugin):
+                designer_plugin_infos.append(plugin.info)
             logger.debug(
                 f"Skipping generation of extra plugin files for {plugin.info.plugin_name_snake} - at least one file out of 'plugin.py', 'pyproject', and 'register_{plugin.info.plugin_name_snake}.py' already exists."
             )
             continue
 
         plugin.run()
+        designer_plugin_infos.append(plugin.info)
+
+    # Write designer_plugins.py with plugin import metadata for all widgets with designer plugins.
+    designer_plugins_path = module_dir / client_subdir / "designer_plugins.py"
+    logger.info(f"Generating designer plugin registry at {designer_plugins_path}")
+    write_designer_plugins(designer_plugin_infos, str(designer_plugins_path))
 
 
 if __name__ == "__main__":  # pragma: no cover

bec_widgets/utils/generate_designer_plugin.py

@@ -63,6 +63,10 @@ class DesignerPluginGenerator:
     def filenames(self):
         return plugin_filenames(self.info.plugin_name_snake)
 
+    @property
+    def excluded(self):
+        return self._excluded
+
     def run(self, validate=True):
         if self._excluded:
             print(f"Plugin {self.widget.__name__} is excluded from generation.")

bec_widgets/utils/plugin_utils.py

@@ -1,56 +1,22 @@
 from __future__ import annotations
 
+import ast
 import importlib
 import inspect
 import os
 from dataclasses import dataclass
+from functools import lru_cache
+from pathlib import Path
 from typing import TYPE_CHECKING, Iterable
 
-from bec_lib.plugin_helper import _get_available_plugins
-from qtpy.QtWidgets import QWidget
-
-from bec_widgets.utils.bec_connector import BECConnector
-from bec_widgets.utils.bec_widget import BECWidget
-
 if TYPE_CHECKING:  # pragma: no cover
+    from qtpy.QtWidgets import QWidget
+
+    from bec_widgets.utils.bec_connector import BECConnector
+    from bec_widgets.utils.bec_widget import BECWidget
     from bec_widgets.widgets.containers.auto_update.auto_updates import AutoUpdates
 
 
-def get_plugin_widgets() -> dict[str, BECConnector]:
-    """
-    Get all available widgets from the plugin directory. Widgets are classes that inherit from BECConnector.
-    The plugins are provided through python plugins and specified in the respective pyproject.toml file using
-    the following key:
-
-        [project.entry-points."bec.widgets.user_widgets"]
-        plugin_widgets = "path.to.plugin.module"
-
-    e.g.
-        [project.entry-points."bec.widgets.user_widgets"]
-        plugin_widgets = "pxiii_bec.bec_widgets.widgets"
-
-        assuming that the widgets module for the package pxiii_bec is located at pxiii_bec/bec_widgets/widgets and
-        contains the widgets to be loaded within the pxiii_bec/bec_widgets/widgets/__init__.py file.
-
-    Returns:
-        dict[str, BECConnector]: A dictionary of widget names and their respective classes.
-    """
-    modules = _get_available_plugins("bec.widgets.user_widgets")
-    loaded_plugins = {}
-    print(modules)
-    for module in modules:
-        mods = inspect.getmembers(module, predicate=_filter_plugins)
-        for name, mod_cls in mods:
-            if name in loaded_plugins:
-                print(f"Duplicated widgets plugin {name}.")
-            loaded_plugins[name] = mod_cls
-    return loaded_plugins
-
-
-def _filter_plugins(obj):
-    return inspect.isclass(obj) and issubclass(obj, BECConnector)
-
-
 def get_plugin_auto_updates() -> dict[str, type[AutoUpdates]]:
     """
     Get all available auto update classes from the plugin directory. AutoUpdates must inherit from AutoUpdate and be
@@ -66,6 +32,8 @@ def get_plugin_auto_updates() -> dict[str, type[AutoUpdates]]:
     Returns:
         dict[str, AutoUpdates]: A dictionary of widget names and their respective classes.
     """
+    from bec_lib.plugin_helper import _get_available_plugins
+
     modules = _get_available_plugins("bec.widgets.auto_updates")
     loaded_plugins = {}
     for module in modules:
@@ -168,6 +136,11 @@ class BECClassContainer:
 
 def _collect_classes_from_package(repo_name: str, package: str) -> BECClassContainer:
     """Collect classes from a package subtree (for example ``widgets`` or ``applications``)."""
+    from qtpy.QtWidgets import QWidget
+
+    from bec_widgets.utils.bec_connector import BECConnector
+    from bec_widgets.utils.bec_widget import BECWidget
+
     collection = BECClassContainer()
     try:
         anchor_module = importlib.import_module(f"{repo_name}.{package}")
@@ -194,17 +167,18 @@ def _collect_classes_from_package(repo_name: str, package: str) -> BECClassConta
 
             for name in dir(module):
                 obj = getattr(module, name)
+                if not isinstance(obj, type):
+                    continue
                 if not hasattr(obj, "__module__") or obj.__module__ != module.__name__:
                     continue
-                if isinstance(obj, type):
-                    class_info = BECClassInfo(name=name, module=module.__name__, file=path, obj=obj)
-                    if issubclass(obj, BECConnector):
-                        class_info.is_connector = True
-                    if issubclass(obj, QWidget) or issubclass(obj, BECWidget):
-                        class_info.is_widget = True
-                    if hasattr(obj, "PLUGIN") and obj.PLUGIN:
-                        class_info.is_plugin = True
-                    collection.add_class(class_info)
+                class_info = BECClassInfo(name=name, module=module.__name__, file=path, obj=obj)
+                if issubclass(obj, BECConnector):
+                    class_info.is_connector = True
+                if issubclass(obj, QWidget) or issubclass(obj, BECWidget):
+                    class_info.is_widget = True
+                if hasattr(obj, "PLUGIN") and obj.PLUGIN:
+                    class_info.is_plugin = True
+                collection.add_class(class_info)
     return collection
 
 
@@ -229,3 +203,89 @@ def get_custom_classes(
     for package in selected_packages:
         collection += _collect_classes_from_package(repo_name, package)
     return collection
+
+
+def _get_designer_registry() -> dict[str, tuple[str, str]]:
+    from bec_widgets.cli.designer_plugins import designer_plugins
+
+    return designer_plugins
+
+
+def _resolve_widget_from_registry(import_path: str, widget_name: str) -> type[QWidget]:
+    widget = importlib.import_module(import_path)
+    return getattr(widget, widget_name)
+
+
+def designer_plugin_exists(name: str) -> bool:
+    from bec_widgets.utils.bec_plugin_helper import get_plugin_designer_registry
+
+    internal_registry = _get_designer_registry()
+    external_registry = get_plugin_designer_registry()
+    return name in internal_registry or name in external_registry
+
+
+def get_designer_plugin(name: str, raise_on_missing: bool = True) -> type[QWidget] | None:
+    from bec_widgets.utils.bec_plugin_helper import get_plugin_designer_registry
+
+    internal_registry = _get_designer_registry()
+    external_registry = get_plugin_designer_registry()
+    if name in external_registry:
+        import_path, widget_name = external_registry[name]
+        return _resolve_widget_from_registry(import_path, widget_name)
+    if name in internal_registry:
+        import_path, widget_name = internal_registry[name]
+        return _resolve_widget_from_registry(import_path, widget_name)
+
+    if raise_on_missing:
+        raise ValueError(
+            f"Designer plugin {name} not found in either internal or external registry."
+        )
+    return None
+
+
+def rpc_widget_registry_from_source(path: str | Path) -> dict[str, tuple[str, str]]:
+    """Parse a generated RPC client module and return its widget registry."""
+    source_path = Path(path)
+    module_node = ast.parse(source_path.read_text(encoding="utf-8"), filename=str(source_path))
+    registry = {}
+    for node in module_node.body:
+        if not isinstance(node, ast.ClassDef):
+            continue
+        for item in node.body:
+            if not isinstance(item, ast.Assign):
+                continue
+            if not any(
+                isinstance(target, ast.Name) and target.id == "_IMPORT_MODULE"
+                for target in item.targets
+            ):
+                continue
+            if isinstance(item.value, ast.Constant) and isinstance(item.value.value, str):
+                registry[node.name] = (item.value.value, node.name)
+            break
+    return registry
+
+
+@lru_cache
+def get_rpc_widget_registry() -> dict[str, tuple[str, str]]:
+    client_path = Path(__file__).resolve().parents[1] / "cli" / "client.py"
+    return rpc_widget_registry_from_source(client_path)
+
+
+@lru_cache
+def rpc_widget_registry() -> dict[str, tuple[str, str]]:
+    from bec_widgets.utils.bec_plugin_helper import get_plugin_rpc_widget_registry
+
+    internal_registry = get_rpc_widget_registry()
+    external_registry = get_plugin_rpc_widget_registry()
+    return {**external_registry, **internal_registry}
+
+
+def get_rpc_widget(name: str, raise_on_missing: bool = True) -> type[QWidget] | None:
+    registry = rpc_widget_registry()
+    if name in registry:
+        import_path, widget_name = registry[name]
+        return _resolve_widget_from_registry(import_path, widget_name)
+
+    if raise_on_missing:
+        raise ValueError(f"RPC widget {name} not found in registry.")
+    return None

bec_widgets/utils/rpc_widget_handler.py

@@ -1,42 +1,34 @@
 from __future__ import annotations
 
-from bec_widgets.cli.client_utils import IGNORE_WIDGETS
-from bec_widgets.utils.bec_plugin_helper import get_all_plugin_widgets
-from bec_widgets.utils.bec_widget import BECWidget
-from bec_widgets.utils.plugin_utils import get_custom_classes
+from typing import TYPE_CHECKING
+
+from bec_widgets.utils.plugin_utils import get_rpc_widget, rpc_widget_registry
+
+if TYPE_CHECKING:  # pragma: no cover
+    from bec_widgets.utils.bec_widget import BECWidget
 
 
 class RPCWidgetHandler:
     """Handler class for creating widgets from RPC messages."""
 
     def __init__(self):
-        self._widget_classes = None
+        self._widget_registry = None
 
     @property
-    def widget_classes(self) -> dict[str, type[BECWidget]]:
+    def widget_classes(self) -> dict[str, tuple[str, str]]:
         """
         Get the available widget classes.
 
         Returns:
             dict: The available widget classes.
         """
-        if self._widget_classes is None:
-            self.update_available_widgets()
-        return self._widget_classes  # type: ignore
+        registry = rpc_widget_registry()
+        if not registry:
+            return {}
+        return registry
 
-    def update_available_widgets(self):
-        """
-        Update the available widgets.
-
-        Returns:
-            None
-        """
-        self._widget_classes = (
-            get_custom_classes("bec_widgets", packages=("widgets", "applications"))
-            + get_all_plugin_widgets()
-        ).as_dict(IGNORE_WIDGETS)
-
-    def create_widget(self, widget_type, **kwargs) -> BECWidget:
+    @staticmethod
+    def create_widget(widget_type, **kwargs) -> BECWidget:
         """
         Create a widget from an RPC message.
 
@@ -48,9 +40,9 @@ class RPCWidgetHandler:
         Returns:
             widget(BECWidget): The created widget.
         """
-        widget_class = self.widget_classes.get(widget_type)  # type: ignore
-        if widget_class:
-            return widget_class(**kwargs)
+        widget = get_rpc_widget(widget_type, raise_on_missing=False)
+        if widget:
+            return widget(**kwargs)
         raise ValueError(f"Unknown widget type: {widget_type}")
 
 

bec_widgets/utils/ui_loader.py

@@ -1,10 +1,8 @@
 from bec_lib.logger import bec_logger
-from qtpy import PYQT6, PYSIDE6
+from qtpy import PYSIDE6
 from qtpy.QtCore import QFile, QIODevice
 
-from bec_widgets.utils.bec_plugin_helper import get_all_plugin_widgets
-from bec_widgets.utils.generate_designer_plugin import DesignerPluginInfo
-from bec_widgets.utils.plugin_utils import get_custom_classes
+from bec_widgets.utils.plugin_utils import get_designer_plugin
 
 logger = bec_logger.logger
 
@@ -12,16 +10,14 @@ if PYSIDE6:
     from qtpy.QtUiTools import QUiLoader
 
     class CustomUiLoader(QUiLoader):
-        def __init__(self, baseinstance, custom_widgets: dict | None = None):
+        def __init__(self, baseinstance):
             super().__init__(baseinstance)
-            self.custom_widgets = custom_widgets or {}
-
             self.baseinstance = baseinstance
 
         def createWidget(self, class_name, parent=None, name=""):
-            if class_name in self.custom_widgets:
-                widget = self.custom_widgets[class_name](self.baseinstance)
-                return widget
+            widget = get_designer_plugin(class_name, raise_on_missing=False)
+            if widget is not None:
+                return widget(self.baseinstance)
             return super().createWidget(class_name, self.baseinstance, name)
 
 
@@ -31,16 +27,9 @@ class UILoader:
     def __init__(self, parent=None):
         self.parent = parent
 
-        self.custom_widgets = (
-            get_custom_classes("bec_widgets") + get_all_plugin_widgets()
-        ).as_dict()
-
-        if PYSIDE6:
-            self.loader = self.load_ui_pyside6
-        elif PYQT6:
-            self.loader = self.load_ui_pyqt6
-        else:
+        if not PYSIDE6:
             raise ImportError("No compatible Qt bindings found.")
+        self.loader = self.load_ui_pyside6
 
     def load_ui_pyside6(self, ui_file, parent=None):
         """
@@ -53,7 +42,7 @@ class UILoader:
             QWidget: The loaded widget.
         """
         parent = parent or self.parent
-        loader = CustomUiLoader(parent, self.custom_widgets)
+        loader = CustomUiLoader(parent)
         file = QFile(ui_file)
         if not file.open(QIODevice.ReadOnly):
             raise IOError(f"Cannot open file: {ui_file}")
@@ -61,71 +50,6 @@ class UILoader:
         file.close()
         return widget
 
-    def load_ui_pyqt6(self, ui_file, parent=None):
-        """
-        Specific loader for PyQt6 using loadUi.
-        Args:
-            ui_file(str): Path to the .ui file.
-            parent(QWidget): Parent widget.
-
-        Returns:
-            QWidget: The loaded widget.
-        """
-        from PyQt6.uic.Loader.loader import DynamicUILoader
-
-        class CustomDynamicUILoader(DynamicUILoader):
-            def __init__(self, package, custom_widgets: dict = None):
-                super().__init__(package)
-                self.custom_widgets = custom_widgets or {}
-
-            def _handle_custom_widgets(self, el):
-                """Handle the <customwidgets> element."""
-
-                def header2module(header):
-                    """header2module(header) -> string
-
-                    Convert paths to C++ header files to according Python modules
-                    >>> header2module("foo/bar/baz.h")
-                    'foo.bar.baz'
-                    """
-
-                    if header.endswith(".h"):
-                        header = header[:-2]
-
-                    mpath = []
-                    for part in header.split("/"):
-                        # Ignore any empty parts or those that refer to the current
-                        # directory.
-                        if part not in ("", "."):
-                            if part == "..":
-                                # We should allow this for Python3.
-                                raise SyntaxError(
-                                    "custom widget header file name may not contain '..'."
-                                )
-
-                            mpath.append(part)
-
-                    return ".".join(mpath)
-
-                for custom_widget in el:
-                    classname = custom_widget.findtext("class")
-                    header = custom_widget.findtext("header")
-                    if header:
-                        header = self._translate_bec_widgets_header(header)
-                    self.factory.addCustomWidget(
-                        classname,
-                        custom_widget.findtext("extends") or "QWidget",
-                        header2module(header),
-                    )
-
-            def _translate_bec_widgets_header(self, header):
-                for name, value in self.custom_widgets.items():
-                    if header == DesignerPluginInfo.pascal_to_snake(name):
-                        return value.__module__
-                return header
-
-        return CustomDynamicUILoader("", self.custom_widgets).loadUi(ui_file, parent)
-
     def load_ui(self, ui_file, parent=None):
         """
         Universal UI loader method.