bec_widgets.utils.bec_connector =============================== .. py:module:: bec_widgets.utils.bec_connector Attributes ---------- .. autoapisummary:: bec_widgets.utils.bec_connector.BECDispatcher Classes ------- .. autoapisummary:: bec_widgets.utils.bec_connector.BECConnector bec_widgets.utils.bec_connector.ConnectionConfig bec_widgets.utils.bec_connector.Worker bec_widgets.utils.bec_connector.WorkerSignals Module Contents --------------- .. py:class:: BECConnector(client=None, config: ConnectionConfig | None = None, gui_id: str | None = None, object_name: str | None = None, root_widget: bool = False, rpc_exposed: bool = True, rpc_passthrough_children: bool = True, **kwargs) Connection mixin class to handle BEC client and device manager BECConnector mixin class to handle BEC client and device manager. :param client: The BEC client. :type client: BECClient, optional :param config: The connection configuration with specific gui id. :type config: ConnectionConfig, optional :param gui_id: The GUI ID. :type gui_id: str, optional :param object_name: The object name. :type object_name: str, optional :param root_widget: If set to True, the parent_id will be always set to None, thus enforcing that the widget is accessible as a root widget of the BECGuiClient object. :type root_widget: bool, optional :param rpc_exposed: If set to False, this instance is excluded from RPC registry broadcast and CLI namespace discovery. :type rpc_exposed: bool, optional :param rpc_passthrough_children: Only relevant when ``rpc_exposed=False``. If True, RPC-visible children rebind to the next visible ancestor. If False (default), children stay hidden behind this widget. :type rpc_passthrough_children: bool, optional :param \*\*kwargs: .. py:method:: _enforce_unique_sibling_name() Enforce that this BECConnector has a unique objectName among its siblings. Sibling logic: - If there's a nearest BECConnector parent, only compare with children of that parent. - If parent is None (i.e., top-level object), compare with all other top-level BECConnectors. .. py:method:: _get_all_rpc() -> dict Get all registered RPC objects. .. py:method:: _get_bec_meta_objects() -> dict Get BEC meta objects for the widget. :returns: BEC meta objects. :rtype: dict .. py:method:: _get_rpc_parent_ancestor() -> BECConnector | None Find the nearest ancestor that is RPC-addressable. Rules: - If an ancestor has ``rpc_exposed=False``, it is an explicit visibility boundary unless ``rpc_passthrough_children=True``. - If an ancestor has ``RPC=False`` (but remains rpc_exposed), it is treated as structural and children continue to the next ancestor. - Lookup always happens through ``WidgetHierarchy.get_becwidget_ancestor`` so plain ``QWidget`` nodes between connectors are ignored. .. py:method:: _set_gui_id(gui_id: str) -> None Set the GUI ID for the widget. :param gui_id: GUI ID. :type gui_id: str .. py:method:: _update_object_name() -> None Enforce a unique object name among siblings and register the object for RPC. This method is called through a single shot timer kicked off in the constructor. .. py:method:: apply_config(config: dict, generate_new_id: bool = True) -> None Apply the configuration to the widget. :param config: Configuration settings. :type config: dict :param generate_new_id: If True, generate a new GUI ID for the widget. :type generate_new_id: bool .. py:method:: change_object_name(name: str) -> None Change the object name of the widget. Unregister old name and register the new one. :param name: The new object name. :type name: str .. py:method:: export_settings() -> dict Export the settings of the widget as dict. :returns: The exported settings of the widget. :rtype: dict .. py:method:: get_bec_shortcuts() Get BEC shortcuts for the widget. .. py:method:: get_config(dict_output: bool = True) -> dict | pydantic.BaseModel Get the configuration of the widget. :param dict_output: If True, return the configuration as a dictionary. If False, return the configuration as a pydantic model. :type dict_output: bool :returns: The configuration of the widget. :rtype: dict | BaseModel .. py:method:: get_obj_by_id(obj_id: str) .. py:method:: load_config(path: str | None = None, gui: bool = False) Load the configuration of the widget from YAML. :param path: Path to the configuration file for non-GUI dialog mode. :type path: str | None :param gui: If True, use the GUI dialog to load the configuration file. :type gui: bool .. py:method:: load_settings(settings: dict) -> None Load the settings of the widget from dict. :param settings: The settings to load into the widget. :type settings: dict .. py:method:: on_config_update(config: ConnectionConfig | dict) -> None Update the configuration for the widget. :param config: Configuration settings. :type config: ConnectionConfig | dict .. py:method:: remove() Cleanup the BECConnector .. py:method:: save_config(path: str | None = None, gui: bool = False) Save the configuration of the widget to YAML. :param path: Path to save the configuration file for non-GUI dialog mode. :type path: str | None :param gui: If True, use the GUI dialog to save the configuration file. :type gui: bool .. py:method:: setObjectName(name: str) -> None Set the object name of the widget. :param name: The new object name. :type name: str .. py:method:: submit_task(fn, *args, on_complete: bec_widgets.utils.error_popups.SafeSlot = None, **kwargs) -> Worker Submit a task to run in a separate thread. The task will run the specified function with the provided arguments and emit the completed signal when done. Use this method if you want to wait for a task to complete without blocking the main thread. :param fn: Function to run in a separate thread. :param \*args: Arguments for the function. :param on_complete: Slot to run when the task is complete. :param \*\*kwargs: Keyword arguments for the function. :returns: The worker object that will run the task. :rtype: worker .. rubric:: Examples >>> def my_function(a, b): >>> print(a + b) >>> self.submit_task(my_function, 1, 2) >>> def my_function(a, b): >>> print(a + b) >>> def on_complete(): >>> print("Task complete") >>> self.submit_task(my_function, 1, 2, on_complete=on_complete) .. py:method:: update_client(client) -> None Update the client and device manager from BEC and create object for BEC shortcuts. :param client: BEC client. .. py:attribute:: EXIT_HANDLERS .. py:attribute:: USER_ACCESS :value: ['_config_dict', '_get_all_rpc', '_rpc_id'] .. py:property:: _config_dict :type: dict Get the configuration of the widget. :returns: The configuration of the widget. :rtype: dict .. py:property:: _rpc_id :type: str Get the RPC ID of the widget. .. py:attribute:: bec_dispatcher .. py:attribute:: client .. py:attribute:: error_utility :value: None .. py:attribute:: name_established .. py:attribute:: object_name .. py:property:: parent_id :type: str | None .. py:attribute:: root_widget :value: False .. py:attribute:: rpc_exposed :value: True .. py:attribute:: rpc_passthrough_children :value: True .. py:attribute:: rpc_register .. py:attribute:: widget_removed .. py:class:: ConnectionConfig(/, **data: Any) Bases: :py:obj:`pydantic.BaseModel` Configuration for BECConnector mixin class Create a new model by parsing and validating input data from keyword arguments. Raises [`ValidationError`][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model. `self` is explicitly positional-only to allow `self` as a field name. .. py:method:: generate_gui_id(v, values) :classmethod: Generate a GUI ID if none is provided. .. py:attribute:: gui_id :type: Optional[str] :value: None .. py:attribute:: model_config :type: dict Configuration for the model, should be a dictionary conforming to [`ConfigDict`][pydantic.config.ConfigDict]. .. py:attribute:: widget_class :type: str :value: None .. py:class:: Worker(func, *args, **kwargs) Bases: :py:obj:`qtpy.QtCore.QRunnable` Worker class to run a function in a separate thread. .. py:method:: run() Run the specified function in the thread. .. py:attribute:: args :value: () .. py:attribute:: func .. py:attribute:: kwargs .. py:attribute:: signals .. py:class:: WorkerSignals Bases: :py:obj:`qtpy.QtCore.QObject` .. py:attribute:: completed .. py:attribute:: progress .. py:data:: BECDispatcher