tool_factory_2: toolfactory/ToolFactory.py comparison

comparison toolfactory/ToolFactory.py @ 3:1c652687a08f draft default tip

Uploaded

author	fubar
date	Fri, 30 Apr 2021 07:06:57 +0000
parents
children

comparison

equal deleted inserted replaced

-:5fc0c9a93072
+:1c652687a08f
+# see https://github.com/fubar2/toolfactory
+#
+# copyright ross lazarus (ross stop lazarus at gmail stop com) May 2012
+#
+# all rights reserved
+# Licensed under the LGPL
+# suggestions for improvement and bug fixes welcome at
+# https://github.com/fubar2/toolfactory
+#
+# April 2021: Refactored into two tools - generate and test/install
+# as part of GTN tutorial development and biocontainer adoption
+# The tester runs planemo on a non-tested archive, creates the test outputs
+# and returns a new proper tool with test.
+# The tester was generated from the ToolFactory_tester.py script
+import argparse
+import copy
+import json
+import logging
+import os
+import re
+import shlex
+import shutil
+import subprocess
+import sys
+import tarfile
+import tempfile
+import time
+import urllib
+from bioblend import ConnectionError
+from bioblend import galaxy
+from bioblend import toolshed
+import galaxyxml.tool as gxt
+import galaxyxml.tool.parameters as gxtp
+import lxml.etree as ET
+import yaml
+myversion = "V2.3 April 2021"
+verbose = True
+debug = True
+toolFactoryURL = "https://github.com/fubar2/toolfactory"
+FAKEEXE = "~~~REMOVE~~~ME~~~"
+# need this until a PR/version bump to fix galaxyxml prepending the exe even
+# with override.
+def timenow():
+"""return current time as a string"""
+return time.strftime("%d/%m/%Y %H:%M:%S", time.localtime(time.time()))
+cheetah_escape_table = {"$": "\\$", "#": "\\#"}
+def cheetah_escape(text):
+"""Produce entities within text."""
+return "".join([cheetah_escape_table.get(c, c) for c in text])
+def parse_citations(citations_text):
+""""""
+citations = [c for c in citations_text.split("**ENTRY**") if c.strip()]
+citation_tuples = []
+for citation in citations:
+if citation.startswith("doi"):
+citation_tuples.append(("doi", citation[len("doi") :].strip()))
+else:
+citation_tuples.append(("bibtex", citation[len("bibtex") :].strip()))
+return citation_tuples
+class Tool_Conf_Updater():
+# update config/tool_conf.xml with a new tool unpacked in /tools
+# requires highly insecure docker settings - like write to tool_conf.xml and to tools !
+# if in a container possibly not so courageous.
+# Fine on your own laptop but security red flag for most production instances
+def __init__(self, args, tool_conf_path, new_tool_archive_path, new_tool_name, tool_dir):
+self.args = args
+self.tool_conf_path = os.path.join(args.galaxy_root,tool_conf_path)
+self.tool_dir = os.path.join(args.galaxy_root, tool_dir)
+self.our_name = 'ToolFactory'
+tff = tarfile.open(new_tool_archive_path, "r:*")
+flist = tff.getnames()
+ourdir = os.path.commonpath(flist) # eg pyrevpos
+self.tool_id = ourdir # they are the same for TF tools
+ourxml = [x for x in flist if x.lower().endswith('.xml')]
+res = tff.extractall()
+tff.close()
+self.run_rsync(ourdir, self.tool_dir)
+self.update_toolconf(ourdir,ourxml)
+def run_rsync(self, srcf, dstf):
+src = os.path.abspath(srcf)
+dst = os.path.abspath(dstf)
+if os.path.isdir(src):
+cll = ['rsync', '-vr', src, dst]
+else:
+cll = ['rsync', '-v', src, dst]
+p = subprocess.run(
+cll,
+capture_output=False,
+encoding='utf8',
+shell=False,
+)
+def install_deps(self):
+gi = galaxy.GalaxyInstance(url=self.args.galaxy_url, key=self.args.galaxy_api_key)
+x = gi.tools.install_dependencies(self.tool_id)
+print(f"Called install_dependencies on {self.tool_id} - got {x}")
+def update_toolconf(self,ourdir,ourxml): # path is relative to tools
+updated = False
+localconf = './local_tool_conf.xml'
+self.run_rsync(self.tool_conf_path,localconf)
+tree = ET.parse(localconf)
+root = tree.getroot()
+hasTF = False
+TFsection = None
+for e in root.findall('section'):
+if e.attrib['name'] == self.our_name:
+hasTF = True
+TFsection = e
+if not hasTF:
+TFsection = ET.Element('section')
+root.insert(0,TFsection) # at the top!
+our_tools = TFsection.findall('tool')
+conf_tools = [x.attrib['file'] for x in our_tools]
+for xml in ourxml:   # may be > 1
+if not xml in conf_tools: # new
+updated = True
+ET.SubElement(TFsection, 'tool', {'file':xml})
+ET.indent(tree)
+newconf = f"{self.tool_id}_conf"
+tree.write(newconf, pretty_print=True)
+self.run_rsync(newconf,self.tool_conf_path)
+if False and self.args.packages and self.args.packages > '':
+self.install_deps()
+class Tool_Factory:
+"""Wrapper for an arbitrary script
+uses galaxyxml
+"""
+def __init__(self, args=None):  # noqa
+"""
+prepare command line cl for running the tool here
+and prepare elements needed for galaxyxml tool generation
+"""
+self.ourcwd = os.getcwd()
+self.collections = []
+if len(args.collection) > 0:
+try:
+self.collections = [
+json.loads(x) for x in args.collection if len(x.strip()) > 1
+]
+except Exception:
+print(
+f"--collections parameter {str(args.collection)} is malformed - should be a dictionary"
+)
+try:
+self.infiles = [
+json.loads(x) for x in args.input_files if len(x.strip()) > 1
+]
+except Exception:
+print(
+f"--input_files parameter {str(args.input_files)} is malformed - should be a dictionary"
+)
+try:
+self.outfiles = [
+json.loads(x) for x in args.output_files if len(x.strip()) > 1
+]
+except Exception:
+print(
+f"--output_files parameter {args.output_files} is malformed - should be a dictionary"
+)
+try:
+self.addpar = [
+json.loads(x) for x in args.additional_parameters if len(x.strip()) > 1
+]
+except Exception:
+print(
+f"--additional_parameters {args.additional_parameters} is malformed - should be a dictionary"
+)
+try:
+self.selpar = [
+json.loads(x) for x in args.selecttext_parameters if len(x.strip()) > 1
+]
+except Exception:
+print(
+f"--selecttext_parameters {args.selecttext_parameters} is malformed - should be a dictionary"
+)
+self.args = args
+self.cleanuppar()
+self.lastxclredirect = None
+self.xmlcl = []
+self.is_positional = self.args.parampass == "positional"
+if self.args.sysexe:
+if ' ' in self.args.sysexe:
+self.executeme = self.args.sysexe.split(' ')
+else:
+self.executeme = [self.args.sysexe, ]
+else:
+if self.args.packages:
+self.executeme = [self.args.packages.split(",")[0].split(":")[0].strip(), ]
+else:
+self.executeme = None
+aXCL = self.xmlcl.append
+assert args.parampass in [
+"0",
+"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.newtool = gxt.Tool(
+self.tool_name,
+self.tool_id,
+self.args.tool_version,
+self.args.tool_desc,
+FAKEEXE,
+)
+self.newtarpath = "%s_toolshed.gz" % self.tool_name
+self.tooloutdir = "./tfout"
+self.repdir = "./TF_run_report"
+self.testdir = os.path.join(self.tooloutdir, "test-data")
+if not os.path.exists(self.tooloutdir):
+os.mkdir(self.tooloutdir)
+if not os.path.exists(self.testdir):
+os.mkdir(self.testdir)
+if not os.path.exists(self.repdir):
+os.mkdir(self.repdir)
+self.tinputs = gxtp.Inputs()
+self.toutputs = gxtp.Outputs()
+self.testparam = []
+if self.args.script_path:
+self.prepScript()
+if self.args.command_override:
+scos = open(self.args.command_override, "r").readlines()
+self.command_override = [x.rstrip() for x in scos]
+else:
+self.command_override = None
+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.script_path:
+for ex in self.executeme:
+aXCL(ex)
+aXCL("$runme")
+else:
+for ex in self.executeme:
+aXCL(ex)
+if self.args.parampass == "0":
+self.clsimple()
+else:
+if self.args.parampass == "positional":
+self.prepclpos()
+self.clpositional()
+else:
+self.prepargp()
+self.clargparse()
+def clsimple(self):
+"""no parameters or repeats - uses < and > for i/o"""
+aXCL = self.xmlcl.append
+if len(self.infiles) > 0:
+aXCL("<")
+aXCL("$%s" % self.infiles[0]["infilename"])
+if len(self.outfiles) > 0:
+aXCL(">")
+aXCL("$%s" % self.outfiles[0]["name"])
+if self.args.cl_user_suffix:  # DIY CL end
+clp = shlex.split(self.args.cl_user_suffix)
+for c in clp:
+aXCL(c)
+def prepargp(self):
+xclsuffix = []
+for i, p in enumerate(self.infiles):
+nam = p["infilename"]
+if p["origCL"].strip().upper() == "STDIN":
+xappendme = [
+nam,
+nam,
+"< $%s" % nam,
+]
+else:
+rep = p["repeat"] == "1"
+over = ""
+if rep:
+over = f'#for $rep in $R_{nam}:\n--{nam} "$rep.{nam}"\n#end for'
+xappendme = [p["CL"], "$%s" % p["CL"], over]
+xclsuffix.append(xappendme)
+for i, p in enumerate(self.outfiles):
+if p["origCL"].strip().upper() == "STDOUT":
+self.lastxclredirect = [">", "$%s" % p["name"]]
+else:
+xclsuffix.append([p["name"], "$%s" % p["name"], ""])
+for p in self.addpar:
+nam = p["name"]
+rep = p["repeat"] == "1"
+if rep:
+over = f'#for $rep in $R_{nam}:\n--{nam} "$rep.{nam}"\n#end for'
+else:
+over = p["override"]
+xclsuffix.append([p["CL"], '"$%s"' % nam, over])
+for p in self.selpar:
+xclsuffix.append([p["CL"], '"$%s"' % p["name"], p["override"]])
+self.xclsuffix = xclsuffix
+def prepclpos(self):
+xclsuffix = []
+for i, p in enumerate(self.infiles):
+if p["origCL"].strip().upper() == "STDIN":
+xappendme = [
+"999",
+p["infilename"],
+"< $%s" % p["infilename"],
+]
+else:
+xappendme = [p["CL"], "$%s" % p["infilename"], ""]
+xclsuffix.append(xappendme)
+for i, p in enumerate(self.outfiles):
+if p["origCL"].strip().upper() == "STDOUT":
+self.lastxclredirect = [">", "$%s" % p["name"]]
+else:
+xclsuffix.append([p["CL"], "$%s" % p["name"], ""])
+for p in self.addpar:
+nam = p["name"]
+rep = p["repeat"] == "1"  # repeats make NO sense
+if rep:
+print(f'### warning. Repeats for {nam} ignored - not permitted in positional parameter command lines!')
+over = p["override"]
+xclsuffix.append([p["CL"], '"$%s"' % nam, over])
+for p in self.selpar:
+xclsuffix.append([p["CL"], '"$%s"' % p["name"], p["override"]])
+xclsuffix.sort()
+self.xclsuffix = xclsuffix
+def prepScript(self):
+rx = open(self.args.script_path, "r").readlines()
+rx = [x.rstrip() for x in rx]
+rxcheck = [x.strip() for x in rx if x.strip() > ""]
+assert len(rxcheck) > 0, "Supplied script is empty. Cannot run"
+self.script = "\n".join(rx)
+fhandle, self.sfile = tempfile.mkstemp(
+prefix=self.tool_name, suffix="_%s" % (self.executeme[0])
+)
+tscript = open(self.sfile, "w")
+tscript.write(self.script)
+tscript.close()
+self.spacedScript = [f"    {x}" for x in rx if x.strip() > ""]
+rx.insert(0,'#raw')
+rx.append('#end raw')
+self.escapedScript = rx
+art = "%s.%s" % (self.tool_name, self.executeme[0])
+artifact = open(art, "wb")
+artifact.write(bytes(self.script, "utf8"))
+artifact.close()
+def cleanuppar(self):
+""" positional parameters are complicated by their numeric ordinal"""
+if self.args.parampass == "positional":
+for i, p in enumerate(self.infiles):
+assert (
+p["CL"].isdigit() or p["CL"].strip().upper() == "STDIN"
+), "Positional parameters must be ordinal integers - got %s for %s" % (
+p["CL"],
+p["label"],
+)
+for i, p in enumerate(self.outfiles):
+assert (
+p["CL"].isdigit() or p["CL"].strip().upper() == "STDOUT"
+), "Positional parameters must be ordinal integers - got %s for %s" % (
+p["CL"],
+p["name"],
+)
+for i, p in enumerate(self.addpar):
+assert p[
+"CL"
+].isdigit(), "Positional parameters must be ordinal integers - got %s for %s" % (
+p["CL"],
+p["name"],
+)
+for i, p in enumerate(self.infiles):
+infp = copy.copy(p)
+infp["origCL"] = infp["CL"]
+if self.args.parampass in ["positional", "0"]:
+infp["infilename"] = infp["label"].replace(" ", "_")
+else:
+infp["infilename"] = infp["CL"]
+self.infiles[i] = infp
+for i, p in enumerate(self.outfiles):
+p["origCL"] = p["CL"]  # keep copy
+self.outfiles[i] = p
+for i, p in enumerate(self.addpar):
+p["origCL"] = p["CL"]
+self.addpar[i] = p
+def clpositional(self):
+# inputs in order then params
+aXCL = self.xmlcl.append
+for (k, v, koverride) in self.xclsuffix:
+aXCL(v)
+if self.lastxclredirect:
+aXCL(self.lastxclredirect[0])
+aXCL(self.lastxclredirect[1])
+if self.args.cl_user_suffix:  # DIY CL end
+clp = shlex.split(self.args.cl_user_suffix)
+for c in clp:
+aXCL(c)
+def clargparse(self):
+"""argparse style"""
+aXCL = self.xmlcl.append
+# inputs then params in argparse named form
+for (k, v, koverride) in self.xclsuffix:
+if koverride > "":
+k = koverride
+aXCL(k)
+else:
+if len(k.strip()) == 1:
+k = "-%s" % k
+else:
+k = "--%s" % k
+aXCL(k)
+aXCL(v)
+if self.lastxclredirect:
+aXCL(self.lastxclredirect[0])
+aXCL(self.lastxclredirect[1])
+if self.args.cl_user_suffix:  # DIY CL end
+clp = shlex.split(self.args.cl_user_suffix)
+for c in clp:
+aXCL(c)
+def getNdash(self, newname):
+if self.is_positional:
+ndash = 0
+else:
+ndash = 2
+if len(newname) < 2:
+ndash = 1
+return ndash
+def doXMLparam(self):  # noqa
+"""Add all needed elements to tool"""
+for p in self.outfiles:
+newname = p["name"]
+newfmt = p["format"]
+newcl = p["CL"]
+test = p["test"]
+oldcl = p["origCL"]
+test = test.strip()
+ndash = self.getNdash(newcl)
+aparm = gxtp.OutputData(
+name=newname, format=newfmt, num_dashes=ndash, label=newname
+)
+aparm.positional = self.is_positional
+if self.is_positional:
+if oldcl.upper() == "STDOUT":
+aparm.positional = 9999999
+aparm.command_line_override = "> $%s" % newname
+else:
+aparm.positional = int(oldcl)
+aparm.command_line_override = "$%s" % newname
+self.toutputs.append(aparm)
+ld = None
+if test.strip() > "":
+if test.startswith("diff"):
+c = "diff"
+ld = 0
+if test.split(":")[1].isdigit:
+ld = int(test.split(":")[1])
+tp = gxtp.TestOutput(
+name=newname,
+value="%s_sample" % newname,
+compare=c,
+lines_diff=ld,
+)
+elif test.startswith("sim_size"):
+c = "sim_size"
+tn = test.split(":")[1].strip()
+if tn > "":
+if "." in tn:
+delta = None
+delta_frac = min(1.0, float(tn))
+else:
+delta = int(tn)
+delta_frac = None
+tp = gxtp.TestOutput(
+name=newname,
+value="%s_sample" % newname,
+compare=c,
+delta=delta,
+delta_frac=delta_frac,
+)
+else:
+c = test
+tp = gxtp.TestOutput(
+name=newname,
+value="%s_sample" % newname,
+compare=c,
+)
+self.testparam.append(tp)
+for p in self.infiles:
+newname = p["infilename"]
+newfmt = p["format"]
+ndash = self.getNdash(newname)
+reps = p.get("repeat", "0") == "1"
+if not len(p["label"]) > 0:
+alab = p["CL"]
+else:
+alab = p["label"]
+aninput = gxtp.DataParam(
+newname,
+optional=False,
+label=alab,
+help=p["help"],
+format=newfmt,
+multiple=False,
+num_dashes=ndash,
+)
+aninput.positional = self.is_positional
+if self.is_positional:
+if p["origCL"].upper() == "STDIN":
+aninput.positional = 9999998
+aninput.command_line_override = "> $%s" % newname
+else:
+aninput.positional = int(p["origCL"])
+aninput.command_line_override = "$%s" % newname
+if reps:
+repe = gxtp.Repeat(name=f"R_{newname}", title=f"Add as many {alab} as needed")
+repe.append(aninput)
+self.tinputs.append(repe)
+tparm = gxtp.TestRepeat(name=f"R_{newname}")
+tparm2 = gxtp.TestParam(newname, value="%s_sample" % newname)
+tparm.append(tparm2)
+self.testparam.append(tparm)
+else:
+self.tinputs.append(aninput)
+tparm = gxtp.TestParam(newname, value="%s_sample" % newname)
+self.testparam.append(tparm)
+for p in self.addpar:
+newname = p["name"]
+newval = p["value"]
+newlabel = p["label"]
+newhelp = p["help"]
+newtype = p["type"]
+newcl = p["CL"]
+oldcl = p["origCL"]
+reps = p["repeat"] == "1"
+if not len(newlabel) > 0:
+newlabel = newname
+ndash = self.getNdash(newname)
+if newtype == "text":
+aparm = gxtp.TextParam(
+newname,
+label=newlabel,
+help=newhelp,
+value=newval,
+num_dashes=ndash,
+)
+elif newtype == "integer":
+aparm = gxtp.IntegerParam(
+newname,
+label=newlabel,
+help=newhelp,
+value=newval,
+num_dashes=ndash,
+)
+elif newtype == "float":
+aparm = gxtp.FloatParam(
+newname,
+label=newlabel,
+help=newhelp,
+value=newval,
+num_dashes=ndash,
+)
+elif newtype == "boolean":
+aparm = gxtp.BooleanParam(
+newname,
+label=newlabel,
+help=newhelp,
+value=newval,
+num_dashes=ndash,
+)
+else:
+raise ValueError(
+'Unrecognised parameter type "%s" for\
+additional parameter %s in makeXML'
+% (newtype, newname)
+)
+aparm.positional = self.is_positional
+if self.is_positional:
+aparm.positional = int(oldcl)
+if reps:
+repe = gxtp.Repeat(name=f"R_{newname}", title=f"Add as many {newlabel} as needed")
+repe.append(aparm)
+self.tinputs.append(repe)
+tparm = gxtp.TestRepeat(name=f"R_{newname}")
+tparm2 = gxtp.TestParam(newname, value=newval)
+tparm.append(tparm2)
+self.testparam.append(tparm)
+else:
+self.tinputs.append(aparm)
+tparm = gxtp.TestParam(newname, value=newval)
+self.testparam.append(tparm)
+for p in self.selpar:
+newname = p["name"]
+newval = p["value"]
+newlabel = p["label"]
+newhelp = p["help"]
+newtype = p["type"]
+newcl = p["CL"]
+if not len(newlabel) > 0:
+newlabel = newname
+ndash = self.getNdash(newname)
+if newtype == "selecttext":
+newtext = p["texts"]
+aparm = gxtp.SelectParam(
+newname,
+label=newlabel,
+help=newhelp,
+num_dashes=ndash,
+)
+for i in range(len(newval)):
+anopt = gxtp.SelectOption(
+value=newval[i],
+text=newtext[i],
+)
+aparm.append(anopt)
+aparm.positional = self.is_positional
+if self.is_positional:
+aparm.positional = int(newcl)
+self.tinputs.append(aparm)
+tparm = gxtp.TestParam(newname, value=newval)
+self.testparam.append(tparm)
+else:
+raise ValueError(
+'Unrecognised parameter type "%s" for\
+selecttext parameter %s in makeXML'
+% (newtype, newname)
+)
+for p in self.collections:
+newkind = p["kind"]
+newname = p["name"]
+newlabel = p["label"]
+newdisc = p["discover"]
+collect = gxtp.OutputCollection(newname, label=newlabel, type=newkind)
+disc = gxtp.DiscoverDatasets(
+pattern=newdisc, directory=f"{newname}", visible="false"
+)
+collect.append(disc)
+self.toutputs.append(collect)
+try:
+tparm = gxtp.TestOutputCollection(newname)  # broken until PR merged.
+self.testparam.append(tparm)
+except Exception:
+print("#### WARNING: Galaxyxml version does not have the PR merged yet - tests for collections must be over-ridden until then!")
+def doNoXMLparam(self):
+"""filter style package - stdin to stdout"""
+if len(self.infiles) > 0:
+alab = self.infiles[0]["label"]
+if len(alab) == 0:
+alab = self.infiles[0]["infilename"]
+max1s = (
+"Maximum one input if parampass is 0 but multiple input files supplied - %s"
+% str(self.infiles)
+)
+assert len(self.infiles) == 1, max1s
+newname = self.infiles[0]["infilename"]
+aninput = gxtp.DataParam(
+newname,
+optional=False,
+label=alab,
+help=self.infiles[0]["help"],
+format=self.infiles[0]["format"],
+multiple=False,
+num_dashes=0,
+)
+aninput.command_line_override = "< $%s" % newname
+aninput.positional = True
+self.tinputs.append(aninput)
+tp = gxtp.TestParam(name=newname, value="%s_sample" % newname)
+self.testparam.append(tp)
+if len(self.outfiles) > 0:
+newname = self.outfiles[0]["name"]
+newfmt = self.outfiles[0]["format"]
+anout = gxtp.OutputData(newname, format=newfmt, num_dashes=0)
+anout.command_line_override = "> $%s" % newname
+anout.positional = self.is_positional
+self.toutputs.append(anout)
+tp = gxtp.TestOutput(name=newname, value="%s_sample" % newname)
+self.testparam.append(tp)
+def makeXML(self):  # noqa
+"""
+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
+cite = gxtp.Citations()
+acite = gxtp.Citation(type="doi", value="10.1093/bioinformatics/bts573")
+cite.append(acite)
+self.newtool.citations = cite
+safertext = ""
+if self.args.help_text:
+helptext = open(self.args.help_text, "r").readlines()
+safertext = "\n".join([cheetah_escape(x) for x in helptext])
+if len(safertext.strip()) == 0:
+safertext = (
+"Ask the tool author (%s) to rebuild with help text please\n"
+% (self.args.user_email)
+)
+if self.args.script_path:
+if len(safertext) > 0:
+safertext = safertext + "\n\n------\n"  # transition allowed!
+scr = [x for x in self.spacedScript if x.strip() > ""]
+scr.insert(0, "\n\nScript::\n")
+if len(scr) > 300:
+scr = (
+scr[:100]
++ ["    >300 lines - stuff deleted", "    ......"]
++ scr[-100:]
+)
+scr.append("\n")
+safertext = safertext + "\n".join(scr)
+self.newtool.help = safertext
+self.newtool.version_command = f'echo "{self.args.tool_version}"'
+std = gxtp.Stdios()
+std1 = gxtp.Stdio()
+std.append(std1)
+self.newtool.stdios = std
+requirements = gxtp.Requirements()
+if self.args.packages:
+try:
+for d in self.args.packages.split(","):
+ver = ""
+d = d.replace("==", ":")
+d = d.replace("=", ":")
+if ":" in d:
+packg, ver = d.split(":")
+else:
+packg = d
+requirements.append(
+gxtp.Requirement("package", packg.strip(), ver.strip())
+)
+except Exception:
+print('### malformed packages string supplied - cannot parse =',self.args.packages)
+sys.exit(2)
+self.newtool.requirements = requirements
+if self.args.parampass == "0":
+self.doNoXMLparam()
+else:
+self.doXMLparam()
+self.newtool.outputs = self.toutputs
+self.newtool.inputs = self.tinputs
+if self.args.script_path:
+configfiles = gxtp.Configfiles()
+configfiles.append(
+gxtp.Configfile(name="runme", text="\n".join(self.escapedScript))
+)
+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.newtool.tests = tests
+self.newtool.add_comment(
+"Created by %s at %s using the Galaxy Tool Factory."
+% (self.args.user_email, timenow())
+)
+self.newtool.add_comment("Source in git at: %s" % (toolFactoryURL))
+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
+part1 = exml.split("<tests>")[0]
+part2 = exml.split("</tests>")[1]
+fixed = "%s\n%s\n%s" % (part1, "\n".join(self.test_override), part2)
+exml = fixed
+# exml = exml.replace('range="1:"', 'range="1000:"')
+xf = open("%s.xml" % self.tool_name, "w")
+xf.write(exml)
+xf.write("\n")
+xf.close()
+# ready for the tarball
+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):
+"""write xmls and input samples into place"""
+if self.args.parampass == 0:
+self.doNoXMLparam()
+else:
+self.makeXML()
+if self.args.script_path:
+stname = os.path.join(self.tooloutdir, 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["name"]
+dest = os.path.join(self.testdir, "%s_sample" % p["infilename"])
+shutil.copyfile(pth, dest)
+dest = os.path.join(self.repdir, "%s_sample.%s" % (p["infilename"],p["format"]))
+shutil.copyfile(pth, dest)
+def makeToolTar(self, report_fail=False):
+"""move outputs into test-data and prepare the tarball"""
+excludeme = "_planemo_test_report.html"
+def exclude_function(tarinfo):
+filename = tarinfo.name
+return None if filename.endswith(excludeme) else tarinfo
+for p in self.outfiles:
+oname = p["name"]
+tdest = os.path.join(self.testdir, "%s_sample" % oname)
+src = os.path.join(self.testdir, oname)
+if not os.path.isfile(tdest):
+if os.path.isfile(src):
+shutil.copyfile(src, tdest)
+dest = os.path.join(self.repdir, "%s.sample" % (oname))
+shutil.copyfile(src, dest)
+else:
+if report_fail:
+print(
+"###Tool may have failed - output file %s not found in testdir after planemo run %s."
+% (tdest, self.testdir)
+)
+tf = tarfile.open(self.newtarpath, "w:gz")
+tf.add(
+name=self.tooloutdir,
+arcname=self.tool_name,
+filter=exclude_function,
+)
+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():
+continue
+if not entry.name.endswith('.html'):
+_, ext = os.path.splitext(entry.name)
+newname = f"{entry.name.replace('.','_')}.txt"
+dest = os.path.join(self.repdir, newname)
+src = os.path.join(self.tooloutdir, entry.name)
+shutil.copyfile(src, dest)
+if self.args.include_tests:
+with os.scandir(self.testdir) as outs:
+for entry in outs:
+if (not entry.is_file()) or entry.name.endswith(
+"_planemo_test_report.html"
+):
+continue
+if "." in entry.name:
+_, ext = os.path.splitext(entry.name)
+if ext in [".tgz", ".json"]:
+continue
+if ext in [".yml", ".xml", ".yaml"]:
+newname = f"{entry.name.replace('.','_')}.txt"
+else:
+newname = entry.name
+else:
+newname = f"{entry.name}.txt"
+dest = os.path.join(self.repdir, newname)
+src = os.path.join(self.testdir, entry.name)
+shutil.copyfile(src, dest)
+def main():
+"""
+This is a Galaxy wrapper.
+It expects to be called by a special purpose tool.xml
+"""
+parser = argparse.ArgumentParser()
+a = parser.add_argument
+a("--script_path", default=None)
+a("--history_test", default=None)
+a("--cl_user_suffix", default=None)
+a("--sysexe", default=None)
+a("--packages", default=None)
+a("--tool_name", default="newtool")
+a("--tool_dir", default=None)
+a("--input_files", default=[], action="append")
+a("--output_files", default=[], action="append")
+a("--user_email", default="Unknown")
+a("--bad_user", default=None)
+a("--help_text", default=None)
+a("--tool_desc", default=None)
+a("--tool_version", default=None)
+a("--citations", default=None)
+a("--command_override", default=None)
+a("--test_override", default=None)
+a("--additional_parameters", action="append", default=[])
+a("--selecttext_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_root", default="/galaxy-central")
+a("--galaxy_venv", default="/galaxy_venv")
+a("--collection", action="append", default=[])
+a("--include_tests", default=False, action="store_true")
+a("--admin_only", default=False, action="store_true")
+a("--install", default=False, action="store_true")
+a("--run_test", default=False, action="store_true")
+a("--local_tools", default='tools') # relative to $__root_dir__
+a("--tool_conf_path", default='config/tool_conf.xml') # relative to $__root_dir__
+a("--galaxy_url", default="http://localhost:8080")
+a("--toolshed_url", default="http://localhost:9009")
+# make sure this is identical to tool_sheds_conf.xml
+# localhost != 127.0.0.1 so validation fails
+a("--toolshed_api_key", default="fakekey")
+a("--galaxy_api_key", default="8993d65865e6d6d1773c2c34a1cc207d")
+args = parser.parse_args()
+if args.admin_only:
+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.yml Galaxy configuration file'
+% (args.bad_user, args.bad_user)
+)
+assert args.tool_name, "## Tool Factory expects a tool name - eg --tool_name=DESeq"
+r = Tool_Factory(args)
+r.writeShedyml()
+r.makeTool()
+r.makeToolTar()
+if args.install:
+#try:
+tcu = Tool_Conf_Updater(args=args, tool_dir=args.local_tools,
+new_tool_archive_path=r.newtarpath, tool_conf_path=args.tool_conf_path,
+new_tool_name=r.tool_name)
+#except Exception:
+#   print("### Unable to install the new tool. Are you sure you have all the required special settings?")
+if __name__ == "__main__":
+main()

Mercurial > repos > fubar > tool_factory_2

comparison toolfactory/ToolFactory.py @ 3:1c652687a08f draft default tip