Merge pull request #2 from bmeg/workflow

Adding workflow builder

Author: khillion

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)

cwlgen/workflow.py

@@ -0,0 +1,137 @@
+
+
+import ruamel.yaml
+import six
+
+from . import CWL_SHEBANG, Parameter
+from .utils import *
+
+
+class Workflow(object):
+    '''
+    Contain all informations to describe a CWL workflow.
+    '''
+    __CLASS__ = 'Workflow'
+
+    def __init__(self):
+        self.steps = []
+        self.inputs = []
+        self.outputs = []
+
+    def export(self, outfile=None):
+        """
+        Export the workflow in CWL either on STDOUT or in outfile.
+        """
+        # First add representer (see .utils.py) for multiline writting
+        ruamel.yaml.add_representer(literal, literal_presenter)
+        cwl_workflow = {k: v for k, v in vars(self).items() if v is not None and\
+                                                           type(v) is str}
+        cwl_workflow['class'] = self.__CLASS__
+        cwl_workflow['cwlVersion'] = 'v1.0'
+
+        cwl_workflow['inputs'] = {}
+        cwl_workflow['outputs'] = {}
+
+        if self.steps:
+            cwl_workflow['steps'] = {}
+            for step in self.steps:
+                cwl_workflow['steps'][step.id] = step.get_dict()
+
+        if self.inputs:
+            cwl_workflow['inputs'] = {}
+            for i in self.inputs:
+                cwl_workflow['inputs'][i.id] = i.get_dict()
+
+        if self.outputs:
+            cwl_workflow['outputs'] = {}
+            for out in self.outputs:
+                cwl_workflow['outputs'][out.id] = out.get_dict()
+
+        # Write CWL file in YAML
+        if outfile is None:
+            six.print_(CWL_SHEBANG, "\n", sep='')
+            six.print_(ruamel.yaml.dump(cwl_workflow))
+        else:
+            out_write = open(outfile, 'w')
+            out_write.write(CWL_SHEBANG + '\n\n')
+            out_write.write(ruamel.yaml.dump(cwl_workflow))
+            out_write.close()
+
+class InputParameter(Parameter):
+    def __init__(self, param_id, label=None, secondary_files=None, param_format=None,
+                 streamable=False, doc=None, input_binding=None, default=None, param_type=None):
+        Parameter.__init__(self, param_id=param_id, label=label,
+                           secondary_files=secondary_files, param_format=param_format,
+                           streamable=streamable, doc=doc, param_type=param_type)
+
+
+class WorkflowStep(object):
+    def __init__(self, id, inputs=None, outputs=None, run=None):
+        self.id = id
+        if inputs:
+            self.inputs = inputs
+        else:
+            self.inputs = []
+        if outputs:
+            self.outputs = outputs
+        else:
+            self.outputs = []
+        self.run = run
+
+    def get_dict(self):
+        '''
+        Transform the object to a [DICT] to write CWL.
+
+        :return: dictionnary of the object
+        :rtype: DICT
+        '''
+        dict_param = {'id' : self.id, "run" : self.run} #{k: v for k, v in vars(self).items() if v is not None and v is not False}
+
+        if self.inputs:
+            dict_param['in'] = {}
+            for i in self.inputs:
+                if i.src:
+                    dict_param['in'][i.id] = i.src
+                elif i.default:
+                    dict_param['in'][i.id] = {"default" : i.default}
+
+        if self.outputs:
+            dict_param['out'] = []
+            for i in self.outputs:
+                dict_param['out'].append(i)
+
+        return dict_param
+
+class WorkflowStepInput(object):
+    def __init__(self, id, src=None, default=None):
+        self.id = id
+        self.src = src
+        self.default = default
+
+    def get_dict(self):
+        '''
+        Transform the object to a [DICT] to write CWL.
+
+        :return: dictionnary of the object
+        :rtype: DICT
+        '''
+        dict_param = {} #{k: v for k, v in vars(self).items() if v is not None and v is not False}
+        if self.src:
+            dict_param["src"] = self.src
+        if self.default:
+            dict_param["default"] = self.default
+        return dict_param
+
+class WorkflowStepOutput(object):
+    def __init__(self, id):
+        self.id = id
+
+
+
+class WorkflowOutputParameter(Parameter):
+    def __init__(self, param_id, outputSource, label=None, secondary_files=None, param_format=None,
+                 streamable=False, doc=None, param_type=None):
+        Parameter.__init__(self, param_id=param_id, label=label,
+                           secondary_files=secondary_files, param_format=param_format,
+                           streamable=streamable, doc=doc, param_type=param_type)
+        self.outputSource = outputSource