Merge remote-tracking branch 'origin/master' into stagefiles

Author: mr-c

cwltool/cwltest.py

@@ -7,8 +7,8 @@
 import sys
 import shutil
 import tempfile
-import yaml
-import yaml.scanner
+import ruamel.yaml as yaml
+import ruamel.yaml.scanner as yamlscanner
 import pipes
 import logging
 import schema_salad.ref_resolver
@@ -112,7 +112,7 @@ def run_test(args, i, t):  # type: (argparse.Namespace, Any, Dict[str,str]) -> i
             _logger.error(t.get("doc"))
             _logger.error("Returned non-zero")
             return 1
-    except (yaml.scanner.ScannerError, TypeError) as e:
+    except (yamlscanner.ScannerError, TypeError) as e:
         _logger.error(u"""Test failed: %s""", " ".join([pipes.quote(tc) for tc in test_command]))
         _logger.error(outstr)
         _logger.error(u"Parse error %s", str(e))

cwltool/expression.py

@@ -5,7 +5,6 @@
 import logging
 import os
 from .errors import WorkflowException
-import yaml
 import schema_salad.validate as validate
 import schema_salad.ref_resolver
 from . import sandboxjs

cwltool/job.py

@@ -4,7 +4,6 @@
 import tempfile
 import glob
 import json
-import yaml
 import logging
 import sys
 import requests

cwltool/main.py

@@ -8,13 +8,13 @@
 import logging
 import copy
 import tempfile
+import ruamel.yaml as yaml
 import urlparse
 import hashlib
 import pkg_resources  # part of setuptools
 import random
 import functools
 
-import yaml
 import rdflib
 from typing import Union, Any, cast, Callable, Dict, Tuple, IO
 

cwltool/process.py

@@ -2,8 +2,6 @@
 import os
 import json
 import copy
-import yaml
-import copy
 import logging
 import pprint
 import stat

cwltool/workflow.py

@@ -242,23 +242,23 @@ def try_make_job(self, step, **kwargs):
                 raise WorkflowException("Workflow step contains valueFrom but StepInputExpressionRequirement not in requirements")
 
             vfinputs = {shortname(k): v for k,v in inputobj.iteritems()}
-            def valueFromFunc(k, v):  # type: (Any, Any) -> Any
-                if k in valueFrom:
-                    return expression.do_eval(
-                            valueFrom[k], vfinputs, self.workflow.requirements,
-                            None, None, {}, context=v)
-                else:
-                    return v
+            def postScatterEval(io):
+                shortio = {shortname(k): v for k,v in io.iteritems()}
+                def valueFromFunc(k, v):  # type: (Any, Any) -> Any
+                    if k in valueFrom:
+                        return expression.do_eval(
+                                valueFrom[k], shortio, self.workflow.requirements,
+                                None, None, {}, context=v)
+                    else:
+                        return v
+                return {k: valueFromFunc(k, v) for k,v in io.items()}
 
             if "scatter" in step.tool:
                 scatter = aslist(step.tool["scatter"])
                 method = step.tool.get("scatterMethod")
                 if method is None and len(scatter) != 1:
                     raise WorkflowException("Must specify scatterMethod when scattering over multiple inputs")
-                kwargs["valueFrom"] = valueFromFunc
-
-                inputobj = {k: valueFromFunc(k, v) if k not in scatter else v
-                            for k,v in inputobj.items()}
+                kwargs["postScatterEval"] = postScatterEval
 
                 if method == "dotproduct" or method is None:
                     jobs = dotproduct_scatter(step, inputobj, scatter,
@@ -280,7 +280,7 @@ def valueFromFunc(k, v):  # type: (Any, Any) -> Any
                                                          callback), 0, **kwargs)
             else:
                 _logger.debug(u"[job %s] job input %s", step.name, json.dumps(inputobj, indent=4))
-                inputobj = {k: valueFromFunc(k, v) for k,v in inputobj.items()}
+                inputobj = postScatterEval(inputobj)
                 _logger.debug(u"[job %s] evaluated job input to %s", step.name, json.dumps(inputobj, indent=4))
                 jobs = step.job(inputobj, callback, **kwargs)
 
@@ -551,7 +551,9 @@ def dotproduct_scatter(process, joborder, scatter_keys, output_callback, **kwarg
     for n in range(0, l):
         jo = copy.copy(joborder)
         for s in scatter_keys:
-            jo[s] = kwargs["valueFrom"](s, joborder[s][n])
+            jo[s] = joborder[s][n]
+
+        jo = kwargs["postScatterEval"](jo)
 
         for j in process.job(jo, functools.partial(rc.receive_scatter_output, n), **kwargs):
             yield j
@@ -571,9 +573,10 @@ def nested_crossproduct_scatter(process, joborder, scatter_keys, output_callback
 
     for n in range(0, l):
         jo = copy.copy(joborder)
-        jo[scatter_key] = kwargs["valueFrom"](scatter_key, joborder[scatter_key][n])
+        jo[scatter_key] = joborder[scatter_key][n]
 
         if len(scatter_keys) == 1:
+            jo = kwargs["postScatterEval"](jo)
             for j in process.job(jo, functools.partial(rc.receive_scatter_output, n), **kwargs):
                 yield j
         else:
@@ -620,9 +623,10 @@ def flat_crossproduct_scatter(process, joborder, scatter_keys, output_callback,
     put = startindex
     for n in range(0, l):
         jo = copy.copy(joborder)
-        jo[scatter_key] = kwargs["valueFrom"](scatter_key, joborder[scatter_key][n])
+        jo[scatter_key] = joborder[scatter_key][n]
 
         if len(scatter_keys) == 1:
+            jo = kwargs["postScatterEval"](jo)
             for j in process.job(jo, functools.partial(rc.receive_scatter_output, put), **kwargs):
                 yield j
             put += 1

setup.py

@@ -37,7 +37,7 @@
                                 'schemas/draft-4/salad/schema_salad/metaschema/*.md']},
       install_requires=[
           'requests',
-          'PyYAML',
+          'ruamel.yaml',
           'rdflib >= 4.1.0',
           'rdflib-jsonld >= 0.3.0',
           'shellescape',