Source code for schrodinger.ui.qt.tasks.taskbar2

from enum import Enum
from typing import Union

from schrodinger import get_maestro
from schrodinger.application.job_monitor.job_monitor_gui import JobMonitor
from schrodinger.models import mappers
from schrodinger.Qt import QtCore
from schrodinger.Qt import QtWidgets
from schrodinger.Qt.QtCore import Qt
from schrodinger.Qt.QtGui import QIcon
from schrodinger.tasks import jobtasks
from schrodinger.tasks import taskmanager
from schrodinger.tasks.tasks import AbstractTask
from schrodinger.ui.qt.standard.icons import icons
from schrodinger.ui.qt.standard_widgets import buttons
from schrodinger.ui.qt.tasks import taskbar_ui
from schrodinger.ui.qt.tasks.taskwidgets import AbstractTaskBar
from schrodinger.utils import mmutil

maestro = get_maestro()
JobMonitorType = Enum('JobMonitorType', ('NEW', 'LEGACY', 'NONE'))


[docs]class TaskBar2(AbstractTaskBar):
    """
    :cvar _TASK_MONITORING_DURATION: time (in ms) to monitor jobs after launch
    :vartype _TASK_MONITORING_DURATION: int
    :cvar _HAS_JM_FAILURE_TEXT: text to use as part of the icon failed tooltip
            if a job monitor is available
    :vartype _HAS_JM_FAILURE_TEXT: str
    :cvar _NO_JM_FAILURE_TEXT: text to use as part of the icon failed tooltip
            if a job monitor is not available
    :vartype _NO_JM_FAILURE_TEXT: str
    :cvar _FAILURE_TOOLTIP_TEMPLATE: the template for the icon failed tooltip
            text
    :vartype _FAILURE_TOOLTIP_TEMPLATE: str
    """
    ui_module = taskbar_ui
    _TASK_MONITORING_DURATION = 60000
    _HAS_JM_FAILURE_TEXT = 'Click to view in Job Monitor'
    _NO_JM_FAILURE_TEXT = 'See log file for details'
    _FAILURE_TOOLTIP_TEMPLATE = ('<font color=red>Job failed</font><br><i>'
                                 '{suggestion}</i>')

[docs]    def initSetOptions(self):
        super().initSetOptions()
        self._most_recent_failed_task = None

[docs]    def initSetUp(self):
        super().initSetUp()
        self.settings_menu = None
        self.config_dialog = None
        self.ui.start_btn.setStyle(buttons.StyledButton.Style.Highlighted)
        self.ui.settings_btn.setStyle(buttons.StyledButton.Style.Highlighted)
        icon = QIcon(icons.EXCLAMATION_ERROR_EB)
        self.ui.task_failed_btn.setIcon(icon)
        self.ui.task_failed_btn.clicked.connect(self._showJobMonitor)
        if self._getJobMonitorType() == JobMonitorType.NONE:
            suggestion = self._NO_JM_FAILURE_TEXT
        else:
            suggestion = self._HAS_JM_FAILURE_TEXT
        text = self._FAILURE_TOOLTIP_TEMPLATE.format(suggestion=suggestion)
        self.ui.task_failed_btn.setToolTip(text)

[docs]    def initFinalize(self):
        super().initFinalize()
        self._updateStatusWidgets()

[docs]    def defineMappings(self):
        M = self.model_class
        mappings = [
            (self.ui.name_le, M),
            (self.ui.settings_btn, M),
            (self.ui.start_btn, M),
        ]
        if M is taskmanager.TaskManager:
            t = mappers.TargetSpec(slot=self._updateSettingsBtnVisibility)
            mappings.append((t, M.TaskClass))
        return mappings

[docs]    def getSignalsAndSlots(self, model):
        return super().getSignalsAndSlots(model) + [
            (self.ui.start_btn.clicked, self.startRequested),
            (self.ui.settings_btn.singleClicked, self.showSettingsMenu),
            (model.taskStarted, self._onTaskStarted),
        ]

[docs]    def setConfigDialog(self, config_dialog: QtWidgets.QDialog):
        """
        Assign a new config dialog.
        """

        config_dialog.startRequested.connect(self.startRequested)
        flags = Qt.CustomizeWindowHint | Qt.WindowTitleHint | Qt.Dialog
        config_dialog.setParent(self, flags)
        self.config_dialog = config_dialog
        self.ui.settings_btn.doubleClicked.connect(self.showConfigDialog)

[docs]    @QtCore.pyqtSlot()
    def showConfigDialog(self):
        if self.config_dialog is None:
            raise RuntimeError("No config dialog has been set for this taskbar")
        self.config_dialog.setModel(self._getCurrentTask().job_config)
        self.config_dialog.run(modal=True)

[docs]    def setModel(self, model):
        super().setModel(model)
        self._updateSettingsBtnVisibility()

    @property
    def start_btn(self):
        return self.ui.start_btn

[docs]    def setSettingsMenuActions(self, text_to_slots):
        """
        Set the actions to show for the settings menu.

        :param text_to_slots: A list of tuples mapping the desired menu item
            text with the function that should be called when the item is
            selected.  If the slot is `None`, then a separator will be added
            instead and the text will be ignored.
        :type  text_to_slots: list[tuple(str, callable) or None]
        """
        menu = QtWidgets.QMenu()
        for text_and_slot in text_to_slots:
            if text_and_slot is None:
                menu.addSeparator()
            else:
                menu.addAction(*text_and_slot)
        self.settings_menu = menu

[docs]    def showSettingsMenu(self):
        """
        Show the settings menu after clicking the settings button.

        Sets the position of the menu to be appear to the top left of the button.
        """
        self.settings_menu.show()
        parent_pos = self.ui.settings_btn.mapToGlobal(QtCore.QPoint(0, 0))
        parent_x = parent_pos.x() + self.ui.settings_btn.width(
        ) - self.settings_menu.width()
        parent_y = parent_pos.y() - self.settings_menu.height()
        self.settings_menu.move(parent_x, parent_y)

    def _showJobMonitor(self):
        """
        Launch an appropriate job monitor panel.
        """

        jm_type = self._getJobMonitorType()
        if jm_type == JobMonitorType.NEW:
            task = self._most_recent_failed_task
            job_id = task.job_id if jobtasks.is_jobtask(task) else None
            jm = JobMonitor.getPanelInstance()
            if job_id is not None:
                jm.selectJob(job_id)
            jm.run()
        elif jm_type == JobMonitorType.LEGACY:
            maestro.command("showpanel monitor")

    def _updateStatusWidgets(self):
        """
        Update the state of widgets that communicate task status information.
        """

        task = self._getMostRecentFailedTask()
        visible = jobtasks.is_jobtask(task)
        self.ui.task_failed_btn.setVisible(visible)
        # TODO: update the status bar text (PANEL-19519)

    def _getMostRecentFailedTask(self) -> Union[AbstractTask, type(None)]:
        """
        :return: the most recent failed task that is still being monitored
        """

        return self._most_recent_failed_task

    def _setMostRecentFailedTask(self, task: Union[AbstractTask, type(None)]):
        """
        :param task: a failed task, or `None` to clear the existing value
        """

        if self._getMostRecentFailedTask() != task:
            self._most_recent_failed_task = task
            self._updateStatusWidgets()

    @QtCore.pyqtSlot(AbstractTask)
    def _onTaskStarted(self, task: AbstractTask):
        """
        Respond to a task starting by monitoring it for a short period of time.
        """

        task.taskFailed.connect(self._onTaskFailed)
        QtCore.QTimer.singleShot(self._TASK_MONITORING_DURATION,
                                 lambda: self._onMonitoringEnded(task))

    @QtCore.pyqtSlot()
    def _onTaskFailed(self):
        """
        Respond to the failure of a monitored task.
        """

        task = self.sender()
        self._setMostRecentFailedTask(task)

    @QtCore.pyqtSlot(AbstractTask)
    def _onMonitoringEnded(self, task: AbstractTask):
        """
        Respond to the end of the period in which a recently-launched task is
        monitored by disconnecting its failure signal.

        :param task: a task that should no longer be monitored
        """

        task.taskFailed.disconnect(self._onTaskFailed)
        if self._getMostRecentFailedTask() == task:
            self._setMostRecentFailedTask(None)

    def _updateSettingsBtnVisibility(self):
        """
        Show the settings button if the task is a jobtask, otherwise hide it.
        """
        task = self._getCurrentTask()
        should_show_settings = jobtasks.is_jobtask(task)
        self.ui.settings_btn.setVisible(should_show_settings)

    def _getJobMonitorType(self) -> JobMonitorType:
        """
        :return: the type of job monitor that this panel can launch
        """

        if mmutil.feature_flag_is_enabled(mmutil.JOB_SERVER):
            return JobMonitorType.NEW
        elif maestro:
            return JobMonitorType.LEGACY
        else:
            return JobMonitorType.NONE