Source code for schrodinger.tasks.gui

from schrodinger.tasks.tasks import CallingContext
from schrodinger.tasks.tasks import TaskDirNotFoundError
from schrodinger.ui.qt import messagebox
from schrodinger.utils import fileutils


def start_task(task, parent=None):
    """
    Start a task using the GUI to interact with preprocessing. Preprocessing
    errors will be shown in a message box, and preprocessing warnings will be
    shown to the user in the form a of a question box that allows the user to
    continue or cancel start.

    :param task: the task to start
    :type tasl: schrodinger.tasks.tasks.AbstractTask

    :param parent: the parent widget, used for positioning the popup messages
    :type parent: QtWidgets.QWidget
    :return: whether the task started
    :rtype: bool
    """

    proceed = _run_preprocessing_with_gui(task, parent)
    if proceed:
        task.start(skip_preprocessing=True)
    else:
        _clean_up_task_dir(task)
    return proceed


def write_task(task, parent=None):
    """
    Write a task using the GUI to interact with preprocessing. Preprocessing
    errors will be shown in a message box, and preprocessing warnings will be
    shown to the user in the form a of a question box that allows the user to
    continue or cancel write.

    :param task: the task to write
    :type tasl: schrodinger.tasks.tasks.AbstractTask

    :param parent: the parent widget, used for positioning the popup messages
    :type parent: QtWidgets.QWidget
    :return: whether the write was successful
    :rtype: bool
    """
    try:
        task._write_mode = True
        # Preprocessing should be run in the context of write mode as we may
        # want to selectively run preprocessors during write operation.
        proceed = _run_preprocessing_with_gui(task, parent)
    finally:
        task._write_mode = False

    if proceed:
        task.write(skip_preprocessing=True)
    else:
        _clean_up_task_dir(task)
    return proceed


def _run_preprocessing_with_gui(task, parent):
    proceed = True

    def callback(result):
        nonlocal proceed
        proceed = result.passed
        if result.message:
            if result.passed:
                proceed = messagebox.show_question(parent,
                                                   text=result.message,
                                                   title='Warning',
                                                   yes_text='Continue',
                                                   no_text='Cancel',
                                                   add_cancel_btn=False)
            else:
                messagebox.show_error(parent, text=result.message)
        return proceed

    task.runPreprocessing(callback, calling_context=CallingContext.GUI)
    return proceed


def _clean_up_task_dir(task):
    """
    Clean up the task dir for `task`. This can be called even if the task
    hasn't created a task directory yet.
    """
    if task.taskDirSetting() is task.AUTO_TASKDIR:
        try:
            taskdir = task.getTaskDir()
        except TaskDirNotFoundError:  # taskdir hasn't been created yet
            pass
        else:
            fileutils.force_rmtree(taskdir)