bec_widgets.widgets.control.device_manager#

Submodules#

Classes#

DMConfigView

Widget to show the config of a selected device in YAML format.

DeviceTable

Custom table to display device configurations.

DocstringView

OphydValidation

Widget to manage and run ophyd device tests.

Package Contents#

class DMConfigView(parent=None)#

Bases: qtpy.QtWidgets.QWidget

Widget to show the config of a selected device in YAML format.

_customize_monaco()#

Customize the Monaco editor for YAML display.

_customize_overlay()#

Customize the overlay widget.

on_select_config(device: list[dict])#

Handle selection of a device from the device table. If more than one device is selected, show an overlay message. Otherwise, display the device config in YAML format.

Parameters:

device (list[dict]) – The selected device configuration.

RPC = False#
monaco_editor#
stacked_layout#
class DeviceTable(parent: qtpy.QtWidgets.QWidget | None = None, client=None)#

Bases: bec_widgets.utils.bec_widget.BECWidget, qtpy.QtWidgets.QWidget

Custom table to display device configurations.

Base class for all BEC widgets. This class should be used as a mixin class for all BEC widgets, e.g.:

>>> class MyWidget(BECWidget, QWidget):
>>>     def __init__(self, parent=None, client=None, config=None, gui_id=None):
>>>         super().__init__(parent=parent, client=client, config=config, gui_id=gui_id)
Parameters:
  • client (BECClient, optional) – The BEC client.

  • config (ConnectionConfig, optional) – The connection configuration.

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

  • theme_update (bool, optional) – Whether to subscribe to theme updates. Defaults to False. When set to True, the widget’s apply_theme method will be called when the theme changes.

__on_client_device_update_event(action: bec_lib.messages.ConfigAction, config: dict[str, dict[str, Any]]) None#

Handle DEVICE_UPDATE events from the BECClient.

__update_device_row_data(row: int, data: dict)#

Update an existing device row with new data.

Parameters:
  • row (int) – The row index to update.

  • data (dict) – The device data to populate the row.

_add_row(data: dict, config_status: bec_widgets.widgets.control.device_manager.components.ophyd_validation.ConfigStatus | int, connection_status: bec_widgets.widgets.control.device_manager.components.ophyd_validation.ConnectionStatus | int)#

Adds a new row at the bottom and populates it with data. The row widgets are stored in self.row_widgets for easy access. Consider to disable sorting when adding rows as this method is not responsible for maintaining sort order.

Parameters:
  • data (dict) – The device data to populate the row.

  • config_status (ConfigStatus | int) – The configuration validation status.

  • connection_status (ConnectionStatus | int) – The connection status.

_apply_row_filter(text_input: str)#

Apply a filter to the table rows based on the filter text.

_clear_table()#

Remove all rows.

_compare_configs(cfg1: dict, cfg2: dict) bool#

Compare two device configurations for equality.

_compare_device_configs(config1: dict, config2: dict) bool#

Compare two device configurations through the Device model in bec_lib.atlas_models.

Parameters:
  • config1 (dict) – The first device configuration.

  • config2 (dict) – The second device configuration.

Returns:

True if the configurations are equivalent, False otherwise.

Return type:

bool

_find_row_by_name(name: str) int | None#

Find a row by device name.

Parameters:

name (str) – The name of the device to find.

Returns:

The row index if found, else None.

Return type:

int | None

_get_cell_data(row: int, column: int) str | bool | None#

Get the data from a specific cell.

Parameters:
  • row (int) – The row index.

  • column (int) – The column index.

_get_overlapping_configs() list[dict[str, Any]]#

Get the device configs that overlap between the table and the config in the current running BEC session. A device will be ignored if it is disabled in the BEC session.

Parameters:

device_configs (Iterable[dict[str, Any]]) – The device configs to check.

Returns:

The list of overlapping device configs.

Return type:

list[dict[str, Any]]

_is_config_in_sync_with_redis()#

Check if the current config is in sync with Redis.

_is_device_in_redis_session(device_name: str, device_config: dict) bool#

Check if a device is in the running section.

_on_cell_double_clicked(row: int, column: int)#

Handle double-click events on table cells.

_on_device_config_update() None#

Handle device configuration updates from the BECClient.

_on_device_row_data_changed(data: dict)#

Handle data change events from device rows.

_on_selection_changed(selected: qtpy.QtCore.QItemSelection, deselected: qtpy.QtCore.QItemSelection)#

Handle selection changes in the table.

_on_table_checkbox_clicked(row: int, column: int, checked: bool)#

Handle checkbox clicks in the table.

_populate_device_row_cells(row: int, device_row: bec_widgets.widgets.control.device_manager.components.device_table.device_table_row.DeviceTableRow)#

Populate the cells of a given row with the widgets from the DeviceTableRow.

_remove_configs_dialog(device_names: list[str]) bool#

Prompt the user to confirm removal of rows and remove them from the model if accepted.

Parameters:

device_names (list[str]) – List of device names to be removed.

Returns:

True if the user confirmed removal, False otherwise.

Return type:

bool

_remove_rows_by_name(device_names: list[str])#

Remove a row by device name.

Parameters:

device_name (str) – The name of the device to remove.

_resize_table_policy(table: qtpy.QtWidgets.QTableWidget, enable: bool)#

Enable or disable column resizing.

_set_table_signals_on_hold(table: qtpy.QtWidgets.QTableWidget, enable: bool)#

Enable or disable table signals.

Create components related to the search functionality

_setup_table()#

Initializes the table configuration and headers.

Handle state changes for the fuzzy search toggle.

_update_device_row_status(row: int, config_status: int, connection_status: int) bool#

Update an existing device row’s validation status.

Parameters:
  • device_name (str) – The name of the device.

  • config_status (int) – The configuration validation status.

  • connection_status (int) – The connection status.

_update_row(data: dict) int | None#

Update an existing row with new data.

Parameters:

data (dict) – The device data to populate the row.

Returns:

The row index if updated, else None.

Return type:

int | None

add_device_configs(device_configs: _DeviceCfgIter, skip_validation: bool = False)#

Add devices to the config. If a device already exists, it will be replaced. If the validation is skipped, the device will be added with UNKNOWN state to the table and has to be manually adjusted by the user later on.

Parameters:
  • device_configs (Iterable[dict[str, Any]]) – The device configs to add.

  • skip_validation (bool) – Whether to skip validation for the added devices.

cleanup()#

Cleanup resources.

clear_device_configs()#

Clear the device configs. Skips validation by default.

eventFilter(source: qtpy.QtCore.QObject, event: qtpy.QtCore.QEvent) bool#

Customize event filtering for table interactions.

get_device_config() list[dict]#

Get the current device configurations in the table.

Returns:

The list of device configurations.

Return type:

list[dict]

get_selected_device_configs() list[dict]#

Get the currently selected device configurations in the table.

Returns:

The list of selected device configurations.

Return type:

list[dict]

get_validation_results() dict[str, Tuple[dict, int, int]]#

Get the current device validation results in the table.

Returns:

Dictionary mapping of device name to

(device config, config status, connection status).

Return type:

dict[str, Tuple[dict, int, int]]

remove_device(device_name: str)#

Remove a device from the config.

Parameters:

device_name (str) – The name of the device to remove.

remove_device_configs(device_configs: _DeviceCfgIter)#

Remove devices from the config.

Parameters:

device_configs (dict[str, dict]) – The device configs to remove.

set_device_config(device_configs: _DeviceCfgIter, skip_validation: bool = False)#

Set the device config. This will clear any existing configs.

Parameters:
  • device_configs (Iterable[dict[str, Any]]) – The device configs to set.

  • skip_validation (bool) – Whether to skip validation for the set devices.

update_device_configs(device_configs: _DeviceCfgIter, skip_validation: bool = False)#

Update devices in the config. If a device does not exist, it will be added.

Parameters:
  • device_configs (Iterable[dict[str, Any]]) – The device configs to update.

  • skip_validation (bool) – Whether to skip validation for the updated devices.

update_device_validation(device_config: dict, config_status: int, connection_status: int, validation_msg: str) None#

Update the validation status of a device. If multiple devices are being updated in a batch, consider using the update_multiple_device_validations method instead for better performance.

Args:

update_multiple_device_validations(validation_results: _ValidationResultIter)#

Slot to update multiple device validation statuses. This is recommended and more efficient than updating individual device validation statuses which may affect the performance of the UI when many devices are being updated in quick succession.

Parameters:

device_configs (Iterable[dict[str, Any]]) – The device configs to update.

RPC = False#
client_callback_id = 0#
device_config_in_sync_with_redis#
device_configs_changed#
device_row_dbl_clicked#
headers_key_map: dict[str, str]#
request_update_after_client_device_update#
request_update_multiple_device_validations#
row_data: dict[str, bec_widgets.widgets.control.device_manager.components.device_table.device_table_row.DeviceTableRow]#
selected_devices#
table#
table_sort_on_hold#
class DocstringView(parent: qtpy.QtWidgets.QWidget | None = None)#

Bases: qtpy.QtWidgets.QTextEdit

_set_text(text: str)#
on_select_config(device: list[dict])#
set_device_class(device_class_str: str) None#
class OphydValidation(parent=None, client=None, hide_legend: bool = False)#

Bases: bec_widgets.utils.bec_widget.BECWidget, qtpy.QtWidgets.QWidget

Widget to manage and run ophyd device tests.

Parameters:
  • parent (QWidget, optional) – Parent widget. Defaults to None.

  • client (BECClient, optional) – BEC client instance. Defaults to None.

  • hide_legend (bool, optional) – Whether to hide the legend. Defaults to False.

Base class for all BEC widgets. This class should be used as a mixin class for all BEC widgets, e.g.:

>>> class MyWidget(BECWidget, QWidget):
>>>     def __init__(self, parent=None, client=None, config=None, gui_id=None):
>>>         super().__init__(parent=parent, client=client, config=config, gui_id=gui_id)
Parameters:
  • client (BECClient, optional) – The BEC client.

  • config (ConnectionConfig, optional) – The connection configuration.

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

  • theme_update (bool, optional) – Whether to subscribe to theme updates. Defaults to False. When set to True, the widget’s apply_theme method will be called when the theme changes.

__delayed_submit_test(widget: bec_widgets.widgets.control.device_manager.components.ophyd_validation.ValidationListItem, connect: bool, force_connect: bool, timeout: float) None#

Delayed submission of device test to ensure UI updates.

_add_device_config(device_config: dict[str, Any], connect: bool, force_connect: bool, timeout: float, skip_validation: bool = False) None#
_compare_device_configs(config1: dict, config2: dict) bool#

Compare two device configurations through the Device model in bec_lib.atlas_models.

Parameters:
  • config1 (dict) – The first device configuration.

  • config2 (dict) – The second device configuration.

Returns:

True if the configurations are equivalent, False otherwise.

Return type:

bool

_create_list_widget_with_label(label_text: str) bec_widgets.utils.bec_list.BECList#

Setup the running validations section.

_device_already_exists(device_name: str) bool#
_emit_device_in_redis_session(device_config: dict) None#
_emit_item_clicked(widget: bec_widgets.widgets.control.device_manager.components.ophyd_validation.ValidationListItem)#
_is_device_in_redis_session(device_name: str, device_config: dict) bool#

Check if a device is in the running section.

_on_current_item_changed(current: qtpy.QtWidgets.QListWidgetItem, previous: qtpy.QtWidgets.QListWidgetItem)#

Handle current item changed.

_on_device_test_completed(device_config: dict, config_status: int, connection_status: int, error_message: str) None#

Handle device test completion.

_on_item_clicked(item: qtpy.QtWidgets.QListWidgetItem)#

Handle click on running item.

_on_request_rerun_validation(device_name: str, device_config: dict[str, Any], connect: bool, force_connect: bool, timeout: float) None#

Handle request to re-run validation for a device.

_remove_device(device_name: str, force_remove: bool = False) None#
_remove_device_config(device_config: dict[str, Any], force_remove: bool = False) None#
_set_running_ophyd_tests(running: bool)#

Set the running ophyd tests state.

_submit_test(widget: bec_widgets.widgets.control.device_manager.components.ophyd_validation.ValidationListItem, connect: bool, force_connect: bool, timeout: float) None#

Submit a device test to the thread pool.

_thread_pool_poll_loop()#

Start the thread pool polling loop.

_trigger_validation_started(device_name: str) None#

Trigger validation started for a specific device.

add_device_to_keep_visible_after_validation(device_name: str) None#

Add a device name to the list of devices to keep visible after validation.

Parameters:

device_name (str) – Name of the device to keep visible.

apply_theme(theme: str)#

Apply the current theme to the widget.

cancel_all_validations() None#

Cancel all running validations.

cancel_validation(device_name: str) None#

Cancel a running validation for a specific device.

Parameters:

device_name (str) – Name of the device to cancel validation for.

change_device_configs(device_configs: list[dict[str, Any]], added: bool, connect: bool = False, force_connect: bool = False, timeout: float = 5.0, skip_validation: bool = False) None#

Change the device configuration to test. If added is False, existing devices are removed. Device tests will be removed based on device names. No duplicates are allowed.

For validation runs, results are emitted via the validation_completed signal. Unless devices are already in the running session with the same config, in which case the combined results of all such devices are emitted via the multiple_validations_completed signal. NOTE Please make sure to connect to both signals if you want to capture all results.

Parameters:
  • device_configs (list[dict[str, Any]]) – List of device configurations.

  • added (bool) – Whether the devices are added to the existing list.

  • connect (bool, optional) – Whether to attempt connection during validation. Defaults to False.

  • force_connect (bool, optional) – Whether to force connection during validation. Defaults to False.

  • timeout (float, optional) – Timeout for connection attempt. Defaults to 5.0.

  • skip_validation (bool, optional) – Whether to skip validation for the added devices. Defaults to False.

clear_all()#

Clear all running and failed validations.

device_table_config_changed(device_configs: list[dict[str, Any]], added: bool, skip_validation: bool) None#

Slot to handle device config changes in the device table.

Parameters:
  • device_configs (list[dict[str, Any]]) – List of device configurations.

  • added (bool) – Whether the devices are added to the existing list.

  • skip_validation (bool) – Whether to skip validation for the added devices.

get_device_configs() list[dict[str, Any]]#

Get the current device configurations being tested.

Returns:

List of device configurations.

Return type:

list[dict[str, Any]]

remove_device_to_keep_visible_after_validation(device_name: str) None#

Remove a device name from the list of devices to keep visible after validation.

Parameters:

device_name (str) – Name of the device to remove.

running_ophyd_tests() bool#

Indicates if validations are currently running.

RPC = False#
item_clicked#
list_widget#
multiple_validations_completed#
validation_completed#
validations_are_running#