crunch_scripts/cwl-runner

   1 #!/usr/bin/env python
   2
   3 import arvados
   4 import arvados_cwl
   5 import arvados.collection
   6 import arvados.util
   7 from cwltool.process import shortname
   8 import cwltool.main
   9 import logging
  10 import os
  11 import json
  12 import argparse
  13 from arvados.api import OrderedJsonModel
  14 from cwltool.process import adjustFiles
  15
  16 api = arvados.api("v1")
  17
  18 try:
  19     job_order_object = arvados.current_job()['script_parameters']
  20
  21     print job_order_object
  22
  23     def keeppath(v):
  24         if arvados.util.keep_locator_pattern.match(v):
  25             return "file://%s/%s" % (os.environ['TASK_KEEPMOUNT'], v)
  26
  27     job_order_object["cwl:tool"] = keeppath(job_order_object["cwl:tool"])
  28
  29     adjustFiles(job_order_object, keeppath)
  30
  31     runner = arvados_cwl.ArvCwlRunner(api_client=arvados.api('v1', model=OrderedJsonModel()))
  32
  33     t = cwltool.main.load_tool(job_order_object, False, True, runner.arvMakeTool, True)
  34
  35     np = argparse.Namespace()
  36     np.project_uuid = arvados.current_job()["owner_uuid"]
  37     np.enable_reuse = True
  38     np.submit = False
  39     np.debug = True
  40     outputObj = runner.arvExecutor(t, job_order_object, "", np, cwl_runner_job=arvados.current_job())
  41
  42     files = {}
  43     def capture(path):
  44         sp = path.split("/")
  45         col = sp[0][5:]
  46         if col not in files:
  47             files[col] = set()
  48         files[col].add("/".join(sp[1:]))
  49         return path
  50
  51     adjustFiles(outputObj, capture)
  52
  53     final = arvados.collection.Collection()
  54
  55     for k,v in files.iteritems():
  56         with arvados.collection.Collection(k) as c:
  57             for f in c:
  58                 final.copy(f, f, c, True)
  59
  60     def makeRelative(path):
  61         return "/".join(path.split("/")[1:])
  62
  63     adjustFiles(outputObj, makeRelative)
  64
  65     with final.open("cwl.output.json", "w") as f:
  66         json.dump(outputObj, f, indent=4)
  67
  68     api.job_tasks().update(uuid=arvados.current_task()['uuid'],
  69                                          body={
  70                                              'output': final.save_new(create_collection_record=False),
  71                                              'success': True,
  72                                              'progress':1.0
  73                                          }).execute()
  74 except Exception as e:
  75     logging.exception("Unhandled exception")
  76     api.job_tasks().update(uuid=arvados.current_task()['uuid'],
  77                                          body={
  78                                              'output': None,
  79                                              'success': False,
  80                                              'progress':1.0
  81                                          }).execute()