Inline expressions

README.rst

@@ -0,0 +1,55 @@
+==================================================================
+Common workflow language tool description reference implementation
+==================================================================
+
+This is the reference implementation of the Common Workflow Language.  It is
+intended to be feature complete and provide comprehensive validation of CWL
+files as well as provide other tools related to working with CWL.
+
+This is written and tested for Python 2.7.
+
+The reference implementation consists of two packages.  The "cwltool" package
+is the primary Python module containing the reference implementation in the
+"cwltool" module and console executable by the same name.
+
+The "cwl-runner" package is optional and provides an additional entry point
+under the alias "cwl-runner", which is the implementation-agnostic name for the
+default CWL interpreter installed on a host.
+
+Install
+-------
+
+Installing the official package from PyPi (will install "cwltool" package as well)::
+
+  pip install cwl-runner
+
+Or from source::
+
+  git clone https://github.com/common-workflow-language/cwltool.git
+  cd cwltool && python setup.py install
+  cd cwl-runner && python setup.py install
+
+Run on the command line
+-----------------------
+
+  ``cwl-runner [tool] [job]``
+
+Import as a module
+----------------
+
+Add::
+
+  import cwltool
+
+to your script.
+
+Use with boot2docker
+--------------------
+boot2docker is running docker inside a virtual machine and it only mounts ``Users``
+on it. The default behavoir of CWL is to create temporary directories under e.g.
+``/Var`` which is not accessible to Docker containers.
+
+To run CWL successfully with boot2docker you need to set the ``--tmpdir-prefix``
+and ``--tmp-outdir-prefix`` to somewhere under ``/Users``::
+
+    $ cwl-runner --tmp-outdir-prefix=/Users/username/project --tmpdir-prefix=/Users/username/project wc-tool.cwl wc-job.json

build-cwl-docker.sh

@@ -0,0 +1,14 @@
+#!/bin/sh
+
+restore=0
+if test -L cwltool/schemas ; then
+  rm cwltool/schemas
+  cp -r ../schemas cwltool/schemas
+  restore=1
+fi
+docker build --file=cwltool_module.Dockerfile --tag=commonworkflowlanguage/cwltool_module .
+docker build --file=cwltool.Dockerfile --tag=commonworkflowlanguage/cwltool .
+if test $restore = 1 ; then
+  rm -r cwltool/schemas
+  ln -s ../../schemas cwltool/schemas
+fi

build-node-docker.sh

@@ -0,0 +1,3 @@
+#!/bin/sh
+
+docker build --file=node-expr-engine.Dockerfile --tag=commonworkflowlanguage/nodejs-engine .

cwl-docker.sh

@@ -0,0 +1,3 @@
+#!/bin/sh
+docker run --name=cwl-docker -v /var/lib/docker -i -t fedora-data true
+docker run --privileged -ti --volume=$PWD:$PWD -w=$PWD commonworkflowlanguage/cwltool $*

cwl-runner/README

@@ -0,0 +1,4 @@
+This an optional companion package to "cwltool" which provides provides an
+additional entry point under the alias "cwl-runner", which is the
+implementation-agnostic name for the default CWL interpreter installed on a
+host.

cwl-runner/setup.py

@@ -0,0 +1,29 @@
+#!/usr/bin/env python
+
+import os
+import sys
+import setuptools.command.egg_info as egg_info_cmd
+import shutil
+
+from setuptools import setup, find_packages
+
+SETUP_DIR = os.path.dirname(__file__)
+README = os.path.join(SETUP_DIR, 'README')
+
+setup(name='cwl_runner',
+      version='1.0',
+      description='Common workflow language reference implementation',
+      long_description=open(README).read(),
+      author='Common workflow language working group',
+      author_email='common-workflow-language@googlegroups.com',
+      url="https://github.com/common-workflow-language/common-workflow-language",
+      download_url="https://github.com/common-workflow-language/common-workflow-language",
+      license='Apache 2.0',
+      install_requires=[
+          'cwltool'
+        ],
+      entry_points={
+          'console_scripts': [ "cwl-runner=cwltool.main:main" ]
+      },
+      zip_safe=True
+)

cwltool.Dockerfile

@@ -0,0 +1,5 @@
+FROM commonworkflowlanguage/cwltool_module
+MAINTAINER peter.amstutz@curoverse.com
+
+VOLUME /var/lib/docker
+ENTRYPOINT ["wrapdocker", "cwltool"]

cwltool/__init__.py

@@ -0,0 +1 @@
+__author__ = 'peter.amstutz@curoverse.com'

cwltool/__main__.py

@@ -0,0 +1,4 @@
+import main
+import sys
+
+sys.exit(main.main())

cwltool/aslist.py

@@ -0,0 +1,5 @@
+def aslist(l):
+    if isinstance(l, list):
+        return l
+    else:
+        return [l]

cwltool/builder.py

@@ -0,0 +1,159 @@
+import copy
+from aslist import aslist
+import expression
+import avro
+import schema_salad.validate as validate
+
+CONTENT_LIMIT = 64 * 1024
+
+def substitute(value, replace):
+    if replace[0] == "^":
+        return substitute(value[0:value.rindex('.')], replace[1:])
+    else:
+        return value + replace
+
+class Builder(object):
+
+    def bind_input(self, schema, datum, lead_pos=[], tail_pos=[]):
+        bindings = []
+        binding = None
+        if "inputBinding" in schema and isinstance(schema["inputBinding"], dict):
+            binding = copy.copy(schema["inputBinding"])
+
+            if "position" in binding:
+                binding["position"] = aslist(lead_pos) + aslist(binding["position"]) + aslist(tail_pos)
+            else:
+                binding["position"] = aslist(lead_pos) + [0] + aslist(tail_pos)
+
+            if "valueFrom" in binding:
+                binding["do_eval"] = binding["valueFrom"]
+            binding["valueFrom"] = datum
+
+        # Handle union types
+        if isinstance(schema["type"], list):
+            for t in schema["type"]:
+                if isinstance(t, basestring) and self.names.has_name(t, ""):
+                    avsc = self.names.get_name(t, "")
+                elif isinstance(t, dict) and "name" in t and self.names.has_name(t["name"], ""):
+                    avsc = self.names.get_name(t["name"], "")
+                else:
+                    avsc = avro.schema.make_avsc_object(t, self.names)
+                if validate.validate(avsc, datum):
+                    schema = copy.deepcopy(schema)
+                    schema["type"] = t
+                    return self.bind_input(schema, datum, lead_pos=lead_pos, tail_pos=tail_pos)
+            raise validate.ValidationException("'%s' is not a valid union %s" % (datum, schema["type"]))
+        elif isinstance(schema["type"], dict):
+            st = copy.deepcopy(schema["type"])
+            if binding and "inputBinding" not in st and "itemSeparator" not in binding and st["type"] in ("array", "map"):
+                st["inputBinding"] = {}
+            bindings.extend(self.bind_input(st, datum, lead_pos=lead_pos, tail_pos=tail_pos))
+        else:
+            if schema["type"] in self.schemaDefs:
+                schema = self.schemaDefs[schema["type"]]
+
+            if schema["type"] == "record":
+                for f in schema["fields"]:
+                    if f["name"] in datum:
+                        bindings.extend(self.bind_input(f, datum[f["name"]], lead_pos=lead_pos, tail_pos=f["name"]))
+                    else:
+                        datum[f["name"]] = f.get("default")
+
+            if schema["type"] == "map":
+                for n, item in datum.items():
+                    b2 = None
+                    if binding:
+                        b2 = copy.deepcopy(binding)
+                        b2["valueFrom"] = [n, item]
+                    bindings.extend(self.bind_input({"type": schema["values"], "inputBinding": b2},
+                                                    item, lead_pos=n, tail_pos=tail_pos))
+                binding = None
+
+            if schema["type"] == "array":
+                for n, item in enumerate(datum):
+                    b2 = None
+                    if binding:
+                        b2 = copy.deepcopy(binding)
+                        b2["valueFrom"] = item
+                    bindings.extend(self.bind_input({"type": schema["items"], "inputBinding": b2},
+                                                    item, lead_pos=n, tail_pos=tail_pos))
+                binding = None
+
+            if schema["type"] == "File":
+                self.files.append(datum)
+                if binding:
+                    if binding.get("loadContents"):
+                        with self.fs_access.open(datum["path"], "rb") as f:
+                            datum["contents"] = f.read(CONTENT_LIMIT)
+
+                    if "secondaryFiles" in binding:
+                        if "secondaryFiles" not in datum:
+                            datum["secondaryFiles"] = []
+                        for sf in aslist(binding["secondaryFiles"]):
+                            if isinstance(sf, dict):
+                                sfpath = self.do_eval(sf, context=datum["path"])
+                            else:
+                                sfpath = {"path": substitute(datum["path"], sf), "class": "File"}
+                            if isinstance(sfpath, list):
+                                datum["secondaryFiles"].extend(sfpath)
+                                self.files.extend(sfpath)
+                            else:
+                                datum["secondaryFiles"].append(sfpath)
+                                self.files.append(sfpath)
+
+        # Position to front of the sort key
+        if binding:
+            for bi in bindings:
+                bi["position"] = binding["position"] + bi["position"]
+            bindings.append(binding)
+
+        return bindings
+
+    def tostr(self, value):
+        if isinstance(value, dict) and value.get("class") == "File":
+            if "path" not in value:
+                raise WorkflowException("File object must have \"path\": %s" % (value))
+            return value["path"]
+        else:
+            return str(value)
+
+    def generate_arg(self, binding):
+        value = binding["valueFrom"]
+        if "do_eval" in binding:
+            value = self.do_eval(binding["do_eval"], context=value)
+
+        prefix = binding.get("prefix")
+        sep = binding.get("separate", True)
+
+        l = []
+        if isinstance(value, list):
+            if binding.get("itemSeparator"):
+                l = [binding["itemSeparator"].join([self.tostr(v) for v in value])]
+            elif binding.get("do_eval"):
+                return ([prefix] if prefix else []) + value
+            elif prefix:
+                return [prefix]
+            else:
+                return []
+        elif isinstance(value, dict) and value.get("class") == "File":
+            l = [value]
+        elif isinstance(value, dict):
+            return [prefix] if prefix else []
+        elif value is True and prefix:
+            return [prefix]
+        elif value is False or value is None:
+            return []
+        else:
+            l = [value]
+
+        args = []
+        for j in l:
+            if sep:
+                args.extend([prefix, self.tostr(j)])
+            else:
+                args.append(prefix + self.tostr(j))
+
+        return [a for a in args if a is not None]
+
+    def do_eval(self, ex, context=None, pull_image=True):
+        return expression.do_eval(ex, self.job, self.requirements, self.outdir, self.tmpdir, context=context, pull_image=pull_image)

cwltool/cwlrdf.py

@@ -0,0 +1,80 @@
+import json
+from rdflib import Graph, plugin
+from rdflib.serializer import Serializer
+
+def printrdf(workflow, wf, ctx, sr):
+    wf["@context"] = ctx
+    g = Graph().parse(data=json.dumps(wf), format='json-ld', location=workflow)
+    print(g.serialize(format=sr))
+
+def printdot(workflow, wf, ctx, sr):
+    wf["@context"] = ctx
+    g = Graph().parse(data=json.dumps(wf), format='json-ld', location=workflow)
+
+    print "digraph {"
+
+    #g.namespace_manager.qname(predicate)
+
+    def lastpart(uri):
+        uri = str(uri)
+        if "/" in uri:
+            return uri[uri.rindex("/")+1:]
+        else:
+            return uri
+
+    qres = g.query(
+        """SELECT ?step ?run
+           WHERE {
+              ?step cwl:run ?run .
+           }""")
+
+    for step, run in qres:
+        print '"%s" [label="%s"]' % (lastpart(step), "%s (%s)" % (lastpart(step), lastpart(run)))
+
+    qres = g.query(
+        """SELECT ?step ?inp ?source
+           WHERE {
+              ?wf cwl:steps ?step .
+              ?step cwl:inputs ?inp .
+              ?inp cwl:source ?source .
+           }""")
+
+    for step, inp, source in qres:
+        print '"%s" [shape=box]' % (lastpart(inp))
+        print '"%s" -> "%s" [label="%s"]' % (lastpart(source), lastpart(inp), "")
+        print '"%s" -> "%s" [label="%s"]' % (lastpart(inp), lastpart(step), "")
+
+    qres = g.query(
+        """SELECT ?step ?out
+           WHERE {
+              ?wf cwl:steps ?step .
+              ?step cwl:outputs ?out .
+           }""")
+
+    for step, out in qres:
+        print '"%s" [shape=box]' % (lastpart(out))
+        print '"%s" -> "%s" [label="%s"]' % (lastpart(step), lastpart(out), "")
+
+    qres = g.query(
+        """SELECT ?out ?source
+           WHERE {
+              ?wf cwl:outputs ?out .
+              ?out cwl:source ?source .
+           }""")
+
+    for out, source in qres:
+        print '"%s" [shape=octagon]' % (lastpart(out))
+        print '"%s" -> "%s" [label="%s"]' % (lastpart(source), lastpart(out), "")
+
+    qres = g.query(
+        """SELECT ?inp
+           WHERE {
+              ?wf rdf:type cwl:Workflow .
+              ?wf cwl:inputs ?inp .
+           }""")
+
+    for (inp,) in qres:
+        print '"%s" [shape=octagon]' % (lastpart(inp))
+
+
+    print "}"