Adding in workflow parsing

Author: kellrott

cwlgen/import_cwl.py

@@ -14,12 +14,25 @@
 import ruamel.yaml as ryaml
 import six
 import cwlgen
+import cwlgen.workflow
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
 #  Class(es)  ------------------------------
 
+def parse_cwl(cwl_path):
+    with open(cwl_path) as yaml_file:
+        cwl_dict = ryaml.load(yaml_file, Loader=ryaml.Loader)
+        cl = cwl_dict['class']
+
+    if cl == "CommandLineTool":
+        p = CWLToolParser()
+        return p.import_cwl(cwl_path)
+    if cl == "Workflow":
+        p = CWLToolParser()
+        return p.import_cwl(cwl_path)
+    return None
 
 class CWLToolParser(object):
     """
@@ -46,6 +59,12 @@ def _load_id(self, tool, id_el):
         """
         tool.id = id_el
 
+    def _load_requirements(self, tool, req_el):
+        pass
+
+    def _load_hints(self, tool, req_el):
+        pass
+
     def _load_baseCommand(self, tool, command_el):
         """
         Load the content of baseCommand into the tool.
@@ -178,6 +197,15 @@ def import_cwl(self, cwl_path):
                 logger.warning(key + " content is not processed (yet).")
         return tool
 
+def dict_or_idlist_items(v):
+    if isinstance(v, dict):
+        return v.items()
+    o = []
+    for i in v:
+        e = dict(i)
+        del e['id']
+        o.append( (i['id'], e) )
+    return o
 
 class InputsParser(object):
     """
@@ -283,7 +311,7 @@ def load_inputs(self, inputs, inputs_el):
         :type inputs_el: LIST or DICT
         """
         # For the moment, only deal with the format exported by cwlgen
-        for key, value in inputs_el.items():
+        for key, value in dict_or_idlist_items(inputs_el):
             input_obj = cwlgen.CommandInputParameter(key)
             for key, element in value.items():
                 try:
@@ -476,6 +504,17 @@ def _load_type(self, output_obj, type_el):
         """
         output_obj.type = type_el
 
+    def _load_outputSource(self, output_obj, type_el):
+        """
+        Load the content of type into the output object.
+
+        :param output_obj: output obj
+        :type output_obj: : :class:`cwlgen.CommandInputParameter`
+        :param type_el: Content of type
+        :type type_el: STRING
+        """
+        output_obj.outputSource = type_el
+
     def load_outputs(self, outputs, outputs_el):
         """
         Load the content of outputs into the outputs list.
@@ -486,7 +525,7 @@ def load_outputs(self, outputs, outputs_el):
         :type outputs_el: LIST or DICT
         """
         # For the moment, only deal with the format exported by cwlgen
-        for key, value in outputs_el.items():
+        for key, value in dict_or_idlist_items(outputs_el):
             output_obj = cwlgen.CommandOutputParameter(key)
             for key, element in value.items():
                 try:
@@ -550,3 +589,160 @@ def load_outbinding(self, output_obj, outbinding_el):
             except AttributeError:
                 logger.warning(key + " content for outputBinding is not processed (yet).")
             output_obj.outputBinding = outbinding_obj
+
+
+class StepsParser(object):
+    """
+    Class to parse content of steps of workflow from existing CWL Workflow.
+    """
+    def __init__(self, basedir=None):
+        self.basedir = basedir
+
+    def _load_in(self, step_obj, in_el):
+        for key, val in in_el.items():
+            o = cwlgen.workflow.WorkflowStepInput(key)
+            if isinstance(val, dict) and 'default' in val:
+                o.default = val['default']
+            else:
+                o.src = val
+            step_obj.inputs.append(o)
+
+    def _load_out(self, step_obj, in_el):
+        for val in in_el:
+            o = cwlgen.workflow.WorkflowStepOutput(val)
+            step_obj.outputs.append(o)
+
+    def _load_run(self, step_obj, in_el):
+        if isinstance(in_el, basestring):
+            path = os.path.join(self.basedir, in_el)
+            #logger.info("Parsing: %s", path)
+            step_obj.run = parse_cwl(path)
+
+
+    def load_steps(self, steps_obj, steps_elm):
+        """
+        Load the content of step into the output object.
+
+        :param output_obj: output object
+        :type output_obj: :class:`cwlgen.CommandOutputParameter`
+        :param outbinding_el: Content of outputBinding element
+        :type outbinding_el: DICT
+        """
+        for key, value in steps_elm.items():
+            step_obj = cwlgen.workflow.WorkflowStep(key)
+            for key, element in value.items():
+                try:
+                    getattr(self, '_load_{}'.format(key))(step_obj, element)
+                except AttributeError:
+                    logger.warning(key + " content for input is not processed (yet).")
+            steps_obj.append(step_obj)
+
+
+class CWLWorkflowParser(object):
+
+    def __init__(self, basedir=None):
+        self.basedir = basedir
+
+    def _init_workflow(self, cwl_dict):
+        """
+        Init tool from existing CWL tool.
+
+        :param cwl_dict: Full content of CWL file
+        :type cwl_dict: DICT
+        """
+        return cwlgen.workflow.Workflow()
+
+    def import_cwl(self, cwl_path):
+        """
+        Load content of cwl into the :class:`cwlgen.workflow.Workflow` object.
+
+        :param cwl_path: Path of the CWL tool to be loaded.
+        :type cwl_path: STRING
+        :return: CWL tool content in cwlgen model.
+        :rtype: :class:`cwlgen.workflow.Workflow`
+        """
+        with open(cwl_path) as yaml_file:
+            cwl_dict = ryaml.load(yaml_file, Loader=ryaml.Loader)
+        tool = self._init_workflow(cwl_dict)
+        self.basedir = os.path.dirname(cwl_path)
+        for key, element in cwl_dict.items():
+            try:
+                getattr(self, '_load_{}'.format(key))(tool, element)
+            except AttributeError:
+                logger.warning(key + " workflow content is not processed (yet).")
+        return tool
+
+    def _load_id(self, tool, id_el):
+        """
+        Load the content of id into the tool.
+
+        :param tool: Tool object from cwlgen
+        :type tool: :class:`cwlgen.CommandLineTool`
+        :param id_el: Content of id
+        :type id_el: STRING or [STRING]
+        """
+        tool.id = id_el
+
+    def _load_cwlVersion(self, tool, cwl_version_el):
+        """
+        Load the content of cwlVersion into the tool.
+
+        :param tool: Tool object from cwlgen
+        :type tool: :class:`cwlgen.workflow.Workflow`
+        :param cwl_version_el: Content of cwlVersion
+        :type cwl_version_el: STRING
+        """
+        tool.cwlVersion = cwl_version_el
+
+
+    def _load_class(self, tool, class_el):
+        """
+        Display message to inform that cwlgen only deal with Workflow for the moment.
+
+        :param tool: Workflow object from cwlgen
+        :type tool: :class:`cwlgen.workflow.Workflow`
+        :param class_el: Content of class
+        :type class_el: STRING
+        """
+        if class_el != 'Workflow':
+            logger.warning('cwlgen library only handle Workflow for the moment')
+
+    def _load_requirements(self, tool, req_el):
+        pass
+
+    def _load_inputs(self, tool, inputs_el):
+        """
+        Load the content of inputs into the tool.
+
+        :param tool: Tool object from cwlgen
+        :type tool: :class:`cwlgen.workflow.Workflow`
+        :param inputs_el: Content of inputs
+        :type inputs_el: LIST or DICT
+        """
+        inp_parser = InputsParser()
+        inp_parser.load_inputs(tool.inputs, inputs_el)
+
+    def _load_outputs(self, tool, outputs_el):
+        """
+        Load the content of inputs into the tool.
+
+        :param tool: Tool object from cwlgen
+        :type tool: :class:`cwlgen.workflow.Workflow`
+        :param outputs_el: Content of outputs
+        :type outputs_el: LIST or DICT
+        """
+        inp_parser = OutputsParser()
+        inp_parser.load_outputs(tool.outputs, outputs_el)
+
+
+    def _load_steps(self, tool, outputs_el):
+        """
+        Load the content of inputs into the tool.
+
+        :param tool: Tool object from cwlgen
+        :type tool: :class:`cwlgen.workflow.Workflow`
+        :param outputs_el: Content of outputs
+        :type outputs_el: LIST or DICT
+        """
+        inp_parser = StepsParser(self.basedir)
+        inp_parser.load_steps(tool.steps, outputs_el)