bec_widgets.widgets.control.device_manager ========================================== .. py:module:: bec_widgets.widgets.control.device_manager Submodules ---------- .. toctree:: :maxdepth: 1 /autoapi/bec_widgets/widgets/control/device_manager/components/index Classes ------- .. autoapisummary:: bec_widgets.widgets.control.device_manager.DMConfigView bec_widgets.widgets.control.device_manager.DeviceTable bec_widgets.widgets.control.device_manager.DocstringView bec_widgets.widgets.control.device_manager.OphydValidation Package Contents ---------------- .. py:class:: DMConfigView(parent=None) Bases: :py:obj:`qtpy.QtWidgets.QWidget` Widget to show the config of a selected device in YAML format. .. py:method:: _customize_monaco() Customize the Monaco editor for YAML display. .. py:method:: _customize_overlay() Customize the overlay widget. .. py:method:: 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. :param device: The selected device configuration. :type device: list[dict] .. py:attribute:: RPC :value: False .. py:attribute:: monaco_editor .. py:attribute:: stacked_layout .. py:class:: DeviceTable(parent: qtpy.QtWidgets.QWidget | None = None, client=None) Bases: :py:obj:`bec_widgets.utils.bec_widget.BECWidget`, :py:obj:`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) :param client: The BEC client. :type client: BECClient, optional :param config: The connection configuration. :type config: ConnectionConfig, optional :param gui_id: The GUI ID. :type gui_id: str, optional :param theme_update: 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. :type theme_update: bool, optional .. py:method:: __on_client_device_update_event(action: bec_lib.messages.ConfigAction, config: dict[str, dict[str, Any]]) -> None Handle DEVICE_UPDATE events from the BECClient. .. py:method:: __update_device_row_data(row: int, data: dict) Update an existing device row with new data. :param row: The row index to update. :type row: int :param data: The device data to populate the row. :type data: dict .. py:method:: _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. :param data: The device data to populate the row. :type data: dict :param config_status: The configuration validation status. :type config_status: ConfigStatus | int :param connection_status: The connection status. :type connection_status: ConnectionStatus | int .. py:method:: _apply_row_filter(text_input: str) Apply a filter to the table rows based on the filter text. .. py:method:: _clear_table() Remove all rows. .. py:method:: _compare_configs(cfg1: dict, cfg2: dict) -> bool Compare two device configurations for equality. .. py:method:: _compare_device_configs(config1: dict, config2: dict) -> bool Compare two device configurations through the Device model in bec_lib.atlas_models. :param config1: The first device configuration. :type config1: dict :param config2: The second device configuration. :type config2: dict :returns: True if the configurations are equivalent, False otherwise. :rtype: bool .. py:method:: _find_row_by_name(name: str) -> int | None Find a row by device name. :param name: The name of the device to find. :type name: str :returns: The row index if found, else None. :rtype: int | None .. py:method:: _get_cell_data(row: int, column: int) -> str | bool | None Get the data from a specific cell. :param row: The row index. :type row: int :param column: The column index. :type column: int .. py:method:: _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. :param device_configs: The device configs to check. :type device_configs: Iterable[dict[str, Any]] :returns: The list of overlapping device configs. :rtype: list[dict[str, Any]] .. py:method:: _is_config_in_sync_with_redis() Check if the current config is in sync with Redis. .. py:method:: _is_device_in_redis_session(device_name: str, device_config: dict) -> bool Check if a device is in the running section. .. py:method:: _on_cell_double_clicked(row: int, column: int) Handle double-click events on table cells. .. py:method:: _on_device_config_update() -> None Handle device configuration updates from the BECClient. .. py:method:: _on_device_row_data_changed(data: dict) Handle data change events from device rows. .. py:method:: _on_selection_changed(selected: qtpy.QtCore.QItemSelection, deselected: qtpy.QtCore.QItemSelection) Handle selection changes in the table. .. py:method:: _on_table_checkbox_clicked(row: int, column: int, checked: bool) Handle checkbox clicks in the table. .. py:method:: _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. .. py:method:: _remove_configs_dialog(device_names: list[str]) -> bool Prompt the user to confirm removal of rows and remove them from the model if accepted. :param device_names: List of device names to be removed. :type device_names: list[str] :returns: True if the user confirmed removal, False otherwise. :rtype: bool .. py:method:: _remove_rows_by_name(device_names: list[str]) Remove a row by device name. :param device_name: The name of the device to remove. :type device_name: str .. py:method:: _resize_table_policy(table: qtpy.QtWidgets.QTableWidget, enable: bool) Enable or disable column resizing. .. py:method:: _set_table_signals_on_hold(table: qtpy.QtWidgets.QTableWidget, enable: bool) Enable or disable table signals. .. py:method:: _setup_search() Create components related to the search functionality .. py:method:: _setup_table() Initializes the table configuration and headers. .. py:method:: _state_change_fuzzy_search(enabled: int) Handle state changes for the fuzzy search toggle. .. py:method:: _update_device_row_status(row: int, config_status: int, connection_status: int) -> bool Update an existing device row's validation status. :param device_name: The name of the device. :type device_name: str :param config_status: The configuration validation status. :type config_status: int :param connection_status: The connection status. :type connection_status: int .. py:method:: _update_row(data: dict) -> int | None Update an existing row with new data. :param data: The device data to populate the row. :type data: dict :returns: The row index if updated, else None. :rtype: int | None .. py:method:: 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. :param device_configs: The device configs to add. :type device_configs: Iterable[dict[str, Any]] :param skip_validation: Whether to skip validation for the added devices. :type skip_validation: bool .. py:method:: cleanup() Cleanup resources. .. py:method:: clear_device_configs() Clear the device configs. Skips validation by default. .. py:method:: eventFilter(source: qtpy.QtCore.QObject, event: qtpy.QtCore.QEvent) -> bool Customize event filtering for table interactions. .. py:method:: get_device_config() -> list[dict] Get the current device configurations in the table. :returns: The list of device configurations. :rtype: list[dict] .. py:method:: get_selected_device_configs() -> list[dict] Get the currently selected device configurations in the table. :returns: The list of selected device configurations. :rtype: list[dict] .. py:method:: 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). :rtype: dict[str, Tuple[dict, int, int]] .. py:method:: remove_device(device_name: str) Remove a device from the config. :param device_name: The name of the device to remove. :type device_name: str .. py:method:: remove_device_configs(device_configs: _DeviceCfgIter) Remove devices from the config. :param device_configs: The device configs to remove. :type device_configs: dict[str, dict] .. py:method:: set_device_config(device_configs: _DeviceCfgIter, skip_validation: bool = False) Set the device config. This will clear any existing configs. :param device_configs: The device configs to set. :type device_configs: Iterable[dict[str, Any]] :param skip_validation: Whether to skip validation for the set devices. :type skip_validation: bool .. py:method:: 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. :param device_configs: The device configs to update. :type device_configs: Iterable[dict[str, Any]] :param skip_validation: Whether to skip validation for the updated devices. :type skip_validation: bool .. py:method:: 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: .. py:method:: 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. :param device_configs: The device configs to update. :type device_configs: Iterable[dict[str, Any]] .. py:attribute:: RPC :value: False .. py:attribute:: client_callback_id :value: 0 .. py:attribute:: device_config_in_sync_with_redis .. py:attribute:: device_configs_changed .. py:attribute:: device_row_dbl_clicked .. py:attribute:: headers_key_map :type: dict[str, str] .. py:attribute:: request_update_after_client_device_update .. py:attribute:: request_update_multiple_device_validations .. py:attribute:: row_data :type: dict[str, bec_widgets.widgets.control.device_manager.components.device_table.device_table_row.DeviceTableRow] .. py:attribute:: selected_devices .. py:attribute:: table .. py:attribute:: table_sort_on_hold .. py:class:: DocstringView(parent: qtpy.QtWidgets.QWidget | None = None) Bases: :py:obj:`qtpy.QtWidgets.QTextEdit` .. py:method:: _set_text(text: str) .. py:method:: on_select_config(device: list[dict]) .. py:method:: set_device_class(device_class_str: str) -> None .. py:class:: OphydValidation(parent=None, client=None, hide_legend: bool = False) Bases: :py:obj:`bec_widgets.utils.bec_widget.BECWidget`, :py:obj:`qtpy.QtWidgets.QWidget` Widget to manage and run ophyd device tests. :param parent: Parent widget. Defaults to None. :type parent: QWidget, optional :param client: BEC client instance. Defaults to None. :type client: BECClient, optional :param hide_legend: Whether to hide the legend. Defaults to False. :type hide_legend: bool, optional 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) :param client: The BEC client. :type client: BECClient, optional :param config: The connection configuration. :type config: ConnectionConfig, optional :param gui_id: The GUI ID. :type gui_id: str, optional :param theme_update: 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. :type theme_update: bool, optional .. py:method:: __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. .. py:method:: _add_device_config(device_config: dict[str, Any], connect: bool, force_connect: bool, timeout: float, skip_validation: bool = False) -> None .. py:method:: _compare_device_configs(config1: dict, config2: dict) -> bool Compare two device configurations through the Device model in bec_lib.atlas_models. :param config1: The first device configuration. :type config1: dict :param config2: The second device configuration. :type config2: dict :returns: True if the configurations are equivalent, False otherwise. :rtype: bool .. py:method:: _create_list_widget_with_label(label_text: str) -> bec_widgets.utils.bec_list.BECList Setup the running validations section. .. py:method:: _device_already_exists(device_name: str) -> bool .. py:method:: _emit_device_in_redis_session(device_config: dict) -> None .. py:method:: _emit_item_clicked(widget: bec_widgets.widgets.control.device_manager.components.ophyd_validation.ValidationListItem) .. py:method:: _is_device_in_redis_session(device_name: str, device_config: dict) -> bool Check if a device is in the running section. .. py:method:: _on_current_item_changed(current: qtpy.QtWidgets.QListWidgetItem, previous: qtpy.QtWidgets.QListWidgetItem) Handle current item changed. .. py:method:: _on_device_test_completed(device_config: dict, config_status: int, connection_status: int, error_message: str) -> None Handle device test completion. .. py:method:: _on_item_clicked(item: qtpy.QtWidgets.QListWidgetItem) Handle click on running item. .. py:method:: _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. .. py:method:: _remove_device(device_name: str, force_remove: bool = False) -> None .. py:method:: _remove_device_config(device_config: dict[str, Any], force_remove: bool = False) -> None .. py:method:: _set_running_ophyd_tests(running: bool) Set the running ophyd tests state. .. py:method:: _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. .. py:method:: _thread_pool_poll_loop() Start the thread pool polling loop. .. py:method:: _trigger_validation_started(device_name: str) -> None Trigger validation started for a specific device. .. py:method:: 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. :param device_name: Name of the device to keep visible. :type device_name: str .. py:method:: apply_theme(theme: str) Apply the current theme to the widget. .. py:method:: cancel_all_validations() -> None Cancel all running validations. .. py:method:: cancel_validation(device_name: str) -> None Cancel a running validation for a specific device. :param device_name: Name of the device to cancel validation for. :type device_name: str .. py:method:: 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. :param device_configs: List of device configurations. :type device_configs: list[dict[str, Any]] :param added: Whether the devices are added to the existing list. :type added: bool :param connect: Whether to attempt connection during validation. Defaults to False. :type connect: bool, optional :param force_connect: Whether to force connection during validation. Defaults to False. :type force_connect: bool, optional :param timeout: Timeout for connection attempt. Defaults to 5.0. :type timeout: float, optional :param skip_validation: Whether to skip validation for the added devices. Defaults to False. :type skip_validation: bool, optional .. py:method:: clear_all() Clear all running and failed validations. .. py:method:: 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. :param device_configs: List of device configurations. :type device_configs: list[dict[str, Any]] :param added: Whether the devices are added to the existing list. :type added: bool :param skip_validation: Whether to skip validation for the added devices. :type skip_validation: bool .. py:method:: get_device_configs() -> list[dict[str, Any]] Get the current device configurations being tested. :returns: List of device configurations. :rtype: list[dict[str, Any]] .. py:method:: 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. :param device_name: Name of the device to remove. :type device_name: str .. py:method:: running_ophyd_tests() -> bool Indicates if validations are currently running. .. py:attribute:: RPC :value: False .. py:attribute:: item_clicked .. py:attribute:: list_widget .. py:attribute:: multiple_validations_completed .. py:attribute:: validation_completed .. py:attribute:: validations_are_running