Source code for schrodinger.ui.qt.appframework2.knime_af2

import os
import subprocess

from schrodinger.ui.qt.appframework2 import af2
from schrodinger.utils import fileutils

# Constants used in AF2 commands written in knime mode
# These constants will then be replaced with actual values by
# KNIME java code, during node execution
POS_ARG_1 = "%POS1%"


class KnimeNodeMixin(object):
    """
    This class is obsolete, and is retained for backwards compatability with
    older KNIME wrappers only.
    """

    def configDialogSettings(self):
        return {}

    def closeKnimePanel(self):
        """
        Called when the user presses "OK" button to close the panel - after
        the panel state is validated. By default, it cases the panel to
        prepare the job in the CWD, and write the <jobname>.sh file.

        Override this method in derived class for custom behavior.

        :return: False if closing of panel should be cancelled. Any other
                return value causes the panel to close afterwards.
        """
        return self.write()

    def writeStateAndClose(self):
        settings_file = self.jobname() + ".json"

        try:
            self.writePanelState(filename=settings_file)
        except TypeError as te:
            import traceback
            print("Error in writing the panel state")
            # NOTE: One disadvantage of using this approach, is that if the
            # unit tests fail, it looks like the failure happened here, as
            # the exception will get printed to the test log file.
            traceback.print_exc()
            fileutils.force_remove(settings_file)

        if self.closeKnimePanel() is not False:
            # Validation passed and command file was written
            self._close()

    def write(self):
        if isinstance(self, af2.JobApp):
            # Panel derived from af2.JobApp

            return af2.JobApp._write(self)
        elif isinstance(self, af2.App):
            # Panel derived from af2.App and use BaseJobRunner
            # to handle jobs
            jobRunner = self.currentTaskRunner()

            # Remove any previous job directories:
            fileutils.force_rmtree(jobRunner.nextJobDir())

            return jobRunner.write()

    def writeCommandFile(self, cmdList):
        # Write the command args to the jobname.cmd file
        with open(self.jobname() + ".cmd", "w") as cmd_file:
            cmd_file.write(subprocess.list2cmdline(cmdList))

    def loadSettings(self, jobname):
        panel_state_file = jobname + '.json'
        if os.path.isfile(panel_state_file) and \
                (os.path.getsize(panel_state_file) > 0):
            self.loadPanelState(filename=panel_state_file)