Source code for devctrl.gui.app.devices_workspace

"""Global devices workspace: shared device list and per-module load/connect control.

The workspace is a host-level page (like the Data workspace, see
:meth:`devctrl.gui.app.main_window.MainWindowBase.add_page`) with two tabs:

- **Devices** hosts the shared :class:`~devctrl.gui.devices.DeviceManagerWidget`
  instance (``app.w_devices``) permanently; the widget is created by the host
  before this workspace and lives here (module workspaces no longer show it).
- **Load/Connect** lists one :class:`.ModuleDevicesRow` per discovered module
  in a shared grid (so all rows align vertically) with buttons to load/unload
  the module's device roles and to connect/disconnect its devices, plus
  auto-load/auto-connect-on-startup checkboxes persisted in the host settings
  (:attr:`.AppSettings.device_module_autoload` / :attr:`.AppSettings.device_module_autoconnect`).
  A final :class:`.CustomDevicesRow` covers manually added ("custom") devices,
  which are always re-added at startup from
  :attr:`.AppSettings.custom_devices_path`.
"""
import logging

from typing import TYPE_CHECKING

from PyQt6.QtCore import QObject, Qt, QCoreApplication
from PyQt6.QtWidgets import (
    QWidget, QVBoxLayout, QLabel, QPushButton, QCheckBox,
    QScrollArea, QTabWidget, QFrame, QGridLayout,
)

from devctrl.devices.device_manager import DeviceManager
from devctrl.devices.device_role import DeviceRole
from devctrl.resources import load_icon

from .blueprint import GuiModule
from .config import AppConfigBase
from .custom_devices import custom_device_roles
from .devices import (
    add_module_devices,
    connect_module_dummies,
    disconnect_module_devices,
    get_module_connect_args,
    resolve_module_roles,
    unload_module_devices,
)

if TYPE_CHECKING:  # pragma: no cover
    from .main_window import MainWindowBase

log = logging.getLogger(__name__)

# Columns of the shared Load/Connect grid (see :meth:`ModuleDevicesRow.add_to_grid`).
_COL_LABEL = 0
_COL_BUTTONS = 1  # columns 1-5: Load, Unload, Connect, Connect dummies, Disconnect
_COL_CHECKBOX = 6
_COL_END = 7  # stretch column after the checkboxes


def _save_host_settings(cfg) -> None:
    """Write the host settings file to disk immediately (best effort).

    The host finalizes the settings on exit anyway; saving here makes the
    per-module startup checkboxes survive a crash.

    :param cfg: The :class:`~.config.AppConfigBase` subclass in use.
    """
    if not getattr(cfg, "APP_CFG_PATH", ""):
        return
    try:
        cfg.APP_CFG.save_yaml(cfg.APP_CFG_PATH)
    except Exception as e:
        log.warning(f"Could not save host settings: {e}")


[docs] class ModuleDevicesRow(QObject): """Load/connect control row for a single module in the Devices workspace. Shows the module name and its device state, the load/unload and connect/disconnect buttons and the auto-load/auto-connect-on-startup checkboxes. The state display refreshes automatically from the device manager's signals. The row is not a widget itself: its widgets are placed into the shared grid of the Load/Connect tab (see :meth:`.add_to_grid`) so that all rows align vertically. """
[docs] def __init__(self, app: "MainWindowBase", module: GuiModule, device_manager: DeviceManager): super().__init__() self.app = app self.module = module self.devmg = device_manager self._actions_enabled = True # -- left: module name + status ----------------------------------- self.w_label_name = QLabel(QCoreApplication.translate(module.pkg, module.name)) font = self.w_label_name.font() font.setBold(True) self.w_label_name.setFont(font) self.w_label_status = QLabel("") # -- middle: action buttons --------------------------------------- self.b_load = QPushButton(self.tr("Load")) self.b_load.clicked.connect(self.load) self.b_unload = QPushButton(self.tr("Unload")) self.b_unload.clicked.connect(self.unload) self.b_connect = QPushButton(self.tr("Connect")) self.b_connect.clicked.connect(self.connect_devices) self.b_connect_dummies = QPushButton(self.tr("Connect dummies")) self.b_connect_dummies.clicked.connect(self.connect_dummies) self.b_disconnect = QPushButton(self.tr("Disconnect")) self.b_disconnect.clicked.connect(self.disconnect) # -- right: startup checkboxes -------------------------------------- self.w_cb_load_on_startup = QCheckBox(self.tr("Load on startup")) self.w_cb_load_on_startup.setChecked(self._load_on_startup_setting()) self.w_cb_load_on_startup.toggled.connect(self._set_load_on_startup) self.w_cb_connect_on_startup = QCheckBox(self.tr("Connect on startup")) self.w_cb_connect_on_startup.setChecked(self._connect_on_startup_setting()) self.w_cb_connect_on_startup.toggled.connect(self._set_connect_on_startup) # -- state updates --------------------------------------------------- self.devmg.device_added.connect(self._on_devices_changed) self.devmg.device_removed.connect(self._on_devices_changed) self.devmg.device_updated.connect(self._on_devices_changed) self.app.on_tasks_started.connect(lambda: self.set_actions_enabled(False)) self.app.on_tasks_stopped.connect(lambda: self.set_actions_enabled(True)) self.update_status()
# -- grid placement -------------------------------------------------------
[docs] def add_to_grid(self, grid: QGridLayout, row: int) -> None: """Place this row's widgets into the shared grid of the Load/Connect tab. The row occupies two grid rows: the module name and the Load-on-startup checkbox in the first, the status label and the Connect-on-startup checkbox in the second. The action buttons span both rows, vertically centered. :param grid: The shared grid of the Load/Connect tab. :param row: First grid row of this module. """ center = Qt.AlignmentFlag.AlignVCenter grid.addWidget(self.w_label_name, row, _COL_LABEL) grid.addWidget(self.w_label_status, row + 1, _COL_LABEL) buttons = [self.b_load, self.b_unload, self.b_connect, self.b_connect_dummies, self.b_disconnect] for col, button in enumerate(buttons): grid.addWidget(button, row, _COL_BUTTONS + col, 2, 1, center) grid.addWidget(self.w_cb_load_on_startup, row, _COL_CHECKBOX) grid.addWidget(self.w_cb_connect_on_startup, row + 1, _COL_CHECKBOX)
# -- settings ----------------------------------------------------------
[docs] def _app_config(self) -> type: """Return the app config class in use (module subclass or host base).""" return self.app.app_config or AppConfigBase
[docs] def _load_on_startup_setting(self) -> bool: """Return the persisted auto-load setting of this module (default True).""" return bool(self._app_config().APP_CFG.device_module_autoload.get(self.module.id, True))
[docs] def _connect_on_startup_setting(self) -> bool: """Return the persisted auto-connect setting of this module (default False).""" return bool(self._app_config().APP_CFG.device_module_autoconnect.get(self.module.id, False))
[docs] def _set_load_on_startup(self, checked: bool) -> None: """Store the auto-load setting of this module and persist the host settings. :param checked: Whether the module's roles are loaded at startup. """ cfg = self._app_config() cfg.APP_CFG.device_module_autoload[self.module.id] = bool(checked) _save_host_settings(cfg)
[docs] def _set_connect_on_startup(self, checked: bool) -> None: """Store the auto-connect setting of this module and persist the host settings. :param checked: Whether the module's devices are connected at startup. """ cfg = self._app_config() cfg.APP_CFG.device_module_autoconnect[self.module.id] = bool(checked) _save_host_settings(cfg)
# -- actions -------------------------------------------------------------
[docs] def load(self) -> None: """Add all of the module's device roles to the device manager (idempotent).""" add_module_devices(self.devmg, self.module) self.update_status()
[docs] def unload(self) -> None: """Remove all of the module's device roles from the device manager. Roles shared with other modules are removed as well; those modules can re-load them (see :func:`~.devices.unload_module_devices`).""" unload_module_devices(self.devmg, self.module, self.app.modules) self.update_status()
[docs] def connect_devices(self) -> None: """Load the module's roles if needed and connect its devices via the startup dialog. Opens the :class:`~devctrl.gui.devices.DeviceStartupWidget` dialog, which connects the module's devices from its device config file in priority order and turns on its power switches. """ from devctrl.gui.devices import DeviceStartupWidget add_module_devices(self.devmg, self.module) DeviceStartupWidget.open_dialog( device_manager=self.devmg, parent=self.app, **get_module_connect_args(self.module)) self.update_status()
[docs] def connect_dummies(self) -> None: """Load the module's roles if needed and connect dummy devices for all of them.""" connect_module_dummies(self.devmg, self.module) self.update_status()
[docs] def disconnect(self) -> None: """Disconnect all devices of the module (roles stay loaded).""" disconnect_module_devices(self.devmg, self.module) self.update_status()
# -- state display ---------------------------------------------------------
[docs] def set_actions_enabled(self, enabled: bool) -> None: """Enable or disable the action buttons (used while tasks run). :param enabled: Whether the action buttons are enabled. """ self._actions_enabled = enabled self.update_status()
[docs] def resolve_roles(self) -> list[DeviceRole]: """Return the device roles of this module (resolving lazy callables).""" return resolve_module_roles(self.module)
[docs] def _on_devices_changed(self, key: str) -> None: """Refresh the state display when the device manager changed. :param key: The changed device key (ignored). """ self.update_status()
[docs] def update_status(self) -> None: """Refresh the status label and the button enabled states from the device manager.""" keys = [r.key for r in self.resolve_roles()] n_loaded = sum(1 for k in keys if k in self.devmg) n_connected = sum(1 for k in keys if self.devmg.is_connected(k)) if not keys: self.w_label_status.setText(self.tr("No device roles")) elif n_loaded == 0: self.w_label_status.setText(self.tr("Not loaded")) else: self.w_label_status.setText( self.tr("{loaded}/{roles} devices loaded, {connected} connected").format( loaded=n_loaded, roles=len(keys), connected=n_connected)) self.b_load.setEnabled(self._actions_enabled) self.b_unload.setEnabled(self._actions_enabled and n_loaded > 0) self.b_connect.setEnabled(self._actions_enabled and len(keys) > 0) self.b_connect_dummies.setEnabled(self._actions_enabled and len(keys) > 0) self.b_disconnect.setEnabled(self._actions_enabled and n_connected > 0)
[docs] class CustomDevicesRow(QObject): """Load/connect control row for manually added ("custom") devices. Custom devices are always re-added at startup (auto-load is always on — otherwise they would be forgotten), so there is no Load/Unload control; the row offers connect/disconnect and the auto-connect-on-startup checkbox (:attr:`.AppSettings.custom_devices_autoconnect`). The state display refreshes automatically from the device manager's signals. The row is not a widget itself: its widgets are placed into the shared grid of the Load/Connect tab (see :meth:`.add_to_grid`). """
[docs] def __init__(self, app: "MainWindowBase", device_manager: DeviceManager): super().__init__() self.app = app self.devmg = device_manager self._actions_enabled = True # -- left: row name + status -------------------------------------- self.w_label_name = QLabel(self.tr("Custom")) font = self.w_label_name.font() font.setBold(True) self.w_label_name.setFont(font) self.w_label_status = QLabel("") # -- middle: action buttons (aligned with the module rows) --------- self.b_connect = QPushButton(self.tr("Connect")) self.b_connect.clicked.connect(self.connect_devices) self.b_connect_dummies = QPushButton(self.tr("Connect dummies")) self.b_connect_dummies.clicked.connect(self.connect_dummies) self.b_disconnect = QPushButton(self.tr("Disconnect")) self.b_disconnect.clicked.connect(self.disconnect) # -- right: startup checkboxes -------------------------------------- # Custom devices are always loaded; the checkbox only communicates this. self.w_cb_load_on_startup = QCheckBox(self.tr("Load on startup")) self.w_cb_load_on_startup.setChecked(True) self.w_cb_load_on_startup.setEnabled(False) self.w_cb_connect_on_startup = QCheckBox(self.tr("Connect on startup")) self.w_cb_connect_on_startup.setChecked(self._connect_on_startup_setting()) self.w_cb_connect_on_startup.toggled.connect(self._set_connect_on_startup) # -- state updates --------------------------------------------------- self.devmg.device_added.connect(self._on_devices_changed) self.devmg.device_removed.connect(self._on_devices_changed) self.devmg.device_updated.connect(self._on_devices_changed) self.app.on_tasks_started.connect(lambda: self.set_actions_enabled(False)) self.app.on_tasks_stopped.connect(lambda: self.set_actions_enabled(True)) self.update_status()
# -- grid placement -------------------------------------------------------
[docs] def add_to_grid(self, grid: QGridLayout, row: int) -> None: """Place this row's widgets into the shared grid of the Load/Connect tab. Like :meth:`.ModuleDevicesRow.add_to_grid`, but the Load/Unload cells stay empty (custom devices are always loaded) and the buttons align with the module rows' Connect / Connect dummies / Disconnect columns. :param grid: The shared grid of the Load/Connect tab. :param row: First grid row of this row. """ center = Qt.AlignmentFlag.AlignVCenter grid.addWidget(self.w_label_name, row, _COL_LABEL) grid.addWidget(self.w_label_status, row + 1, _COL_LABEL) buttons = [self.b_connect, self.b_connect_dummies, self.b_disconnect] for col, button in enumerate(buttons): grid.addWidget(button, row, _COL_BUTTONS + 2 + col, 2, 1, center) grid.addWidget(self.w_cb_load_on_startup, row, _COL_CHECKBOX) grid.addWidget(self.w_cb_connect_on_startup, row + 1, _COL_CHECKBOX)
# -- settings ----------------------------------------------------------
[docs] def _app_config(self) -> type: """Return the app config class in use (module subclass or host base).""" return self.app.app_config or AppConfigBase
[docs] def _connect_on_startup_setting(self) -> bool: """Return the persisted auto-connect setting for custom devices (default False).""" return bool(self._app_config().APP_CFG.custom_devices_autoconnect)
[docs] def _set_connect_on_startup(self, checked: bool) -> None: """Store the custom-device auto-connect setting and persist the host settings. :param checked: Whether custom devices are connected at startup. """ cfg = self._app_config() cfg.APP_CFG.custom_devices_autoconnect = bool(checked) _save_host_settings(cfg)
# -- actions -------------------------------------------------------------
[docs] def resolve_roles(self) -> list[DeviceRole]: """Return the currently registered custom device roles.""" return custom_device_roles(self.devmg, self.app.modules)
[docs] def _on_devices_changed(self, key: str) -> None: """Refresh the state display when the device manager changed. :param key: The changed device key (ignored). """ self.update_status()
[docs] def connect_devices(self) -> None: """Connect the custom devices via the startup dialog. Opens the :class:`~devctrl.gui.devices.DeviceStartupWidget` dialog for the custom devices (no power-switch handling), which connects them from the device config file and prompts for devices without a stored connection. """ from devctrl.gui.devices import DeviceStartupWidget DeviceStartupWidget.open_dialog( device_manager=self.devmg, parent=self.app, power_switch_autoconnect_keys=[], power_switch_autoconnect_devices_timeout_s=self._app_config().APP_CFG.power_switch_autoconnect_devices_timeout_s, device_keys=[r.key for r in self.resolve_roles()]) self.update_status()
[docs] def connect_dummies(self) -> None: """Connect dummy devices for all custom devices (not persisted).""" for role in self.resolve_roles(): self.devmg.connect_dummy(role.key, persist=False) self.update_status()
[docs] def disconnect(self) -> None: """Disconnect all connected custom devices (roles stay loaded).""" for role in self.resolve_roles(): if self.devmg.is_connected(role.key): self.devmg.disconnect_device(role.key) self.update_status()
# -- state display ---------------------------------------------------------
[docs] def set_actions_enabled(self, enabled: bool) -> None: """Enable or disable the action buttons (used while tasks run). :param enabled: Whether the action buttons are enabled. """ self._actions_enabled = enabled self.update_status()
[docs] def update_status(self) -> None: """Refresh the status label and the button enabled states from the device manager.""" keys = [r.key for r in self.resolve_roles()] n_loaded = sum(1 for k in keys if k in self.devmg) n_connected = sum(1 for k in keys if self.devmg.is_connected(k)) if not keys: self.w_label_status.setText(self.tr("No custom devices")) else: self.w_label_status.setText( self.tr("{loaded}/{roles} devices loaded, {connected} connected").format( loaded=n_loaded, roles=len(keys), connected=n_connected)) self.b_connect.setEnabled(self._actions_enabled and len(keys) > 0) self.b_connect_dummies.setEnabled(self._actions_enabled and len(keys) > 0) self.b_disconnect.setEnabled(self._actions_enabled and n_connected > 0)
[docs] class DevicesWorkspaceWidget(QWidget): """Global devices workspace with the shared device list and per-module load/connect control. The "Devices" tab permanently hosts the shared device manager widget (``app.w_devices``); the "Load/Connect" tab lists the per-module controls and the custom devices row. """
[docs] def __init__(self, app: "MainWindowBase", parent=None): super().__init__(parent) self.app = app self.l_main = QVBoxLayout(self) self.l_main.setContentsMargins(0, 0, 0, 0) self.w_tabs = QTabWidget(self) self.l_main.addWidget(self.w_tabs) # The shared Devices widget (app.w_devices) is created by the host # before this workspace; it lives here permanently now. self.w_tabs.addTab(app.w_devices, load_icon("devices"), self.tr("Devices")) self.w_loadconnect = self._create_load_connect_tab() self.w_tabs.addTab(self.w_loadconnect, load_icon("list-add"), self.tr("Load/Connect"))
[docs] def _create_load_connect_tab(self) -> QWidget: """Build the per-module load/connect tab (one shared grid with a block per module).""" w = QWidget(self) l_tab = QVBoxLayout(w) l_tab.setContentsMargins(4, 4, 4, 4) scroll = QScrollArea(w) scroll.setWidgetResizable(True) scroll.setFrameShape(QFrame.Shape.NoFrame) w_rows = QWidget(scroll) self.l_modules = QGridLayout(w_rows) self.l_modules.setColumnStretch(_COL_END, 1) self.module_rows: list[ModuleDevicesRow] = [] row = 0 for i, mod in enumerate(self.app.modules): if i > 0: # separator line between the module blocks separator = QFrame(w_rows) separator.setFrameShape(QFrame.Shape.HLine) separator.setFrameShadow(QFrame.Shadow.Sunken) self.l_modules.addWidget(separator, row, 0, 1, _COL_END) row += 1 md_row = ModuleDevicesRow(self.app, mod, self.app.devmg) self.module_rows.append(md_row) md_row.add_to_grid(self.l_modules, row) row += 2 # Custom (manually added) devices; always shown so the row invites # adding devices via the Devices tab, even when the list is empty. if self.module_rows: separator = QFrame(w_rows) separator.setFrameShape(QFrame.Shape.HLine) separator.setFrameShadow(QFrame.Shadow.Sunken) self.l_modules.addWidget(separator, row, 0, 1, _COL_END) row += 1 self.custom_row = CustomDevicesRow(self.app, self.app.devmg) self.custom_row.add_to_grid(self.l_modules, row) row += 2 self.l_modules.setRowStretch(row, 1) scroll.setWidget(w_rows) l_tab.addWidget(scroll) return w