No commit message

No commit message

src/ng6/analysis.py

@@ -122,9 +122,6 @@ class Analysis (Component):
                 if not os.path.isdir(save_dir) and not os.path.isdir(work_dir):
                     break
                 directory_name = uuid.uuid4().hex[:9]
-            # Then make directories
-            work_dir = os.path.join(ng6conf.get_work_directory()+self.DIRECTORIES_STRUCTURE, directory_name)
-            os.makedirs(work_dir, 0751)
             self.directory = os.path.join(self.DIRECTORIES_STRUCTURE, directory_name)
             logging.getLogger("Analysis.__init__").debug("Building brand new analysis [" + str(self) + "]")
     
@@ -144,8 +141,11 @@ class Analysis (Component):
         # first create the output directory
         if not os.path.isdir(self.output_directory):
             os.makedirs(self.output_directory, 0751)
-            
-        # first add analysis information
+        # and analysis output
+        if not os.path.isdir(self.__get_work_directory()):
+            os.makedirs(self.__get_work_directory(), 0751)
+
+        # then add analysis information
         self.define_analysis()
         
         # serialized the object
@@ -182,24 +182,26 @@ class Analysis (Component):
         """
         Synchronize data related to the Analysis between temporary folder and long term storage folder.
         """
-        logging.getLogger("Analysis.sync").debug("Synchronizing analysis id=" + str(self.id) + " from " + self.__get_work_directory() + " to " + self.__get_save_directory())
-        try :
-            # First make the save directory
-            os.makedirs(self.__get_save_directory(), 0751)
-            if os.path.exists(self.__get_work_directory()) and self.__get_work_directory() != self.__get_save_directory():
-                cmd = "rsync -avh --remove-sent-files "+ self.__get_work_directory() + "/ " + self.__get_save_directory()
-                logging.getLogger("Analysis.sync").debug(cmd)
-                retcode = call(["rsync", "-avh", "--remove-sent-files",self.__get_work_directory() + "/", self.__get_save_directory()], shell=False)
-                if retcode < 0:
-                    logging.getLogger("Analysis.sync").error("Error when synchronizing analysis id=" + str(self.id) + "(retcode=" + str(retcode) + ")")
+        if not os.path.isdir(self.__get_save_directory()) and len(os.listdir(self.__get_work_directory())):
+            logging.getLogger("Analysis.sync").debug("Synchronizing analysis id=" + str(self.id) + " from " + self.__get_work_directory() + " to " + self.__get_save_directory())
+            try :
+                # First make the save directory
+                os.makedirs(self.__get_save_directory(), 0751)
+                if os.path.exists(self.__get_work_directory()) and self.__get_work_directory() != self.__get_save_directory():
+                    cmd = "rsync -avh --remove-sent-files "+ self.__get_work_directory() + "/ " + self.__get_save_directory()
+                    logging.getLogger("Analysis.sync").debug(cmd)
+                    retcode = call(["rsync", "-avh", "--remove-sent-files",self.__get_work_directory() + "/", self.__get_save_directory()], shell=False)
+                    if retcode < 0:
+                        logging.getLogger("Analysis.sync").error("Error when synchronizing analysis id=" + str(self.id) + "(retcode=" + str(retcode) + ")")
+                    else:
+                        try: os.rmdir(self.__get_work_directory())
+                        except: pass
+                        logging.getLogger("Analysis.sync").debug("Synchronization done for analysis id=" + str(self.id) + "(retcode=" + str(retcode) + ")")
                 else:
-                    logging.getLogger("Analysis.sync").debug("Synchronization done for analysis id=" + str(self.id) + "(retcode=" + str(retcode) + ")")
-            else:
-                logging.getLogger("Analysis.sync").info("The synchronisation has not been performed, either because " + self.__get_work_directory() + " does not exists or because the source and destination folders are the same.")
-        except Exception, err:
-            raise Exception(str(err))
-
-
+                    logging.getLogger("Analysis.sync").info("The synchronisation has not been performed, either because " + self.__get_work_directory() + " does not exists or because the source and destination folders are the same.")
+            except Exception, err:
+                raise Exception(str(err))
+        
     def list_or_copy_analysis_files(self, outputdir, pattern):
         """
         Return list of file corresponding to the pattern and copy unzipped files into output dir if it's zipped

src/ng6/run.py

@@ -230,20 +230,24 @@ class Run(object):
         except Exception, err:
             logging.getLogger("Run.sync").error("Error when synchronizing analysis that belongs to run id=" + str(self.id))
             raise Exception(str(err))
-        # Finally the Run itself
-        try :
-            # First make the save directory
-            os.makedirs(self.__get_save_directory(), 0751)            
-            if os.path.exists(self.__get_work_directory()) and self.__get_work_directory() != self.__get_save_directory():
-                cmd = "rsync -avh --remove-sent-files "+ self.__get_work_directory() + "/ " + self.__get_save_directory()
-                logging.getLogger("Run.sync").debug(cmd)
-                retcode = call(["rsync", "-avh", "--remove-sent-files", self.__get_work_directory() + "/", self.__get_save_directory()], shell=False)
-                if retcode < 0:
-                    logging.getLogger("Run.sync").error("Error when synchronizing run id=" + str(self.id) + "(retcode=" + str(retcode) + ")")
-                else:
-                    logging.getLogger("Run.sync").debug("Synchronization done for run id=" + str(self.id) + "(retcode=" + str(retcode) + ")")
-        except Exception, err:
-            raise Exception(str(err))
+        # If a sync is required
+        if not os.path.isdir(self.__get_save_directory()) and len(os.listdir(self.__get_work_directory())):
+            # Finally the Run itself
+            try :
+                # First make the save directory
+                os.makedirs(self.__get_save_directory(), 0751)            
+                if os.path.exists(self.__get_work_directory()) and self.__get_work_directory() != self.__get_save_directory():
+                    cmd = "rsync -avh --remove-sent-files "+ self.__get_work_directory() + "/ " + self.__get_save_directory()
+                    logging.getLogger("Run.sync").debug(cmd)
+                    retcode = call(["rsync", "-avh", "--remove-sent-files", self.__get_work_directory() + "/", self.__get_save_directory()], shell=False)
+                    if retcode < 0:
+                        logging.getLogger("Run.sync").error("Error when synchronizing run id=" + str(self.id) + "(retcode=" + str(retcode) + ")")
+                    else:
+                        try: os.rmdir(self.__get_work_directory())
+                        except: pass
+                        logging.getLogger("Run.sync").debug("Synchronization done for run id=" + str(self.id) + "(retcode=" + str(retcode) + ")")
+            except Exception, err:
+                raise Exception(str(err))
 
     def add_analysis(self, my_analysis):
         """

workflows/addanalysis/__init__.py

+#
+# Copyright (C) 2012 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 os
+import sys
+
+from ng6.ng6workflow import NG6Workflow
+
+
+class AddAnalysis (NG6Workflow):
+    
+    def process(self):
+        """ 
+        Run the workflow
+        """
+        # First check if files provided exists
+        files_to_save = [] 
+        for file in self.args["data_file"]:
+            if os.path.isfile(file):
+                files_to_save.append(file)
+            else:
+                sys.exit("error: %s file does not exists\n" % (file))
+        # Check if user provided a project or a run
+        if self.project and self.runobj:
+            sys.exit("An analysis should be added to a run OR a project not both of them!\n")
+        elif self.project is None and self.runobj is None :
+            sys.exit("One of run_id or project_id parameter is required!\n")
+
+        addto = "run"
+        if self.project: addto = "project"
+        basicanalysis = self.add_component("BasicAnalyse", [files_to_save, self.args["analysis_name"], 
+                        self.args["analysis_description"], self.args["analysis_software"], self.args["analysis_software_options"], 
+                        self.args["analysis_software_version"], self.args["compression"], self.args["delete"], 
+                        self.args["archive_name"]], addto=addto)        
+        
\ No newline at end of file

workflows/addanalysis/workflow.properties

+#
+# Copyright (C) 2012 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/>.
+#
+
+[global]
+name = addanalysis
+description = add a brand new analysis to a run or a project
+
+#
+# Parameter section
+# param.name: the parameter display name
+#      .flag: the command line flag to use the argument
+#      .help: a brief description of what the parameter does
+#      .default [None]: the value produced if the parameter is not provided
+#      .type [str]: the parameter type that should be tested (str|int|date|file|bool)
+#      .choices [None]: a container of the allowable values for the parameter
+#      .required [False]: whether or not the command-line option may be omitted
+#      .action [store]: the basic type of action to be taken (store|append)
+#
+[parameters]
+data_file.name = data_file
+data_file.flag = --data-file
+data_file.help = Which data files should be added to the analysis
+data_file.required = True
+data_file.action = append
+
+compression.name = compression
+compression.flag = --compression
+compression.help = How should data be compressed once archived (none|gz|bz2)
+compression.default = none
+compression.choices = none|gz|bz2
+
+delete.name = delete
+delete.flag = --delete
+delete.help = Should the input data be deleted once the process is over
+delete.default = False
+delete.type = bool
+
+archive_name.name = archive_name
+archive_name.flag = --archive-name
+archive_name.help = Give a name to the final archive
+archive_name.default = ng6_archive
+
+project_id.name = project_id
+project_id.flag = --project-id
+project_id.help = The project id the analysis belongs to
+project_id.type = int
+
+run_id.name = run_id
+run_id.flag = --run-id
+run_id.help = The run id the run belongs to
+run_id.type = int
+
+analysis_name.name = analysis_name
+analysis_name.flag = --analysis-name
+analysis_name.help = Give a name to your analysis
+analysis_name.required = True
+
+analysis_description.name = analysis_description
+analysis_description.flag = --analysis-description
+analysis_description.help = Give a description to your analysis
+analysis_description.required = True
+
+analysis_software.name = analysis_software
+analysis_software.flag = --analysis-software
+analysis_software.help = Which software was used for this analysis
+analysis_software.required = True
+
+analysis_software_options.name = analysis_software_options
+analysis_software_options.flag = --analysis-software-options
+analysis_software_options.help = Which software options were used for this analysis
+analysis_software_options.required = True
+
+analysis_software_version.name = analysis_software_version
+analysis_software_version.flag = --analysis-software-version
+analysis_software_version.help = Which version of the software was used for this analysis
+analysis_software_version.required = True

workflows/components/basicanalysis.py

+#
+# Copyright (C) 2012 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/>.
+#
+
+from jflow.component import Component
+from jflow.iotypes import OutputFile, InputFileList, Formats
+
+from ng6.analysis import Analysis
+
+
+class BasicAnalyse (Analysis):
+    
+    def define_parameters(self, files, name, description, software, options, version, compression="none", 
+                          delete=False, archive_name=None):
+        self.files = files
+        self.name = name
+        self.description = description
+        self.software = software
+        self.options = options
+        self.version = version
+        self.compression = compression
+        self.delete = delete
+        self.archive_name = archive_name
+        
+    def define_analysis(self): pass
+        
+    def process(self): pass
+        
+    def get_version(self): return ""
+    
+    def post_process(self):
+        self._archive_files(self.files, self.compression, self.archive_name, self.delete)
+        
\ No newline at end of file