Upgrade jflow v3-git

src/jflow/__init__.py

@@ -22,8 +22,6 @@ import os
 
 from jflow.config_reader import JFlowConfigReader
 
-# Define some Error classes
-class InvalidFormatError(Exception): pass
 
 jflowconf = JFlowConfigReader()
 

src/jflow/abstraction.py

@@ -27,7 +27,66 @@ from weaver.options import Options
 from weaver.abstraction import Abstraction
 
 
-class MultiMap(Abstraction):
+class AbstractionWargs(Abstraction):
+    """ The base Abstraction class.
+
+    **Positional Arguments**:
+
+    - `function`    -- Function to apply (Function, string, string format)
+
+    **Keyword Arguments**:
+
+    - `inputs`      -- Inputs to function
+    - `outputs`     -- Output of function
+    - `includes`    -- Files to include for each task.
+    - `native`      -- Whether or not to use native abstraction if available.
+    - `group`       -- Number of tasks to inline.
+    - `collect`     -- Whether or not to mark files for garbage collection.
+    - `local`       -- Whether or not to force local execution.
+
+    `inputs` and `includes` are parsed using
+    :func:`~weaver.data.parse_input_list` and must be in a form acceptable tot
+    hat function.  Likewise, `outputs` is parsed by
+    :func:`~weaver.data.parse_output_list` and `function` is parsed by
+    :func:`~weaver.function.parse_function`.
+    """
+    Counter = None
+
+    def __init__(self, function, inputs=None, outputs=None, includes=None,
+                 native=False, group=None, collect=False, local=False, arguments=None):
+        Abstraction.__init__(self, function, inputs, outputs, includes, native, group, collect, local)
+        self.arguments = arguments
+
+
+class Map(AbstractionWargs):
+    """ Weaver Map Abstraction.
+
+    This Abstraction enables the following pattern of execution:
+
+        Map(f, inputs, outputs)
+
+    In this case, the :class:`Function` *f* is applied to each item in
+    *inputs* to generate the corresponding *outputs*.
+    """
+
+    Counter = itertools.count()
+
+    @cache_generation
+    def _generate(self):
+        with self:
+            debug(D_ABSTRACTION, 'Generating Abstraction {0}'.format(self))
+
+            function = parse_function(self.function)
+            inputs   = parse_input_list(self.inputs)
+            outputs  = parse_output_list(self.outputs, inputs)
+            includes = parse_input_list(self.includes)
+
+            for i, o in zip(inputs, outputs):
+                with Options(local=self.options.local, collect=[i] if self.collect else None):
+                    yield function(i, o, self.arguments, includes)
+
+
+class MultiMap(AbstractionWargs):
     """ Weaver MultiMap Abstraction.
 
     This Abstraction enables the following pattern of execution:
@@ -110,5 +169,5 @@ class MultiMap(Abstraction):
                     iteration_outputs = parse_output_list(self.outputs, input_pattern)
                 
                 with Options(local=self.options.local):
-                    yield function(iteration_inputs, iteration_outputs, None, includes)
+                    yield function(iteration_inputs, iteration_outputs, self.arguments, includes)
 

src/jflow/argument_parser.py

+#
+# Copyright (C) 2015 INRA
+# 
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+# 
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+# 
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+import argparse
+
+class JflowArgumentParser (argparse.ArgumentParser):
+    def _read_args_from_files(self, arg_strings):
+        # expand arguments referencing files
+        new_arg_strings = []
+        for arg_string in arg_strings:
+            # if it's not a comment or an empty line
+            if not arg_string.startswith("#") and arg_string:
+                # for regular arguments, just add them back into the list
+                if not arg_string or arg_string[0] not in self.fromfile_prefix_chars:
+                    new_arg_strings.append(arg_string)
+                # replace arguments referencing files with the file content
+                else:
+                    try:
+                        with open(arg_string[1:]) as args_file:
+                            arg_strings = []
+                            # give to the convert_arg_line_to_args a table of lines instead of line per line
+                            for arg in self.convert_arg_line_to_args(args_file.read().splitlines()):
+                                arg_strings.append(arg)
+                            arg_strings = self._read_args_from_files(arg_strings)
+                            new_arg_strings.extend(arg_strings)
+                    except OSError:
+                        err = _sys.exc_info()[1]
+                        self.error(str(err))
+        # return the modified argument list
+        return new_arg_strings
\ No newline at end of file

src/jflow/component.py

@@ -21,6 +21,7 @@ import sys
 import inspect
 import tempfile
 import types
+import shutil
 
 from operator import attrgetter
 
@@ -33,23 +34,32 @@ from jflow.abstraction import MultiMap
 
 from weaver.util import parse_string_list
 from weaver.function import ShellFunction
-from weaver.abstraction import Map
+from jflow.abstraction import Map
 from weaver.function import PythonFunction
 
+from pathlib import Path
+import inspect
+
 
 class Component(object):
     
     TRACE_FILE_NAME = "trace.txt"
     
     def __init__(self):
-        self.prefix = "default"
+        self.__prefix = "default"
         self.params_order = []
         self.output_directory = None
+        self.description = None
         self.config_reader = JFlowConfigReader()
+        try:
+            self.prg_name = self.get_command()
+        except NotImplementedError:
+            self.prg_name = self.__class__.__name__
         self.version = self.get_version()
         if isinstance(self.version, bytes):
             self.version = self.version.decode()
         self.batch_options = self.config_reader.get_component_batch_options(self.__class__.__name__)
+        self.modules = self.config_reader.get_component_modules(self.__class__.__name__)
         # in case of SGE, parse the cpu and memory parameter
         self.__cpu=None
         self.__memory=None
@@ -68,14 +78,16 @@ class Component(object):
             try:
                 self.__memory = re.match( r'.*\s?mem=(\d+\w)\s?.*', self.batch_options).group(1)
             except: pass
-        elif type.lower() == "slurm" :
-            try:
-                self.__cpu = int(re.match( r'.*-c\s+(\w+)\s+(\d+)\s?.*', self.batch_options).group(2))
-            except: pass
-            try:
-                self.__memory = re.match( r'.*--mem=(\d+\S+)\s?.*', self.batch_options).group(1)
-            except: pass
-            
+
+    def get_prefix(self):
+        return self.__prefix
+
+    def set_prefix(self, prefix):
+        self.__prefix = prefix
+
+    def get_description(self):
+        return self.description
+
     def get_cpu(self):
         return self.__cpu
     
@@ -99,11 +111,13 @@ class Component(object):
         outputs = {}
         for attribute_value in list(self.__dict__.values()):
             if ( issubclass( attribute_value.__class__, DynamicOutput ) or
-                 issubclass( attribute_value.__class__, OutputFileList) ):
+                 issubclass( attribute_value.__class__, OutputFileList)):
                 for f in attribute_value:
                     outputs[os.path.basename(f)] = f
             elif issubclass( attribute_value.__class__, OutputFile):
                 outputs[os.path.basename(attribute_value)] = attribute_value
+            elif issubclass( attribute_value.__class__, OutputDirectory):
+                outputs[os.path.basename(attribute_value)] = attribute_value
         return outputs
 
     def add_input_directory(self, name, help, default=None, required=False, flag=None, 
@@ -135,7 +149,11 @@ class Component(object):
     
     def reset(self):
         for file in os.listdir(self.output_directory):
-            os.remove(os.path.join(self.output_directory, file))
+            final_file = os.path.join(self.output_directory, file)
+            if os.path.isdir(final_file):
+                shutil.rmtree(final_file)
+            else:
+                os.remove(final_file)
     
     def add_input_file_list(self, name, help, file_format="any", default=None, type="inputfile", 
                             required=False, flag=None, group="default", display_name=None,
@@ -243,6 +261,17 @@ class Component(object):
         # add it to the class itself
         self.params_order.append(name)
         self.__setattr__(name, new_param)
+
+    def add_output_directory(self, name, help, dirname=None, group="default", display_name=None,
+                            cmd_format="", argpos=-1):
+        dirname = os.path.basename(dirname)
+        new_param = OutputDirectory(name, help, default=os.path.join(self.output_directory, dirname),
+                               group=group, display_name=display_name, cmd_format=cmd_format, argpos=argpos)
+        # store where the parameter is coming from
+        new_param.linkTrace_nameid = self.get_nameid()
+        # add it to the class itself
+        self.params_order.append(name)
+        self.__setattr__(name, new_param)
         
     def add_output_file(self, name, help, file_format="any", filename=None, group="default", display_name=None,
                          cmd_format="", argpos=-1):
@@ -400,7 +429,7 @@ class Component(object):
                     for e in file :
                         commandline += ' %s $%s ' % (file.cmd_format, cpt)
                         cpt+=1
-            function = ShellFunction( commandline,  cmd_format='{EXE} {IN} {OUT}')
+            function = ShellFunction( commandline,  cmd_format='{EXE} {IN} {OUT}', modules=self.modules)
             function(inputs=inputs, outputs=outputs) 
         # weaver map abstraction
         elif abstraction == 'map' :
@@ -417,7 +446,7 @@ class Component(object):
                 if isinstance(file, ParameterList) :
                     outputs = file 
             
-            function = ShellFunction( commandline,  cmd_format='{EXE} {IN} {OUT}')
+            function = ShellFunction( commandline,  cmd_format='{EXE} {IN} {OUT}', modules=self.modules)
             exe = Map(function, inputs=inputs, outputs=outputs)
         
         # jflow multimap
@@ -429,7 +458,7 @@ class Component(object):
                 commandline += ' %s $%s ' % (file.cmd_format, cpt)
                 cpt+=1
             
-            function  = ShellFunction( commandline,  cmd_format='{EXE} {IN} {OUT}')
+            function  = ShellFunction( commandline,  cmd_format='{EXE} {IN} {OUT}', modules=self.modules)
             exe = MultiMap(function, inputs=inputs, outputs=outputs)
         # anything other than that will be considered errored
         else :
@@ -484,16 +513,16 @@ class Component(object):
             raise Exception("'" + software + "' path connot be retrieved either in the PATH and in the application.properties file!")
         elif exec_path is None and which(software) != None: 
             exec_path = software
-        elif exec_path != None and not os.path.isfile(exec_path):
+        elif exec_path != None and not os.path.exists(exec_path):
             logging.getLogger("jflow").exception("'" + exec_path + "' set for '" + software + "' does not exists, please provide a valid path!")
             raise Exception("'" + exec_path + "' set for '" + software + "' does not exists, please provide a valid path!")
         return exec_path
     
     def get_nameid(self):
-        return self.__class__.__name__ + "." + self.prefix
+        return self.__class__.__name__ + "." + self.__prefix
     
     def __eq__(self, other):
-        return self.__class__ == other.__class__ and self.prefix == other.prefix
+        return self.__class__ == other.__class__ and self.__prefix == other.get_prefix()
     
     def __getattribute__(self, attr):
         # an IOobject is a specific object defined by the presence of the dump_path attribute
@@ -547,12 +576,13 @@ class Component(object):
         return new_ios,includes
     
     def add_python_execution(self, function, inputs=[], outputs=[], arguments=[], includes=[], 
-                             add_path=None, collect=False, local=False, map=False, cmd_format=""):
+                             add_path=set(), collect=False, local=False, map=False, cmd_format=""):
+
+        workflow_dir = Path(os.path.dirname(inspect.getfile(self.__class__))).parent
+        lib_dir = str(workflow_dir) + os.path.sep + "lib"
+        add_path.add(lib_dir)
                
         if map:
-            if arguments != [] :
-                logging.getLogger("jflow").exception("add_python_execution: '" + function.__name__ + "' arguments parameter not allowed with map!")
-                raise Exception("add_python_execution: '" + function.__name__ + "' arguments parameter not allowed with map!" )
             if not issubclass(inputs.__class__, list) or not issubclass(outputs.__class__, list):
                 logging.getLogger("jflow").exception("add_python_execution: '" + function.__name__ + "' map requires a list as inputs and output!")
                 raise Exception("add_python_execution: '" + function.__name__ + "' map requires a list as inputs and output!")
@@ -565,7 +595,7 @@ class Component(object):
                 cmd_format += " {IN}"
             if (isinstance(outputs, list) and len(outputs)>0) or (outputs is not None and outputs != []):
                 cmd_format += " {OUT}"
-        py_function = PythonFunction(function, add_path=add_path, cmd_format=cmd_format)
+        py_function = PythonFunction(function, add_path=add_path, cmd_format=cmd_format, modules=self.modules)
 
         
         new_inputs,includes_in = self.__generate_iolist(inputs, map)
@@ -573,25 +603,22 @@ class Component(object):
         if not isinstance(includes, list):
             includes=[includes]
         if map:
-            MultiMap(py_function, inputs=new_inputs, outputs=new_outputs, includes=includes+includes_in, collect=collect, local=local)
+            MultiMap(py_function, inputs=new_inputs, outputs=new_outputs, includes=includes+includes_in,
+                     collect=collect, local=local, arguments=arguments)
         else:
             py_function(inputs=new_inputs, outputs=new_outputs, arguments=arguments, includes=includes+includes_in)
-        
+
         self.__write_trace(function.__name__, inputs, outputs, arguments, cmd_format, map, "PythonFunction")
         
     def add_shell_execution(self, source, inputs=[], outputs=[], arguments=[], includes=[], 
                             cmd_format=None, map=False, shell=None, collect=False, local=False):
-        shell_function  = ShellFunction( source, shell=shell, cmd_format=cmd_format )
+        shell_function  = ShellFunction(source, shell=shell, cmd_format=cmd_format, modules=self.modules)
        
         # if abstraction is map or multimap
         if map :
             # if input and output are list or filelist
             if issubclass(inputs.__class__, list) and issubclass(outputs.__class__, list) :
-                # arguments cannot be set with 
-                if arguments != [] :
-                    logging.getLogger("jflow").exception("add_shell_execution: '" + source + "' arguments parameter not allowed with map")
-                    raise Exception("add_shell_execution: '" + source + "' arguments parameter not allowed with map" )
-                MultiMap(shell_function,inputs=inputs, outputs=outputs, includes=includes, collect=collect, local=local)
+                MultiMap(shell_function,inputs=inputs, outputs=outputs, includes=includes, collect=collect, local=local, arguments=arguments)
             else :
                 logging.getLogger("jflow").exception("add_shell_execution: '" + source + "' map requires a list as inputs and output")
                 raise Exception("add_shell_execution: '" + source + "'  map requires a list as inputs and output")
@@ -614,19 +641,17 @@ class Component(object):
         trace_fh.close()
         
     def __write_element(self,fh, title, element):
-        logging.getLogger("jflow").debug("element = "+str(element))
         to_write=''
         if isinstance(element, list):
             if len (element)> 0 :
                 if isinstance(element[0], list):
                     for i in range(len(element)) : 
                         to_write+="List"+str(i+1)+": \n"
-                        to_write+="\n".join(element[i])+"\n"
+                        to_write+="\n".join([str(x) for x in element[i]])+"\n"
                 else : 
-                    to_write+="\n".join(element)+"\n"
+                    to_write+="\n".join([str(x) for x in element])+"\n"
         else :
             to_write+= str(element)+"\n"
         if to_write != "" :
             fh.write(title+" :\n")
             fh.write(to_write)
-            
\ No newline at end of file

src/jflow/config_reader.py

@@ -16,6 +16,7 @@
 #
 
 import os
+import re
 import sys
 import inspect
 import logging
@@ -29,7 +30,7 @@ class JFlowConfigReader(object):
     """
     
     CONFIG_FILE_PATH = "../../application.properties"
-    
+    USER_PATTERN = re.compile("###USER###")
     def __init__(self):
         """ 
         """
@@ -37,12 +38,18 @@ class JFlowConfigReader(object):
         self.reader.read(os.path.join(os.path.dirname(inspect.getfile(self.__class__)), self.CONFIG_FILE_PATH))
 
     def get_tmp_directory(self):
-        if not os.path.isdir(self.reader.get("storage", "tmp_directory").replace("###USER###",os.getenv("USER"))):
-            os.makedirs(self.reader.get("storage", "tmp_directory").replace("###USER###",os.getenv("USER")), 0o751)
-        return self.reader.get("storage", "tmp_directory").replace("###USER###",os.getenv("USER"))
-        
+        tmp_dir=self.reader.get("storage", "tmp_directory")
+        if self.USER_PATTERN.search(self.reader.get("storage", "tmp_directory")) is not None :
+            tmp_dir=tmp_dir.replace("###USER###",os.getenv("USER"))
+        if not os.path.isdir(tmp_dir):
+            os.makedirs(tmp_dir, 0o751)
+        return tmp_dir
+
     def get_work_directory(self):
-        return self.reader.get("storage", "work_directory").replace("###USER###",os.getenv("USER"))
+        if self.USER_PATTERN.search(self.reader.get("storage", "work_directory")) is None :
+            return self.reader.get("storage", "work_directory")
+        else:
+            return self.reader.get("storage", "work_directory").replace("###USER###",os.getenv("USER"))
     
     def get_exec(self, software):
         try:
@@ -59,9 +66,13 @@ class JFlowConfigReader(object):
           @return: the path to the log file
         """
         try:
-            return self.reader.get('storage', 'log_file').replace("###USER###",os.getenv("USER"))
+            if self.USER_PATTERN.search(self.reader.get("storage", "log_file")) is None :
+                return self.reader.get('storage', 'log_file')
+            else :
+                return self.reader.get('storage', 'log_file').replace("###USER###",os.getenv("USER"))
         except :
             raise NoOptionError("Failed when parsing the config file, no section logging found!")
+
         
     def get_makeflow_path(self):
         try:
@@ -118,11 +129,24 @@ class JFlowConfigReader(object):
             return self.reader.get("components", component_class+".batch_options")
         except:
             return ""
+
+    def get_component_modules(self, component_class):
+        try:
+            return self.reader.get("components", component_class+".modules").split(",")
+        except:
+            return []
     
     def get_workflow_group(self, workflow_class):
         try:
             return self.reader.get("workflows", workflow_class+".group")
         except:
             return ""
-        
-         
\ No newline at end of file
+
+    def get_browse_root_dir(self):
+        return self.reader.get("storage", "browse_root_dir")
+
+    def get_debug(self):
+        try:
+            return self.reader.get("global", "debug") == "True"
+        except NoOptionError:
+            return False

src/jflow/exceptions.py

+#
+# Copyright (C) 2015 INRA
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+
+
+class InvalidFormatError(Exception): 
+    pass
+
+class RuleException (Exception):
+    pass
+
+class RuleIgnore (Exception):
+    pass

src/jflow/featureiolib/gff3.py

@@ -15,7 +15,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 
-import sys, re
+import re
 
 class GFF3Record:
     """
@@ -171,7 +171,7 @@ class GFF3IO:
         for line in self._handle:
             line = line.rstrip()
             self._line += 1
-            if line.startswith('#') :
+            if line.startswith('#') or line == "":
                 continue
             try:
                 gff_record = GFF3Record.fromGff(line)

src/jflow/rules.py

+#
+# Copyright (C) 2015 INRA
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+
+import re
+
+from abc import ABC, abstractmethod
+
+from jflow.parameter import *
+
+from jflow.exceptions import RuleException, RuleIgnore
+
+
+#################
+# Basic classes #
+#################
+
+
+class SimpleRule (ABC):
+
+    def get_parameter(self, src_arg):
+        if ">" not in src_arg:
+            wf_parameter = getattr(self.wf_instance, src_arg)
+        else:
+            wf_parameter = getattr(self.wf_instance, src_arg[:src_arg.index(">")]).\
+                get_sub_parameters_by_name()[src_arg[src_arg.index(">")+1:]][0]
+        return wf_parameter
+
+    def __init__(self, user_args, wf_instance, src_arg, nb_rows):
+        self.user_args = user_args
+        self.wf_instance = wf_instance
+        self.parameter_name = src_arg
+        self.parameter_value = user_args[src_arg]
+        self.nb_rows = nb_rows
+        if ">" not in src_arg:
+            self.wf_parameter = getattr(wf_instance, src_arg)
+        else:
+            self.wf_parameter = getattr(wf_instance, src_arg[:src_arg.index(">")]).\
+                get_sub_parameters_by_name()[src_arg[src_arg.index(">")+1:]][0]
+
+        self.is_file_list = isinstance(self.wf_parameter, InputFileList)
+        self.is_a_file = isinstance(self.wf_parameter, InputFile)
+        self.is_file = self.is_a_file or self.is_file_list
+        self.is_directory = isinstance(self.wf_parameter, InputDirectory)
+        self.is_file_or_directory = self.is_file or self.is_directory
+
+    def check_allowed_types(self, allowed_types):
+        if self.parameter_name.find(">") == -1:
+            parameter_obj = getattr(self.wf_instance, self.parameter_name)
+        else:
+            parts = self.parameter_name.split(">")
+            parent_name = parts[0]
+            sub_param_name = parts[1]
+            parent_obj = getattr(self.wf_instance, parent_name)
+            parameter_obj = parent_obj.get_sub_parameters_by_name()[sub_param_name]
+        good_type = False
+        for type_param in allowed_types:
+            if isinstance(parameter_obj, type_param):
+                good_type = True
+                break
+        if not good_type:
+            self.warning("Rule " + type(self).__name__ + " ignored on parameter " + self.parameter_name + ": rule not "
+                         "allowed here")
+            raise RuleIgnore()
+
+
+    @staticmethod
+    def error(message):
+        raise RuleException(message)
+
+    @staticmethod
+    def warning(message):
+        print("\033[93mWarning: " + message + "\033[0m")
+
+    @abstractmethod
+    def check(self):
+        pass
+
+
+class LinkRule (SimpleRule):
+    def __init__(self, user_args, wf_instance, src_arg, targets_args, nb_rows):
+        SimpleRule.__init__(self, user_args, wf_instance, src_arg, nb_rows)
+        self.require_src = True
+        self.targets_args = targets_args
+
+    @abstractmethod
+    def check(self):
+        pass
+
+class ValueRule (SimpleRule):
+    def __init__(self, user_args, wf_instance, src_arg, values_arg, nb_rows):
+        SimpleRule.__init__(self, user_args, wf_instance, src_arg, nb_rows)
+        values = values_arg.replace("\,", "###virgule###")
+        values = values.split(",")