tool_factory_2: toolfactory/rgToolFactory2.py comparison

comparison toolfactory/rgToolFactory2.py @ 63:b3254219d4fb draft

Uploaded

author	fubar
date	Tue, 10 Nov 2020 02:33:46 +0000
parents	881bd8c81ac9
children	ed1ecd161295

comparison

equal deleted inserted replaced

-:881bd8c81ac9
+:b3254219d4fb
 # 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 copy
 import logging
 import os
 import re
 import shutil
 import subprocess
 import sys
 import tarfile
 import tempfile
 import time
+from bioblend import galaxy
+from bioblend import toolshed
 import galaxyxml.tool as gxt
 import galaxyxml.tool.parameters as gxtp
 import lxml
 self.lastclredirect = None
 self.lastxclredirect = None
 self.cl = []
 self.xmlcl = []
 self.is_positional = self.args.parampass == "positional"
-if self.args.packages:
+if self.args.sysexe:
-self.executeme = self.args.packages.split(",")[0].split(":")[0]
-else:
 self.executeme = self.args.sysexe
-assert (
+else:
-self.executeme is not None
+if self.args.packages:
-), "No system or managed executable passed in. Cannot build"
+self.executeme = self.args.packages.split(",")[0].split(":")[0]
+else:
+self.executeme = None
 aCL = self.cl.append
 aXCL = self.xmlcl.append
 assert args.parampass in [
 "0",
 "argparse",
 additional parameter %s in makeXML'
 % (newtype, newname)
 )
 aparm.positional = self.is_positional
 if self.is_positional:
-aninput.positional = int(oldcl)
+aparm.positional = int(oldcl)
 self.tinputs.append(aparm)
-self.tparm = gxtp.TestParam(newname, value=newval)
+tparm = gxtp.TestParam(newname, value=newval)
 self.testparam.append(tparm)
 def doNoXMLparam(self):
 """filter style package - stdin to stdout"""
 if len(self.infiles) > 0:
 aninput.command_line_override = "< $%s" % newname
 aninput.positional = self.is_positional
 self.tinputs.append(aninput)
 tp = gxtp.TestParam(name=newname, value="%s_sample" % newname)
 self.testparam.append(tp)
-if len(self.outfiles > 0):
+if len(self.outfiles) > 0:
 newname = self.outfiles[0][OCLPOS]
 newfmt = self.outfiles[0][OFMTPOS]
 anout = gxtp.OutputData(newname, format=newfmt, num_dashes=0)
 anout.command_line_override = "> $%s" % newname
 anout.positional = self.is_positional
 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:
+if False and self.args.script_path:
-scrpt = self.script.split('\n')
+scrp = self.script.split('\n')
+scrpt = ['   %s' % x for x in scrp] # try to stop templating
 scrpt.insert(0,"```\n")
 if len(scrpt) > 300:
 safertext = safertext + scrpt[:100] + ['>500 lines - stuff deleted','......'] + scrpt[-100:]
 else:
 safertext = safertext + scrpt
 ):  # 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, 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
 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")
 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
 if len(self.infiles) > 0:
 sti = open(self.infiles[0][IPATHPOS], "rb")
 else:
 sti = sys.stdin
 if len(self.outfiles) > 0:
 sto = open(self.outfiles[0][ONAMEPOS], "wb")
 else:
 sto = sys.stdout
 p = subprocess.run(self.cl, shell=False, stdout=sto, stdin=sti)
+sto.write(
+"## Executing Toolfactory generated command line = %s\n" % scl
+)
 retval = p.returncode
 sto.close()
 sti.close()
 if os.path.isfile(self.tlog) and os.stat(self.tlog).st_size == 0:
 os.unlink(self.tlog)
 if retval != 0 and err:  # problem
 sys.stderr.write(err)
 logging.debug("run done")
 return retval
-def planemo_shedload(self):
-"""
+def shedLoad(self):
-planemo shed_create --shed_target testtoolshed
+"""
-planemo shed_update --check_diff --shed_target testtoolshed
+{'deleted': False,
+'description': 'Tools for manipulating data',
+'id': '175812cd7caaf439',
+'model_class': 'Category',
+'name': 'Text Manipulation',
+'url': '/api/categories/175812cd7caaf439'}]
 """
 if os.path.exists(self.tlog):
-tout = open(self.tlog, "a")
+sto = open(self.tlog, "a")
 else:
-tout = open(self.tlog, "w")
+sto = open(self.tlog, "w")
-cll = ["planemo", "shed_create", "--shed_target", "local"]
-try:
+ts = toolshed.ToolShedInstance(url=self.args.toolshed_url,key=self.args.toolshed_api_key,verify=False)
-p = subprocess.run(
+repos = ts.repositories.get_repositories()
-cll, shell=True, cwd=self.tooloutdir, stdout=tout, stderr=tout
+rnames = [x.get('name','?') for x in repos]
-)
+rids = [x.get('id','?') for x in repos]
-except:
+sto.write(f'############names={rnames} rids={rids}')
-pass
+cat = 'ToolFactory generated tools'
-if p.returncode != 0:
+if self.args.tool_name not in rnames:
-print("Repository %s exists" % self.args.tool_name)
+tscat = ts.categories.get_categories()
-else:
+cnames = [x.get('name','?') for x in tscat]
-print("initiated %s" % self.args.tool_name)
+cids = [x.get('id','?') for x in tscat]
-cll = [
+catID = None
-"planemo",
+if cat in cnames:
-"shed_upload",
+ci = cnames.index(cat)
-"--shed_target",
+catID = cids[ci]
-"local",
+res = ts.repositories.create_repository(name=self.args.tool_name, synopsis='Synopsis:%s' % self.args.tool_desc, description=self.args.tool_desc,
-"--owner",
+type='unrestricted', remote_repository_url=self.args.toolshed_url,
-"fubar",
+homepage_url=None, category_ids=catID)
-"--name",
+tid = res.get('id',None)
-self.args.tool_name,
+sto.write(f'##########create res={res}')
-"--shed_key",
+else:
-self.args.toolshed_api_key,
+i = rnames.index(self.args.tool_name)
-"--tar",
+tid = rids[i]
-self.newtarpath,
+res = ts.repositories.update_repository(id=tid, tar_ball_path=self.newtarpath, commit_message=None)
-]
+sto.write(f'#####update res={res}')
-p = subprocess.run(cll, shell=False)
+sto.close()
-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 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")
-else:
-tout = open(self.tlog, "w")
-if genoutputs:
-cll = [
-"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():
-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):
 self.args.tool_name,
 "--owner",
 "fubar",
 "--toolshed",
 self.args.toolshed_url,
 "--section_label",
-"Generated Tools",
+"ToolFactory",
 ]
-print("running\n", " ".join(cll))
+tout.write("running\n%s\n" % " ".join(cll))
 p = subprocess.run(cll, shell=False, stderr=tout, stdout=tout)
-if p.returncode != 0:
+tout.write("installed %s - got retcode %d\n" % (self.args.tool_name,p.returncode))
-print(
-"Repository %s installation returned %d"
-% (self.args.tool_name, p.returncode)
-)
-else:
-print("installed %s" % self.args.tool_name)
 tout.close()
 return p.returncode
+def planemo_shedload(self):
+"""
+planemo shed_create --shed_target testtoolshed
+planemo shed_init --name=<name>
+--owner=<shed_username>
+--description=<short description>
+[--remote_repository_url=<URL to .shed.yml on github>]
+[--homepage_url=<Homepage for tool.>]
+[--long_description=<long description>]
+[--category=<category name>]*
+planemo shed_update --check_diff --shed_target testtoolshed
+"""
+if os.path.exists(self.tlog):
+tout = open(self.tlog, "a")
+else:
+tout = open(self.tlog, "w")
+ts = toolshed.ToolShedInstance(url=self.args.toolshed_url,key=self.args.toolshed_api_key,verify=False)
+repos = ts.repositories.get_repositories()
+rnames = [x.get('name','?') for x in repos]
+rids = [x.get('id','?') for x in repos]
+tout.write(f'############names={rnames} rids={rids}')
+cat = 'ToolFactory generated tools'
+if self.args.tool_name not in rnames:
+cll = ["planemo", "shed_create", "--shed_target", "local",
+"--owner","fubar","--name",
+self.args.tool_name,"--shed_key",
+self.args.toolshed_api_key,]
+try:
+p = subprocess.run(
+cll, shell=False, cwd=self.tooloutdir, stdout=tout, stderr=tout
+)
+except:
+pass
+if p.returncode != 0:
+tout.write("Repository %s exists" % self.args.tool_name)
+else:
+tout.write("initiated %s" % self.args.tool_name)
+cll = [
+"planemo",
+"shed_upload",
+"--shed_target",
+"local",
+"--owner",
+"fubar",
+"--name",
+self.args.tool_name,
+"--shed_key",
+self.args.toolshed_api_key,
+"--tar",
+self.newtarpath,
+]
+p = subprocess.run(cll, shell=False, stdout=tout, stderr=tout)
+tout.write("Ran %s got %d" %  (" ".join(cll), p.returncode))
+tout.close()
+return p.returncode
+def eph_test(self):
+"""
+"""
+if os.path.exists(self.tlog):
+tout = open(self.tlog, "a")
+else:
+tout = open(self.tlog, "w")
+cll = [
+"shed-tools",
+"test",
+"-g",
+self.args.galaxy_url,
+"-a",
+self.args.galaxy_api_key,
+"--name",
+self.args.tool_name,
+"--owner",
+"fubar",
+]
+p = subprocess.run(
+cll, shell=False, cwd=self.tooloutdir, stderr=tout, stdout=tout
+)
+tout.write("eph_test Ran %s got %d" % (" ".join(cll), p.returncode))
+tout.close()
+return p.returncode
+def planemo_test(self, genoutputs=True):
+"""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")
+else:
+tout = open(self.tlog, "w")
+if genoutputs:
+dummy,tfile = tempfile.mkstemp()
+cll = [
+"planemo",
+"test",
+"--galaxy_root",
+self.args.galaxy_root,
+"--update_test_data",
+xreal,
+]
+p = subprocess.run(
+cll, shell=False, cwd=self.tooloutdir, stderr=dummy, stdout=dummy,
+)
+dummy.close() # throw all the log away as it will be rerun after outputs are generated
+else:
+cll = ["planemo", "test", "--galaxy_root",
+self.args.galaxy_root,
+xreal,]
+p = subprocess.run(
+cll, shell=False, cwd=self.tooloutdir, stderr=tout, stdout=tout
+)
+tout.close()
+return p.returncode
 def writeShedyml(self):
 """for planemo
 """
 yuser = self.args.user_email.split("@")[0]
 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("--toolshed_url", default="http://localhost:9009") # make sure this is NOT 127.0.0.1 - it won't work if tool_sheds_conf.xml has localhost
+#a("--galaxy_api_key", default="1e62ddad74fe9bf112859f4e9efea48b")
+#a("--toolshed_api_key", default="9154c91f2a162bf12fda15764f43846c")
+a("--toolshed_api_key", default="fakekey")
 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")
 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.command_override or args.test_override:
+retcode = r.planemo_test(genoutputs=True)  # this fails :( - see PR
-retcode = r.planemo_test(genoutputs=True)  # this fails :( - see PR
+r.moveRunOutputs()
-r.moveRunOutputs()
+r.makeToolTar()
-r.makeToolTar()
+retcode = r.planemo_test(genoutputs=False)
-retcode = r.planemo_test(genoutputs=False)
+r.moveRunOutputs()
-r.moveRunOutputs()
+if args.make_Tool == "gentestinstall":
-if args.make_Tool == "gentestinstall":
+retcode = r.planemo_shedload() #r.shedLoad()
-r.planemo_shedload()
+print(f'planemo_shedload returned {retcode}')
 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()
-r.makeToolTar()
-if args.make_Tool in ["gentestinstall", "gentest"]:
-r.planemo_test(genoutputs=False)
-r.moveRunOutputs()
-if args.make_Tool == "gentestinstall":
-r.planemo_shedload()
-r.eph_galaxy_load()
 if __name__ == "__main__":
 main()

Mercurial > repos > fubar > tool_factory_2

comparison toolfactory/rgToolFactory2.py @ 63:b3254219d4fb draft