tool_factory_2: toolfactory/rgToolFactory2.py comparison

comparison toolfactory/rgToolFactory2.py @ 50:bf432f4486c7 draft

Uploaded

author	fubar
date	Sun, 30 Aug 2020 01:45:27 -0400
parents	35a912ce0c83
children	4b71b252eacb

comparison

equal deleted inserted replaced

-:35a912ce0c83
+:bf432f4486c7
 # essential problem is to create two command lines - one for the tool xml and a different
 # one to run the executable with the supplied test data and settings
 # Be simpler to write the tool, then run it with planemo and soak up the test outputs.
 import argparse
 import logging
 import os
 import re
 import shutil
 myversion = "V2.1 July 2020"
 verbose = True
 debug = True
 toolFactoryURL = "https://github.com/fubar2/toolfactory"
 ourdelim = "~~~"
-ALOT = 10000000 # srsly. command or test overrides use read() so just in case
+ALOT = 10000000  # srsly. command or test overrides use read() so just in case
 STDIOXML = """<stdio>
 <exit_code range="100:" level="debug" description="shite happens" />
 </stdio>"""
 # --input_files="$input_files~~~$CL~~~$input_formats~~~$input_label
 "argparse",
 "positional",
 ], 'args.parampass must be "0","positional" or "argparse"'
 self.tool_name = re.sub("[^a-zA-Z0-9_]+", "", args.tool_name)
 self.tool_id = self.tool_name
-self.tool = gxt.Tool(
+self.newtool = gxt.Tool(
 self.args.tool_name,
 self.tool_id,
 self.args.tool_version,
 self.args.tool_desc,
 FAKEEXE,
 )
-if self.args.script_path:
-self.tool.interpreter = self.executeme
 self.tooloutdir = "tfout"
 self.repdir = "TF_run_report_tempdir"
 self.testdir = os.path.join(self.tooloutdir, "test-data")
 if not os.path.exists(self.tooloutdir):
 os.mkdir(self.tooloutdir)
 if self.args.test_override:
 stos = open(self.args.test_override, "r").readlines()
 self.test_override = [x.rstrip() for x in stos]
 else:
 self.test_override = None
-if self.args.cl_prefix: # DIY CL start
+if self.args.cl_prefix:  # DIY CL start
 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.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)
+self.tlog = os.path.join(self.repdir, "%s_runner_log.txt" % self.tool_name)
 if self.args.parampass == "0":
 self.clsimple()
 else:
 clsuffix = []
 aparm.positional = int(oldcl)
 aparm.command_line_override = "$%s" % newcl
 self.toutputs.append(aparm)
 usetest = None
 ld = None
-if test > '':
+if test > "":
-if test.startswith('diff'):
+if test.startswith("diff"):
-usetest = 'diff'
+usetest = "diff"
-if test.split(':')[1].isdigit:
+if test.split(":")[1].isdigit:
-ld = int(test.split(':')[1])
+ld = int(test.split(":")[1])
 else:
 usetest = test
-tp = gxtp.TestOutput(name=newcl, value="%s_sample" % newcl, format=newfmt,
+tp = gxtp.TestOutput(
-compare=usetest, lines_diff=ld, delta=None,)
+name=newcl,
+value="%s_sample" % newcl,
+format=newfmt,
+compare=usetest,
+lines_diff=ld,
+delta=None,
+)
 self.testparam.append(tp)
 for p in self.infiles:
 newname = p[ICLPOS]
 newfmt = p[IFMTPOS]
 ndash = self.getNdash(newname)
 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.tool.command_line_override = self.command_override # config file
+self.newtool.command_line_override = self.command_override  # config file
 else:
-self.tool.command_line_override = self.xmlcl
+self.newtool.command_line_override = self.xmlcl
-# if self.args.interpreter_name:
-#    self.tool.interpreter = self.args.interpreter_name
 if self.args.help_text:
 helptext = open(self.args.help_text, "r").readlines()
-helptext = [html_escape(x) for x in helptext]
+safertext = [html_escape(x) for x in helptext]
-self.tool.help = "".join([x for x in helptext])
+if self.args.script_path:
-else:
+scrpt = self.script.split('\n')
-self.tool.help = (
+scrpt.append("```\n")
+if len(scrpt) > 300:
+safertext = safertext + scrpt[:100] + ['>500 lines - stuff deleted','......'] + scrpt[-100:]
+else:
+safertext = safertext + scrpt
+safertext.append("\n```")
+self.newtool.help = "".join([x for x in safertext])
+else:
+self.newtool.help = (
 "Please ask the tool author (%s) for help \
 as none was supplied at tool generation\n"
 % (self.args.user_email)
 )
-self.tool.version_command = None  # do not want
+self.newtool.version_command = None  # do not want
 requirements = gxtp.Requirements()
 if self.args.packages:
 for d in self.args.packages.split(","):
 if ":" in d:
 packg, ver = d.split(":")
 else:
 packg = d
 ver = ""
-requirements.append(gxtp.Requirement("package", packg.strip(), ver.strip()))
+requirements.append(
-self.tool.requirements = requirements
+gxtp.Requirement("package", packg.strip(), ver.strip())
+)
+self.newtool.requirements = requirements
 if self.args.parampass == "0":
 self.doNoXMLparam()
 else:
 self.doXMLparam()
-self.tool.outputs = self.toutputs
+self.newtool.outputs = self.toutputs
-self.tool.inputs = self.tinputs
+self.newtool.inputs = self.tinputs
-if (
+if self.args.script_path:
-self.args.script_path
-):
 configfiles = gxtp.Configfiles()
 configfiles.append(gxtp.Configfile(name="runme", text=self.script))
-self.tool.configfiles = configfiles
+self.newtool.configfiles = configfiles
 tests = gxtp.Tests()
 test_a = gxtp.Test()
 for tp in self.testparam:
 test_a.append(tp)
 tests.append(test_a)
-self.tool.tests = tests
+self.newtool.tests = tests
-self.tool.add_comment(
+self.newtool.add_comment(
 "Created by %s at %s using the Galaxy Tool Factory."
 % (self.args.user_email, timenow())
 )
-self.tool.add_comment("Source in git at: %s" % (toolFactoryURL))
+self.newtool.add_comment("Source in git at: %s" % (toolFactoryURL))
-self.tool.add_comment(
+self.newtool.add_comment(
 "Cite: Creating re-usable tools from scripts doi: \
 10.1093/bioinformatics/bts573"
 )
-exml0 = self.tool.export()
+exml0 = self.newtool.export()
 exml = exml0.replace(FAKEEXE, "")  # temporary work around until PR accepted
-if self.test_override: # cannot do this inside galaxyxml as it expects lxml objects for tests
+if (
-part1 = exml.split('<tests>')[0]
+self.test_override
-part2 = exml.split('</tests>')[1]
+):  # cannot do this inside galaxyxml as it expects lxml objects for tests
-fixed = '%s\n%s\n%s' % (part1,self.test_override,part2)
+part1 = exml.split("<tests>")[0]
+part2 = exml.split("</tests>")[1]
+fixed = "%s\n%s\n%s" % (part1, self.test_override, part2)
 exml = fixed
 xf = open("%s.xml" % self.tool_name, "w")
 xf.write(exml)
 xf.write("\n")
 xf.close()
 # ready for the tarball
-def makeTool(self):
-"""write xmls and samples into place
-"""
-self.makeXML()
-if self.args.script_path:
-stname = os.path.join(self.tooloutdir, "%s" % (self.sfile))
-if not os.path.exists(stname):
-shutil.copyfile(self.sfile, stname)
-xreal = "%s.xml" % self.tool_name
-xout = os.path.join(self.tooloutdir, xreal)
-shutil.copyfile(xreal, xout)
-for p in self.infiles:
-pth = p[IPATHPOS]
-dest = os.path.join(self.testdir, "%s_sample" % p[ICLPOS])
-shutil.copyfile(pth, dest)
-def makeToolTar(self):
-self.newtarpath = "toolfactory_%s.tgz" % self.tool_name
-tf = tarfile.open(self.newtarpath, "w:gz")
-tf.add(name=self.tooloutdir, arcname=self.tool_name)
-tf.close()
-shutil.copyfile(self.newtarpath, self.args.new_tool)
-if os.path.exists(self.tlog) and os.stat(self.tlog).st_size > 0:
-shutil.copyfile(
-self.tlog, os.path.join(self.tooloutdir, "test1_log_outfiletxt")
-)
-def moveRunOutputs(self):
-"""need to move files into toolfactory collection after any run - planemo or not
-"""
-for p in self.outfiles:
-naym = p[ONAMEPOS]
-src = os.path.join(self.tooloutdir,naym)
-if os.path.isfile(src):
-dest = os.path.join(self.testdir, "%s_sample" % naym)
-shutil.copyfile(naym, dest)
-else:
-print('### problem - output file %s not found in tooloutdir %s' % (src,self.tooloutdir))
-with os.scandir(self.tooloutdir) as outs:
-for entry in outs:
-if not entry.is_file() or entry.name.startswith('.'):
-continue
-if "." in entry.name:
-nayme,ext = os.path.splitext(entry.name)
-else:
-ext = ".txt"
-ofn = "%s%s" % (entry.name.replace(".", "_"), ext)
-dest = os.path.join(self.repdir, ofn)
-src = os.path.join(self.tooloutdir,entry.name)
-shutil.copyfile(src, dest)
 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")
 """
 planemo shed_create --shed_target testtoolshed
 planemo shed_update --check_diff --shed_target testtoolshed
 """
 if os.path.exists(self.tlog):
-tout = open(self.tlog,'a')
+tout = open(self.tlog, "a")
 else:
-tout = open(self.tlog,'w')
+tout = open(self.tlog, "w")
 cll = ["planemo", "shed_create", "--shed_target", "local"]
 try:
-p = subprocess.run(cll, shell=False, cwd=self.tooloutdir, stdout=tout, stderr = tout)
+p = subprocess.run(
+cll, shell=True, cwd=self.tooloutdir, stdout=tout, stderr=tout
+)
 except:
 pass
 if p.returncode != 0:
 print("Repository %s exists" % self.args.tool_name)
 else:
 "--shed_key",
 self.args.toolshed_api_key,
 "--tar",
 self.newtarpath,
 ]
-print("Run", " ".join(cll))
+p = subprocess.run(cll, shell=True)
-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 is a requirement so is available
+"""planemo is a requirement so is available for testing
+and for generating test outputs if command or test overrides are supplied
+test outputs are sent to repdir for display
 """
 xreal = "%s.xml" % self.tool_name
 if os.path.exists(self.tlog):
-tout = open(self.tlog,'a')
+tout = open(self.tlog, "a")
 else:
-tout = open(self.tlog,'w')
+tout = open(self.tlog, "w")
 if genoutputs:
 cll = [
 "planemo",
 "test",
 "--galaxy_root",
 self.args.galaxy_root,
 "--update_test_data",
+"--galaxy_python_version",
+"3.6",
+"--no_conda_auto_install",
+"--no_conda_auto_init",
 xreal,
 ]
 else:
-cll = ["planemo", "test", "--galaxy_root", self.args.galaxy_root, xreal]
+cll = ["planemo", "test", "--galaxy_python_version",
-try:
+"3.6", "--galaxy_root",
-p = subprocess.run(cll, shell=False, cwd=self.tooloutdir, stderr=tout, stdout=tout)
+self.args.galaxy_root,
-except:
+"--no_conda_auto_install",
-pass
+"--no_conda_auto_init",
+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():
+dest = os.path.join(self.repdir, entry.name)
+src = os.path.join(self.testdir, entry.name)
+shutil.copyfile(src, dest)
+tout.write(
+"Copied output %s to %s after planemo test\n" % (src, dest)
+)
 tout.close()
 return p.returncode
 def eph_galaxy_load(self):
-"""
+"""load the new tool from the local toolshed after planemo uploads it
 """
 if os.path.exists(self.tlog):
-tout = open(self.tlog,'a')
+tout = open(self.tlog, "a")
 else:
-tout = open(self.tlog,'w')
+tout = open(self.tlog, "w")
 cll = [
 "shed-tools",
 "install",
 "-g",
 self.args.galaxy_url,
 print("installed %s" % self.args.tool_name)
 tout.close()
 return p.returncode
 def writeShedyml(self):
+"""for planemo
+"""
 yuser = self.args.user_email.split("@")[0]
 yfname = os.path.join(self.tooloutdir, ".shed.yml")
 yamlf = open(yfname, "w")
 odict = {
 "name": self.tool_name,
 "owner": yuser,
 "type": "unrestricted",
 "description": self.args.tool_desc,
+"synopsis": self.args.tool_desc,
+"category": "TF Generated Tools",
 }
 yaml.dump(odict, yamlf, allow_unicode=True)
 yamlf.close()
+def makeTool(self):
-def install_load(self):
+"""write xmls and input samples into place
-_ = self.planemo_test(genoutputs=True)
+"""
-testres = self.planemo_test(genoutputs=False)
+self.makeXML()
-if testres == 0:
+if self.args.script_path:
-if self.args.make_Tool == "install":
+stname = os.path.join(self.tooloutdir, "%s" % (self.sfile))
-self.planemo_shedload()
+if not os.path.exists(stname):
-self.eph_galaxy_load()
+shutil.copyfile(self.sfile, stname)
-else:
+xreal = "%s.xml" % self.tool_name
-os.stderr.write(
+xout = os.path.join(self.tooloutdir, xreal)
-"Planemo test failed - tool %s was not installed" % self.args.tool_name
+shutil.copyfile(xreal, xout)
-)
+for p in self.infiles:
+pth = p[IPATHPOS]
+dest = os.path.join(self.testdir, "%s_sample" % p[ICLPOS])
+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)
+shutil.copyfile(src, dest)
+dest = os.path.join(self.repdir, "%s.%s" % (src, p[OFMTPOS]))
+shutil.copyfile(src, dest)
+else:
+print(
+"### problem - output file %s not found in tooloutdir %s"
+% (src, self.tooloutdir)
+)
+self.newtarpath = "toolfactory_%s.tgz" % self.tool_name
+tf = tarfile.open(self.newtarpath, "w:gz")
+tf.add(name=self.tooloutdir, arcname=self.tool_name)
+tf.close()
+shutil.copyfile(self.newtarpath, self.args.new_tool)
+def moveRunOutputs(self):
+"""need to move planemo or run outputs into toolfactory collection
+"""
+with os.scandir(self.tooloutdir) as outs:
+for entry in outs:
+if not entry.is_file() or entry.name.startswith("."):
+continue
+if "." in entry.name:
+nayme, ext = os.path.splitext(entry.name)
+else:
+ext = ".txt"
+ofn = "%s%s" % (entry.name.replace(".", "_"), ext)
+dest = os.path.join(self.repdir, ofn)
+src = os.path.join(self.tooloutdir, entry.name)
+shutil.copyfile(src, dest)
 def main():
 """
 This is a Galaxy wrapper. It expects to be called by a special purpose tool.xml as:
 <command interpreter="python">rgBaseScriptWrapper.py --script_path "$scriptPath"
 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="fbdd3c2eecd191e88939fffc02eeeaf8")
+a("--galaxy_api_key", default="fakekey")
 a("--toolshed_url", default="http://localhost:9009")
-a("--toolshed_api_key", default="d46e5ed0e242ed52c6e1f506b5d7f9f7")
+a("--toolshed_api_key", default="fakekey")
-a("--galaxy_root", default="/home/ross/galaxy")
+a("--galaxy_root", default="/galaxy-central")
 args = parser.parse_args()
 assert not args.bad_user, (
 'UNAUTHORISED: %s is NOT authorized to use this tool until Galaxy admin adds %s to "admin_users" in the Galaxy configuration file'
 % (args.bad_user, args.bad_user)
 for i, x in enumerate(args.additional_parameters):
 args.additional_parameters[i] = args.additional_parameters[i].replace('"', "")
 r = ScriptRunner(args)
 r.writeShedyml()
 r.makeTool()
-if args.make_Tool == "runonly":
+if args.command_override or args.test_override:
+retcode = r.planemo_test(genoutputs=True)  # this fails :( - see PR
+r.moveRunOutputs()
+r.makeToolTar()
+retcode = r.planemo_test(genoutputs=False)
+r.moveRunOutputs()
+if args.make_Tool == "gentestinstall":
+r.planemo_shedload()
+r.eph_galaxy_load()
+else:
 retcode = r.run()
 if retcode:
 sys.stderr.write(
 "## Run failed with return code %d. Cannot build yet. Please fix and retry"
 % retcode
 )
 sys.exit(1)
 else:
 r.moveRunOutputs()
-elif args.make_Tool in ["gentestinstall", "generate", "gentest"]:
+r.makeToolTar()
-retcode = r.run()
+if args.make_Tool in ["gentestinstall", "gentest"]:
-if retcode:
+r.planemo_test(genoutputs=False)
-sys.stderr.write(
+r.moveRunOutputs()
-"## Run failed with return code %d. Cannot build yet. Please fix and retry"
+if args.make_Tool == "gentestinstall":
-% retcode
+r.planemo_shedload()
-)
+r.eph_galaxy_load()
-sys.exit(1)
-r.moveRunOutputs()
-r.makeToolTar()
-if args.make_Tool in ["gentestinstall","gentest"]:
-r.planemo_test(genoutputs=False)
-r.moveRunOutputs()
-r.planemo_shedload()
-r.eph_galaxy_load()
-else:
-retcode = r.planemo_test(genoutputs=True)  # this fails :(
-r.moveRunOutputs()
-r.makeToolTar()
-retcode = r.planemo_test(genoutputs=False)
-r.moveRunOutputs()
-if args.make_Tool == "planemotestinstall":
-r.planemo_shedload()
-r.eph_galaxy_load()
-# if retcode:
-# sys.exit(retcode)  # indicate failure to job runner
 if __name__ == "__main__":
 main()

Mercurial > repos > fubar > tool_factory_2

comparison toolfactory/rgToolFactory2.py @ 50:bf432f4486c7 draft