diff commons/tools/TEclassifierPE_parallelized.py @ 18:94ab73e8a190

Uploaded

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/commons/tools/TEclassifierPE_parallelized.py	Mon Apr 29 03:20:15 2013 -0400
@@ -0,0 +1,261 @@
+#!/usr/bin/env python
+
+# Copyright INRA (Institut National de la Recherche Agronomique)
+# http://www.inra.fr
+# http://urgi.versailles.inra.fr
+#
+# This software is governed by the CeCILL license under French law and
+# abiding by the rules of distribution of free software.  You can  use, 
+# modify and/ or redistribute the software under the terms of the CeCILL
+# license as circulated by CEA, CNRS and INRIA at the following URL
+# "http://www.cecill.info". 
+#
+# As a counterpart to the access to the source code and  rights to copy,
+# modify and redistribute granted by the license, users are provided only
+# with a limited warranty  and the software's author,  the holder of the
+# economic rights,  and the successive licensors  have only  limited
+# liability. 
+#
+# In this respect, the user's attention is drawn to the risks associated
+# with loading,  using,  modifying and/or developing or reproducing the
+# software by the user in light of its specific status of free software,
+# that may mean  that it is complicated to manipulate,  and  that  also
+# therefore means  that it is reserved for developers  and  experienced
+# professionals having in-depth computer knowledge. Users are therefore
+# encouraged to load and test the software's suitability as regards their
+# requirements in conditions enabling the security of their systems and/or 
+# data to be ensured and,  more generally, to use and operate it in the 
+# same conditions as regards security. 
+#
+# The fact that you are presently reading this means that you have had
+# knowledge of the CeCILL license and that you accept its terms.
+
+import os
+import sys
+import shutil
+
+if not "REPET_PATH" in os.environ.keys():
+    print "ERROR: no environment variable REPET_PATH"
+    sys.exit(1)
+sys.path.append(os.environ["REPET_PATH"])
+if not "PYTHONPATH" in os.environ.keys():
+    os.environ["PYTHONPATH"] = os.environ["REPET_PATH"]
+else:
+    os.environ["PYTHONPATH"] = "%s:%s" % (os.environ["REPET_PATH"], os.environ["PYTHONPATH"])
+    
+from commons.core.LoggerFactory import LoggerFactory
+from commons.core.utils.RepetOptionParser import RepetOptionParser
+from commons.core.utils.FileUtils import FileUtils
+from commons.core.checker.ConfigChecker import ConfigRules
+from commons.core.checker.ConfigChecker import ConfigChecker
+from commons.core.seq.FastaUtils import FastaUtils
+from commons.core.sql.DbFactory import DbFactory
+from commons.core.sql.TableJobAdaptatorFactory import TableJobAdaptatorFactory
+from commons.core.launcher.Launcher import Launcher
+from denovo_pipe.ReverseComplementAccordingToClassif import ReverseComplementAccordingToClassif
+from denovo_pipe.DetectTEFeatures_parallelized import DetectTEFeatures_parallelized
+from denovo_pipe.RenameHeaderClassif import RenameHeaderClassif
+from denovo_pipe.LaunchPASTEC import LaunchPASTEC
+from PASTEC.StatPastec import StatPastec
+
+LOG_DEPTH = "repet.tools"
+#LOG_FORMAT = "%(message)s"
+
+####TEclassifier PASTEC Edition - parallelized
+#
+class TEclassifierPE_parallelized(object):
+    
+    def __init__(self, fastaFileName = "", configFileName = "", addWickerCode = False, reverseComp = False, doClean = False, verbosity = 0):
+        self._fastaFileName = fastaFileName
+        self._addWickerCode = addWickerCode
+        self._reverseComp = reverseComp
+        self._configFileName = configFileName
+        self._doClean = doClean
+        self._verbosity = verbosity
+        self._projectName = ""
+        self._log = LoggerFactory.createLogger("%s.%s" % (LOG_DEPTH, self.__class__.__name__), self._verbosity)
+        
+    def setAttributesFromCmdLine(self):
+        description = "TE classifier PASTEC Edition.\n"
+        description += "Detect TE features on consensus and classify them. Give some classification statistics.\n"
+        description += "Can rename headers with classification info and Wicker's code at the beginning.\n"
+        description += "Can reverse-complement consensus if they are detected in reverse strand.\n"
+        epilog = "\n"
+        epilog += "Example 1: launch and clean temporary files\n"
+        epilog += "\t$ python TEclassifierPE.py -i consensus.fa -C TEclassifier.cfg -c\n"
+        epilog += "\n"
+        epilog += "Example 2: launch with 'rename headers' and 'reverse-complement' options\n"
+        epilog += "\t$ python TEclassifierPE.py -i consensus.fa -C TEclassifier.cfg -c -w -r\n"
+        parser = RepetOptionParser(description = description, epilog = epilog)
+        parser.add_option("-i", "--fasta",      dest = "fastaFileName", action = "store",       type = "string", help = "input fasta file name [compulsory] [format: fasta]", default = "")
+        parser.add_option("-C", "--config",     dest = "configFileName",action = "store",       type = "string", help = "configuration file name (e.g. TEclassifier.cfg) [compulsory]", default = "")
+        parser.add_option("-w", "--wicker",     dest = "addWickerCode", action = "store_true",                   help = "add classification info and Wicker's code at the beginning of the headers [optional] [default: False]", default = False)
+        parser.add_option("-r", "--reverse",    dest = "reverseComp",   action = "store_true",                   help = "reverse-complement consensus if they are detected in reverse strand [optional] [default: False]", default = False)
+        parser.add_option("-c", "--clean",      dest = "doClean",       action = "store_true",                   help = "clean temporary files [optional] [default: False]", default = False)
+        parser.add_option("-v", "--verbosity",  dest = "verbosity",     action = "store",       type = "int",    help = "verbosity [optional] [default: 3, from 1 to 4]", default = 3)
+        options = parser.parse_args()[0]
+        self._setAttributesFromOptions(options)
+        
+    def _setAttributesFromOptions(self, options):
+        self.setFastaFileName(options.fastaFileName)
+        self.setAddWickerCode(options.addWickerCode)
+        self.setReverseComp(options.reverseComp)
+        self.setConfigFileName(options.configFileName)
+        self.setDoClean(options.doClean)
+        self.setVerbosity(options.verbosity)
+
+    def _checkConfig(self):       
+        iConfigRules = ConfigRules()
+        iConfigRules.addRuleOption(section="project", option ="project_name", mandatory=True, type="string")
+        sectionName = "classif_consensus"
+        iConfigRules.addRuleOption(section=sectionName, option ="clean", mandatory=True, type="bool")
+        iConfigRules.addRuleOption(section=sectionName, option ="limit_job_nb", type="int")
+        iConfigRules.addRuleOption(section=sectionName, option ="resources", type="string")
+        iConfigRules.addRuleOption(section=sectionName, option ="tmpDir", type="string")
+        iConfigChecker = ConfigChecker(self._configFileName, iConfigRules)
+        iConfig = iConfigChecker.getConfig()
+        self._setAttributesFromConfig(iConfig)
+        
+    def _setAttributesFromConfig(self, iConfig):
+        self.setProjectName(iConfig.get("project", "project_name"))
+        sectionName = "classif_consensus"
+        self.setDoClean(iConfig.get(sectionName, "clean"))
+        self._maxJobNb = iConfig.get(sectionName, "limit_job_nb")
+        self._resources = iConfig.get(sectionName, "resources")
+        self._tmpDir = iConfig.get(sectionName, "tmpDir")
+        
+    def setFastaFileName(self, fastaFileName):
+        self._fastaFileName = fastaFileName
+        
+    def setConfigFileName(self, configFileName):
+        self._configFileName = configFileName
+        
+    def setAddWickerCode(self, addWickerCode):
+        self._addWickerCode = addWickerCode
+        
+    def setReverseComp(self, reverseComp):
+        self._reverseComp = reverseComp
+        
+    def setDoClean(self, doClean):
+        self._doClean = doClean
+        
+    def setVerbosity(self, verbosity):
+        self._verbosity = verbosity
+        
+    def setProjectName(self, projectName):
+        self._projectName = projectName
+        
+    def _checkOptions(self):
+        if self._fastaFileName == "":
+            self._logAndRaise("ERROR: Missing input fasta file name")
+            
+    def _logAndRaise(self, errorMsg):
+        self._log.error(errorMsg)
+        raise Exception(errorMsg)
+    
+#    def setup_env(config):
+#        os.environ["REPET_HOST"] = config.get("repet_env", "repet_host")
+#        os.environ["REPET_USER"] = config.get("repet_env", "repet_user")
+#        os.environ["REPET_PW"] = config.get("repet_env", "repet_pw")
+#        os.environ["REPET_DB"] = config.get("repet_env", "repet_db")
+#        os.environ["REPET_PORT"] = config.get("repet_env", "repet_port")
+#        os.environ["REPET_JOB_MANAGER"] = config.get("repet_env", "repet_job_manager")
+#        os.environ["REPET_QUEUE"] = config.get("repet_env", "repet_job_manager")
+#        os.environ["REPET_JOBS"] = "MySQL"
+    
+    def getPASTECcommand(self, iLauncher, fileName):
+        lArgs = []
+        lArgs.append("-C %s" % self._configFileName)
+        lArgs.append("-P %s" % self._projectName)
+        lArgs.append("-S 2")
+        lArgs.append("-i %s" % fileName)
+        lArgs.append("-v %s" % self._verbosity)
+        return iLauncher.getSystemCommand("LaunchPASTEC.py", lArgs)
+                    
+    def run(self):
+        LoggerFactory.setLevel(self._log, self._verbosity)
+        if self._configFileName:
+            self._checkConfig()
+        self._checkOptions()
+        self._log.info("START TEclassifier PASTEC Edition")
+        self._log.debug("Fasta file name: %s" % self._fastaFileName)
+        nbSeq = FastaUtils.dbSize(self._fastaFileName)
+        self._log.debug("Total number of sequences: %i)" % nbSeq)
+
+        self._log.debug("Launch DetectTEFeatures on each batch")
+        iDF = DetectTEFeatures_parallelized(self._fastaFileName, self._projectName, self._configFileName, self._doClean, self._verbosity)
+        iDF.run()
+        
+        self._log.debug("Insert banks in database")
+        iLP = LaunchPASTEC(self._configFileName, "1", projectName = self._projectName, verbose = self._verbosity)
+        iLP.run()
+        
+        self._log.info("Split fasta file")
+        if self._maxJobNb == 0 or nbSeq / self._maxJobNb <= 1.0:
+            nbSeqPerBatch = nbSeq
+        else:
+            nbSeqPerBatch = nbSeq / self._maxJobNb + 1
+        FastaUtils.dbSplit(self._fastaFileName, nbSeqPerBatch, True, verbose = self._verbosity - 2)
+        
+        self._log.info("Launch PASTEC on each batch")
+        queue = self._resources
+        cDir = os.getcwd()
+        if self._tmpDir != "":
+            tmpDir = self._tmpDir
+        else:
+            tmpDir = cDir
+            
+        #TODO: allow not to parallelize
+        groupid = "%s_PASTEC" % self._projectName
+        acronym = "PASTEC"
+        iDb = DbFactory.createInstance()
+        iTJA = TableJobAdaptatorFactory.createInstance(iDb, "jobs")
+        iLauncher = Launcher(iTJA, os.getcwd(), "", "", cDir, tmpDir, "jobs", queue, groupid)
+        lCmdsTuples = []
+        lFiles = FileUtils.getFileNamesList("%s/batches" % cDir, "batch_")
+        if len(lFiles) == 0:
+            self._logAndRaise("ERROR: directory 'batches' is empty")
+        classifFileName = "%s.classif" % self._projectName
+        count = 0
+        for file in lFiles:
+            count += 1
+            lCmds = [self.getPASTECcommand(iLauncher, file)]    
+            lCmdStart = []
+            lCmdStart.append("shutil.copy(\"%s/batches/%s\", \".\")" % (cDir, file))
+            lCmdStart.append("shutil.copy(\"%s/%s\", \".\")" % (cDir, self._configFileName))
+            lCmdFinish = []
+            lCmdFinish.append("shutil.move(\"%s\", \"%s/%s_%i\")" % (classifFileName, cDir, classifFileName, count))
+            lCmdsTuples.append(iLauncher.prepareCommands_withoutIndentation(lCmds, lCmdStart, lCmdFinish))
+        iLauncher.runLauncherForMultipleJobs(acronym, lCmdsTuples, self._doClean)  
+        
+        FileUtils.catFilesByPattern("%s_*" % classifFileName, classifFileName)
+        if self._doClean:
+            FileUtils.removeFilesByPattern("%s_*" % classifFileName)
+            shutil.rmtree("batches")
+
+        self._log.debug("Compute stats about classification")
+        iSP = StatPastec(classifFileName)
+        iSP.run()
+        
+        if self._reverseComp:
+            self._log.debug("Reverse complement")
+            iRevComplAccording2Classif = ReverseComplementAccordingToClassif()
+            iRevComplAccording2Classif.setFastaFile(self._fastaFileName)
+            iRevComplAccording2Classif.setClassifFile(classifFileName)
+            iRevComplAccording2Classif.run()
+            newFastaFileName = "%s_negStrandReversed.fa" % os.path.splitext(self._fastaFileName)[0]
+        else:
+            newFastaFileName = self._fastaFileName
+
+        if self._addWickerCode:
+            self._log.debug("Rename headers according to Wicker's code")
+            iRHC = RenameHeaderClassif(classifFileName, newFastaFileName, self._projectName)
+            iRHC.setOutputFileName("")
+            iRHC.run()
+        
+        self._log.info("END TEclassifier PASTEC Edition")
+
+if __name__ == "__main__":
+    iLaunch = TEclassifierPE_parallelized()
+    iLaunch.setAttributesFromCmdLine()
+    iLaunch.run()        
\ No newline at end of file