import os
import re
import shlex
import sys
import warnings
import zipfile
import schrodinger
# Other Schrodinger modules
from schrodinger import structure
# Install the appropriate exception handler
from schrodinger.infra import exception_handler
from schrodinger.infra import jobhub
from schrodinger.job import jobcontrol
from schrodinger.job import jobhandler
from schrodinger.job import jobwriter
from schrodinger.job import launcher
from schrodinger.job import launchparams
from schrodinger.Qt import QtCore
from schrodinger.Qt import QtGui
from schrodinger.Qt import QtWidgets
from schrodinger.Qt.QtCore import Qt
from schrodinger.ui import maestro_ui
# Original Appframework modules
from schrodinger.ui.qt import config_dialog
from schrodinger.ui.qt import input_selector
from schrodinger.ui.qt import jobwidgets
from schrodinger.ui.qt import swidgets
from schrodinger.ui.qt import utils as qt_utils
from schrodinger.ui.qt.appframework2 import appmethods
from schrodinger.ui.qt.appframework2 import baseapp
from schrodinger.ui.qt.appframework2 import debug
from schrodinger.ui.qt.appframework2 import jobnames
from schrodinger.ui.qt.appframework2 import jobs
from schrodinger.ui.qt.appframework2 import maestro_callback
from schrodinger.ui.qt.appframework2 import settings
from schrodinger.ui.qt.appframework2 import tasks
from schrodinger.ui.qt.appframework2 import validation
from schrodinger.ui.qt.appframework2 import validators
# Appframework2 modules
from schrodinger.ui.qt.appframework2.application import \
start_application # noqa: F401
from schrodinger.ui.qt.appframework2.jobnames import JobnameType # noqa: F401
from schrodinger.ui.qt.appframework2.markers import MarkerMixin
from schrodinger.ui.qt.appframework2.validation import validator # noqa: F401
# For use by panels that import af2:
from schrodinger.ui.qt.config_dialog import DISP_APPEND # noqa: F401
from schrodinger.ui.qt.config_dialog import DISP_FLAG_FIT # noqa: F401
from schrodinger.ui.qt.config_dialog import DISP_IGNORE # noqa: F401
from schrodinger.ui.qt.config_dialog import ConfigDialog # noqa: F401
from schrodinger.ui.qt.forcefield import ffselector
from schrodinger.ui.qt.forcefield import forcefield
from schrodinger.ui.qt.standard import constants
from schrodinger.ui.qt.standard_widgets import statusbar
from schrodinger.ui.qt.utils import AcceptsFocusPushButton
from schrodinger.ui.qt.utils import ButtonAcceptsFocusMixin
from schrodinger.utils import cmdline
from schrodinger.utils import fileutils
from schrodinger.utils import mmutil # noqa: F401
maestro = schrodinger.get_maestro()
exception_handler.set_exception_handler()
STU_URL = 'https://stu.schrodinger.com/test/add/?automated_cmd=%s'
LAUNCHSCRIPT = 0
LAUNCHDRIVER = 1
LAUNCHMANUAL = 2
FULL_START = 0
ONLY_WRITE = 1
DARK_GREEN = QtGui.QColor(Qt.darkGreen)
#=========================================================================
# Appframework2 App Class
#=========================================================================
AppSuper = baseapp.ValidatedPanel
[docs]class App(maestro_callback.MaestroCallbackMixin, MarkerMixin,
settings.SettingsPanelMixin, AppSuper):
_singleton = None
[docs] @classmethod
def panel(cls, run=True):
"""
Launch a singleton instance of this class. If the panel has already
been instantiated, the existing panel instance will be re-opened and
brought to the front.
:param run: Whether to launch the panel
:type run: bool
:return: The singleton panel instance
:rtype: App
"""
if cls._singleton is None or not isinstance(cls._singleton, cls):
# The isinstance check covers cases of panel inheritance
cls._singleton = cls()
if run:
cls._singleton.run()
return cls._singleton
[docs] def __init__(self, **kwargs):
self.bottom_bar = None
self.app_methods = None
self.input_selector = None
self.main_taskwidgets = []
self.main_runners = []
self.all_runners = []
self.current_runner_index = None
super(App, self).__init__(**kwargs)
self.start_mode = FULL_START
[docs] @classmethod
def runKnime(cls,
input_selector_file=None,
workspace_st_file=None,
jobname=None,
run=True,
load_settings=True,
panel_state_file=None):
"""
Call this static method to instantiate this panel in KNIME mode - where
OK & Cancel buttons are shown at the bottom. Pressing OK button cases
the job files to be written to the CWD.
:param input_selector_file: the filename to be fed into the input
selector, replacing interactive input from the user. Required if
the panel contains an input selector.
:type input_selector_file: str
:param workspace_st_file: the filename containing the
`schrodinger.structure.Structure` that replaces the workspace
structure in a Maestro session.
:type workspace_st_file: str
:param jobname: Jobname for the panel
:type jobname: str
:param run: Whether to launch the panel. If False, just returns the
panel instance without starting the event loop.
:type run: bool
:param load_settings: Whether to load previous settings for the
given jobname from the CWD.
:type load_settings: bool
:param panel_state_file: Unused (added for backwards compatability)
"""
instance = cls(in_knime=True, workspace_st_file=workspace_st_file)
# Set input file
if input_selector_file:
instance.input_selector.setFile(input_selector_file)
# When we call setFile() in above line, the input_changed signal is
# not getting emitted. Setting 'tracking' to True also does not
# help as it is only applicable for visible widgets and this
# input_selector is hidden in knime. So emitting input_changed
# explicitly here
instance.input_selector.input_changed.emit()
# Set the jobname and load settings
if jobname:
instance.setJobname(jobname)
if load_settings:
# Load panel settings if there any in the CWD for the jobname:
instance.loadSettings(jobname)
# Clear any status message set
instance.status_bar.clearMessage()
# Set MODE_SUBPANEL as allowed mode to be used from KNIME nodes and
# additionally set MODE_STANDALONE also for testing from commandline
instance.allowed_run_modes = [
baseapp.MODE_SUBPANEL, baseapp.MODE_STANDALONE
]
if run:
instance.run()
return instance
[docs] def setPanelOptions(self):
"""
Configure the panel by setting instance variables here. Always call the
parent method. Panel options:
self.maestro_dockable - whether this panel should be dockable in the
Maestro main window. Setting to false will prevent the panel from
docking regardless of Maestro preference. When setting it to true, if
Maestro Preference allows docking of panels, it will dock the panel
on the right-hand side of the main window if "Location" is set to
"Main window", or a floating window if "Location" is set to
"Floating window". Default is False.
self.title - string to display in the window title bar
self.ui - a Ui_Form instance defining the main ui, default None
self.allowed_run_modes - subset of [MODE_MAESTRO, MODE_STANDALONE,
MODE_SUBPANEL, MODE_CANVAS] defining how the panel may be run.
Default is all.
self.help_topic - string defining the help topic. Default ''
self.input_selector_options - dict of options for the common input
selector widget. Default is an empty dict, meaning do not add an
input selector
self.add_main_layout_stretch - bool of whether to add a stretch to the
main layout under the main ui (if self.ui exists). Default is True
"""
AppSuper.setPanelOptions(self)
self.input_selector_options = {}
self.help_topic = ''
self.add_main_layout_stretch = True
[docs] def setup(self):
AppSuper.setup(self)
self.input_selector_layout = swidgets.SVBoxLayout()
self.main_layout = swidgets.SVBoxLayout()
self.bottom_layout = swidgets.SVBoxLayout()
self.app_methods = appmethods.MethodsDict(self)
self.status_bar = StatusBar(self)
self.status_bar.status_shrunk.connect(self._statusShrunk)
self.progress_bar = self.status_bar.progress_bar
if self.input_selector_options:
self.createInputSelector()
if self.in_knime:
self._createKnimeBottomBar()
elif self.app_methods:
self.createBottomBar()
# For jobs that were not launched using launchJobCmd() but
# that are being tracked by a call to trackJobProgress(), we need to
# use a timer to periodically query job control about the progress
# of the job
timer = QtCore.QTimer()
timer.timeout.connect(self._periodicUpdateProgressBar)
timer.setInterval(1000)
self.progress_bar_timer = timer
[docs] def setDefaults(self):
self._configurePanelSettings()
AppSuper.setDefaults(self)
[docs] def layOut(self):
AppSuper.layOut(self)
if self.main_taskwidgets:
self.setCurrentTask(0)
self.panel_layout.addLayout(self.input_selector_layout)
self.panel_layout.addLayout(self.main_layout)
self.panel_layout.addLayout(self.bottom_layout)
if self.input_selector:
self.input_selector_layout.addWidget(self.input_selector)
self.input_selector_layout.addWidget(swidgets.SHLine())
if self.bottom_bar:
self.bottom_line = swidgets.SHLine()
self.bottom_layout.addWidget(self.bottom_line)
self.bottom_layout.addWidget(self.bottom_bar)
self.bottom_layout.addWidget(self.status_bar)
if self.ui:
self.main_layout.addWidget(self.ui_widget)
if self.add_main_layout_stretch:
self.main_layout.insertStretch(-1)
#===========================================================================
# Task Runner support
#===========================================================================
[docs] def addMainTaskRunner(self, runner, taskwidget):
"""
A "main" task runner is a runner that is operated by a task widget
(generally a job bar) at the very bottom of the panel. A panel may
have more than one main task, but there is always one that is the
"current" task. This is useful for panels that have multiple modes, with
each mode launching a different job.
The related method, self.setCurrentTask(), is used to switch between
main runners that have been added via this function.
:param runner: the task runner
:type runner: tasks.AbstractTaskRuner
:param taskwidget: the associated task widget
:type taskwidget: taskwidgets.TaskUIMixin
"""
self.setupTaskRunner(runner, taskwidget)
self.main_runners.append(runner)
self.main_taskwidgets.append(taskwidget)
self.bottom_layout.addWidget(taskwidget)
return runner
[docs] def setCurrentTask(self, index):
"""
Selects the current main task for the panel. Switching to a new task
involves several steps. These are 1) saving the current panel state to
the task runner, 2) hiding the current task widget (and all others), 3)
showing the widget for the new task, and 4) setting the panel state to
correspond to the new task runner's settings.
:param index: the index of the task to be selected. The index for each
main task is set sequentially from 0 as each task as added using
self.addMainTaskRunner()
:type index: int
"""
for widget in self.main_taskwidgets:
widget.setVisible(False)
current_widget = self.main_taskwidgets[index]
current_widget.setVisible(True)
if self.current_runner_index is not None:
runner = self.currentTaskRunner()
runner.pullSettings()
self.current_runner_index = index
runner = self.currentTaskRunner()
runner.pushSettings()
runner.updateStatusText()
if self.in_knime:
# Hide the job settings widgets
current_widget = self.main_taskwidgets[index]
current_widget.setVisible(False)
[docs] def currentTaskRunner(self):
if self.current_runner_index is None:
return None
return self.main_runners[self.current_runner_index]
[docs] def processTaskMessage(self, message_type, text, options=None, runner=None):
"""
This method is meant to be used as a callback to a task runner, and
provides a single point of interaction from the runner to the user.
:param message_type: the type of message being sent
:type message_type: int
:param text: the main text to show the user
:type text: str
:param options: extra options
:type caption: dict
"""
if options is None:
options = {}
caption = options.get('caption', '')
if message_type == tasks.WARNING:
QtWidgets.QMessageBox.warning(self, caption, text)
elif message_type == tasks.ERROR:
QtWidgets.QMessageBox.critical(self, caption, text)
elif message_type == tasks.QUESTION:
return self.question(text, title=caption)
elif message_type == tasks.INFO:
return self.info(text, title=caption)
elif message_type == tasks.STATUS:
if runner != self.currentTaskRunner():
return
timeout = options.get('timeout', 3000)
color = options.get('color')
self.status_bar.showMessage(text, timeout, color)
else:
raise ValueError('Unexpected message_type %d for message:\n%s' %
(message_type, text))
[docs] def setupTaskRunner(self, runner, taskwidget):
"""
Connects a task widget to a task runner and associates the runner with
this af2 panel via the panel callbacks.
This method is called by self.addMainTaskRunner() and does not need to
be called for main tasks; however, it is useful for setting up other
tasks that are not main tasks - for example, if there is a smaller job
that gets launched from a button in the middle of the panel somewhere.
:param runner: the task runner
:type runner: tasks.AbstractTaskRuner
:param taskwidget: the associated task widget
:type taskwidget: taskwidgets.TaskUIMixin
"""
runner.setCallbacks(messaging_callback=self.processTaskMessage,
settings_callback=self.processSettings)
runner.resetAllRequested.connect(self._reset)
self.all_runners.append(runner)
taskwidget.connectRunner(runner)
[docs] def resetAllRunners(self):
"""
Resets all task runners associated with this panel (main tasks and other
tasks added via setupTaskRunner). This is called from _reset() and
normally does not need to be called directly.
"""
for runner in self.all_runners:
runner.reset()
[docs] def processSettings(self, settings=None, runner=None):
"""
This method is meant to be used as a callback to a task runner. If it
is called with no arguments, it returns a dictionary of all the alieased
settings. If settings are passed, the settings are first applied to
self, and then the newly modified settings are returned.
:param settings: a settings dictionary to apply to this object
:type settings: dict or None
:param runner: the task runner that is invoking this callback. This
optional argument is necessary for per-runner grouping of settings
:type runner: tasks.AbstractTaskRuner
"""
if runner:
group = runner.runner_name
else:
group = ''
if settings is not None:
self._applySettingsFromGroup(group, settings)
return self._getSettingsForGroup(group)
def _getSettingsForGroup(self, group):
settings = self.getAliasedSettings()
filtered_settings = reduce_settings_for_group(settings, group)
return filtered_settings
def _applySettingsFromGroup(self, group, settings):
all_aliases = list(self.settings_aliases)
expanded_settings = expand_settings_from_group(settings, group,
all_aliases)
self.applyAliasedSettings(expanded_settings)
#===========================================================================
# Panel setup
#===========================================================================
def _createKnimeBottomBar(self):
self.bottom_bar = OKAndCancelBottomBar(self.app_methods)
self.bottom_bar.ok_bn.clicked.connect(self.writeStateAndClose)
self.status_bar.removeWidget(self.status_bar.status_lb)
[docs] def createBottomBar(self):
self.bottom_bar = BottomBar(self.app_methods)
self.bottom_bar.hideToolbarStyle()
def _close(self):
# This method is no longer used and will be removed in the future.
if self.dock_widget:
self.dock_widget.close()
else:
self.close()
def _prestart(self):
"""
Needed for the appmethod @prestart decorator to work
"""
def _prewrite(self):
"""
Needed for the appmethod @prewrite decorator to work
"""
def _start(self):
"""
:return: Returns False upon failure, otherwise returns nothing (None)
:rtype: False or None
"""
self.start_mode = FULL_START
if self.app_methods.preStart() is False:
return False
if not self.runValidation():
return False
self.app_methods.start()
def _read(self):
self.app_methods.read()
def _reset(self):
"""
:return: Returns False upon failure, otherwise returns nothing (None)
:rtype: False or None
"""
if self.app_methods.reset() is False: # Only False should abort
return False
self.setDefaults()
if self.input_selector:
self.input_selector._reset()
self.removeAllMarkers()
self.resetAllRunners()
def _help(self):
"""
Display the help dialog (or a warning dialog if no help can be found).
This function requires help_topic to have been given when the class was
initialized.
"""
qt_utils.help_dialog(self.help_topic, parent=self)
[docs] def closeEvent(self, event):
"""
Receives the close event and calls the panel's 'close'-decorated
appmethod. If the appmethod specifically returns False, the close event
will be ignored and the panel will remain open. All other return values
(including None) will allow the panel to proceed with closing.
This is a PyQT slot method and should not be explicitly called.
"""
if self.app_methods:
proceed_with_close = self.app_methods.close()
if proceed_with_close is False:
event.ignore()
return
super(App, self).closeEvent(event)
[docs] def showEvent(self, event):
"""
When the panel is shown, call the panel's 'show'-decorated methods.
Note that restoring a minimized panel will not trigger the 'show'
methods.
"""
super(App, self).showEvent(event)
if not event.spontaneous() and self.app_methods:
self.app_methods.show()
[docs] def cleanup(self):
if self.app_methods:
self.app_methods.source_obj = None
self.app_methods = None
self.bottom_bar.app_methods = None
self.bottom_bar = None
AppSuper.cleanup(self)
[docs] def showProgressBarForJob(self, job, show_lbl=True, start_timer=True):
"""
Show a progress bar that tracks the progress of the specified job
:param job: The job to track
:type job: `schrodinger.job.jobcontrol.Job`
:param show_lbl: If True, the job progress text description will be
shown above the progress bar. If False, the text description will not
be shown.
:type show_lbl: bool
:param start_timer: If True, the progress bar will automatically be
updated and removed when the job is complete. If False, it is the
caller's responsibility to periodically call
self.progress_bar.readJobAndUpdateProgress() and to call
self.status_bar.hideProgress() when the job is complete.
:type start_timer: bool
"""
self.status_bar.showProgress()
self.progress_bar.trackJobProgress(job, show_lbl)
if start_timer:
self.progress_bar_timer.start()
def _periodicUpdateProgressBar(self):
"""
Update the progress bar and remove it if the job has completed.
"""
complete = self.progress_bar.readJobAndUpdateProgress()
if complete:
self.progress_bar_timer.stop()
self.status_bar.hideProgress()
def _statusShrunk(self, size_diff):
"""
If the panel had to be enlarged to show the progress bar, shrink it back
down once the progress bar is hidden.
:note: If the panel wasn't at minimum height when the progress bar was
shown, then it most likely wasn't enlarged since the progress bar would
have been given existing free space. As a result, we only shrink the
panel if it is at minimum height.
"""
cur_height = self.height()
if cur_height == self.minimumHeight():
new_height = cur_height - size_diff
width = self.width()
resize = lambda: self.resize(width, new_height)
# If we call resize immediately, the panel won't "know" about the
# status bar size change and will reject the resize() call
QtCore.QTimer.singleShot(25, resize)
[docs] def getWorkspaceStructure(self):
"""
If panel is open in Maestro session, returns the
current workspace `schrodinger.strucutre.Structure`.
If panel is open from outside of Maestro, returns the self.workspace_st
if self.workspace_st_file is available. Used while running from command
line or starting the panel from KNIME.
Returns None otherwise.
:rtype: `schrodinger.structure.Structure` or None
:return: Maestro workspace structure or None
"""
if maestro:
return maestro.workspace_get()
elif hasattr(self, 'workspace_st'):
return self.workspace_st
elif self.workspace_st_file:
self.workspace_st = structure.Structure.read(self.workspace_st_file)
return self.workspace_st
else:
# This can happen when the opening the panel outside of Maestro
# without specifying workspace file
return None
[docs] def hideLayoutElements(self, layout):
"""
Hide all elements from the given layout. Used for customizing KNIME
panel wrappers.
"""
for i in reversed(list(range(layout.count()))):
item = layout.itemAt(i)
if item.layout():
self.hideLayoutElements(item.layout())
elif item.widget():
item.widget().hide()
elif item.spacerItem():
layout.removeItem(item.spacerItem())
[docs] def loadSettings(self, jobname):
"""
Load the GUI state for the job in the CWD with the given name.
Default implementation will return False.
Each KNIME panel will need to implement a custom version.
For example, the panel may want to read the <jobname.sh> file, parse
the list of command-line options, and populate the GUI accordintly.
If a panel writes key/value file, then it would need to read it here.
:return: True if panel state was restored, False if saved state was not
found.
:rtype: bool
"""
return False
[docs] def jobname(self):
"""
Return the job name currently set for the current task.
"""
if len(self.all_runners) == 0:
raise AttributeError("App.jobname() only works with tasks.")
return self.currentTaskRunner().nextName()
[docs] def setJobname(self, jobname):
"""
Set the job name for the current task.
"""
if len(self.all_runners) == 0:
raise AttributeError("App.setJobname() only works with tasks.")
self.currentTaskRunner().setCustomName(jobname)
[docs] def writeStateAndClose(self):
"""
Called when OK button button is pressed when running in KNIME mode.
Will "write" the job files for current task, and close the panel.
"""
if len(self.all_runners) == 0:
raise AttributeError(
"App.writeStateAndClose() only works with tasks.")
if self.currentTaskRunner().write() is not False:
# Validation passed and command file was written
self._close()
[docs] def readShFile(self, jobname):
"""
Reads the jobname.sh file (written by _write()) and returns the list of
command line arguments
"""
cmd_file = os.path.join(jobname + ".sh")
if not os.path.isfile(cmd_file):
return None
# Parse the jobname.sh file
with open(cmd_file, 'r') as fh:
return shlex.split(fh.read())
[docs] def validateOPLSDir(self, opls_dir=None):
"""
See `forcefield.validate_opls_dir()`
:param opls_dir: the opls dir to validate
:type opls_dir: str or None
:return: the validation result
:rtype: forcefield.OPLSDirResult
"""
return forcefield.validate_opls_dir(opls_dir, parent=self)
#=========================================================================
# Appframework2 JobApp Class
#=========================================================================
JobAppSuper = App
[docs]class JobApp(JobAppSuper):
jobCompleted = QtCore.pyqtSignal(jobcontrol.Job)
lastJobCompleted = QtCore.pyqtSignal(jobcontrol.Job)
[docs] def __init__(self, **kwargs):
self.config_dlg = None
self._old_jobname_data = None
self.last_job = None
super(JobApp, self).__init__(**kwargs)
self.orig_dir = ''
self.showing_progress_for_job = None
[docs] def setPanelOptions(self):
"""
See parent class for more options.
self.use_mini_jobbar - whether this panel use the narrow version of the
bottom job bar. This is useful for narrow panels where the regular job
bar is too wide to fit. Default: False
self.viewname - this identifier is used by the job status button so that
it knows which jobs belong to this panel. This is automatically
generated from the module and class name of the panel and so it does not
need to be set unless the module/class names are generic.
self.program_name - a human-readable text name for the job this panel
launches. This shows up in the main job monitor to help the user
identify the job. Example: "Glide grid generation". Default: "Job"
self.omit_one_from_standard_jobname - see documentation in jobnames.py
add_driverhost - If True, the backend supports running -DRIVERHOST
to specify a different host for the driver job than subjobs. Only
certain workflows support this option.
"""
JobAppSuper.setPanelOptions(self)
self.viewname = str(self)
self.use_mini_jobbar = False
self.program_name = None
self.default_jobname = 'Job'
self.omit_one_from_standard_jobname = False
self.add_driverhost = False
[docs] def getConfigDialog(self):
return None
[docs] def setup(self):
# These lines need to be executed before calling the super class'
# method
if jobhandler.is_auto_download_active():
jobhub.get_job_manager().jobDownloaded.connect(self._onJobDone)
else:
# FIXME PANEL-18802: under JOB_SERVER, panel jobs outside maestro
# won't be downloaded
jobhub.get_job_manager().jobCompleted.connect(self._onJobDone)
self.config_dlg = self.getConfigDialog()
JobAppSuper.setup(self)
if self.use_mini_jobbar:
self.status_bar.hide()
[docs] def setDefaults(self):
JobAppSuper.setDefaults(self)
if self.app_methods:
self.updateJobname()
[docs] def layOut(self):
JobAppSuper.layOut(self)
if self.config_dlg:
self.updateStatusBar()
[docs] def createBottomBar(self):
if self.use_mini_jobbar:
self.bottom_bar = MiniJobBottomBar(self.app_methods)
else:
self.bottom_bar = JobBottomBar(self.app_methods)
self.bottom_bar.jobname_le.editingFinished.connect(
self._populateEmptyJobname)
[docs] def syncConfigDialog(self):
jobname = self.jobname()
self.setConfigDialogSettings({'jobname': jobname})
self.config_dlg.getSettings()
[docs] def configDialogSettings(self):
self.config_dlg.getSettings()
return self.config_dlg.kw
[docs] def setConfigDialogSettings(self, new_values):
settings = config_dialog.StartDialogParams()
settings.__dict__.update(new_values)
self.config_dlg.applySettings(settings)
self.config_dlg.getSettings()
def _settings(self):
cd_settings = self.configDialogSettings() # Get previous settings
# Instantiate new config dialog
self.config_dlg = self.getConfigDialog()
self.setConfigDialogSettings(cd_settings) # Apply previous settings
self.syncConfigDialog() # Update the job name
orig_jobname = self.jobname()
# We don't use setJobname here because that would send the updated job
# name back to the config dialog after updating jobname_le
set_jobname = self.bottom_bar.jobname_le.setText
try:
# The jobnameChanged signal won't exist if the config dialog doesn't
# have a job name line edit
self.config_dlg.jobnameChanged.connect(set_jobname)
except AttributeError:
pass
if not self.config_dlg.activate():
self.setJobname(orig_jobname)
return
cd_settings = self.configDialogSettings()
self.updateStatusBar()
ra = self.config_dlg.requested_action
if ra == config_dialog.RequestedAction.Run:
self._start()
elif ra == config_dialog.RequestedAction.Write:
self._write()
def _start(self):
"""
Called when the "Run" button is pressed in the panel or in the
config dialog.
:return: Returns False upon failure, returns None on success.
:rtype: False or None
"""
self.start_mode = FULL_START
if not self.validForceFieldSelectorCustomOPLSDir():
return False
if self.app_methods.preStart() is False: # Only False should abort
return False
ret = self._startOrWrite()
if ret is None:
# Increment the job name
self.updateJobname()
return ret
def _writeJobFiles(self):
"""
Write job files for the current job without incrementing the jobname
field.
:return: Returns False upon failure, returns None on success.
:rtype: False or None
"""
self.start_mode = ONLY_WRITE
if not self.validForceFieldSelectorCustomOPLSDir():
return False
if self.app_methods.preWrite() is False: # Only False should abort
return False
if self._startOrWrite() is False:
return False
return None
def _write(self):
"""
Called when the "Write" action is selected by the user, and by
writeStateAndClose() and _writeSTU() methods.
:return: Returns False upon failure, returns None on success.
:rtype: False or None
"""
ret = self._writeJobFiles()
if ret is not False:
# Increment the job name on success
self.updateJobname()
return ret
def _startOrWrite(self):
"""
Combined method for starting a job or writing it to a .sh file. The
value of self.start_mode determines which to do.
:return: Returns False upon failure, returns None on success.
:rtype: False or None
"""
if not fileutils.is_valid_jobname(self.jobname()):
msg = fileutils.INVALID_JOBNAME_ERR % self.jobname()
self.warning(msg)
return False
if not self.runValidation(stop_on_fail=True):
return False
if self.config_dlg:
if not self.config_dlg.validate():
return False
is_dummy = config_dialog.DUMMY_GPU_HOSTNAME in self.status_bar.status(
)
if self.start_mode == FULL_START and is_dummy:
self.error(
"Cannot start job with dummy GPU host set. Please set a valid CPU or GPU host."
)
return False
if not jobs.CHDIR_MUTEX.tryLock():
self.warning(jobs.CHDIR_LOCKED_TEXT)
return False
self.orig_dir = os.getcwd()
if not self.in_knime:
if self.createJobDir() is False:
# User has cancelled the job start/write; we don't chdir into jobdir
jobs.CHDIR_MUTEX.unlock()
self.orig_dir = ''
return False
if self.start_mode == FULL_START:
msg = 'Submitting Job...'
elif self.start_mode == ONLY_WRITE:
msg = 'Writing Job...'
self.status_bar.showMessage(msg)
start_bn = self.bottom_bar.start_bn
start_bn.setEnabled(False)
settings_bn = self.bottom_bar.settings_bn
settings_bn.setEnabled(False)
# Force some QT event processing to ensure these state changes show up
# in the GUI - PANEL-7556
self.application.processEvents(QtCore.QEventLoop.ExcludeUserInputEvents)
if self.start_mode == FULL_START:
failed_status = 'Failed to start job'
ok_status = 'Job started'
message_duration = 3000
else:
failed_status = 'Failed to write job'
ok_status = 'Job written to ' + self.jobDir()
message_duration = 6000
failed = False
job = None
try:
if not self.in_knime:
os.chdir(self.jobDir())
if self.input_selector:
self.input_selector.original_cwd = self.orig_dir
if not self.in_knime:
# Write the <jobname>.maegz file to the job dir:
self.input_selector.setup(self.jobname())
if self.start_mode == FULL_START:
# Call the panel's start method here. It will return a Job
# object or None on success, and False on failure.
job = self.app_methods.start()
if job is False:
failed = True
job = None
elif self.start_mode == ONLY_WRITE:
# Typically the write method is the same as the start method
# for most panels. It will return None on success, and False
# on failure.
if self.app_methods.write() is False:
failed = True
except jobcontrol.JobLaunchFailure:
# NOTE: launchJobCmd() by this point has shown the error
# to the user via a warning dialog box.
failed = True
except:
# Re-raise other exceptions - typically from the start method
self.status_bar.showMessage(failed_status)
raise
finally:
if not self.in_knime:
os.chdir(self.orig_dir)
jobs.CHDIR_MUTEX.unlock()
start_bn.setEnabled(True)
settings_bn.setEnabled(True)
self.orig_dir = ''
if self.input_selector:
self.input_selector.original_cwd = None
if failed:
# Start/write method returned False (failure)
if not self.in_knime:
fileutils.force_rmtree(self.jobDir())
self.status_bar.showMessage(failed_status)
return False
# If got here, then start/write was successful
self.status_bar.showMessage(ok_status, message_duration, DARK_GREEN)
previous_jobname = self.jobname()
if self.start_mode == ONLY_WRITE: # Done with everything for writing
return None
if job is not None and maestro:
if not isinstance(job, jobcontrol.Job):
raise TypeError('Return value of start method must be a Job '
'object, None, or False.')
self.last_job = job
self.addProjectJobNote(job.JobId, previous_jobname)
elif self.last_job is not None and maestro:
# In case the start method did not return a job object, but did
# call registerJob():
self.addProjectJobNote(self.last_job.JobId, previous_jobname)
return None
[docs] def addProjectJobNote(self, job_id, jobname):
"""
Adds a note to the project annotation file.
:param job_id: The ID of the job, as assigned by Maestro
:type job_id: string
:param jobname: The name of the job, as shown in the job panel
:type jobname: string
"""
note_text = 'Starting ' + self.title + '\nJob name: ' + jobname + '\nJob ID: ' + job_id
if maestro:
maestro_hub = maestro_ui.MaestroHub.instance()
maestro_hub.addProjectLogRequested.emit(note_text)
[docs] def jobnameData(self):
"""
Provides panel settings that are to be incorporated into job names. If
self.default_jobname includes string formatting characters (i.e. %s,
{0}, etc.), then this method must be implemented. It should return a
tuple or a dictionary to be interpolated into the job name.
"""
err = ("To include panel settings in the job name, jobnameData must be"
"implemented")
raise NotImplementedError(err)
[docs] def jobnameDataChanged(self):
"""
If the job name includes panel settings, then this method should be
called whenever the relevant panel settings are modified
"""
self.updateJobname(False)
[docs] def updateJobname(self, uniquify_custom=True):
"""
Generate a new job name based on the current panel settings
:param uniquify_custom: Whether we should uniquify custom job name by
adding integers to the end. If False, only standard and modified job
names will be uniquified. (See `JobnameType` for an explanation of job
name types.)
:type uniquify_custom: bool
"""
current_jobname = self.jobname()
old_standard_jobname, new_standard_jobname = self._getStandardJobnames()
new_jobname, jobtype = jobnames.determine_jobtype(
current_jobname, old_standard_jobname, new_standard_jobname,
uniquify_custom)
uniq_jobname = jobnames.uniquify(new_jobname, jobtype, uniquify_custom,
self.omit_one_from_standard_jobname)
self.setJobname(uniq_jobname)
def _getStandardJobnames(self):
"""
Get the old and new standard job names
:rtype: tuple
Returns a tuple of:
- The standard job name using the panel settings from the last time we
ran this method. (Needed to search for the standard job name in the
current job name.)
- The standard job name using the current panel settings. (Needed to
generate the new job name.)
"""
percent_found = "%" in self.default_jobname
bracket_found = "{" in self.default_jobname
formatting_needed = percent_found or bracket_found
if percent_found:
format_name = lambda name, data: name % data
elif bracket_found:
format_name = lambda name, data: (name.format(**data) if isinstance(
data, dict) else name.format(*data))
# The first time we run this method, self._old_jobname_data will be
# None, which means we don't have anything to interpolate
if formatting_needed and self._old_jobname_data is not None:
old = format_name(self.default_jobname, self._old_jobname_data)
else:
old = self.default_jobname
if formatting_needed:
new_jobname_data = self.jobnameData()
new = format_name(self.default_jobname, new_jobname_data)
self._old_jobname_data = new_jobname_data
else:
new = self.default_jobname
return old, new
[docs] def sanitizeJobnameText(self, text):
"""
Modify the given text so it can be used in a job name. White space is
replaced with underscores and all other disallowed characters are
removed.
:param text: The text to sanitize
:type text: basestring
:return: The sanitized text
:rtype: basestring
"""
text = re.sub(r"\s+", "_", text)
text = re.sub(r"[^\w_\-\.]", "", text)
return text
def _populateEmptyJobname(self):
"""
If the user clears the job name line edit, populate it with the standard
job name
"""
jobname = self.jobname()
if not jobname:
self.updateJobname()
[docs] def writeStateAndClose(self):
"""
Will "write" the job files and close the panel.
"""
if self._write() is not False:
# Validation passed and command file was written
self._close()
[docs] def cleanup(self):
if self.app_methods:
self.mini_monitor = None
JobAppSuper.cleanup(self)
#=========================================================================
# Job Launching - General
#=========================================================================
def _getSHFilename(self):
return os.path.join(self.jobDir(), self.jobname() + '.sh')
[docs] def jobname(self):
try:
return str(self.bottom_bar.jobname_le.text())
except AttributeError:
return None
[docs] def setJobname(self, jobname):
self.bottom_bar.jobname_le.setText(jobname)
if self.config_dlg:
self.syncConfigDialog()
[docs] def writeJobCmd(self, cmdlist, job_spec=None, launch_params=None):
"""
Writes the job invocation command to a file named "<jobname>.sh" Removes
options from the command that are maestro-specific.
Note this may modify the contents of `cmdlist`
:param schrodinger.job.launchapi.JobSpecification job_spec: The job
specification for the command you want to write. This is NOT used
to write the command that is run; it is used to write a
commented-out, un-hashed, human-readable command in the `sh` file.
If `None` (which is also the default), then the human-readable
comment is not written. If it is present, launch_params must be
present too.
:param job.launchparams.LaunchParameters launch_params: Job launch
params
"""
jobwriter.write_job_cmd(cmdlist, self._getSHFilename(), self.jobDir())
# See MATSCI-10844 and MATSCI-10976
if job_spec is not None:
readable_cmdlist = get_readable_cmd_list(cmdlist, job_spec,
launch_params)
self.writeReadableCmdComment(readable_cmdlist)
[docs] def getCmdListArgValue(self, cmdlist, arg):
return cmdlist[cmdlist.index(arg) + 1]
[docs] def jobDir(self):
if self.in_knime:
return self.orig_dir
return os.path.join(self.orig_dir, self.jobname())
[docs] def createJobDir(self):
dirname = self.jobDir()
if os.path.exists(dirname):
qtext = ('The job directory, %s, already exists.\nWould you like '
'to delete its contents and continue?' % dirname)
overwrite = self.question(qtext, title='Overwrite contents?')
if not overwrite:
return False
fileutils.force_rmtree(dirname)
try:
os.mkdir(dirname)
except PermissionError:
# User has no write permissions to CWD
self.error('Permission denied; unable to create directory:\n%s' %
self.jobDir())
return False
[docs] def registerJob(self, job, show_progress_bar=False):
"""
Registers a job with the periodic job check callback and starts timer.
:param job: job to register
:type job: jobcontrol.Job
:param show_progress_bar: Whether or not to show a progress bar tracking
the job's status.
:type show_progress_bar: bool
"""
if not job:
return
self.last_job = job
self.showing_progress_for_job = job.JobId
if show_progress_bar:
show_text = False
self.showProgressBarForJob(job, show_text, start_timer=False)
else:
# Make sure we hide the progress bar in case the previous job had a
# progress bar and hasn't finished
self.status_bar.hideProgress()
self.showing_progress_for_job = None
[docs] def updateStatusBar(self):
"""
Updates the status bar.
"""
text = self.generateStatus()
self.status_bar.setStatus(text)
[docs] def generateStatus(self):
"""
Generate the text to put into the status bar
:return: The text to put into the status bar
:rtype: str
"""
cd_params = self.configDialogSettings()
if not cd_params:
return
cpus = config_dialog.get_num_nprocs(cd_params)
cd = self.config_dlg
text_items = []
if cd.options.get('host') and not cd.options.get('host_products'):
# We are not showing "Host=" status for panels that have
# multiple host menus yet.
# This is true in IFD, where CPUs are specified on a per product basis
if not cpus:
host = 'Host={0}'.format(cd_params.get('host', ''))
else:
host = 'Host={0}:{1}'.format(cd_params.get('host', ''), cpus)
text_items.append(host)
disp = cd_params.get('disp', None)
if disp and cd and cd.options['incorporation']:
first_disp = disp.split(config_dialog.DISP_FLAG_SEPARATOR)[0]
dispname = config_dialog.DISP_NAMES[first_disp]
incorporate = 'Incorporate={0}'.format(dispname)
text_items.append(incorporate)
text = ', '.join(text_items)
return text
#=========================================================================
# Job Launching - Scripts via launcher
#=========================================================================
[docs] def launchScript(
self,
script,
script_args=None,
input_files=[], # noqa: M511
structure_output_file=None,
output_files=[], # noqa: M511
aux_modules=[], # noqa: M511
show_progress_bar=False,
**kwargs):
"""
DEPRECATED, add get_job_spec_from_args() to the backend script and
launch it using launchJobCmd() or also add getJobSpec() to the panel
and launch using launchFromJobSpec().
Creates and launches a script using makeLauncher. For documentation on
method parameters, see makeLauncher below. Use this method for scripts
that do not themselves integrate with job control.
This method honors self.start_mode; it can either launch the script or
write out a job file to the job directory.
:param show_progress_bar: Whether or not to show a progress bar tracking
the job's status.
:type show_progress_bar: bool
"""
msg = ("AF2's launchScript() and makeLauncher() are deprecated. "
"Add get_job_spec_from_args() to the backend script and launch "
"it using launchJobCmd() or also add getJobSpec() to the panel "
"and launch using launchFromJobSpec().")
warnings.warn(msg, DeprecationWarning, stacklevel=2)
slauncher = self.makeLauncher(
script=script,
script_args=script_args,
input_files=input_files,
structure_output_file=structure_output_file,
output_files=output_files,
aux_modules=aux_modules,
**kwargs)
return self.launchLauncher(slauncher, show_progress_bar)
[docs] def launcherToCmdList(self, slauncher):
cmdlist = slauncher.getCommandArgs()
expandvars = slauncher._expandvars
if expandvars is None:
expandvars = True
cmdlist = jobcontrol.fix_cmd(cmdlist, expandvars)
return cmdlist
[docs] def makeLauncher(
self,
script,
script_args=[], # noqa: M511
input_files=[], # noqa: M511
structure_output_file=None,
output_files=[], # noqa: M511
aux_modules=[], # noqa: M511
**kwargs):
"""
DEPRECATED, add get_job_spec_from_args() to the backend script and
launch it using launchJobCmd() or also add getJobSpec() to the panel
and launch using launchFromJobSpec().
Create a launcher.Launcher instance using the settings defined by the
panel, its config dialog, and specified arguments. Returns a launcher
instance ready to be launched or further modified. Use this method for
scripts that do not themselves integrate with job control.
Only use this method if you need to modify the launcher before launching
it. Otherwise, the method launchScript() is preferred to create the
launcher and launch it.
:param script: Remote path to the script to be launched. See Launcher
documentation for more info. If only launching to localhost is desired,
then a local path can be specified.
:type script: str
:param script_args: arguments to be added to the script's command line
:type script_args: list of str
:param input_files: input files that will be copied to the temporary
job directory.
:type input_files: list of str
:param structure_output_file: this is the file that will be registered
with job control to incorporate at the end of the job
:type structure_output_file: str
:param output_files: additional output files to be copied back from the
temporary job directory
:type output_files: list of str
:param aux_modules: Additional modules required by the script
:type aux_modules: list of modules
:return: A prepped launcher
:rtype: Launcher
"""
if hasattr(script, '__file__'): # script is a module
import warnings
msg = ("Ability to launch scripts via imported module object "
"is deprecated. Please give the full remote path instead. "
"(if script is in search path for $SCHRODINGER/run, then "
"just the name of the script can be passed in)")
warnings.warn(msg, DeprecationWarning, stacklevel=2)
# This join is needed because under certain conditions, __file__
# is a relative path. In most cases, __file__ is a full path, in
# which case os.path.join will ignore self.orig_dir
filename = os.path.join(self.orig_dir, script.__file__)
# Fix for PANEL-5149; Tell Launcher to copy the script, as
# <filename> is a local path instead of a remote path
kwargs['copyscript'] = True
else: # script is a filename
filename = script
cd_params = self.configDialogSettings()
more_scriptargs = []
host = cd_params.get('host', 'localhost')
# af1.get_num_nprocs returns int or None (for 1 subjob)
njobs = config_dialog.get_num_nprocs(cd_params) or 1
host += ":{}".format(njobs)
threads = cd_params.get('threads')
if threads:
more_scriptargs.extend(['-TPP', str(threads)])
queue_resources = cd_params.get('queue_resources', '')
if queue_resources:
more_scriptargs.append(queue_resources)
if 'njobs' in cd_params:
more_scriptargs.extend(['-NJOBS', str(cd_params['njobs'])])
if self.runMode() == baseapp.MODE_MAESTRO:
proj = cd_params.get('proj', None)
disp = cd_params.get('disp', None)
viewname = self.viewname
else:
proj = None
disp = None
viewname = None
slauncher = launcher.Launcher(script=filename,
runtoplevel=True,
prog=self.program_name,
jobname=self.jobname(),
host=host,
proj=proj,
disp=disp,
viewname=viewname,
**kwargs)
slauncher.addScriptArgs(more_scriptargs)
for inputfile in input_files:
slauncher.addInputFile(inputfile)
if structure_output_file:
slauncher.setStructureOutputFile(structure_output_file)
for outputfile in output_files:
slauncher.addOutputFile(outputfile)
if script_args:
slauncher.addScriptArgs(script_args)
for aux_module in aux_modules:
filename = os.path.join(self.orig_dir, aux_module.__file__)
slauncher.addForceInputFile(filename)
return slauncher
[docs] def launchLauncher(self, slauncher, show_progress_bar=False):
"""
Either launches a launcher instance or writes the job invocation
command, depending on the state of self.start_mode. This allows the
panel's start method to double as a write method.
Calling launchLauncher() is only necessary if creating a customized
launcher using makeLauncher().
:param show_progress_bar: Whether or not to show a progress bar tracking
the job's status.
:type show_progress_bar: int
"""
if self.start_mode == FULL_START:
job = slauncher.launch()
self.registerJob(job, show_progress_bar)
return job
elif self.start_mode == ONLY_WRITE:
cmdlist = self.launcherToCmdList(slauncher)
self.writeJobCmd(cmdlist)
[docs] def getJobSpec(self):
raise NotImplementedError
def _addJaguarOptions(self):
"""
Returns list of cmdline options. Useful when you need to construct a
cmd for a job specification that will use parallel options for a future
jaguar job.
"""
cmd = []
cd_params = self.configDialogSettings()
threads = cd_params['threads']
cpus = cd_params['openmpcpus']
if threads:
cmd.extend(["-TPP", "{}".format(threads)])
# FIXME: Jaguar would expect -PARALLEL options, but matsci jaguar
# workflows can't accept this. Should they even support threads +
# subjobs?
#else:
# cmd.extend(["-PARALLEL", "{}".format(cpus)])
return cmd
[docs] def validForceFieldSelectorCustomOPLSDir(self):
"""
Check whether a force field selector exists and if so whether it is set
to use a custom OPLS directory that is valid.
:return: whether OPLS directory has issues
:rtype: bool
"""
child = self.findChild(ffselector.ForceFieldSelector)
if child:
return child.sanitizeCustomOPLSDir()
return True
[docs] def launchFromJobSpec(self, oplsdir=None):
"""
Call this function in start method if the calling script implements
the launch api. This function requires implementation of getJobSpec
to return the job specification.
:type oplsdir: None, False or str
:param oplsdir: If None (default), search widgets on the panel for a
`ffselector.ForceFieldSelector` (or subclass thereof)
and get any custom OPLS directory information from that widget. If
False, do not use a custom OPLS directory. If a str, this is the path to
use for the custom OPLS directory. Note that the OPLSDIR setting found
by oplsdir=None is ambiguous if there is more than one
ForceFieldSelector child widget, and that ForceFieldSelector widgets
that are NOT child widgets of this panel - such as a widget on a dialog
- will not be found. Setting this parameter to False for a panel that
does not use a ForceFieldSelector widget avoids the widget search but
will only shave a few thousandths of a second off job startup time even
for very complex panels.
"""
try:
job_spec = self.getJobSpec()
except SystemExit as e:
self.error('Error launching job {}'.format(e))
return
launch_params = launchparams.LaunchParameters()
launch_params.setJobname(self.jobname())
cd_params = self.configDialogSettings()
host = None
if 'host' in cd_params:
host = cd_params['host']
launch_params.setHostname(host)
if self.config_dlg.PRODUCT_HOSTS_KEY in cd_params:
status = self.config_dlg.setLaunchParams(job_spec, launch_params)
# Error already was shown if status is False
if status is False:
return False
if 'openmpcpus' in cd_params:
threads = cd_params['threads']
cpus = cd_params['openmpcpus']
if threads:
launch_params.setNumberOfSubjobs(cd_params['openmpsubjobs'])
if job_spec.jobUsesTPP():
launch_params.setNumberOfProcessorsOneNode(threads)
#NOTE: If the driver is not using the TPP option, but passing
#to subjobs, this needs to go as part of command in getJobSpec
# (use _addJaguarOptions)
else:
# NOTE: this is the right thing to do for matsci GUIs but
# maybe be the wrong thing to do for jaguar GUIs, since
# they may want ONLY the -PARALLEL N option and not also
# -HOST foo:N as well
launch_params.setNumberOfSubjobs(cpus)
elif 'cpus' in cd_params:
launch_params.setNumberOfSubjobs(cd_params['cpus'])
if self.runMode() == baseapp.MODE_MAESTRO:
if 'proj' in cd_params:
launch_params.setMaestroProjectName(cd_params['proj'])
# Setting disposition is only valid if we have a project
if 'disp' in cd_params:
launch_params.setMaestroProjectDisposition(
cd_params['disp'])
launch_params.setMaestroViewname(self.viewname)
if maestro and maestro.get_command_option(
"prefer", "enablejobdebugoutput") == "True":
launch_params.setDebugLevel(2)
# PANEL-8401 has been filed to improve the AF2 infrastructure for using
# the FF Selector. That case may eventually result in changes here.
# Detect any forcefield selector if requested
sanitized_opls_dir = False
if oplsdir is None:
child = self.findChild(ffselector.ForceFieldSelector)
if child:
if not child.sanitizeCustomOPLSDir():
return
sanitized_opls_dir = True
oplsdir = child.getCustomOPLSDIR()
# verify the oplsdir method argument's validity and allow using default
if oplsdir and not sanitized_opls_dir:
opls_dir_result = self.validateOPLSDir(oplsdir)
if opls_dir_result == forcefield.OPLSDirResult.ABORT:
return False
if opls_dir_result == forcefield.OPLSDirResult.DEFAULT:
oplsdir = None
if oplsdir:
launch_params.setOPLSDir(oplsdir)
launch_params.setDeleteAfterIncorporation(True)
launch_params.setLaunchDirectory(self.jobDir())
# Call private function here because there's not guaranteed a great analog
# for cmdline launching.
cmdlist = jobcontrol._get_job_spec_launch_command(job_spec,
launch_params,
write_output=True)
self.writeJobCmd(cmdlist,
job_spec=job_spec,
launch_params=launch_params)
if self.start_mode == FULL_START:
try:
with qt_utils.JobLaunchWaitCursorContext():
job = jobcontrol.launch_from_job_spec(
job_spec,
launch_params,
display_commandline=jobs.cmdlist_to_cmd(cmdlist))
except jobcontrol.JobLaunchFailure:
# NOTE: By this point, launch_job() would already have shown
# the error to the user in a dialog box.
return
self.registerJob(job)
return job
#=========================================================================
# Job Launching - Drivers via jobcontrol.launch_job
#=========================================================================
[docs] def launchJobCmd(self,
cmdlist,
show_progress_bar=False,
auto_add_host=True,
use_parallel_flag=True,
jobdir=None):
"""
Launches a job control command. Use this to launch scripts that accept
the standard job control options arguments like -HOST, -DISP, etc. By
default, automatically populates standard arguments from the config
dialog, but will not overwrite if they are already found in cmdlist. For
example, if -HOST is found in cmdlist, launchJobCmd will ignore the host
specified in the config dialog.
This method honors self.start_mode; it can either launch the script or
write out a job file to the job directory.
:param cmdlist: the command list
:type cmdlist: list
:param show_progress_bar: Whether or not to show a progress bar tracking
the job's status.
:type show_progress_bar: bool
:param auto_add_host: Whether or not to automatically add -HOST flag to
command when it is not already included.
:type auto_add_host: bool
:type use_parallel_flag: bool
:param use_parallel_flag: Whether requesting CPUs > 1 without
specifying threads > 1 should be represented by the use of the
-PARALLEL X flag (True, default) or -HOST host:X (False). -PARALLEL is a
Jaguar flag and may not be appropriate for other programs.
:param jobdir: launch the job from this dir, if provided.
:type jobdir: str
:returns: Job object for started job, or None if job start failed
or if writing instead of starting. Panels should in general
ignore the return value.
"""
cmd = self.setupJobCmd(cmdlist,
auto_add_host,
use_parallel_flag=use_parallel_flag)
assert len(cmd) > 0
# Automatically pre-pend $SCHRODINGER/run to the command if the first
# argument is a Python list. Use brackets to allow SCHRODINGER with
# spaces, and use forward slash on all platforms when writing.
write_cmd = list(cmd)
if write_cmd[0].endswith('.py') or write_cmd[0].endswith('.pyc'):
write_cmd.insert(0, '${SCHRODINGER}/run')
self.writeJobCmd(write_cmd)
if self.start_mode == FULL_START:
# jobdir allows customized launch dir (e.g. multiJobStart creates
# subjob_dir inside job_dir and launches there)
if jobdir is None:
jobdir = self.jobDir()
# Keep a reference for jobProgressChanged to get emitted:
self._jhandler = jobhandler.JobHandler(cmd, self.viewname, jobdir)
self._jhandler.jobProgressChanged.connect(
self._onJobProgressChanged)
# NOTE: launch_job() will automatically add "${SCHRODINGER}/run"
# with platform-specific separator when launching a Python script.
# NOTE: This will run an event loop while the job launches:
try:
with qt_utils.JobLaunchWaitCursorContext():
job = self._jhandler.launchJob()
except jobcontrol.JobLaunchFailure as err:
qt_utils.show_job_launch_failure_dialog(err, self)
raise
self.registerJob(job, show_progress_bar)
return job
def _onJobDone(self, job):
if job.Viewname == self.viewname:
self.jobCompleted.emit(job)
job_id = job.JobId
if self.last_job is not None and job_id == self.last_job.JobId:
self.lastJobCompleted.emit(job)
if self.showing_progress_for_job == job_id:
self.status_bar.hideProgress()
def _onJobProgressChanged(self, job, current_step, total_steps,
progress_msg):
"""
Called by JobHub when progress of a job changes.
"""
if self.showing_progress_for_job == job.job_id:
# This make an additional query to job DB, but makes the
# code simpler, with less duplication.
self.progress_bar.readJobAndUpdateProgress()
[docs] def setupJobCmd(self, cmdlist, auto_add_host=True, use_parallel_flag=True):
"""
Adds standard arguments HOST, NJOBS, PROJ, DISP, VIEWNAME to the
cmdlist if they are set in the config dialog. Settings pre-existing
in the cmdlist take precedence over the config dialog settings.
:param cmdlist: the command list
:type cmdlist: list
:param auto_add_host: Whether or not to automatically add -HOST flat to
command when it is not already included.
:type auto_add_host: bool
:type use_parallel_flag: bool
:param use_parallel_flag: Whether requesting CPUs > 1 without
specifying threads > 1 should be represented by the use of the
-PARALLEL X flag (True, default) or -HOST host:X (False). -PARALLEL is a
Jaguar flag and may not be appropriate for other programs.
"""
cd_params = self.configDialogSettings()
host = ""
if 'host' in cd_params and '-HOST' not in cmdlist and auto_add_host:
host = cd_params['host']
if 'openmpcpus' in cd_params:
cmdlist.extend(
self.config_dlg._formJaguarCPUFlags(
use_parallel_flag=use_parallel_flag))
elif 'cpus' in cd_params:
cmdlist.extend(['-HOST', '%s:%s' % (host, cd_params['cpus'])])
else:
cmdlist.extend(['-HOST', host])
self._addCmdParam(cmdlist, cd_params, 'njobs') # Adds -NJOBS option
if self.runMode() == baseapp.MODE_MAESTRO:
self._addCmdParam(cmdlist, cd_params, 'proj') # Adds -PROJ option
self._addCmdParam(cmdlist, cd_params, 'disp') # Adds -DISP option
# Add -VIEWNAME even when outside of Maestro, for job status button
cmdlist.extend(['-VIEWNAME', self.viewname])
# For SET_QUEUE_RESOURCES featureflag
if 'queue_resources' in cd_params:
cmdlist.extend(['-QARGS', cd_params['queue_resources']])
# Tell job control that launch directory should be removed as well
# when removing all job files, PANEL-2164:
cmdlist.append('-TMPLAUNCHDIR')
if self.add_driverhost:
if maestro:
remote_driver = maestro.get_command_option(
'prefer', 'useremotedriver')
else:
remote_driver = "True"
if remote_driver == "True":
driverhost = jobs.get_first_hostname(host)
if driverhost and driverhost != "localhost":
cmdlist.extend(["-DRIVERHOST", driverhost])
return cmdlist
def _addCmdParam(self, cmdlist, cd_params, cdname, cmdname=None):
if cmdname is None:
cmdname = '-' + cdname.upper()
if cdname in cd_params and cmdname not in cmdlist:
cmdlist.extend([cmdname, str(cd_params[cdname])])
#=========================================================================
# Write STU test file
#=========================================================================
def _getSTUZIPFilename(self, jobname):
return os.path.join(os.getcwd(), jobname) + ".zip"
[docs] def showSTUDialog(self, sh_txt, jobname):
"""
Shows dialog with information necessary to start a STU test, including
a label that links to the test suite.
:param sh_txt: Text contained within the .sh file
:type sh_txt: str
"""
stu_qd = QtWidgets.QDialog()
stu_qd.setWindowTitle("STU Test Zipfile Created")
stu_layout = QtWidgets.QVBoxLayout(stu_qd)
stu_url = STU_URL % sh_txt
stu_lbl = QtWidgets.QLabel(("<a href='%s'>" % stu_url) +
"Add STU Test...</a>")
stu_lbl.setOpenExternalLinks(True)
stu_lbl.setTextInteractionFlags(Qt.TextBrowserInteraction)
stu_zip_lbl = QtWidgets.QLabel(
"Select this as STU 'Test Directory' ZIP File:")
stu_le = QtWidgets.QLineEdit(self._getSTUZIPFilename(jobname))
stu_le.setReadOnly(True)
stu_layout.addWidget(stu_lbl)
stu_layout.addWidget(stu_zip_lbl)
stu_layout.addWidget(stu_le)
stu_qd.exec()
def _writeSTU(self):
"""
This function writes the jobdir using normal af2 methods, then
processes the .sh file and jobdir into a zip, so that it can be
easily used by STU.
:return: Returns False upon failure, otherwise returns nothing (None)
:rtype: False or None
"""
jobdir = self.jobDir()
jobname = self.jobname()
if self._writeJobFiles() is False:
return False
with open(self._getSHFilename(), 'r') as sh_file:
sh_txt = sh_file.read()
#Zip the jobdir into jobdir.zip
with zipfile.ZipFile(self._getSTUZIPFilename(jobname), 'w') as stu_zip:
for base, dirs, files in os.walk(jobdir):
base = os.path.relpath(base)
for infile in files:
fn = os.path.join(base, infile)
stu_zip.write(fn)
fileutils.force_rmtree(self.jobDir())
self.showSTUDialog(sh_txt, jobname)
# Increment the job name:
self.updateJobname()
[docs] def startDebug(self):
debug.start_gui_debug(self, globals(), locals())
#=========================================================================
# Bottom button bar
#=========================================================================
[docs]class BaseBottomBar(QtWidgets.QFrame, validation.ValidationMixin):
[docs] def __init__(self, app_methods, **kwargs):
QtWidgets.QFrame.__init__(self, **kwargs)
validation.ValidationMixin.__init__(self)
self.app_methods = app_methods
self.button_height = constants.BOTTOM_TOOLBUTTON_HEIGHT
self.custom_buttons = {}
self.setup()
self.layOut()
[docs] def setup(self):
self.layout = swidgets.SHBoxLayout()
self.custom_tb = QtWidgets.QToolBar()
self.standard_tb = QtWidgets.QToolBar()
self.buildCustomBar()
self.buildStandardBar()
[docs] def layOut(self):
self.setLayout(self.layout)
if self.custom_tb.isEnabled():
self.layout.addWidget(self.custom_tb)
[docs] def buildCustomBar(self):
buttons = self.makeCustomButtons()
if not buttons:
return
self.custom_tb.setEnabled(True)
for button in buttons:
self.custom_tb.addWidget(button)
self.custom_buttons[str(button.text())] = button
[docs]class BottomBar(BaseBottomBar):
[docs] def layOut(self):
BaseBottomBar.layOut(self)
if self.standard_tb.isEnabled():
self.layout.addStretch()
self.layout.addWidget(self.standard_tb)
[docs] def buildStandardBar(self):
methods = self.app_methods
am = appmethods # Module alias for brevity
app = methods.source_obj
self.standard_tb.setEnabled(True)
if am.READ in methods:
self.read_bn = self.makeButton(methods[am.READ], app._read)
self.standard_tb.addWidget(self.read_bn)
if am.WRITE in methods:
self.write_bn = self.makeButton(methods[am.WRITE], app._write)
self.standard_tb.addWidget(self.write_bn)
if am.RESET in methods:
self.reset_bn = self.makeButton(methods[am.RESET], app._reset)
self.standard_tb.addWidget(self.reset_bn)
if am.START in methods:
self.start_bn = self.makeButton(methods[am.START], app._start)
self.standard_tb.addWidget(self.start_bn)
# NOTE: Close button is not added to the bottom bar as of PANEL-7429
[docs]class JobBottomBar(BottomBar):
[docs] def layOut(self):
BaseBottomBar.layOut(self)
am = appmethods # Module alias for brevity
methods = self.app_methods
if self.standard_tb.isEnabled():
self.layout.addWidget(self.standard_tb)
if am.START in methods:
self.layout.addWidget(self.monitor_bn)
self.layout.addWidget(self.start_bn)
[docs] def buildStandardBar(self):
am = appmethods # Module alias for brevity
methods = self.app_methods
app = methods.source_obj
self.settings_bn_act = None
self.jobname_lb_act = None
if (am.READ in methods or am.WRITE in methods or am.RESET in methods or
am.START in methods):
self.settings_bn = SettingsButton(methods)
else:
self.standard_tb.setEnabled(False)
self.jobname_le = QtWidgets.QLineEdit()
self.jobname_lb = QtWidgets.QLabel('Job name:')
validators.JobName(self.jobname_le)
self.jobname_le.setToolTip('Enter job name here')
self.jobname_le.setContentsMargins(2, 2, 2, 2)
if am.START in methods:
self.start_bn = self.makeButton(methods[am.START], app._start)
self.jobname_lb_act = self.standard_tb.addWidget(self.jobname_lb)
self.standard_tb.addWidget(self.jobname_le)
self.monitor_bn = jobwidgets.JobStatusButton(parent=self,
viewname=app.viewname)
self.monitor_bn.setFixedHeight(self.button_height)
self.monitor_bn.setFixedWidth(self.button_height)
if (am.READ in methods or am.WRITE in methods or am.RESET in methods or
am.START in methods):
self.settings_bn = SettingsButton(methods)
self.settings_bn_act = self.standard_tb.addWidget(self.settings_bn)
self.settings_bn.setFixedHeight(self.button_height)
self.settings_bn.setFixedWidth(self.button_height)
[docs]class MiniJobBottomBar(JobBottomBar):
"""
This is just an alternate layout of the regular job bar, optimized for
narrow panels.
"""
[docs] def setup(self):
JobBottomBar.setup(self)
self.layout = swidgets.SVBoxLayout()
self.middle_layout = swidgets.SHBoxLayout()
self.lower_layout = swidgets.SHBoxLayout()
methods = self.app_methods
app = methods.source_obj
self.help_bn = None
if app.help_topic:
self.help_bn = swidgets.HelpButton()
self.help_bn.clicked.connect(app._help)
[docs] def buildStandardBar(self):
"""
Constructs the parent standard bar, then removes widgets that need to be
relocated for the mini layout. When a widget is removed from a toolbar,
it needs to be re-instantiated, as the old instance becomes unusable.
"""
JobBottomBar.buildStandardBar(self)
if self.jobname_lb_act:
self.standard_tb.removeAction(self.jobname_lb_act)
self.jobname_lb = QtWidgets.QLabel('Job name:')
if self.settings_bn_act:
self.standard_tb.removeAction(self.settings_bn_act)
self.settings_bn = SettingsButton(self.app_methods)
self.settings_bn.setFixedHeight(self.button_height)
self.settings_bn.setFixedWidth(self.button_height)
[docs] def layOut(self):
self.layout.addWidget(self.jobname_lb)
self.layout.addLayout(self.middle_layout)
self.layout.addLayout(self.lower_layout)
self.setLayout(self.layout)
if self.custom_tb.isEnabled():
self.lower_layout.addWidget(self.custom_tb)
am = appmethods # Module alias for brevity
methods = self.app_methods
app = methods.source_obj
if self.standard_tb.isEnabled():
self.middle_layout.addWidget(self.standard_tb)
if am.START in methods:
self.middle_layout.addWidget(self.monitor_bn)
self.middle_layout.addWidget(self.start_bn)
if am.RESET in methods:
self.reset_bn = self.makeButton(methods[am.RESET], app._reset)
self.lower_layout.addWidget(self.reset_bn)
self.lower_layout.addStretch()
if self.settings_bn_act:
self.lower_layout.addWidget(self.settings_bn)
if self.help_bn:
self.lower_layout.addWidget(self.help_bn)
[docs]class OKAndCancelBottomBar(BaseBottomBar):
"""
Version of the bottom bar - which shows OK and Cancel buttons.
"""
[docs] def layOut(self):
BaseBottomBar.layOut(self)
self.layout.addStretch()
self.layout.addWidget(self.ok_bn)
self.layout.addWidget(self.cancel_button)
[docs] def buildStandardBar(self):
methods = self.app_methods
app = methods.source_obj
self.ok_bn = QtWidgets.QPushButton('OK')
self.cancel_button = QtWidgets.QPushButton('Cancel')
self.cancel_button.clicked.connect(app._close)
self.jobname_le = QtWidgets.QLineEdit()
# AF2 expects these buttons, as it will try to disable them while
# the *.sh file is getting written:
self.start_bn = self.ok_bn
self.settings_bn = self.ok_bn
#=========================================================================
# Settings Menu
#=========================================================================
#=========================================================================
# Status bar
#=========================================================================
[docs]class StatusBar(statusbar.StatusBar):
status_shrunk = QtCore.pyqtSignal(int)
"""
A signal emitted when the status bar has been shrunk due to hiding the
progress bar. The signal is emitted with the number of pixels the status
bar has been shrunk by.
"""
[docs] def __init__(self, app):
QtWidgets.QStatusBar.__init__(self)
self.status_lb = QtWidgets.QLabel()
self.progress_bar = ProgressFrame()
self.addWidget(self.status_lb)
self.progress_shown = False
if app.help_topic:
self.addHelpButton(app)
[docs] def setStatus(self, text):
self.status_lb.setText(text)
[docs] def status(self):
return self.status_lb.text()
[docs] def hideProgress(self):
"""
Hide the progress bar and re-display the status message
"""
if not self.progress_shown:
return
self.progress_shown = False
pre_height = self.sizeHint().height()
self.addWidget(self.status_lb)
self.status_lb.show()
self.removeWidget(self.progress_bar)
self.clearMessage()
post_height = self.sizeHint().height()
shrinkage = pre_height - post_height
self.status_shrunk.emit(shrinkage)
[docs] def showProgress(self):
"""
Show the progress bar in place of the status message
"""
if self.progress_shown:
return
self.progress_shown = True
self.addWidget(self.progress_bar, 1)
self.progress_bar.show()
self.removeWidget(self.status_lb)
self.clearMessage()
#===============================================================================
# Progress Bar
#===============================================================================
[docs]class ProgressFrame(QtWidgets.QFrame):
"""
A progress bar. Job progress can be tracked using `trackJobProgress` and
`readJobandUpdateProgress`. The progress bar can be also be used
"manually" for non-job-control tasks: It can be shown and hidden from the
panel via `self.status_bar.showProgress` and
`self.status_bar.hideProgress` and can be updated using
`self.progress_bar.setValue`, `self.progress_bar.setMaximum`,
`self.progress_bar.setText`, and `self.progress_bar.setTextVisible`.
"""
[docs] def __init__(self, parent=None):
super(ProgressFrame, self).__init__(parent)
self._layout = QtWidgets.QVBoxLayout(self)
self._layout.setContentsMargins(0, 0, 0, 0)
self._lbl = QtWidgets.QLabel(self)
self._lbl.hide()
self._bar = QtWidgets.QProgressBar(self)
self._layout.addWidget(self._lbl)
self._layout.addWidget(self._bar)
self._job = None
[docs] def trackJobProgress(self, job, show_lbl=False):
"""
Track the progress of the specified job
:param job: The job to track
:type job: `schrodinger.job.jobcontrol.Job`
:param show_lbl: If True, the job progress text description will be
shown above the progress bar. If False, the text description will not
be shown. Defaults to False.
:type show_lbl: bool
"""
self._bar.setValue(0)
self._bar.setMaximum(100)
self._lbl.setText("")
self._lbl.setVisible(show_lbl)
self._job = job
[docs] def readJobAndUpdateProgress(self):
"""
Update the status bar based on the current job's progress. The job
progress will be re-read.
:return: True if the job has completed. False otherwise.
:rtype: bool
"""
# Preemptive job ID
try:
job = jobhub.get_cached_job(self._job.job_id)
except jobhub.StdException:
# Job record is missing, so do not update progress
pass
else:
self._job = job
self.updateProgress()
return self._job.isComplete()
[docs] def updateProgress(self):
"""
Update the status bar based on the current job's progress. Note that
the job database will not be re-read. Use `readJobAndUpdateProgress`
instead if you have not already updated the job object.
"""
job_percent = self._job.getProgressAsPercentage()
self._bar.setValue(job_percent)
job_msg = self._job.getProgressAsString()
if job_msg == "The job has not yet started.":
job_msg = "Job submitted..."
self._lbl.setText(job_msg)
[docs] def setValue(self, value):
self._bar.setValue(value)
[docs] def setMaximum(self, value):
self._bar.setMaximum(value)
[docs] def setText(self, text):
self._lbl.setText(text)
[docs] def setTextVisibile(self, visible):
self._lbl.setVisible(visible)
[docs] def mouseDoubleClickEvent(self, event):
"""
If the user double clicks and there is a job loaded, launch the Monitor
panel
:note: If self._job is None or if refers to a completed job, then we
assume the progress bar is currently tracking progress for something not
job-related (such as reading input files into the panel), so we don't
launch the Monitor panel.
"""
if maestro and self._job is not None and not self._job.isComplete():
maestro.command("showpanel monitor")
else:
super(ProgressFrame, self).mouseDoubleClickEvent(event)
#===============================================================================
# Utility functions
#===============================================================================
[docs]def reduce_settings_for_group(settings, group):
"""
Reduces a full settings dictionary to a dictionary targeted for a specific
group. The function does two things:
1) Strips off the group prefix for this group from aliases. Example:
For 'group_A', 'group_A.num_atoms' becomse just 'num_atoms'. In the
case of resultant name collisions, the group-specific setting takes
priority. Ex. "group_A.num_atoms" takes priority over just
"num_atoms".
2) Removes settings for other groups. For example, if 'group_A' is
passed in, settings like 'group_B.num_atoms' will be excluded.
:param settings: settings dictionary mapping alias to value
:type settings: dict
:param group: the desired group
:type group: str
"""
filtered_settings = {}
prefix = '%s.' % group
used_aliases = []
for alias, value in settings.items():
if alias in used_aliases:
continue
if alias.startswith(prefix):
alias = alias.split(prefix, 1)[1]
used_aliases.append(alias)
elif '.' in alias:
continue
filtered_settings[alias] = value
return filtered_settings
[docs]def expand_settings_from_group(settings, group, all_aliases):
expanded_settings = {}
for alias, value in settings.items():
prefixed_alias = '%s.%s' % (group, alias)
if (prefixed_alias) in all_aliases:
alias = prefixed_alias
expanded_settings[alias] = value
return expanded_settings
[docs]def get_readable_cmd_list(jlaunch_cmdlist, job_spec, launch_params):
"""
Generate a portable command for launching the job defined
by the given job specification, for writing out as a comment to
<jobname.sh> file that user can re-use when submitting jobs, without having
to base64 decode the command from the jlaunch command list.
The returned command will use $SCHRODINGER/run <script.py> instead of
jlaunch.pl, so the script must define a get_job_spec_from_args() function
for the command to work.
:param list jlaunch_cmdlist: the list of top-level commands passed to
`jlaunch.pl` to launch/submit the job
:param schrodinger.job.launchapi.JobSpecification job_spec: The job
specification for the command you want to write.
:param job.launchparams.LaunchParameters launch_params: Used to check for
the subhost
:rtype: list[str]
:return: Command to launch/submit the job
"""
run_cmdlist = job_spec.getCommand()
# Map certain jalaunch.pl arguments (created by launchapi.py) to
# corresponding run command arguments
flag_mappings = {'-name': '-JOBNAME', '-OPLSDIR': '-OPLSDIR'}
if launch_params.getSubHostName():
# If subhost is set, HOST is HOST, SUBHOST is nodelist
flag_mappings[cmdline.FLAG_HOST] = cmdline.FLAG_HOST
flag_mappings[cmdline.FLAG_NODELIST] = cmdline.FLAG_SUBHOST
else:
flag_mappings[cmdline.FLAG_NODELIST] = cmdline.FLAG_HOST
for jlaunch_flag, run_flag in flag_mappings.items():
try:
idx = jlaunch_cmdlist.index(jlaunch_flag)
except ValueError:
continue
value = jlaunch_cmdlist[idx + 1]
run_cmdlist += [run_flag, value]
# Formatting & enabling of cross-platform execution
exe = run_cmdlist[0].replace('%SCHRODINGER%', '${SCHRODINGER}')
run_cmdlist[0] = jobwriter._normalize_schrodinger_exec(exe)
return run_cmdlist