import copy
import os
import traceback
import IPython
from schrodinger import get_maestro
from schrodinger.job import jobcontrol
from schrodinger.models import json
from schrodinger.models import mappers
from schrodinger.models import paramtools
from schrodinger.models import presets
from schrodinger.Qt import QtCore
from schrodinger.Qt import QtGui
from schrodinger.tasks import gui
from schrodinger.tasks import jobtasks
from schrodinger.tasks import taskmanager
from schrodinger.tasks import tasks
from schrodinger.ui.qt import utils as qt_utils
from schrodinger.ui.qt.widgetmixins.basicmixins import StatusBarMixin
from schrodinger.utils import preferences
from schrodinger.utils import scollections
maestro = get_maestro()
IN_DEV_MODE = 'SCHRODINGER_SRC' in os.environ
DEFAULT = object()
DARK_GREEN = QtGui.QColor(QtCore.Qt.darkGreen)
LOAD_PANEL_OPTIONS = 'Load Panel Options...'
SAVE_PANEL_OPTIONS = 'Save Panel Options...'
JOB_SETTINGS = 'Job Settings...'
PREFERENCES = 'Preferences...'
WRITE_STU_FILE = 'Write STU ZIP File'
WRITE = 'Write'
RESET = 'Reset'
RESET_THIS_TASK = 'Reset This Task'
RESET_ENTIRE_PANEL = 'Reset Entire Panel'
START_DEBUGGER = 'Start debugger...'
START_DEBUGGER_GUI = 'Start debugger GUI...'
[docs]class PanelMixin(mappers.MapperMixin):
"""
PanelMixin makes a widget act as a panel - it supports a panel singleton,
and expects to be shown as a window rather than an embedded widget.
Requires ExecutionMixin
"""
_singleton = None
SHOW_AS_WINDOW = True
# Whether to enable panel presets. The panel presets feature isn't
# complete yet so we hide it by default.
PRESETS_FEATURE_FLAG = False
[docs] @classmethod
def getPanelInstance(cls, create=True):
"""
Return the singleton instance of this panel, creating one if necessary.
:param create: Whether to create an instance of the panel if none exists
:type create: bool
:return: instance of this panel.
:rtype: `PanelMixin`
"""
if not isinstance(cls._singleton, cls):
# If the singleton hasn't been initialized or if it has been
# initialized as a superclass instance via inheritance.
cls._singleton = None
if cls._singleton is None and create:
cls._singleton = cls()
return cls._singleton
[docs] @classmethod
def panel(cls, blocking=False, modal=False, finished_callback=None):
"""
Open an instance of this class.
For full argument documentation, see `ExecutionMixin.run`.
"""
singleton = cls.getPanelInstance()
singleton.run(blocking=blocking,
modal=modal,
finished_callback=finished_callback)
return singleton
[docs] def initSetUp(self):
super().initSetUp()
if self.PRESETS_FEATURE_FLAG:
self._preset_mgr = self._makePresetManager()
[docs] def initSetDefaults(self):
super().initSetDefaults()
if self.PRESETS_FEATURE_FLAG and self._preset_mgr.getDefaultPreset():
try:
self._preset_mgr.loadDefaultPreset(self.model)
except Exception:
self.error("Encountered an error while trying to load the "
"settings for this panel. Unsetting the default "
"presets.")
if IN_DEV_MODE:
traceback.print_stack()
self._preset_mgr.clearDefaultPreset()
self.initSetDefaults()
def _makePresetManager(self):
panel_name = type(self).__name__
return presets.PresetManager(panel_name, self.model_class)
[docs]class CleanStateMixin:
"""
Mixin for use with `PanelMixin`. Implements two methods for saving and
reverting changes to the model. Automatically saves the state of the model
when a panel is run. Subclasses are responsible for calling `discardChanges`
at the right time (e.g. when a user hits the cancel button)
"""
[docs] def run(self, *args, **kwargs):
self.saveCleanState()
super().run(*args, **kwargs)
[docs] def setModel(self, model):
super().setModel(model)
self.saveCleanState()
[docs] def saveCleanState(self):
"""
Copy the model as a clean state. Next time `discardChanges` is called,
the model will be updated to reflect this current state.
"""
self._clean_state = copy.deepcopy(self.model)
[docs] def discardChanges(self):
"""
Revert the model to the value it was the last time `saveCleanState`
was called.
"""
if self._clean_state is None:
raise RuntimeError('No restore state set.')
else:
self.model.setValue(self._clean_state)
[docs]class TaskPanelMixin(PanelMixin, StatusBarMixin):
"""
OVERVIEW
========
A panel where the overall panel is associated with one or more panel tasks.
One task is active at any time; this task will be sync'ed to the panel state
and the bottom taskbar of the panel will be associated with the active task
(i.e. job options will pertain to the active task, and clicking the run
button will start that task).
PANEL TASKS
===========
A panel task is a task that is launched via the taskbar at the bottom of the
panel and generally represents the main purpose of the panel. There may be
multiple panel tasks for panels that have more than one mode, and there is
always one "active" panel task. The UX for selecting the active task of the
panel is determined by each panel, but typically is done with a combobox or
set of radio buttons near the top of the panel.
There is a taskmanager for each panel task. The taskmanager handles naming
of the tasks and creating a new instance of the panel task each time a panel
task is started. The taskmanager also provides access to previously started
tasks as well as signals for when any instance of that panel task changes
status or finshes.
A panel task's naming can be customized with `setStandardBasename`.
The TaskPanelMixin provides a standard taskbar for each panel task; a custom
taskbar can be set by overriding `_makeTaskBar`.
Similarly, a standard config dialog is provided for any panel tasks that are
jobtasks. The config dialog may be customized by overriding
`_makeConfigDialog`
PREFERENCES_KEY
===============
TaskPanelMixin persists job settings (accessible through the "Job Settings"
item in the gear menu) between sessions. The job settings are saved whenever
the config dialog is closed and loaded in `initFinalize`. The settings
are saved using the key `PREFERENCES_KEY` which defaults to the class name.
Subclasses should overwrite `PREFERENCES_KEY` so the key is stable and
unique in case the class name changes or another panel is created with
the same name.
DEPENDENCIES: widgetmixins.MessageBoxMixin
:ivar taskWritten: A signal emitted when a task is successfully written.
Emits the task instance that was written.
"""
PANEL_TASKS = tuple()
taskWritten = QtCore.pyqtSignal(tasks.AbstractTask)
[docs] def initSetUp(self):
super().initSetUp()
# Initialize the preference handler
pref_handler = preferences.Preferences(preferences.SCRIPTS)
pref_handler.beginGroup(self.PREFERENCES_KEY)
self._pref_handler = pref_handler
self._taskmans = scollections.IdDict()
self._taskbars = scollections.IdDict()
self._active_task = self.PANEL_TASKS[0]
for panel_task in self.PANEL_TASKS:
if not isinstance(panel_task, tasks.AbstractTask):
err_msg = ("All tasks in PANEL_TASKS must be abstract tasks. "
f'Got {panel_task} instead.')
raise ValueError(err_msg)
self._addPanelTask(panel_task)
self.setActiveTask(self.PANEL_TASKS[0])
[docs] def initSetDefaults(self):
old_base_names = []
for panel_task in self.PANEL_TASKS:
old_base_names.append(
self.getTaskManager(panel_task).namer.base_name)
super().initSetDefaults()
for panel_task, name in zip(self.PANEL_TASKS, old_base_names):
self.getTaskManager(panel_task).namer.base_name = name
self.getTaskManager(panel_task).uniquifiyTaskName()
[docs] def initFinalize(self):
super().initFinalize()
for task in self.PANEL_TASKS:
if self.getTaskBar(task) is None:
continue
config_dialog = self.getTaskBar(task).config_dialog
if (config_dialog is not None and
config_dialog.windowTitle() == ''):
window_title = f'{self.windowTitle()} - Job Settings'
config_dialog.setWindowTitle(window_title)
self._loadJobConfigPreferences()
@property
def PREFERENCES_KEY(self):
return type(self).__name__
def _loadJobConfigPreferences(self):
"""
Load persistent job config settings.
If the job config settings are from an older version then any error
raised during deserialization will be suppressed.
"""
try:
config_json_str = self._pref_handler.get("job_config")
except KeyError:
return
current_job_config = self.getTask().job_config
try:
configs = json.loads(config_json_str)
for idx, config in enumerate(configs):
panel_task = self.PANEL_TASKS[idx]
JobConfigClass = type(panel_task.job_config)
current_job_config = self.getTask(panel_task).job_config
deserialized_config = json.loads(config,
DataClass=JobConfigClass)
paramtools.selective_set_value(deserialized_config,
current_job_config,
exclude=[JobConfigClass.jobname])
except Exception:
print("Error while loading saved job settings. Default job "
"settings have been restored.")
config_json_str = self._pref_handler.remove("job_config")
traceback.print_exc()
return
def _saveJobConfigPreferences(self):
pref_handler = self._pref_handler
job_configs = []
for abstract_panel_task in self.PANEL_TASKS:
task = self.getTask(abstract_panel_task)
job_configs.append(json.dumps(task.job_config))
config_json_str = json.dumps(job_configs)
pref_handler.set('job_config', config_json_str)
def _onJobLaunchFailure(self, task):
qt_utils.show_job_launch_failure_dialog(task.failure_info.exception)
def _onTaskStatusChanged(self):
task = self.sender()
if task.status is task.FAILED and isinstance(
task.failure_info.exception, jobcontrol.JobLaunchFailure):
self._onJobLaunchFailure(task)
[docs] def setModel(self, model):
#TODO: remove callbacks from old model
super().setModel(model)
if model is None:
return
for abstract_task in self.PANEL_TASKS:
self._taskmans[abstract_task].setNextTask(
abstract_task.getParamValue(model))
for task, func, order in self._getTaskPreprocessors(model):
task.addPreprocessor(func, order=order)
for task, func, order in self._getTaskPostprocessors(model):
task.addPostprocessor(func, order=order)
def _savePanelOptionsSlot(self):
# To prevent circular import, import here.
from schrodinger.ui.qt.presets import save_presets_dialog
dlg = save_presets_dialog.SavePresetsDialog(self._preset_mgr,
self.model)
dlg.setWindowTitle(f'Save {self.windowTitle()} Options')
dlg.run(modal=True, blocking=True)
def _managePanelOptionsSlot(self):
from schrodinger.ui.qt.presets import manage_presets_dialog
dlg = manage_presets_dialog.ManagePresetsDialog(self._preset_mgr,
self.model)
dlg.setWindowTitle(f'Manage {self.windowTitle()} Options')
dlg.run(modal=True, blocking=True)
def _jobSettingsSlot(self):
self.getTaskBar().showConfigDialog()
def _preferencesSlot(self):
if maestro:
maestro.command("showpanel prefer:jobs_starting")
def _runSlot(self):
taskbar = self.getTaskBar()
taskbar.start_btn.setEnabled(False)
self.status_bar.showMessage("Submitting job...")
try:
task = self.getTask()
with qt_utils.JobLaunchWaitCursorContext():
task_started = gui.start_task(task, parent=self)
if task_started:
self.status_bar.showMessage("Job started", 3000, DARK_GREEN)
else:
self.status_bar.clearMessage()
except:
self.status_bar.clearMessage()
raise
finally:
taskbar.start_btn.setEnabled(True)
def _writeSlot(self):
task = self.getTask()
success = gui.write_task(task, parent=self)
if success:
self.status_bar.showMessage(f"Job written to {task.getTaskDir()}",
5000, DARK_GREEN)
self.getTaskManager().loadNewTask()
self.taskWritten.emit(task)
def _writeStuZipFileSlot(self):
task = self.getTask()
task.writeStuZipFile()
def _resetCurrentTask(self):
self.getTask().reset()
def _resetEntirePanelSlot(self):
self.initSetDefaults()
def _startDebuggerSlot(self):
self._startDebugger()
def _startDebuggerGuiSlot(self):
self.debug()
def _startDebugger(self):
QtCore.pyqtRemoveInputHook()
IPython.embed()
QtCore.pyqtRestoreInputHook()
[docs] def setStandardBaseName(self, base_name, panel_task=None):
"""
Set the base name used for naming tasks.
:param base_name: The new base name to use
:type base_name: str
:param index: The abstract panel task to set the standard basename for.
Must be a member of `PANEL_TASKS`. If not provided, set the basename
for the currently active task.
:type index: tasks.AbstractTask
"""
self.getTaskManager(panel_task).setStandardBaseName(base_name)
def _getTaskPreprocessors(self, model):
"""
Return a list of tuples used to connect or disconnect preprocessors.
"""
return self._getTaskProcessors(model, tasks.preprocessor)
def _getTaskPostprocessors(self, model):
"""
Return a list of tuples used to connect or disconnect postprocessors.
"""
return self._getTaskProcessors(model, tasks.postprocessor)
def _getTaskProcessors(self, model, marker):
"""
Return a list of tuples used to connect or disconnect processors.
:raises RuntimeError: if the defined processor callback tuples are
invalid
:param model: a model instance
:type model: parameters.CompoundParam
:param marker: a processor marker, either `tasks.preprocessor` or
`tasks.postprocessor`
:type marker: tasks._ProcessorMarker
:return: a list of (task, callback, order) tuples, where:
- `task` is the task associated with the processor
- `callback` is the processor itself
- `order` is a number used to determine the order in which the
processors are run
:rtype: list[tuple[tasks.AbstractTask, typing.Callable, float]]
"""
if marker == tasks.preprocessor:
callback_tuples = self.defineTaskPreprocessors(model)
proc_str = 'Preprocess'
elif marker == tasks.postprocessor:
callback_tuples = self.defineTaskPostprocessors(model)
proc_str = 'Postprocess'
cleaned_callback_tuples = []
for callback_tuple in callback_tuples:
if len(callback_tuple) not in (2, 3):
msg = (f'{proc_str} callbacks must be defined as a tuple of'
' (task, callback, order), with the order optional.'
f' Instead, got {callback_tuple}.')
raise RuntimeError(msg)
task, callback = callback_tuple[0:2]
try:
order = callback_tuple[2]
except IndexError:
order = None
cleaned_callback_tuples.append((task, callback, order))
return cleaned_callback_tuples
[docs] def defineTaskPreprocessors(self, model):
"""
Return a list of tuples containing a task and an associated preprocesor.
To include preprocessors, override this method in a subclass. Example::
def defineTaskPreprocessors(self, model):
return [
(model.search_task, self._validateSearchTerms),
(model.email_task, self._compileAddresses)
]
:param model: a model instance
:type model: parameters.CompoundParam
:return: a list of (task, method) tuples
:rtype: list[tuple[tasks.AbstractTask, typing.Callable]]
"""
return []
[docs] def defineTaskPostprocessors(self, model):
"""
Return a list of tuples containing a task and an associated
postprocessor.
The signature of this method is identical to that of
`defineTaskPreprocessors()`.
:param model: a model instance
:type model: parameters.CompoundParam
:return: a list of (task, method) tuples
:rtype: list[tuple[tasks.AbstractTask, typing.Callable]]
"""
return []
def _addPanelTask(self, abstract_task):
taskman = taskmanager.TaskManager(type(abstract_task),
directory_management=True)
taskman.nextTask().statusChanged.connect(self._onTaskStatusChanged)
def connect_task(task):
task.statusChanged.connect(self._onTaskStatusChanged)
taskman.newTaskLoaded.connect(connect_task)
self._taskmans[abstract_task] = taskman
taskbar = self._makeTaskBar(abstract_task)
if not taskbar:
return
taskbar.startRequested.connect(self._runSlot)
config_dialog = self._makeConfigDialog(abstract_task)
if config_dialog is not None:
taskbar.setConfigDialog(config_dialog)
config_dialog.finished.connect(self._onConfigDialogClosed)
self._taskbars[abstract_task] = taskbar
taskbar.hide()
self.bottom_middle_layout.addWidget(taskbar)
actions = self.getSettingsMenuActions(abstract_task)
if actions is not None:
taskbar.setSettingsMenuActions(actions)
taskbar.setModel(taskman)
def _makePresetManager(self):
panel_name = type(self).__name__
return presets.TaskPanelPresetManager(panel_name, self.model_class,
self.PANEL_TASKS)
def _makeTaskBar(self, panel_task):
"""
Create and return the taskbar to be used for `panel_task`. This is
called once per task in `self.PANEL_TASKS`.
Subclasses can override this method to return customized taskbars.
Returned taskbars must be a subclass of `taskwidgets.AbstractTaskBar`.
Subclasses can also return `None` if no taskbar should be used for
a particular panel task.
:param panel_task: The panel task to create a task bar for.
:type panel_task: tasks.AbstractTask
"""
from schrodinger.ui.qt.tasks import taskwidgets
if jobtasks.is_jobtask(panel_task):
taskbar = taskwidgets.JobTaskBar(parent=self)
else:
taskbar = taskwidgets.TaskBar(parent=self)
return taskbar
def _makeConfigDialog(self, panel_task):
"""
Create and return the config dialog to be used for `panel_task`. This
is called once per task in `self.PANEL_TASKS`. If this returns `None`,
then the `panel_task` will not have a config dialog.
Subclasses can override this method to return customized config dialogs.
:param panel_task: The panel task to create a config dialog for.
:type panel_task: tasks.AbstractTask
"""
from schrodinger.ui.qt.tasks import configwidgets
if jobtasks.is_jobtask(panel_task):
config_dlg = configwidgets.ConfigDialog()
return config_dlg
else:
return None
def _onConfigDialogClosed(self, accepted):
if accepted:
self._saveJobConfigPreferences()
[docs] def setActiveTask(self, new_active_task):
"""
Set the currently active task. Expects a task from `PANEL_TASKS`.
:param new_active_task: Abstract task
"""
for index, task in enumerate(self.PANEL_TASKS):
if task is new_active_task:
break
else:
raise ValueError("Unexpected value: ", new_active_task)
taskbar = self.getTaskBar(self._active_task)
if taskbar:
taskbar.setVisible(False)
taskbar = self.getTaskBar(new_active_task)
if taskbar:
taskbar.setVisible(True)
self.getTaskManager(new_active_task).uniquifiyTaskName()
self._active_task = new_active_task
[docs] def activeTask(self):
"""
Return the currently active task.
:return: The currently active task from `PANEL_TASKS`
"""
return self._active_task
[docs] def getTask(self, panel_task=None):
"""
Gets the task instance for a specified panel task. This is the task
instance that will be run the next time the corresponding start button
is clicked.
:param panel_task: The abstract task from `PANEL_TASKS` for which to
get the next task instance. If None, will return the next task
instance for the active task.
"""
return self.getTaskManager(panel_task).nextTask()
[docs] def getTaskBar(self, panel_task=None):
"""
Gets the taskbar for a panel task.
:param panel_task: The abstract task from `PANEL_TASKS` for which to
get the taskbar. If None, will return the taskbar for the active
panel task.
"""
if panel_task is None:
panel_task = self.activeTask()
return self._taskbars.get(panel_task)
[docs] def getTaskManager(self, panel_task=None):
"""
Gets the taskmanager for a panel task.
:param panel_task: The abstract task from `PANEL_TASKS` for which to get
the taskmanager. If None, will return the taskmanager for the active
task.
"""
if panel_task is None:
panel_task = self.activeTask()
return self._taskmans[panel_task]
[docs]class KnimeMixin:
"""
A mixin for creating Knime versions of panels. This mixin will hide the
taskbars and replace them with an "OK" and "Cancel" button.
SUBCLASSING
===========
All Knime panels _must_ implement panel presets. Generally, subclasses
will also implement some behavior to hide widgets that are not relevant
to a Knime node as well, though this is not strictly required.
If a Knime panel needs additional arguments, they can be specified through
an override of `runKnime`, passing any additional arguments to the
super method as a keyword arguent. All `runKnime` arguments will be
accessible in the instance variable `_knime_args` dictionary.
BUTTON BEHAVIOR
===============
The "OK" button will attempt to write a task. If a preprocessor fails then
an error will show similar to the normal panel. If the preprocessing
succeeds, then a job will be written and the settings for the panel
will be saved.
If the "Cancel" button is pressed, the panel will just be closed without
writing a job or saving settings.
KNIME
=====
The `runKnime` static method is the entry point to using Knime panels,
and are generally invoked here:
https://opengrok.schrodinger.com/xref/knime-src/scripts/service.py?r=61deff03
"""
gui_closed = QtCore.pyqtSignal()
[docs] def __init__(self, *args, _knime_args=None, **kwargs):
if _knime_args is None:
_knime_args = {}
self._knime_args = _knime_args
super().__init__(*args, **kwargs)
settings_fname = _knime_args['settings_fname']
if settings_fname and os.path.exists(settings_fname):
self._preset_mgr.loadPresetFromFile(settings_fname, self.model)
[docs] def initSetOptions(self):
super().initSetOptions()
self.std_btn_specs = {
self.StdBtn.Ok: self._okSlot,
self.StdBtn.Cancel: self._cancelSlot
}
def _okSlot(self):
task = self.getTask()
task.name = self._knime_args['jobname']
success = gui.write_task(task, parent=self)
if success:
self._preset_mgr.savePresetToFile(
self._knime_args['settings_fname'], self.model)
self.close()
def _cancelSlot(self):
pass
def _makeTaskBar(self, panel_task):
return None
[docs] def closeEvent(self, event):
super().closeEvent(event)
self.gui_closed.emit()
[docs] @classmethod
def runKnime(cls, *, jobname: str, settings_fname: str, **kwargs):
"""
Call this static method to instantiate this panel in KNIME mode.
:param jobname: The basename to save the written job to.
The job file will be written as "<jobname>.sh"
:param settings_fname: The filename to use save the settings of the panel.
If the settings already exist, then they are used to populate the
panel.
"""
if not cls.PRESETS_FEATURE_FLAG:
err_msg = (
f"{cls.__name__} does not have presets implemented. "
"Knime panels must have presets implemented in order to run "
"correctly.")
raise RuntimeError(err_msg)
kwargs['jobname'] = jobname
kwargs['settings_fname'] = settings_fname
inst = cls(_knime_args=kwargs)
inst.run()
return inst