Source code for schrodinger.application.job_monitor.job_monitor_gui
import os
from schrodinger import get_maestro
from schrodinger.application.job_monitor import job_monitor_models
from schrodinger.application.job_monitor import job_monitor_stylesheets
from schrodinger.application.job_monitor import job_monitor_widgets
from schrodinger.infra import jobhub
from schrodinger.Qt import QtCore
from schrodinger.ui import maestro_ui
from schrodinger.ui.qt import basewidgets
from schrodinger.ui.qt.appframework2 import maestro_callback
from schrodinger.ui.qt.widgetmixins import basicmixins
maestro = get_maestro()
[docs]class JobMonitor(maestro_callback.MaestroCallbackMixin,
basicmixins.StatusBarMixin, basewidgets.Panel):
model_class = job_monitor_models.JobMonitorPanelModel
[docs] def initSetOptions(self):
super().initSetOptions()
self.setWindowTitle("Job Monitor (Beta)")
self.setStyleSheet(job_monitor_stylesheets.STYLE)
self.help_topic = "JOB_MONITOR"
[docs] def initSetUp(self):
super().initSetUp()
self.jobs_list_pane = job_monitor_widgets.JobsListPane(self)
self.job_details_pane = job_monitor_widgets.JobDetailsPane(self)
maestro_hub = maestro_ui.MaestroHub.instance()
maestro_hub.projectOpened.connect(self.updatePanelTitle)
# Every 2 seconds, refresh the tables (to recalculate last updated text)
self.refresh_timer = QtCore.QTimer()
self.refresh_timer.setInterval(2000)
self.refresh_timer.timeout.connect(self.refreshTables)
[docs] def initLayOut(self):
super().initLayOut()
self.widget_layout.setContentsMargins(0, 0, 0, 0)
self.main_layout.setContentsMargins(0, 0, 0, 0)
self.main_layout.addWidget(self.jobs_list_pane)
self.main_layout.addWidget(self.job_details_pane)
self.resize(650, 500)
self.setMinimumWidth(500)
[docs] def initFinalize(self):
if maestro:
pt = maestro.project_table_get()
self.model.current_project_name = pt.project_name
job_manager = jobhub.get_job_manager()
job_manager.jobsChanged.connect(self.updateJobs)
job_manager.jobsDeleted.connect(self.model.deleteJobs)
job_manager.updateView.connect(self.updateJobs)
job_manager.jobCompleted.connect(self._updateJob)
job_manager.aboutToStartJobsDownload.connect(
self._onDownloadAboutToStart)
job_manager.jobDownloaded.connect(self._onDownloadCompleted)
self.refresh_timer.start()
self.updateJobs(job_manager.getJobs(jobhub.JobOption.ALL_JOBS).values())
self.jobs_list_pane.updateStatusBar.connect(self.onStatusBarUpdate)
self.job_details_pane.updateStatusBar.connect(self.onStatusBarUpdate)
def _onDownloadAboutToStart(self, jobs):
"""
Set the download_status for the given jobs as 'DOWNLOADING'
:param jobs: list of job objects
:type jobs: List[schrodinger.job.jobcontrol.Job]
"""
download_status = job_monitor_models.DownloadStatus.DOWNLOADING
for job in jobs:
self.model.setJobDownloadStatus(job.job_id, download_status)
def _onDownloadCompleted(self, job):
"""
Set the download_status for the given job as 'DOWNLOADED'
:param job: job object
:type job: schrodinger.job.jobcontrol.Job
"""
download_status = job_monitor_models.DownloadStatus.DOWNLOADED
self.model.setJobDownloadStatus(job.job_id, download_status)
[docs] def onStatusBarUpdate(self, status):
self.model.status = status
def _updateJob(self, job):
"""
Convenience slot to update a single job
"""
self.updateJobs([job])
[docs] def updateJobs(self, jobs_to_update):
"""
Update the model with the new or updated jobs and incorporate
these jobs in JobsListPane and JobDetailsPane.
:param jobs_to_update: list of all the new or updated jobs
:type jobs_to_update: list[schrodinger.job.jobcontrol.Job]
"""
job_models = []
for job in jobs_to_update:
job_model = job_monitor_models.JobModel.fromJobObject(job)
job_models.append(job_model)
if not job_models:
return
self.model.updateJobs(job_models)
self.requestRestoreSelection()
self.refreshTables()
[docs] def requestRestoreSelection(self):
self.jobs_list_pane.restoreSelectionRequested.emit()
self.job_details_pane.restoreSelectionRequested.emit()
[docs] def refreshTables(self):
self.job_details_pane.refreshJobTable()
self.jobs_list_pane.refreshJobTable()
[docs] def defineMappings(self):
M = self.model_class
return [(self.jobs_list_pane, M),
(self.job_details_pane, M),
(self.onCurrentJobChanged, M.current_job),
(self.updatePanelTitle, M.current_project_jobs_only),
(self.updatePanelTitle, M.current_project_name),
(self.status_lbl, M.status)] # yapf:disable
[docs] def onCurrentJobChanged(self):
self.jobs_list_pane.setVisible(self.model.on_jobs_list)
self.job_details_pane.setVisible(not self.model.on_jobs_list)
[docs] @maestro_callback.project_changed
def syncPanelInfoWithMaestro(self):
"""
Synchronize panel information with Maestro.
"""
pt = maestro.project_table_get()
self.model.current_project_name = pt.project_name
[docs] def updatePanelTitle(self):
"""
Update the panel title based on whether currently showing jobs from
the current project or all projects.
"""
model = self.model
maestro_hub = maestro_ui.MaestroHub.instance()
model.is_curr_project_scratch = maestro_hub.isCurrentProjectScratch()
if model.current_project_jobs_only:
project_name = ''
if model.current_project_name is not None:
if model.is_curr_project_scratch:
project_name = "Scratch Project"
else:
project_name = os.path.basename(
self.model.current_project_name)
else:
project_name = "All Projects"
title = f"Job Monitor (Beta){' - ' if project_name else ''}{project_name}"
self.setWindowTitle(title)
[docs] def selectJob(self, job_id):
"""
Open the panel to the job with jobid `job_id`.
:param job_id: The Job ID
:type job_id: str
"""
job_model = self.model.getJob(job_id)
if not job_model:
return
self.jobs_list_pane.model.setCurrentTopLevelJob(job_model)
self.jobs_list_pane.model.setCurrentJob(job_model)
self.jobs_list_pane.model.active_jobs_only = job_model.is_active
panel = JobMonitor.panel