"""
<<<<< DEPRECATED >>>>>
This module should not be used for new code. Instead, consider using
`schrodinger.ui.qt.tasks`
<<<<< !!!!!!!!! >>>>>
"""
import os
from schrodinger.infra import util
from schrodinger.Qt import QtCore
from schrodinger.Qt import QtGui
from schrodinger.Qt import QtWidgets
from schrodinger.ui.qt import swidgets
from schrodinger.ui.qt import table_helper
from schrodinger.ui.qt import utils as ui_qt_utils
from schrodinger.ui.qt.appframework2 import baseapp
from schrodinger.ui.qt.appframework2 import debug
from schrodinger.ui.qt.appframework2 import tasks
from schrodinger.ui.qt.standard import constants
DEFAULT_START_LATENCY = 750
IMAGE_PATH = os.path.join(os.path.dirname(__file__), "images")
ICON_PATH_TEMPLATE = os.path.join(IMAGE_PATH, "{}.png")
SPIN_ICONS = None
ERROR_ICON = None
[docs]def get_spin_icons():
global SPIN_ICONS
if SPIN_ICONS is None:
SPIN_ICONS = [
QtGui.QIcon(ICON_PATH_TEMPLATE.format(num)) for num in range(0, 9)
]
return SPIN_ICONS
[docs]def get_error_icon():
global ERROR_ICON
ERROR_ICON = QtGui.QIcon(ICON_PATH_TEMPLATE.format('error'))
return ERROR_ICON
#===============================================================================
# Base Classes
#===============================================================================
[docs]class TaskUIMixin(object):
"""
This mixin provides the framework for making a user interface for a task
runner. In MVC parlance, this mixin is used for creating a view/controller
for a task runner model.
"""
start_latency = DEFAULT_START_LATENCY
[docs] def connectRunner(self, runner):
"""
Sets the task runner object for this UI and connects signals. If there
is already a runner connected to this UI, it will first be disconnected
before connecting the new runner. Passing in None will leave the UI
not connected to anything.
:param runner: the task runner to act as a model for this UI
:type runner: tasks.AbstractTaskRunner
"""
if not hasattr(self, 'start_btn'):
self.start_btn = QtWidgets.QPushButton('Dummy')
if not hasattr(self, 'reset_btn'):
self.reset_btn = QtWidgets.QPushButton('Dummy')
self.disconnectRunner()
if not runner:
return
self.runner = runner
self.start_btn.clicked.connect(self.onStartPressed)
self.reset_btn.clicked.connect(self.onResetPressed)
self.runner.stateChanged.connect(self.onRunnerStateChanged)
self.runner.startRequested.connect(self.onStartRequested)
self.runner.startFailed.connect(self.onStartFailed)
self.runner.taskStarted.connect(self.onTaskStarted)
self.runner.taskEnded.connect(self.onTaskEnded)
# Update after all subclass connection logic
QtCore.QTimer.singleShot(0, self.onRunnerStateChanged)
[docs] def disconnectRunner(self):
"""
Disconnects the current runner from this UI. When subclassing, first
perform any subclass-specific disconnection logic before calling the
parent class' disconnectRunner(). If there is no runner connected, this
method will do nothing.
"""
if not self.runner:
return
self.runner.stateChanged.disconnect(self.onRunnerStateChanged)
self.runner.startRequested.disconnect(self.onStartRequested)
self.runner.startFailed.disconnect(self.onStartFailed)
self.runner.taskStarted.disconnect(self.onTaskStarted)
self.runner.taskEnded.disconnect(self.onTaskEnded)
self.runner = None
# Update after all subclass disconnection logic
QtCore.QTimer.singleShot(0, self.onRunnerStateChanged)
[docs] def onResetPressed(self):
self.runner.reset()
[docs] def onStartPressed(self):
self.runner.start()
[docs] def onStartRequested(self):
self.start_btn.setEnabled(False)
[docs] def onStartFailed(self):
self.start_btn.setEnabled(True)
[docs] def onTaskStarted(self, task):
"""
Start latency is the small delay after a task is started before another
task can be started.
"""
if self.runner.allow_concurrent:
QtCore.QTimer.singleShot(self.start_latency, self._restoreReady)
def _restoreReady(self):
self.start_btn.setEnabled(True)
[docs] def onTaskEnded(self, task):
"""
This slot will only be called if the task was run within a single
Maestro session.
"""
self.start_btn.setEnabled(True)
[docs] def onRunnerStateChanged(self):
pass
#===============================================================================
# Task Name Line Edit
#===============================================================================
skip_if_editing_name = util.skip_if('_editing_name')
[docs]class TaskNameLineEdit(TaskUIMixin, QtWidgets.QLineEdit):
"""
A line edit interface for task names (i.e. job names). This widget will
automatically respond to changes in the job runner. It will also set itself
read-only if the job runner does not allow custom job names.
"""
_editingName = util.flag_context_manager('_editing_name')
[docs] def __init__(self, runner=None):
QtWidgets.QLineEdit.__init__(self)
self._editing_name = False
self.runner = None
self.setContentsMargins(2, 2, 2, 2)
self.textChanged.connect(self.onTextChanged)
self.editingFinished.connect(self.onNameChanged)
self.connectRunner(runner)
[docs] def setText(self, text):
"""
Overrides parent method so that programmatic modification of the text
will trigger an update of the runner.
"""
old_text = self.text()
QtWidgets.QLineEdit.setText(self, text)
if old_text != text: # Prevents infinite loop of nameChanged signals
self.onNameChanged()
[docs] def connectRunner(self, runner):
TaskUIMixin.connectRunner(self, runner)
if not runner:
return
self.setReadOnly(not runner.allow_custom_name)
self.setText(self.runner.nextName())
self.runner.nameChanged.connect(self.onRunnerNameChanged)
[docs] def onTextChanged(self):
"""
We need to respond to the textChanged signal because if a user edits the
name and directly clicks the start button, the editingFinished signal
can come *after* the start button clicked signal, resulting in the new
name not being assigned to the task that gets launched.
Because textChanged is emitted while the user is editing the field, we
don't want to process the empty-name case (which re-populates the field
with the default job name).
"""
if self.text() == '':
return
# We don't want this widget to respond to this name change as it is
# redundant and it will result in the cursor jumping to the end of the
# line.
with self._editingName():
self.onNameChanged()
[docs] def onNameChanged(self):
self.runner.setCustomName(self.text())
[docs] @skip_if_editing_name
def onRunnerNameChanged(self):
self.setText(self.runner.nextName())
[docs] def onTaskStarted(self, task):
TaskUIMixin.onTaskStarted(self, task)
self.setText(self.runner.nextName())
#===============================================================================
# Spinner Widgets
#===============================================================================
[docs]class SpinLabel(TaskUIMixin, QtWidgets.QLabel):
"""
This is a simple label that displays a spinner animation. Whenever a task
from the connected task launcher is running, the spinner will be animated.
It stops automatically when the last task ends. Other than connecting a
runner, nothing generally needs to be done with this label.
Note that QLabel uses a pixmap, not an icon, so the SpinnerIconMixin will
not work here.
"""
[docs] def __init__(self, runner=None):
QtWidgets.QLabel.__init__(self)
self.runner = None
self.height = 16
self.current_num = 0
self._loadImages()
self.timer = QtCore.QTimer()
self.timer.timeout.connect(self._advanceSpinner)
self.connectRunner(runner)
def _loadImages(self):
height = self.height
icons = get_spin_icons()
self.pics = [icon.pixmap(height, height) for icon in icons[1:]]
self.stop_pic = icons[0].pixmap(height, height)
def _advanceSpinner(self):
pixmap = self.pics[self.current_num]
self.current_num += 1
if self.current_num == len(self.pics):
self.current_num = 0
self.setPixmap(pixmap)
[docs] def onRunnerStateChanged(self):
if self.runner.isRunning():
self.startSpinner()
else:
self.stopSpinner()
[docs] def startSpinner(self):
if not self.timer.isActive():
self.timer.start(250)
[docs] def stopSpinner(self):
self.timer.stop()
self.setPixmap(self.stop_pic)
[docs]class SpinnerIconMixin(object):
"""
Contains common code for widgets with spinners that use icons.
"""
[docs] def setupSpinner(self):
icons = get_spin_icons()
self.spin_icons = icons[1:]
self.spin_idle_icon = icons[0]
self.spin_error_icon = get_error_icon()
self.spin_error_state = False
self.spin_current_num = 0
self.spin_timer = QtCore.QTimer()
self.spin_timer.timeout.connect(self._advanceSpinner)
def _advanceSpinner(self):
if self.spin_error_state:
self.setIcon(self.spin_error_icon)
return
icon = self.spin_icons[self.spin_current_num]
self.spin_current_num += 1
if self.spin_current_num == len(self.spin_icons):
self.spin_current_num = 0
self.setIcon(icon)
[docs] def startSpinner(self):
if not self.spin_timer.isActive():
self.spin_timer.start(250)
[docs] def stopSpinner(self):
self.spin_timer.stop()
self.updateSpinIcon()
[docs] def updateSpinIcon(self):
if self.spin_error_state:
self.setIcon(self.spin_error_icon)
else:
self.setIcon(self.spin_idle_icon)
[docs] def setError(self, state=False):
self.spin_error_state = state
self.updateSpinIcon()
#===============================================================================
# Task Bar
#===============================================================================
[docs]class TaskBar(TaskUIWidget):
"""
A compound widget with a task name label, a spinner, and a run button.
"""
[docs] def __init__(self,
runner=None,
label_text='Task name:',
button_text='Run',
show_spinner=True,
task_reset=True):
"""
:param runner: the runner to connect to this task bar
:type runner: tasks.AbstractTaskRunner
:param label_text: text label associated with the task name field
:type label_text: str
:param button_text: text on the "start" button
:type button_text: str
:param show_spinner: whether to show a progress spinner
:type show_spinner: bool
:param task_reset: whether to include a separate task reset action.
Otherwise, reset will emit a global reset signal
:type task_reset: bool
"""
self.label_text = label_text
self.button_text = button_text
self.show_spinner = show_spinner
self.task_reset = task_reset
TaskUIWidget.__init__(self, runner)
[docs] def setOptions(self):
TaskUIWidget.setOptions(self)
self.start_latency = DEFAULT_START_LATENCY
[docs] def setup(self):
TaskUIWidget.setup(self)
self.name_lbl = QtWidgets.QLabel(self.label_text)
self.name_le = TaskNameLineEdit()
self.spinner = SpinLabel()
self.start_btn = ui_qt_utils.AcceptsFocusPushButton(self.button_text)
self.start_timer = QtCore.QTimer()
self.settings_btn = SettingsButton(task_reset=self.task_reset)
[docs] def layOut(self):
TaskUIWidget.layOut(self)
self.main_layout.addWidget(self.name_lbl)
self.main_layout.addWidget(self.name_le)
self.main_layout.addWidget(self.settings_btn)
if self.show_spinner:
self.main_layout.addWidget(self.spinner)
self.main_layout.addWidget(self.start_btn)
[docs] def connectRunner(self, runner):
TaskUIWidget.connectRunner(self, runner)
self.spinner.connectRunner(runner)
self.name_le.connectRunner(runner)
self.settings_btn.connectRunner(runner)
[docs] def disconnectRunner(self):
self.settings_btn.disconnectRunner()
self.spinner.disconnectRunner()
self.name_le.disconnectRunner()
TaskUIWidget.disconnectRunner(self)
[docs] def onNameChanged(self):
self.runner.setCustomName(self.name_le.text())
[docs] def onRunnerStateChanged(self):
if not self.runner.allow_concurrent:
self.start_btn.setEnabled(not self.runner.isRunning())
[docs]class MiniTaskBar(TaskBar):
"""
A narrow version of the TaskBar for narrow or dockable panels. No functional
difference.
"""
[docs] def setup(self):
TaskBar.setup(self)
self.main_layout = QtWidgets.QVBoxLayout()
[docs] def layOut(self):
TaskUIWidget.layOut(self)
self.main_layout.addWidget(self.name_lbl)
self.run_layout = QtWidgets.QHBoxLayout()
self.main_layout.addLayout(self.run_layout)
self.run_layout.addWidget(self.name_le)
self.run_layout.addWidget(self.start_btn)
self.line = QtWidgets.QFrame()
self.line.setFrameShape(QtWidgets.QFrame.HLine)
self.line.setFrameShadow(QtWidgets.QFrame.Sunken)
self.main_layout.addWidget(self.line)
self.bottom_layout = QtWidgets.QHBoxLayout()
self.bottom_layout.addStretch()
self.bottom_layout.addWidget(self.settings_btn)
if self.show_spinner:
self.bottom_layout.addWidget(self.spinner)
self.main_layout.addLayout(self.bottom_layout)
self.main_layout.setSpacing(3)
self.bottom_layout.setSpacing(0)
self.run_layout.setSpacing(0)
#===============================================================================
# Settings Button
#===============================================================================
#===============================================================================
# Task table
#===============================================================================
[docs]class TaskTableColumns(object):
"""
Columns object expected by table_helper
"""
HEADERS = ['Task Name', 'Status']
NUM_COLS = len(HEADERS)
NAME, STATUS = list(range(NUM_COLS))
[docs]class TaskTableModel(TaskUIMixin, table_helper.RowBasedTableModel):
"""
The table model for representing multiple tasks and their current statuses
"""
COLUMN = TaskTableColumns
ROW_CLASS = tasks.AbstractTaskWrapper
[docs] def __init__(self):
table_helper.RowBasedTableModel.__init__(self)
self.runner = None
[docs] def onRunnerStateChanged(self):
# Whenever the runner state is changed, the entire table is reloaded.
# This shouldn't be a problem as the table is generally small.
TaskUIMixin.onRunnerStateChanged(self)
self.loadData(self.runner.tasks())
@table_helper.data_method(QtCore.Qt.DisplayRole)
def _data(self, col, task, role):
if col == self.COLUMN.NAME:
return task.getName()
if col == self.COLUMN.STATUS:
return task.status()
# settings are disabled since we never need to serialize the Task Table and
# it causes for certain types of tasks.
[docs] def af2SettingsGetValue(self):
return
[docs] def af2SettingsSetValue(self, value):
return
[docs]class TaskTableView(QtWidgets.QTableView):
pass