import os
import re
from schrodinger.application.job_monitor import job_info_view_ui
from schrodinger.application.job_monitor import job_monitor_diagnostics_dialog
from schrodinger.application.job_monitor import job_monitor_file_browser
from schrodinger.application.job_monitor import job_monitor_icons_rc # noqa: F401
from schrodinger.application.job_monitor import job_monitor_models
from schrodinger.application.job_monitor import job_monitor_progress_bar
from schrodinger.application.job_monitor import job_monitor_table
from schrodinger.application.job_monitor import top_level_view_ui
from schrodinger.application.job_monitor import util
from schrodinger.application.job_monitor.buttons import BackButton
from schrodinger.application.job_monitor.job_monitor_table import \
get_icon_path_for_status
from schrodinger.infra import jobhub
from schrodinger.infra import mm
from schrodinger.job import jobcontrol
from schrodinger.job.jobcontrol import Job
from schrodinger.models import mappers
from schrodinger.Qt import QtCore
from schrodinger.Qt import QtGui
from schrodinger.Qt import QtWidgets
from schrodinger.ui.qt import basewidgets
from schrodinger.ui.qt.mapperwidgets import plptable
from schrodinger.ui.qt.standard_widgets import flat_button
from schrodinger.ui.qt.swidgets import StyleMixin
from schrodinger.utils import fileutils
ICON_PATH = ":/job_monitor/icons/"
SUB_JOBS_TAB = "Sub-jobs"
FILE_BROWSER_TAB = "Job Files"
TIME_DISPLAY_FORMAT = "%I:%M%p"
DATE_DISPLAY_FORMAT = "%d %b %Y"
def _get_elided_text(font, text, width, mode=QtCore.Qt.ElideLeft):
"""
Return the elided text for the given text, font, mode and width.
:param font: font of the widget that will use the elided text
:type font: QFont
:param text: text that will be elided
:type text: str
:param width: if the text is longer than this width then it will be elided
:type width: int
:param mode: end of the text that should be elided (left or right)
:type mode: QtCore.Qt.ElideLeft or QtCore.Qt.ElideRight
"""
metrics = QtGui.QFontMetrics(font)
return metrics.elidedText(text, mode, width)
class _BasePane(StyleMixin, mappers.MapperMixin, basewidgets.BaseWidget):
"""
:ivar updateStatusBar: a signal to update the status of `job_monitor_diagnostics_dialog
.CollectJobDiagnosticsTask`.
:vartype updateStatusBar: QtCore.pyqtSignal
:ivar restoreSelectionRequested: Signal to restore the selection in table
when the model updates the jobs.
:vartype restoreSelectionRequested: QtCore.pyqtSignal
"""
updateStatusBar = QtCore.pyqtSignal(str)
restoreSelectionRequested = QtCore.pyqtSignal()
def initSetUp(self):
super().initSetUp()
self.job_table = self._getJobTable()
self.job_table.collectDiagnosticsRequested.connect(self.postmortem)
self.job_table.stopRequested.connect(self._stopJobs)
self.job_table.cancelRequested.connect(self._cancelJobs)
self.job_table.view.doubleClicked.connect(self.viewJobDetails)
self.job_table.job_details_btn_delegate.clicked.connect(
self.viewJobDetails)
self.job_table.deleteJobRecordRequested.connect(self._deleteJobRecord)
self.restoreSelectionRequested.connect(self.job_table.restoreSelection)
def _getJobTable(self):
raise NotImplementedError
def postmortem(self, job_ids):
self.dlg = job_monitor_diagnostics_dialog.CollectJobDiagnosticsDialog(
job_ids, self)
self.dlg.updateStatusBar.connect(self.updateStatusBar)
self.dlg.startTask()
def _stopJobs(self, job_ids):
"""
:type job_ids: list[str]
:param job_ids: Job IDs
"""
jobhub.get_job_manager().stopJobs(job_ids)
for job_id in job_ids:
job_model = self.model.getJob(job_id)
job_model.requested_status = job_monitor_models.RequestedStatus.STOP
def _cancelJobs(self, job_ids):
jobhub.get_job_manager().killJobs(job_ids)
for job_id in job_ids:
job_model = self.model.getJob(job_id)
job_model.requested_status = job_monitor_models.RequestedStatus.CANCEL
def _deleteJobRecord(self, job_ids):
"""
Remove the given jobs from both Job DB.
:param job_ids: List of job ids
:type: List[str]
"""
job_manager = jobhub.get_job_manager()
job_manager.cleanupJobs(job_ids, jobhub.RemoveJobRecordAndMonitorFiles)
def refreshJobTable(self):
self.job_table.refresh()
def getSelectedJobIds(self):
"""
Return jobs ids of all selected jobs in table
:return: Selected job IDs
:rtype: list[str]
"""
return [m.job_id for m in self.job_table.selectedParams()]
[docs]class JobsListPane(_BasePane):
"""
This is the pane you start on when opening the job monitor. It contains
widgets to stop or cancel jobs, filter jobs, open preferences,
and a table that shows top level jobs
"""
ui_module = top_level_view_ui
model_class = job_monitor_models.JobMonitorPanelModel
[docs] def initSetUp(self):
super().initSetUp()
self.ui.stop_btn.clicked.connect(self.stopSelected)
self.ui.stop_btn.setIconPath(ICON_PATH + "stop.png")
self.ui.stop_btn.setHoverIconPath(ICON_PATH + "stop-h.png")
self.ui.stop_btn.setDisabledIconPath(ICON_PATH + "stop-d.png")
self.ui.stop_btn.setIconSize_(25, 25)
self.ui.stop_btn.setEnabled(False)
self.ui.cancel_btn.clicked.connect(self.cancelSelected)
self.ui.cancel_btn.setIconPath(ICON_PATH + "cancel.png")
self.ui.cancel_btn.setHoverIconPath(ICON_PATH + "cancel-h.png")
self.ui.cancel_btn.setDisabledIconPath(ICON_PATH + "cancel-d.png")
self.ui.cancel_btn.setDisabledIconPath(ICON_PATH + "cancel-d.png")
self.ui.cancel_btn.setIconSize_(25, 25)
self.ui.cancel_btn.setEnabled(False)
self.ui.active_jobs_btn.clicked.connect(self.toggleActiveFilter)
self.ui.all_jobs_btn.clicked.connect(self.toggleActiveFilter)
self.ui.project_filter_btn.clicked.connect(self.toggleProjectFilter)
self.job_table.view.selectionModel().selectionChanged.connect(
self.updateStopCancelButtons)
[docs] def initLayOut(self):
super().initLayOut()
self.ui.table_layout.addWidget(self.job_table)
self.job_table.setSizePolicy(QtWidgets.QSizePolicy.Expanding,
QtWidgets.QSizePolicy.Expanding)
[docs] def defineMappings(self):
M = self.model_class
return [
(self.job_table, M.top_level_jobs),
(self.onProjFilterChanged, M.current_project_jobs_only),
(self.onProjFilterChanged, M.current_project_name),
(self.onActiveJobsOnlyChanged, M.active_jobs_only),
]
def _getJobTable(self):
return job_monitor_table.JobTableWidget(self)
[docs] def getSignalsAndSlots(self, model):
return [
(model.filtersInvalidated, self.refilterJobs),
(model.current_project_jobs_onlyChanged, self.refilterJobs),
(model.current_project_nameChanged, self.refilterJobs),
(model.active_jobs_onlyChanged, self.refilterJobs),
]
[docs] def refreshJobTable(self):
"""
Update the stop and cancel buttons after the job table
gets refreshed.
"""
super().refreshJobTable()
self.updateStopCancelButtons()
[docs] def viewJobDetails(self, index):
job_model = index.data(plptable.ROW_OBJECT_ROLE)
job_model.job_index = index.row()
self.model.setCurrentTopLevelJob(job_model)
self.model.setCurrentJob(job_model)
[docs] def onProjFilterChanged(self):
if self.model.current_project_jobs_only:
new_text = "Show All Projects"
proj_filter_btn_tooltip = "Show jobs from all projects"
else:
new_text = "Show Current Project Only"
proj_filter_btn_tooltip = "Show jobs from current project only"
self.ui.project_filter_btn.setText(new_text)
self.ui.project_filter_btn.setToolTip(proj_filter_btn_tooltip)
[docs] def refilterJobs(self):
"""
Filter jobs based on whether we're showing active jobs only, current
project jobs only.
"""
cur_proj_only = self.model.current_project_jobs_only
active_only = self.model.active_jobs_only
filtered_jobs = []
for job_model in self.model.getAllTopLevelJobs():
if cur_proj_only:
if job_model.project_name != self.model.current_project_name:
continue
if active_only:
if not job_model.is_active:
continue
filtered_jobs.append(job_model)
self.model.setTopLevelJobs(filtered_jobs)
[docs] def toggleActiveFilter(self):
self.model.active_jobs_only = not self.model.active_jobs_only
self.job_table.saveSelection()
[docs] def onActiveJobsOnlyChanged(self):
self.ui.active_jobs_btn.setEnabled(not self.model.active_jobs_only)
self.ui.all_jobs_btn.setEnabled(self.model.active_jobs_only)
[docs] def toggleProjectFilter(self):
self.model.current_project_jobs_only = not self.model.current_project_jobs_only
self.job_table.saveSelection()
[docs] def stopSelected(self):
self._stopJobs(self.getSelectedJobIds())
[docs] def cancelSelected(self):
self._cancelJobs(self.getSelectedJobIds())
[docs]class JobDetailsPane(_BasePane):
"""
The job details pane is opened when a user double clicks a job in a job
table. It shows info on the selected job and the selected job's top level job
ancestor. It has back buttons to return to the Jobs List Pane.
"""
model_class = job_monitor_models.JobMonitorPanelModel
[docs] def initSetUp(self):
super().initSetUp()
self.top_level_job_bar = TopLevelJobBar(self)
self.top_level_job_bar.setObjectName("top_level_job_bar")
self.top_level_job_bar.back_button.clicked.connect(self.onTopLevelBack)
self.top_level_job_bar.next_job_btn.clicked.connect(
lambda: self.switchTopLevelJob(True))
self.top_level_job_bar.prev_job_btn.clicked.connect(
lambda: self.switchTopLevelJob(False))
# # JOB SUB FRAME
self.job_info_frame = QtWidgets.QFrame()
self.job_info_frame.setObjectName("job_info_frame")
self.sub_job_bar = SubJobBar(self)
self.sub_job_bar.setObjectName("sub_job_bar")
self.sub_job_bar.back_button.clicked.connect(self.onSubJobBack)
self.sub_job_bar.next_job_btn.clicked.connect(
lambda: self.switchSubJob(True))
self.sub_job_bar.prev_job_btn.clicked.connect(
lambda: self.switchSubJob(False))
self.job_info_widget = JobInfoWidget(self)
self.no_subjobs_lbl = QtWidgets.QLabel("(no sub-jobs)")
self.no_subjobs_lbl.setObjectName("no_subjobs_lbl")
self.file_browser = job_monitor_file_browser.JobMonitorFileBrowser()
self.file_browser.updateStatusBar.connect(self.updateStatusBar)
self.no_subjobs_widget = QtWidgets.QWidget()
self.no_subjobs_widget.setObjectName("no_subjobs_widget")
self.job_details_tab = QtWidgets.QTabWidget()
self.job_details_tab.setObjectName("job_details_tab")
self.job_details_tab.addTab(self.job_table, SUB_JOBS_TAB)
self.job_details_tab.addTab(self.file_browser, FILE_BROWSER_TAB)
self.job_details_tab.currentChanged.connect(self._onActiveTabChanged)
# True if job table is shown in sub-jobs tab else False
self._job_table_active = True
[docs] def initLayOut(self):
super().initLayOut()
self.no_subjobs_widget.layout = QtWidgets.QVBoxLayout()
self.no_subjobs_widget.layout.addWidget(
self.no_subjobs_lbl,
QtCore.Qt.AlignHCenter | QtCore.Qt.AlignVCenter)
self.no_subjobs_lbl.setAlignment(QtCore.Qt.AlignCenter)
self.no_subjobs_widget.setLayout(self.no_subjobs_widget.layout)
job_info_frame_layout = QtWidgets.QVBoxLayout()
job_info_frame_layout.setSpacing(0)
job_info_frame_layout.setContentsMargins(0, 0, 0, 0)
job_info_frame_layout.addWidget(self.sub_job_bar)
job_info_frame_layout.addWidget(self.job_info_widget)
job_info_frame_layout.addWidget(self.job_details_tab)
self.job_info_frame.setLayout(job_info_frame_layout)
self.main_layout.addWidget(self.top_level_job_bar)
self.main_layout.addWidget(self.job_info_frame)
self.main_layout.setContentsMargins(4, 2, 4, 4)
[docs] def defineMappings(self):
M = self.model_class
return [
(self.top_level_job_bar, M.current_top_level_job),
(self.sub_job_bar, M.current_job),
(self.job_info_widget, M),
(self.job_table, M.subjobs),
(self.onCurrentJobChanged, M.current_job),
(self.file_browser, M.current_job)
] # yapf: disable
[docs] def switchTopLevelJob(self, to_next):
"""
Switch to the next toplevel job if `to_next` is True else
to the previous toplevel job.
:param to_next: value indicator for next or previous
:type to_next: bool
"""
index = self.model.current_job.job_index
index = index + 1 if to_next else index - 1
job = self.model.top_level_jobs[index]
job.job_index = index
self.model.setCurrentTopLevelJob(job)
self.model.setCurrentJob(job)
[docs] def switchSubJob(self, to_next):
"""
Switch to the next sub job if `to_next` is True else
to the previous sub job.
:param to_next: value indicator for next or previous
:type to_next: bool
"""
index = self.model.current_job.job_index
index = index + 1 if to_next else index - 1
job_id = self.model.current_top_level_job.sub_job_ids[index]
job = job_monitor_models.JobModel.fromJobObject(Job(job_id))
job.job_index = index
self.model.setCurrentJob(job)
def _getJobTable(self):
return job_monitor_table.SubJobTableWidget(self)
def _onActiveTabChanged(self, index):
# Clear the status message when active tab is changed
self.updateStatusBar.emit("")
[docs] def onCurrentJobChanged(self):
"""
Insert sub-jobs table or `no_subjobs_widget` in sub-jobs tab
based on whether the job has atleast a sub-job or not.
"""
if self.model.current_job.is_null_job:
return
show_job_table = bool(self.model.current_job.sub_job_ids)
if not self.model.current_top_level_job.is_null_job:
curr_idx = self.job_details_tab.currentIndex()
# Replace with no-subjobs widget
if not show_job_table and self._job_table_active:
self.job_details_tab.removeTab(0)
self.job_details_tab.insertTab(0, self.no_subjobs_widget,
SUB_JOBS_TAB)
self._job_table_active = False
# Replace with job table
elif show_job_table and not self._job_table_active:
self.job_details_tab.removeTab(0)
self.job_details_tab.insertTab(0, self.job_table, SUB_JOBS_TAB)
self._job_table_active = True
self.job_details_tab.setCurrentIndex(curr_idx)
# Show the job files tab if the job has no sub-job
if not show_job_table:
self.job_details_tab.setCurrentIndex(1)
self.sub_job_bar.setVisible(not self.model.current_job.is_top_level_job)
self.top_level_job_bar.next_job_btn.setVisible(
self.model.current_job.is_top_level_job)
self.top_level_job_bar.prev_job_btn.setVisible(
self.model.current_job.is_top_level_job)
if self.model.current_job.is_top_level_job:
self.top_level_job_bar.next_job_btn.setEnabled(
self.model.current_job.job_index <
len(self.model.top_level_jobs) - 1)
self.top_level_job_bar.prev_job_btn.setEnabled(
bool(self.model.current_job.job_index))
else:
self.sub_job_bar.next_job_btn.setEnabled(
self.model.current_job.job_index <
len(self.model.current_top_level_job.sub_job_ids) - 1)
self.sub_job_bar.prev_job_btn.setEnabled(
bool(self.model.current_job.job_index))
[docs] def viewJobDetails(self, index):
job_model = index.data(plptable.ROW_OBJECT_ROLE)
job_model.job_index = index.row()
self.model.setCurrentJob(job_model)
self.sub_job_bar.incrementNestingLevel()
[docs] def onSubJobBack(self):
current_job = self.model.current_job
assert not current_job.is_top_level_job
self.model.setCurrentJob(self.model.getParentJob(current_job))
self.job_details_tab.setCurrentWidget(self.job_table)
self.sub_job_bar.decrementNestingLevel()
[docs] def onTopLevelBack(self):
self.model.showJobsList()
self.sub_job_bar.resetNestingLevel()
[docs]class JobBar(StyleMixin, mappers.MapperMixin, basewidgets.BaseWidget):
model_class = job_monitor_models.JobModel
BUTTON_TEXT = NotImplemented
[docs] def initSetUp(self):
super().initSetUp()
self.back_button = BackButton(self.BUTTON_TEXT)
self.status_icon = StatusIcon(self)
sp = QtWidgets.QSizePolicy(QtWidgets.QSizePolicy.Expanding,
QtWidgets.QSizePolicy.Preferred)
self.lbl = QtWidgets.QLabel()
self.lbl.setSizePolicy(sp)
sp = QtWidgets.QSizePolicy(QtWidgets.QSizePolicy.Fixed,
QtWidgets.QSizePolicy.Fixed)
self.progress_bar = job_monitor_progress_bar.ProgressBar(self)
self.progress_bar.setSizePolicy(sp)
self.prev_job_btn = self.getButtonWithIcon(ICON_PATH + "back-arrow.png",
"prev_job_btn")
self.next_job_btn = self.getButtonWithIcon(ICON_PATH + "next-arrow.png",
"next_job_btn")
[docs] def initLayOut(self):
super().initLayOut()
h_layout = QtWidgets.QHBoxLayout()
h_layout.addWidget(self.back_button)
h_layout.addStretch()
h_layout.addWidget(self.status_icon)
h_layout.addWidget(self.lbl)
h_layout.addStretch()
h_layout.addWidget(self.progress_bar)
h_layout.addStretch()
h_layout.addWidget(self.prev_job_btn)
h_layout.addWidget(self.next_job_btn)
self.main_layout.addLayout(h_layout)
self.widget_layout.setContentsMargins(0, 0, 0, 0)
self.main_layout.setContentsMargins(0, 0, 0, 0)
[docs] def defineMappings(self):
M = self.model_class
lbl_tgt = mappers.TargetSpec(getter=self.lbl.text,
setter=self._updateLbl)
return [
(lbl_tgt, M.job_name),
(self.progress_bar, M),
(self.status_icon, M.status),
]
def _updateLbl(self, text):
"""
Set label text. Extend/override this method if you want to add/modify
some functionality.
"""
self.lbl.setText(text)
[docs]class TopLevelJobBar(JobBar):
"""
This widget will show the back button to take the GUI back to the top
level job table, and info (name, progress) of the top level job
"""
BUTTON_TEXT = "Jobs List"
[docs] def initLayOut(self):
super().initLayOut()
self.main_layout.setContentsMargins(2, 0, 5, 5)
def _updateLbl(self, text):
"""
Set a tooltip on the label along with text.
"""
super()._updateLbl(text)
self.lbl.setToolTip(f'Job name: {text}')
[docs]class SubJobBar(JobBar):
"""
This widget will show the back button to take the GUI back to the
parent's subjob table, and info (name, progress) of the selected sub job
"""
BUTTON_TEXT = "Sub-Jobs List"
[docs] def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._nesting_level = 0
[docs] def initLayOut(self):
super().initLayOut()
self.main_layout.setContentsMargins(12, 5, 12, 5)
def _updateLbl(self, text):
"""
Set a text and tooltip on the label. Also elide the text from left end
if the text is too long.
"""
# text shouldn't be longer than `self.width() // 3` after elidation
elided_text = _get_elided_text(self.lbl.font(), text, self.width() // 3)
self.lbl.setText(elided_text)
self.lbl.setToolTip(f'Sub-job name: {text}')
[docs] def resizeEvent(self, event):
super().resizeEvent(event)
self._updateLbl(self.model.job_name)
[docs] def updateButtonText(self):
btn_text = self.BUTTON_TEXT
if self._nesting_level > 1:
btn_text += f" ({self._nesting_level})"
self.back_button.setText(btn_text)
[docs] def incrementNestingLevel(self):
self._nesting_level += 1
self.updateButtonText()
[docs] def decrementNestingLevel(self):
self._nesting_level -= 1
self.updateButtonText()
[docs] def resetNestingLevel(self):
self._nesting_level = 0
self.updateButtonText()
[docs]class StatusIcon(mappers.TargetMixin, flat_button.FlatButton):
[docs] def targetSetValue(self, new_status):
icon_path = get_icon_path_for_status(new_status)
self.setIconPath(icon_path)
def _is_project_scratch(project_name):
"""
Check if the given project is scratch project by checking if it follows the
Maestro temp project naming pattern as in 'Tmp_04Jun2021_0919_49688'.
This function would give false positive result if the saved project name
follows the specific pattern.
:param project_name: Name of the project.
:type project_path: str
:return: True is the project is scratch else False
:rtype: bool
"""
temp_project_pattern = mm.mmcommon_get_scratch_project_regular_expression()
return bool(re.match(temp_project_pattern, project_name))