V1.1.0-dev1 support (#694)

Support for v1.1.0-dev1 end of March milestone.

Author: tetron

cwltool/builder.py

@@ -2,7 +2,8 @@
 import copy
 import os
 import logging
-from typing import Any, Callable, Dict, List, Text, Type, Union
+import json
+from typing import Any, Callable, Dict, List, Text, Type, Union, Set
 
 import six
 from six import iteritems, string_types
@@ -11,6 +12,9 @@
 import schema_salad.validate as validate
 from schema_salad.sourceline import SourceLine
 
+from rdflib import Graph, URIRef
+from rdflib.namespace import OWL, RDFS
+
 from . import expression
 from .errors import WorkflowException
 from .mutation import MutationManager
@@ -32,6 +36,52 @@ def substitute(value, replace):  # type: (Text, Text) -> Text
     else:
         return value + replace
 
+def formatSubclassOf(fmt, cls, ontology, visited):
+    # type: (Text, Text, Graph, Set[Text]) -> bool
+    """Determine if `fmt` is a subclass of `cls`."""
+
+    if URIRef(fmt) == URIRef(cls):
+        return True
+
+    if ontology is None:
+        return False
+
+    if fmt in visited:
+        return False
+
+    visited.add(fmt)
+
+    uriRefFmt = URIRef(fmt)
+
+    for s, p, o in ontology.triples((uriRefFmt, RDFS.subClassOf, None)):
+        # Find parent classes of `fmt` and search upward
+        if formatSubclassOf(o, cls, ontology, visited):
+            return True
+
+    for s, p, o in ontology.triples((uriRefFmt, OWL.equivalentClass, None)):
+        # Find equivalent classes of `fmt` and search horizontally
+        if formatSubclassOf(o, cls, ontology, visited):
+            return True
+
+    for s, p, o in ontology.triples((None, OWL.equivalentClass, uriRefFmt)):
+        # Find equivalent classes of `fmt` and search horizontally
+        if formatSubclassOf(s, cls, ontology, visited):
+            return True
+
+    return False
+
+def checkFormat(actualFile, inputFormats, ontology):
+    # type: (Union[Dict[Text, Any], List, Text], Union[List[Text], Text], Graph) -> None
+    for af in aslist(actualFile):
+        if not af:
+            continue
+        if "format" not in af:
+            raise validate.ValidationException(u"Missing required 'format' for File %s" % af)
+        for inpf in aslist(inputFormats):
+            if af["format"] == inpf or formatSubclassOf(af["format"], inpf, ontology, set()):
+                return
+        raise validate.ValidationException(
+            u"Incompatible file format, expected format(s) %s but file object is: %s" % (inputFormats, json.dumps(af, indent=4)))
 
 class Builder(object):
     def __init__(self):  # type: () -> None
@@ -54,6 +104,7 @@ def __init__(self):  # type: () -> None
         self.js_console = False  # type: bool
         self.mutation_manager = None  # type: MutationManager
         self.force_docker_pull = False  # type: bool
+        self.formatgraph = None  # type: Graph
 
         # One of "no_listing", "shallow_listing", "deep_listing"
         # Will be default "no_listing" for CWL v1.1
@@ -70,8 +121,8 @@ def build_job_script(self, commands):
         else:
             return None
 
-    def bind_input(self, schema, datum, lead_pos=None, tail_pos=None):
-        # type: (Dict[Text, Any], Any, Union[int, List[int]], List[int]) -> List[Dict[Text, Any]]
+    def bind_input(self, schema, datum, lead_pos=None, tail_pos=None, discover_secondaryFiles=False):
+        # type: (Dict[Text, Any], Any, Union[int, List[int]], List[int], bool) -> List[Dict[Text, Any]]
         if tail_pos is None:
             tail_pos = []
         if lead_pos is None:
@@ -105,9 +156,9 @@ def bind_input(self, schema, datum, lead_pos=None, tail_pos=None):
                     schema = copy.deepcopy(schema)
                     schema["type"] = t
                     if not value_from_expression:
-                        return self.bind_input(schema, datum, lead_pos=lead_pos, tail_pos=tail_pos)
+                        return self.bind_input(schema, datum, lead_pos=lead_pos, tail_pos=tail_pos, discover_secondaryFiles=discover_secondaryFiles)
                     else:
-                        self.bind_input(schema, datum, lead_pos=lead_pos, tail_pos=tail_pos)
+                        self.bind_input(schema, datum, lead_pos=lead_pos, tail_pos=tail_pos, discover_secondaryFiles=discover_secondaryFiles)
                         bound_input = True
             if not bound_input:
                 raise validate.ValidationException(u"'%s' is not a valid union %s" % (datum, schema["type"]))
@@ -119,17 +170,17 @@ def bind_input(self, schema, datum, lead_pos=None, tail_pos=None):
                 if k in schema:
                     st[k] = schema[k]
             if value_from_expression:
-                self.bind_input(st, datum, lead_pos=lead_pos, tail_pos=tail_pos)
+                self.bind_input(st, datum, lead_pos=lead_pos, tail_pos=tail_pos, discover_secondaryFiles=discover_secondaryFiles)
             else:
-                bindings.extend(self.bind_input(st, datum, lead_pos=lead_pos, tail_pos=tail_pos))
+                bindings.extend(self.bind_input(st, datum, lead_pos=lead_pos, tail_pos=tail_pos, discover_secondaryFiles=discover_secondaryFiles))
         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"]))
+                        bindings.extend(self.bind_input(f, datum[f["name"]], lead_pos=lead_pos, tail_pos=f["name"], discover_secondaryFiles=discover_secondaryFiles))
                     else:
                         datum[f["name"]] = f.get("default")
 
@@ -147,15 +198,14 @@ def bind_input(self, schema, datum, lead_pos=None, tail_pos=None):
                         if k in schema:
                             itemschema[k] = schema[k]
                     bindings.extend(
-                        self.bind_input(itemschema, item, lead_pos=n, tail_pos=tail_pos))
+                        self.bind_input(itemschema, item, lead_pos=n, tail_pos=tail_pos, discover_secondaryFiles=discover_secondaryFiles))
                 binding = None
 
             if schema["type"] == "File":
                 self.files.append(datum)
-                if binding:
-                    if binding.get("loadContents"):
-                        with self.fs_access.open(datum["location"], "rb") as f:
-                            datum["contents"] = f.read(CONTENT_LIMIT)
+                if (binding and binding.get("loadContents")) or schema.get("loadContents"):
+                    with self.fs_access.open(datum["location"], "rb") as f:
+                        datum["contents"] = f.read(CONTENT_LIMIT)
 
                 if "secondaryFiles" in schema:
                     if "secondaryFiles" not in datum:
@@ -175,14 +225,20 @@ def bind_input(self, schema, datum, lead_pos=None, tail_pos=None):
                             if not found:
                                 if isinstance(sfname, dict):
                                     datum["secondaryFiles"].append(sfname)
-                                else:
+                                elif discover_secondaryFiles:
                                     datum["secondaryFiles"].append({
                                         "location": datum["location"][0:datum["location"].rindex("/")+1]+sfname,
                                         "basename": sfname,
                                         "class": "File"})
+                                else:
+                                    raise WorkflowException("Missing required secondary file '%s' from file object: %s" % (
+                                        sfname, json.dumps(datum, indent=4)))
 
                     normalizeFilesDirs(datum["secondaryFiles"])
 
+                if "format" in schema:
+                    checkFormat(datum, self.do_eval(schema["format"]), self.formatgraph)
+
                 def _capture_files(f):
                     self.files.append(f)
                     return f

cwltool/command_line_tool.py

@@ -470,7 +470,7 @@ def register_mut(f):
             def register_reader(f):
                 if f["location"] not in muts:
                     builder.mutation_manager.register_reader(j.name, f)
-                    readers[f["location"]] = f
+                    readers[f["location"]] = copy.copy(f)
 
             for li in j.generatefiles["listing"]:
                 li = cast(Dict[Text, Any], li)
@@ -630,8 +630,6 @@ def collect_output(self, schema, builder, outdir, fs_access, compute_checksum=Tr
                             f.seek(0, 2)
                             filesize = f.tell()
                         files["size"] = filesize
-                        if "format" in schema:
-                            files["format"] = builder.do_eval(schema["format"], context=files)
 
             optional = False
             single = False
@@ -687,6 +685,10 @@ def collect_output(self, schema, builder, outdir, fs_access, compute_checksum=Tr
                                         sfitem["class"] = "Directory"
                                         primary["secondaryFiles"].append(sfitem)
 
+            if "format" in schema:
+                for primary in aslist(r):
+                    primary["format"] = builder.do_eval(schema["format"], context=primary)
+
             # Ensure files point to local references outside of the run environment
             adjustFileObjs(r, cast(  # known bug in mypy
                 # https://github.com/python/mypy/issues/797
@@ -702,4 +704,4 @@ def collect_output(self, schema, builder, outdir, fs_access, compute_checksum=Tr
                     f, builder, outdir, fs_access,
                     compute_checksum=compute_checksum)
             return out
-        return r
+        return r

cwltool/executors.py

@@ -56,6 +56,7 @@ def execute(self, t,  # type: Process
             "tmp_outdir_prefix") else tempfile.mkdtemp()
         self.output_dirs.add(kwargs["outdir"])
         kwargs["mutation_manager"] = MutationManager()
+        kwargs["toplevel"] = True
 
         jobReqs = None
         if "cwl:requirements" in job_order_object:

cwltool/main.py

@@ -459,6 +459,7 @@ def main(argsl=None,  # type: List[str]
                                                                         tool_file_uri)
         except Exception as e:
             _logger.error(Text(e), exc_info=args.debug)
+            return 1
 
         if args.overrides:
             overrides.extend(load_overrides(file_uri(os.path.abspath(args.overrides)), tool_file_uri))

cwltool/process.py

@@ -352,55 +352,6 @@ def cleanIntermediate(output_dirs):  # type: (Set[Text]) -> None
             shutil.rmtree(a, True)
 
 
-def formatSubclassOf(fmt, cls, ontology, visited):
-    # type: (Text, Text, Graph, Set[Text]) -> bool
-    """Determine if `fmt` is a subclass of `cls`."""
-
-    if URIRef(fmt) == URIRef(cls):
-        return True
-
-    if ontology is None:
-        return False
-
-    if fmt in visited:
-        return False
-
-    visited.add(fmt)
-
-    uriRefFmt = URIRef(fmt)
-
-    for s, p, o in ontology.triples((uriRefFmt, RDFS.subClassOf, None)):
-        # Find parent classes of `fmt` and search upward
-        if formatSubclassOf(o, cls, ontology, visited):
-            return True
-
-    for s, p, o in ontology.triples((uriRefFmt, OWL.equivalentClass, None)):
-        # Find equivalent classes of `fmt` and search horizontally
-        if formatSubclassOf(o, cls, ontology, visited):
-            return True
-
-    for s, p, o in ontology.triples((None, OWL.equivalentClass, uriRefFmt)):
-        # Find equivalent classes of `fmt` and search horizontally
-        if formatSubclassOf(s, cls, ontology, visited):
-            return True
-
-    return False
-
-
-def checkFormat(actualFile, inputFormats, ontology):
-    # type: (Union[Dict[Text, Any], List, Text], Union[List[Text], Text], Graph) -> None
-    for af in aslist(actualFile):
-        if not af:
-            continue
-        if "format" not in af:
-            raise validate.ValidationException(u"Missing required 'format' for File %s" % af)
-        for inpf in aslist(inputFormats):
-            if af["format"] == inpf or formatSubclassOf(af["format"], inpf, ontology, set()):
-                return
-        raise validate.ValidationException(
-            u"Incompatible file format %s required format(s) %s" % (af["format"], inputFormats))
-
-
 def fillInDefaults(inputs, job):
     # type: (List[Dict[Text, Text]], Dict[Text, Union[Dict[Text, Any], List, Text]]) -> None
     for e, inp in enumerate(inputs):
@@ -606,6 +557,7 @@ def _init_job(self, joborder, **kwargs):
         builder.debug = kwargs.get("debug")
         builder.js_console = kwargs.get("js_console")
         builder.mutation_manager = kwargs.get("mutation_manager")
+        builder.formatgraph = self.formatgraph
 
         builder.make_fs_access = kwargs.get("make_fs_access") or StdFsAccess
         builder.fs_access = builder.make_fs_access(kwargs["basedir"])
@@ -640,13 +592,7 @@ def _init_job(self, joborder, **kwargs):
                 builder.tmpdir = builder.fs_access.realpath(kwargs.get("tmpdir") or tempfile.mkdtemp())
                 builder.stagedir = builder.fs_access.realpath(kwargs.get("stagedir") or tempfile.mkdtemp())
 
-        if self.formatgraph:
-            for i in self.tool["inputs"]:
-                d = shortname(i["id"])
-                if d in builder.job and i.get("format"):
-                    checkFormat(builder.job[d], builder.do_eval(i["format"]), self.formatgraph)
-
-        builder.bindings.extend(builder.bind_input(self.inputs_record_schema, builder.job))
+        builder.bindings.extend(builder.bind_input(self.inputs_record_schema, builder.job, discover_secondaryFiles=kwargs.get("toplevel")))
 
         if self.tool.get("baseCommand"):
             for n, b in enumerate(aslist(self.tool["baseCommand"])):

cwltool/schemas/.travis.yml

@@ -0,0 +1,7 @@
+language: python
+python:
+  - 2.7
+before_script:
+  - git clone https://github.com/common-workflow-language/schema_salad.git salad && cd ./salad && pip install . && cd ..
+script:
+  - for target in v1.0/v1.0/*.cwl; do schema-salad-tool ./salad/schema_salad/tests/test_schema/CommonWorkflowLanguage.yml ${target} || exit 1; done

cwltool/schemas/CITATION

@@ -3,10 +3,11 @@ To cite the Common Workflow Language standard in a publication, please use:
 Amstutz, Peter; Crusoe, Michael R; Tijanić, Nebojša; Chapman, Brad;
 Chilton, John; Heuer, Michael; Kartashov, Andrey; Kern, John; Leehr, Dan;
 Ménager, Hervé; Nedeljkovich, Maya; Scales, Matt; Soiland-Reyes, Stian;
-Stojanovic, Luka (2016): Common Workflow Language, v1.0. figshare.
+Stojanovic, Luka (2016): Common Workflow Language, v1.0. Specification,
+Common Workflow Language working group. https://w3id.org/cwl/v1.0/
 https://dx.doi.org/10.6084/m9.figshare.3115156.v2
 
-@data{common-workflow-language-draft3,
+@data{cwl,
   doi = {10.6084/m9.figshare.3115156.v2},
   url = {http://dx.doi.org/10.6084/m9.figshare.3115156.v2},
   author = {Peter Amstutz; Michael R. Crusoe; Nebojša Tijanić; Brad Chapman;
@@ -15,6 +16,13 @@ Hervé Ménager; Maya Nedeljkovich; Matt Scales; Stian Soiland-Reyes;
 Luka Stojanovic
  },
   publisher = {Figshare},
+  institution = {Common Workflow Language working group},
   title = {Common Workflow Language, v1.0},
   year = {2016}
 }
+
+# Are you editing this file?
+# Synchronize any changes made with
+# README.md
+# and
+# https://github.com/common-workflow-language/user_guide/blob/gh-pages/CITATION