comparison toolfactory/rgToolFactory2.py @ 56:68fbdbe35f08 draft

Uploaded

55:dcf59d4e4515

            clp = self.args.cl_prefix.split(" ")
            for c in clp:
                aCL(c)
                aXCL(c)
        else:
            if self.args.runmode == "Executable":
                if self.args.script_path:
                    aCL(self.executeme)
                    aCL(self.sfile)
                    aXCL(self.executeme)
                    aXCL("$runme")
                else:
                    aCL(self.executeme)  # this little CL will just run
                    aXCL(self.executeme)
            else:
                if self.args.script_path:
                    aCL(self.executeme)
                    aCL(self.sfile)
                    aXCL(self.executeme)
                    aXCL("$runme")
                else:
                    aCL(self.executeme)  # this little CL will just run
                    aXCL(self.executeme)
        self.elog = os.path.join(self.repdir, "%s_error_log.txt" % self.tool_name)
        self.tlog = os.path.join(self.repdir, "%s_runner_log.txt" % self.tool_name)

        if self.args.parampass == "0":
            self.clsimple()

        Create a Galaxy xml tool wrapper for the new script
        Uses galaxyhtml
        Hmmm. How to get the command line into correct order...
        """
        if self.command_override:
            self.newtool.command_line_override = self.command_override  # config file
        else:
            self.newtool.command_line_override = self.xmlcl
        if self.args.help_text:
            helptext = open(self.args.help_text, "r").readlines()
            safertext = [html_escape(x) for x in helptext]
            if self.args.script_path:
                scrpt = self.script.split('\n')

        # ready for the tarball

    def run(self):
        """
        generate test outputs by running a command line
        won't work if command or test override in play - planemo is the 
        easiest way to generate test outputs for that case so is
        automagically selected
        """
        s = "run cl=%s" % str(self.cl)
        logging.debug(s)
        scl = " ".join(self.cl)
        err = None
        if self.args.parampass != "0":
            ste = open(self.elog, "wb")
            if self.lastclredirect:
                sto = open(self.lastclredirect[1], "wb")  # is name of an output file
            else:
                sto = open(self.tlog, "wb")
                sto.write(
                    bytes(
                        "## Executing Toolfactory generated command line = %s\n" % scl,
                        "utf8",
                    )
                )
            sto.flush()
            p = subprocess.run(self.cl, shell=False, stdout=sto, stderr=ste)
            sto.close()
            ste.close()
            tmp_stderr = open(self.elog, "rb")
            err = ""
            buffsize = 1048576
            try:
                while True:
                    err += str(tmp_stderr.read(buffsize))
                    if not err or len(err) % buffsize != 0:
                        break
            except OverflowError:
                pass
            tmp_stderr.close()
            retval = p.returncode
        else:  # work around special case - stdin and write to stdout
            sti = open(self.infiles[0][IPATHPOS], "rb")
            sto = open(self.outfiles[0][ONAMEPOS], "wb")
            # must use shell to redirect

            "--shed_key",
            self.args.toolshed_api_key,
            "--tar",
            self.newtarpath,
        ]
        p = subprocess.run(cll, shell=True)
        print("Ran", " ".join(cll), "got", p.returncode)
        tout.close()
        return p.returncode

    def planemo_test(self, genoutputs=True):

                "planemo",
                "test",
                "--galaxy_root",
                self.args.galaxy_root,
                "--update_test_data",
                "--galaxy_python_version",
                "3.6",
                xreal,
            ]
        else:
            cll = ["planemo", "test", "--galaxy_python_version",
                "3.6", "--galaxy_root",
                self.args.galaxy_root,
                xreal,]
        p = subprocess.run(
                cll, shell=True, cwd=self.tooloutdir, stderr=tout, stdout=tout
            )
        if genoutputs:
            with os.scandir(self.testdir) as outs:
                for entry in outs:
                    if entry.is_file():

            shutil.copyfile(pth, dest)
            dest = os.path.join(self.repdir, "%s.%s" % (p[ICLPOS], p[IFMTPOS]))
            shutil.copyfile(pth, dest)

    def makeToolTar(self):
        """ move outputs into test-data and prepare the tarball 
        """
        for p in self.outfiles:
            src = p[ONAMEPOS]
            if os.path.isfile(src):
                dest = os.path.join(self.testdir, "%s_sample" % src)

    a("--additional_parameters", action="append", default=[])
    a("--edit_additional_parameters", action="store_true", default=False)
    a("--parampass", default="positional")
    a("--tfout", default="./tfout")
    a("--new_tool", default="new_tool")
    a("--runmode", default=None)
    a("--galaxy_url", default="http://localhost:8080")
    a("--galaxy_api_key", default="fakekey")
    a("--toolshed_url", default="http://localhost:9009")
    a("--toolshed_api_key", default="fakekey")
    a("--galaxy_root", default="/galaxy-central")

56:68fbdbe35f08

            clp = self.args.cl_prefix.split(" ")
            for c in clp:
                aCL(c)
                aXCL(c)
        else:
            if self.args.script_path:
                aCL(self.executeme)
                aCL(self.sfile)
                aXCL(self.executeme)
                aXCL("$runme")
            else:
                aCL(self.executeme)  # this little CL will just run
                aXCL(self.executeme)
        self.elog = os.path.join(self.repdir, "%s_error_log.txt" % self.tool_name)
        self.tlog = os.path.join(self.repdir, "%s_runner_log.txt" % self.tool_name)

        if self.args.parampass == "0":
            self.clsimple()

        Create a Galaxy xml tool wrapper for the new script
        Uses galaxyhtml
        Hmmm. How to get the command line into correct order...
        """
        if self.command_override:
            self.newtool.command_override = self.command_override  # config file
        else:
            self.newtool.command_override = self.xmlcl
        if self.args.help_text:
            helptext = open(self.args.help_text, "r").readlines()
            safertext = [html_escape(x) for x in helptext]
            if self.args.script_path:
                scrpt = self.script.split('\n')

        # ready for the tarball

    def run(self):
        """
        generate test outputs by running a command line
        won't work if command or test override in play - planemo is the
        easiest way to generate test outputs for that case so is
        automagically selected
        """
        s = "run cl=%s" % str(self.cl)
        logging.debug(s)
        scl = " ".join(self.cl)
        err = None
        if self.args.parampass != "0":
            if os.path.exists(self.elog):
                ste = open(self.elog, "a")
            else:
                ste = open(self.elog, "w")
            if self.lastclredirect:
                sto = open(self.lastclredirect[1], "wb")  # is name of an output file
            else:
                if os.path.exists(self.tlog):
                    sto = open(self.tlog, "a")
                else:
                    sto = open(self.tlog, "w")
                sto.write(
                        "## Executing Toolfactory generated command line = %s\n" % scl
                )
            sto.flush()
            p = subprocess.run(self.cl, shell=False, stdout=sto, stderr=ste)
            sto.close()
            ste.close()
            retval = p.returncode
        else:  # work around special case - stdin and write to stdout
            sti = open(self.infiles[0][IPATHPOS], "rb")
            sto = open(self.outfiles[0][ONAMEPOS], "wb")
            # must use shell to redirect

            "--shed_key",
            self.args.toolshed_api_key,
            "--tar",
            self.newtarpath,
        ]
        p = subprocess.run(cll, shell=False)
        print("Ran", " ".join(cll), "got", p.returncode)
        tout.close()
        return p.returncode

    def planemo_test(self, genoutputs=True):

                "planemo",
                "test",
                "--galaxy_root",
                self.args.galaxy_root,
                "--update_test_data",
                xreal,
            ]
        else:
            cll = ["planemo", "test", "--galaxy_root",
                self.args.galaxy_root,
                xreal,]
        p = subprocess.run(
                cll, shell=False, cwd=self.tooloutdir, stderr=tout, stdout=tout
            )
        if genoutputs:
            with os.scandir(self.testdir) as outs:
                for entry in outs:
                    if entry.is_file():

            shutil.copyfile(pth, dest)
            dest = os.path.join(self.repdir, "%s.%s" % (p[ICLPOS], p[IFMTPOS]))
            shutil.copyfile(pth, dest)

    def makeToolTar(self):
        """ move outputs into test-data and prepare the tarball
        """
        for p in self.outfiles:
            src = p[ONAMEPOS]
            if os.path.isfile(src):
                dest = os.path.join(self.testdir, "%s_sample" % src)

    a("--additional_parameters", action="append", default=[])
    a("--edit_additional_parameters", action="store_true", default=False)
    a("--parampass", default="positional")
    a("--tfout", default="./tfout")
    a("--new_tool", default="new_tool")
    a("--galaxy_url", default="http://localhost:8080")
    a("--galaxy_api_key", default="fakekey")
    a("--toolshed_url", default="http://localhost:9009")
    a("--toolshed_api_key", default="fakekey")
    a("--galaxy_root", default="/galaxy-central")