Source code for schrodinger.application.job_monitor.job_monitor_diagnostics_dialog

import os
from datetime import datetime

from schrodinger.application.job_monitor import diagnostics_dialog_ui
from schrodinger.models import mappers
from schrodinger.models import parameters
from schrodinger.Qt import QtCore
from schrodinger.Qt import QtGui
from schrodinger.tasks import tasks
from schrodinger.ui.qt.basewidgets import BaseDialog
from schrodinger.utils import subprocess

DIAGNOSTICS_STATUS = "Collecting diagnostics..."

MULTI_JOB_DIAGNOSTICS_INFO = """
<p>Archive files containing the diagnostics information have <br/>been added
to the current working directory for each of the<br/>selected jobs.
</p><p>No structures were included in the archives, and all directory <br/>
names were hidden.
</p>
"""

SINGLE_JOB_DIAGNOSTICS_INFO = """
<p>An archive file containing the diagnostics information has<br/>been added to
 the current working directory.</p><p>No structures were included in the
 archive, and all directory<br/>names were hidden.</p>
"""


class CollectJobDiagnosticsModel(parameters.CompoundParam):
    job_ids: list


class CollectJobDiagnosticsDialog(mappers.MapperMixin, BaseDialog):
    """
    :ivar updateStatusBar: a signal to update the status of `CollectJobDiagnosticsTask`.
    :vartype updateStatusBar: QtCore.pyqtSignal
    """
    model_class = CollectJobDiagnosticsModel
    ui_module = diagnostics_dialog_ui

    updateStatusBar = QtCore.pyqtSignal(str)

    def __init__(self, job_ids, parent=None):
        super().__init__(parent)
        self.model.job_ids = job_ids

    def initSetUp(self):
        super().initSetUp()
        self.setWindowTitle("Job Diagnostics Collected")
        self.ui.info_icon.setPixmap(
            QtGui.QPixmap(":/job_monitor/icons/info-h.png"))
        self.ui.ok_btn.clicked.connect(self.close)

    def startTask(self):
        self.updateStatusBar.emit(DIAGNOSTICS_STATUS)
        self.task = CollectJobDiagnosticsTask(input=self.model)
        self.task.start()
        self.task.statusChanged.connect(self.onTaskStatusChanged)
        self.close()

    def onTaskStatusChanged(self, status):
        t = self.task
        if status is t.DONE and t.output.success:
            status = MULTI_JOB_DIAGNOSTICS_INFO if len(
                self.model.job_ids) > 1 else SINGLE_JOB_DIAGNOSTICS_INFO
            self.ui.info_label.setText(status)
            self.run()
        elif status is t.DONE and not t.output.success:
            self.warning(f"Postmortem generation failed. Log at "
                         f"{t.getLogPath()}")
        elif status is t.FAILED:
            self.warning("Postmortem generation failed.")
        self.updateStatusBar.emit("")


class CollectJobDiagnosticsTask(tasks.ThreadFunctionTask):
    input: CollectJobDiagnosticsModel

    class output(parameters.CompoundParam):
        success: bool

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        if len(self.input.job_ids) > 1:
            now = datetime.now()
            time_stamp = now.strftime("%d%b%Y_%H%M")
            self.name = f"Diagnostic_Archives_{time_stamp}"

            task_dir_spec = self.AUTO_TASKDIR
            if os.path.exists(self.name):
                task_dir_spec = self.name

            self.specifyTaskDir(task_dir_spec)

    def mainFunction(self):
        o = self.output
        o.success = generate_postmortem(self.input.job_ids, self.getLogPath(),
                                        self.getTaskDir())

    def getLogPath(self):
        return self.getTaskFilename("postmortem-error.log")


def generate_postmortem(job_ids, log_path, task_dir):
    args = create_postmortem_args(job_ids)
    try:
        with open(log_path, "w") as error_output:
            exit_code = subprocess.call(args,
                                        stdout=error_output,
                                        stderr=error_output,
                                        cwd=task_dir)
    except OSError:
        # Raised when OS fails to open file in write mode
        return False
    success = exit_code == 0
    return success


def create_postmortem_args(job_ids):
    args = ["jsc", "postmortem"]
    args.extend(job_ids)
    return args