Source code for devctrl.gui.task.run.device_action

from .base import TaskRunWidget
from ....task import DeviceActionTask
from ....utility.threading import ThreadWithExceptionSupport, Queue
from ....devices.device_manager import DeviceManager

from typing import Callable, Optional
from PyQt6.QtCore import QTimer
from PyQt6.QtWidgets import QHBoxLayout, QLabel, QPushButton, QVBoxLayout
import logging
log = logging.getLogger(__name__)





class DeviceActionTaskRunWidget(TaskRunWidget):
    """Widget for :class:`.DeviceActionTask` that runs the configured device action in a separate thread.

    The device must be managed by a py:class:`device manager <.DeviceManager>`. 
    The device manager can be assigned to the static py:attr:`.DEVICE_MAMANGER` variable, or be passed in the constructor of the widget.

    The action is run in a separate thread to make sure the GUI does not hang when a longer operation is performed.
    """

    DEVICE_MANAGER: Optional[DeviceManager] = None



    def __init__(self, task: DeviceActionTask, device_manager: Optional[DeviceManager]=None, parent=None):
        super().__init__(task, parent)
        if device_manager is not None:
            self.devmg = device_manager
        elif self.DEVICE_MANAGER is not None:
            self.devmg = self.DEVICE_MANAGER
        else:
            raise ValueError(f"A device manager must be either given to the constructor or be assigned to `DeviceActionTaskRunWidget.DEVICE_MANAGER`, but both are None.")
        l = QVBoxLayout()
        self.setLayout(l)
        self.w_label = QLabel(f"")
        l.addWidget(self.w_label)
        self._thread: Optional[ThreadWithExceptionSupport] = None
        self._ex_queue: Queue = Queue()





    def start_task(self) -> bool:
        self.w_label.setText(f"Getting SMTP settings")
        device = self.devmg.get_device(self.task.device_key)
        try:
            func = getattr(device, self.task.device_parameter)
        except AttributeError as e:
            raise AttributeError(f"Task is misconfigured: The device {device.get_device_type_name()} does not have attribute '{self.task.device_parameter}'.")

        if self.task.arguments:
            args_str = ", "+ ", ".join(self.task.arguments)
        else:
            args_str = ""
        self.w_label.setText(f"Running: {device.get_device_name()}.{self.task.device_parameter}({args_str})")
        log.debug(f"Running: {device.get_device_name()}.{self.task.device_parameter}({args_str})")
        self._thread = ThreadWithExceptionSupport(queue=self._ex_queue, target=func, args=self.task.arguments)
        self._thread.run()
        return True




    def update_task(self) -> Optional[bool]:
        """Check if thread is still running"""
        if self._thread is None:
            raise RuntimeError(f"DeviceAction thread not running")
        if self._thread.is_alive():
            return None
        else:
            return True




    def stop_task(self) -> bool:
        if self._thread is None:
            raise RuntimeError(f"DeviceAction thread not running")
        errs = False
        while not self._ex_queue.empty():
            ex = self._ex_queue.get()
            log.error(f"Exception occured while performing device action: {type(ex)}: {ex}")
            errs = True
        if self._thread.is_alive():
            self._thread.join()
        self._thread = None

        return not errs