bec_widgets.utils.bec_connector#

Attributes#

BECDispatcher

Classes#

BECConnector

Connection mixin class to handle BEC client and device manager

ConnectionConfig

Configuration for BECConnector mixin class

Worker

Worker class to run a function in a separate thread.

WorkerSignals

Module Contents#

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.

Parameters:

  • client (BECClient, optional) – The BEC client.

  • config (ConnectionConfig, optional) – The connection configuration with specific gui id.

  • gui_id (str, optional) – The GUI ID.

  • object_name (str, optional) – The object name.

  • root_widget (bool, optional) – 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.

  • rpc_exposed (bool, optional) – If set to False, this instance is excluded from RPC registry broadcast and CLI namespace discovery.

  • rpc_passthrough_children (bool, optional) – 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.

  • **kwargs

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

_get_all_rpc() → dict#

Get all registered RPC objects.

_get_bec_meta_objects() → dict#

Get BEC meta objects for the widget.

Returns:

BEC meta objects.

Return type:

dict

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

_set_gui_id(gui_id: str) → None#

Set the GUI ID for the widget.

Parameters:

gui_id (str) – GUI ID.

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

apply_config(config: dict, generate_new_id: bool = True) → None#

Apply the configuration to the widget.

Parameters:

  • config (dict) – Configuration settings.

  • generate_new_id (bool) – If True, generate a new GUI ID for the widget.

change_object_name(name: str) → None#

Change the object name of the widget. Unregister old name and register the new one.

Parameters:

name (str) – The new object name.

export_settings() → dict#

Export the settings of the widget as dict.

Returns:

The exported settings of the widget.

Return type:

dict

get_bec_shortcuts()#

Get BEC shortcuts for the widget.

get_config(dict_output: bool = True) → dict | pydantic.BaseModel#

Get the configuration of the widget.

Parameters:

dict_output (bool) – If True, return the configuration as a dictionary. If False, return the configuration as a pydantic model.

Returns:

The configuration of the widget.

Return type:

dict | BaseModel

get_obj_by_id(obj_id: str)#
load_config(path: str | None = None, gui: bool = False)#

Load the configuration of the widget from YAML.

Parameters:

  • path (str | None) – Path to the configuration file for non-GUI dialog mode.

  • gui (bool) – If True, use the GUI dialog to load the configuration file.

load_settings(settings: dict) → None#

Load the settings of the widget from dict.

Parameters:

settings (dict) – The settings to load into the widget.

on_config_update(config: ConnectionConfig | dict) → None#

Update the configuration for the widget.

Parameters:

config (ConnectionConfig | dict) – Configuration settings.

remove()#

Cleanup the BECConnector

save_config(path: str | None = None, gui: bool = False)#

Save the configuration of the widget to YAML.

Parameters:

  • path (str | None) – Path to save the configuration file for non-GUI dialog mode.

  • gui (bool) – If True, use the GUI dialog to save the configuration file.

setObjectName(name: str) → None#

Set the object name of the widget.

Parameters:

name (str) – The new object name.

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.

Parameters:

  • fn – Function to run in a separate thread.

  • *args – Arguments for the function.

  • on_complete – Slot to run when the task is complete.

  • **kwargs – Keyword arguments for the function.

Returns:

The worker object that will run the task.

Return type:

worker

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)
update_client(client) → None#

Update the client and device manager from BEC and create object for BEC shortcuts.

Parameters:

client – BEC client.

EXIT_HANDLERS#
USER_ACCESS = ['_config_dict', '_get_all_rpc', '_rpc_id']#
property _config_dict: dict#

Get the configuration of the widget.

Returns:

The configuration of the widget.

Return type:

dict

property _rpc_id: str#

Get the RPC ID of the widget.

bec_dispatcher#
client#
error_utility = None#
name_established#
object_name#
property parent_id: str | None#
root_widget = False#
rpc_exposed = True#
rpc_passthrough_children = True#
rpc_register#
widget_removed#
class ConnectionConfig(/, **data: Any)#

Bases: 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.

classmethod generate_gui_id(v, values)#

Generate a GUI ID if none is provided.

gui_id: str | None = None#
model_config: dict#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

widget_class: str = None#
class Worker(func, *args, **kwargs)#

Bases: qtpy.QtCore.QRunnable

Worker class to run a function in a separate thread.

run()#

Run the specified function in the thread.

args = ()#
func#
kwargs#
signals#
class WorkerSignals#

Bases: qtpy.QtCore.QObject

completed#
progress#
BECDispatcher#