"""
A dialog for obtaining a job directory of a previously submitted job
Copyright Schrodinger, LLC. All rights reserved.
"""
import os.path
import time
from schrodinger.infra import mm
from schrodinger.infra import jobhub
from schrodinger.job import jobcontrol
from schrodinger.Qt import QtCore
from schrodinger.Qt import QtGui
from schrodinger.Qt import QtWidgets
from schrodinger.Qt.QtCore import Qt
from schrodinger.ui.qt import utils as qtutils
from schrodinger.ui.qt import filedialog
from schrodinger.ui.qt import swidgets
[docs]def start_job_manager():
"""
Start the job manager so that it begins an asynchronous update of the job
list in the background. It is best to do this as soon as possible in a new
process because obtaining the job list may take some time after the first
access of the jobhub job manager.
In Maestro, this call doesn't have any affect because Maestro starts the job
manager at startup. But this call does have an effect for panels run from
the command line.
This call cannot be done on import of this module because a QApplication
needs to be started first, which often isn't done at the point this module
is imported.
"""
JobManager.getManager()
[docs]class JobManager(QtCore.QObject):
""" A class for managing job information and alerts when new jobs start """
_singleton = None
TIME_FORMAT = '%a %b %d %I:%M:%S %p'
UPDATED_TEXT = 'Last updated: {utime}'
NEW_DATA_TEXT = '<font color=green>New jobs available</font>'
UPDATING_TEXT = 'Updating job information...'
new_data_available = QtCore.pyqtSignal()
[docs] def __init__(self):
""" Create a JobManager instance """
super().__init__()
self.jobhub_manager = jobhub.get_job_manager()
# Check to see if the job manager knows about jobs yet
# status_text has two states:
# UPDATING_TEXT: If no job information is available yet
# UPDATED_TEXT: The last time new job information was updated
jobs = self.getAllJobs()
if jobs:
self.update_time = self.getTime()
self.status_text = self.UPDATED_TEXT.format(utime=self.update_time)
else:
self.update_time = None
self.status_text = self.UPDATING_TEXT
self.prev_num_jobs = len(jobs)
[docs] def connectToJobHub(self):
""" Connect to the job hub manager's periodic callback """
# This connection is done in a method rather than directly so that it is
# easy to mock in a session fixture. We must prevent actually connecting
# to this signal in unit tests otherwise the number of tests that use
# this combined with the short periodicity of the callback causes
# problems with Windows builds
#
# The session fixture (curretly in
# schrodinger.test.pytest.sessionfixture) mocks this method out
# automatically for the entire test framework so that # unit tests don't
# have to do it individually for every panel that uses this class.
self.jobhub_manager.jobsChanged.connect(self.updateStatus)
[docs] @staticmethod
def getManager(*args, **kwargs):
"""
Get JobManager singleton
Calling this method the first time starts the jobhub manager loading job
information
"""
if not JobManager._singleton:
JobManager._singleton = JobManager(*args, **kwargs)
return JobManager._singleton
[docs] def getTime(self):
"""
Get the current time in user-format
:rtype: str
:return: The current time
"""
return time.strftime(self.TIME_FORMAT)
[docs] def updateStatus(self, jobids):
"""
Callback for the jobhub jobsChanged signal. This gets called
periodically whether any job changed or not.
:param list jobids: The job ids for jobs changing state
"""
if not jobids:
# This function is called as a periodic callback (despite the name
# of the calling signal, it is *not* called only when jobs change.
# Do nothing if no jobids have changed.
return
# Check if there are more jobs available than before
self.update_time = self.getTime()
self.status_text = self.UPDATED_TEXT.format(utime=self.update_time)
num_jobs = len(self.getAllJobs())
if num_jobs > self.prev_num_jobs:
self.new_data_available.emit()
self.prev_num_jobs = num_jobs
[docs] def getAllJobs(self):
"""
Get all jobs in the database
:rtype: view
:return: Each item in the view is a job object for a job in the database
"""
# Note that getJobs can return an empty list of jobs if called early in
# a session before the jobhub manager has updated the jobs list. After
# discussions with the jobcontrol team (Python GChat room 12/7/20) it
# looks like this is just something we'll have to live with.
return self.jobhub_manager.getJobs(jobhub.JobOption.ALL_JOBS).values()
[docs] def getRunningJobs(self):
"""
Get all running jobs
:rtype: view
:return: Each item in the view is a job object for a running job
"""
return self.jobhub_manager.getJobs(
jobhub.JobOption.ACTIVE_JOBS).values()
[docs]class NewJobDirFrame(swidgets.SFrame):
"""
A collection of widgets that reads in the JobDB and puts jobs in a table
that can be selected. It also allows the user to specify a job directory
manually.
"""
[docs] def __init__(self, master, layout=None, dclick_callback=None):
"""
Create a NewJobDirFrame instance
:type master: QWidget
:param master: The master panel for this frame, should have a warning
method
:type layout: QBoxLayout
:param layout: The layout to place this frame into
:type dclick_callback: callable
:param dclick_callback: The function to call when the user double-clicks
on a table cell
"""
self.manager = JobManager.getManager()
self.manager.new_data_available.connect(self.newDataAlert)
self.master = master
self.programs = set()
swidgets.SFrame.__init__(self, layout=layout)
layout = self.mylayout
self.table_rb = swidgets.SRadioButton(
'Load from job database',
command=self.loadToggled,
layout=layout,
checked=True)
# Status update
ulayout = swidgets.SHBoxLayout(layout=layout)
self.update_button = swidgets.SPushButton(
'Refresh', layout=ulayout, command=self.loadJobsIntoTable)
self.update_label = swidgets.SLabel(
self.manager.status_text, layout=ulayout)
ulayout.addStretch()
# Job database table
tlayout = swidgets.SHBoxLayout(layout=layout, indent=True)
self.table = QtWidgets.QTableWidget()
table_headers = [
'Job Name', 'Launched', 'Status', 'Program', 'Directory'
]
self.table.setColumnCount(len(table_headers))
self.table.setHorizontalHeaderLabels(table_headers)
self.table.setSelectionBehavior(self.table.SelectRows)
self.table.setSelectionMode(self.table.SingleSelection)
self.table.setSortingEnabled(True)
if dclick_callback:
self.table.cellDoubleClicked.connect(dclick_callback)
self.loadJobsIntoTable()
tlayout.addWidget(self.table)
# Manual directory browsing
self.browse_rb = swidgets.SRadioButton(
'Manually locate job directory',
command=self.loadToggled,
layout=layout)
blayout = swidgets.SHBoxLayout(layout=layout, indent=True)
self.browse_le = swidgets.SLabeledEdit(
'Path:', stretch=False, layout=blayout)
self.browse_button = swidgets.SPushButton(
'Browse...', command=self.browseDirectory, layout=blayout)
self.loadToggled()
[docs] def newDataAlert(self):
""" Set the text label to alert user there are new jobs available """
self.update_label.setText(self.manager.NEW_DATA_TEXT)
[docs] def reset(self, load_jobs=True):
"""
Reset the entire frame
:type load_jobs: bool
:param load_jobs: Whether the job database should be loaded back into
the table after reset
"""
self.programs = set()
self.resetTable()
self.table_rb.reset()
self.browse_rb.reset()
self.browse_le.reset()
self.loadToggled()
if load_jobs:
self.loadJobsIntoTable()
[docs] def resetTable(self):
"""
Remove all rows from the table
"""
self.table.setRowCount(0)
[docs] def setAllowedPrograms(self, programs):
"""
Set the programs whose jobs are allowed to show up in the table
:type programs: set
:param programs: The strings that show up in the job.Program field for
programs whose jobs should show in the dialog. Use None to show all
active jobs.
"""
self.programs = programs
[docs] def readJobsFromDatabase(self):
"""
Read the jobs from the JobDB database
:rtype: list, list
:return: Two lists. The first contains the completed jobs, the second
currently running jobs. All list items are
`schrodinger.job.jobcontrol.Job` objects and the lists are sorted so
that the newest jobs appear first
"""
# Read in all the jobs in the job database and find the jobs of
# interest - parse them into running and completed
running = []
completed = []
all_jobs = self.manager.getAllJobs()
all_running_jobs = set(self.manager.getRunningJobs())
self.update_label.setText(self.manager.status_text)
if self.programs:
jobs = []
for job in all_jobs:
try:
if job.Program in self.programs:
jobs.append(job)
except AttributeError:
# Some jobs may not have a program set (MATSCI-1713)
pass
else:
jobs = all_jobs
for ajob in jobs:
try:
exists = os.path.exists(ajob.Dir)
except AttributeError:
# It is possible for jobs to not have a Dir attribute if they
# are not fully fleshed out yet
exists = False
if not exists:
continue
if ajob in all_running_jobs:
running.append(ajob)
else:
completed.append(ajob)
# Sort all the jobs by reversed Launch Time
running.sort(reverse=True, key=lambda x: x.LaunchTime)
completed.sort(reverse=True, key=lambda x: x.LaunchTime)
return running, completed
[docs] @qtutils.wait_cursor
def loadJobsIntoTable(self):
"""
Load all the desired jobs from the job database into the table
"""
# Turn off sorting because adding data to a table with sorting on will
# mess up the rows
self.table.setSortingEnabled(False)
self.resetTable()
# Standard item for the table - used to clone new items
base_item = QtWidgets.QTableWidgetItem()
base_item.setFlags(Qt.ItemIsSelectable | Qt.ItemIsEnabled)
def set_table_text(row, column, text, jobid=None):
"""
Put new text in the table in row, column
:type row: int
:param row: The row of the table cell
:type column: int
:param column: the column of the table cell
:type text: str
:param text: The text to insert into row, column
:type jobid: str
:param jobid: The job id of the job for this row
"""
item = QtWidgets.QTableWidgetItem(base_item)
item.setText(text)
if jobid:
item.setData(Qt.UserRole, jobid)
self.table.setItem(row, column, item)
running, completed = self.readJobsFromDatabase()
# Put all the jobs in the table, running first, then completed
for row, ajob in enumerate(running + completed):
self.table.insertRow(row)
set_table_text(row, 0, ajob.Name, jobid=ajob.job_id)
set_table_text(row, 1, ajob.LaunchTime)
set_table_text(row, 2, ajob.Status)
try:
set_table_text(row, 3, ajob.Program)
except AttributeError:
set_table_text(row, 3, "")
set_table_text(row, 4, ajob.Dir)
header = self.table.horizontalHeader()
header.resizeSections(header.ResizeToContents)
self.table.setSortingEnabled(True)
[docs] def loadToggled(self):
"""
Whether to load from the job table or a manual directory has been
toggled - react to that
"""
browsing = self.browse_rb.isChecked()
if browsing:
sranges = self.table.selectedRanges()
for srange in sranges:
self.table.setRangeSelected(srange, False)
self.table.setEnabled(not browsing)
self.browse_le.setEnabled(browsing)
self.browse_button.setEnabled(browsing)
[docs] def getCurrentJobDir(self):
"""
Get the currently selected job directory and the associated job if
applicable
:rtype: (str, `schrodinger.job.jobcontrol.Job` or (str, None) or (None,
None)
:return: The path to the selected job directory and if possible, the
associated Job object. If the user specified a job directory
manually, the Job object will be None. If no job directory has been
specified, the return value is (None, None)
"""
if self.browse_rb.isChecked():
# Manual directory
path = str(self.browse_le.text())
job = None
if not path:
raise RuntimeError('A directory must be specified manually.')
else:
# Read a directory from the job database
path = ""
for item in self.table.selectedItems():
if item.column() == 0:
jobid = str(item.data(Qt.UserRole))
job = jobcontrol.Job(jobid)
path = job.Dir
break
if not path:
raise RuntimeError('Please select a job in the table to load.')
return path, job
[docs] def browseDirectory(self):
"""
Allow the user to browse to a new directory via file dialog
"""
job_dir = filedialog.get_existing_directory(id='jobdirdlg')
if not job_dir:
return
self.browse_le.setText(job_dir)
[docs] def setWaitCursor(self, app_wide=True):
"""
Set the cursor to the wait cursor. This will be an hourglass, clock or
similar. Call restoreCursor() to return to the default cursor.
If 'app_wide' is True then it will apply to the entire application
(including Maestro if running there). If it's False then it will apply
only to this panel.
Added for the wait_cursor decorator
"""
if app_wide:
QtWidgets.QApplication.instance().setOverrideCursor(
QtGui.QCursor(QtCore.Qt.WaitCursor))
else:
self.setCursor(QtGui.QCursor(QtCore.Qt.WaitCursor))
[docs] def restoreCursor(self, app_wide=True):
"""
Restore the application level cursor to the default. If 'app_wide' is
True then if will be restored for the entire application, if it's
False, it will be just for this panel.
Added for the wait_cursor decorator
"""
if app_wide:
QtWidgets.QApplication.instance().restoreOverrideCursor()
else:
self.setCursor(QtGui.QCursor(QtCore.Qt.ArrowCursor))
[docs]class NewJobDirDialog(swidgets.SDialog):
"""
SDialog to allow the user to read in information for a new job. Should be
created with the user_accept_function parameter set to a function that
accepts (path, job=job) api, where path is the path to the job directory and
job is a Job object if available.
"""
def __getattr__(self, attribute):
"""
Pass on any requests for unknown attributes to the job loader. If the
attribute is still unknown, raise an AttributeError for this class
"""
try:
return getattr(self.jdir_loader, attribute)
except AttributeError:
raise AttributeError('%s has no attribute %s' %
(self.__class__.__name__, attribute))
[docs] def layOut(self):
"""
Lay out the widgets for the dialog
We use a wait cursor because searching the job database may take a few
seconds (or more)
"""
layout = self.mylayout
self.jdir_loader = NewJobDirFrame(
self, layout, dclick_callback=self.accept)
size = QtCore.QSize(800, 400)
self.resize(size)
[docs] def accept(self):
"""
The user has pressed the Accept button. Call the user_accept_function
with the chosen job information.
"""
try:
path, job = self.jdir_loader.getCurrentJobDir()
except RuntimeError as msg:
self.warning(str(msg))
return
if not path:
return
self.user_accept_function(path, job=job)
return swidgets.SDialog.accept(self)