Mercurial > repos > yufei-luo > s_mart

comparison commons/pyRepetUnit/hmmer/tests/TestAcceptanceHmmpfamAndParse2alignInparallel.py @ 18:94ab73e8a190

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Uploaded
author m-zytnicki
date Mon, 29 Apr 2013 03:20:15 -0400
parents
children
comparison
equal deleted inserted replaced
17:b0e8584489e6 18:94ab73e8a190
1 import os
2 import sys
3 import unittest
4 import pyRepet.seq.fastaDB
5 from sets import Set
6 import ConfigParser
7 from pyRepet.launcher.programLauncher import *
8 from pyRepet.launcher.Launcher import HmmpfamAndParse2alignLauncher
9 from pyRepet.sql.RepetJobMySQL import *
10 from pyRepet.util.file.FileUtils import *
11 import commons.pyRepetUnit.dbSplit.LauncherDbSplit
12 import commons.pyRepetUnit.align.hmmOutputParsing.HmmpfamOutput2align
13
14 CONFIG_FILE = "datas/configTestAcceptanceHmmpfamAndParse2alignLauncherInparallel.cfg"
15 CURRENT_DIR = os.getcwd()
16
17 class TestAcceptanceHmmpfamLauncherAndParse2alignInparallel(unittest.TestCase):
18
19 def setUp(self):
20 self._config = ConfigParser.ConfigParser()
21 self._config.readfp( open(CONFIG_FILE) )
22 self._batchFilesList = []
23 self._batchFilesBatchesList = []
24 self._inputFile = self._config.get("dbSplit_config", "input_file")
25 self._programLauncherInstance = programLauncher()
26 self._outputFileNotInparallel = self._config.get("hmmer_config", "output_file")
27 self._hmmpfamOutput2align = commons.pyRepetUnit.align.hmmOutputParsing.HmmpfamOutput2align.HmmpfamOutput2align()
28 os.chdir(CURRENT_DIR)
29
30 def _launchHmmpfamAndParseNotInParallel(self):
31 self._programLauncherInstance.reset(self._inputFile)
32 self._programLauncherInstance.setOutputFileName(self._outputFileNotInparallel)
33 self._programLauncherInstance.launchHmmpfam(evalFilter=self._config.get("profil_search", "Evalue"), inputFormat=self._config.get("profil_search", "InputFormat"), profilDatabank=self._config.get("profil_search", "ProfilDatabank"))
34 self._hmmpfamOutput2align.setInputFile(self._outputFileNotInparallel)
35 self._outputFileParsingNotInparallel = self._config.get("hmmpfam2align_config", "output_file")
36 self._hmmpfamOutput2align.setOutputFile(self._outputFileParsingNotInparallel)
37 self._hmmpfamOutput2align.run()
38
39
40 def _countNblinesInAllResultsFilesInParallel(self, fileUtils):
41 result_dir = self._config.get("hmmer_config", "result_dir")
42 dirFiles = fileUtils.listFilesInDir(result_dir)
43 listPath = []
44 for file in dirFiles:
45 listPath.append(file)
46
47 nbLinesInAlignFilesInparallel = fileUtils.countLinesInFiles(listPath)
48 return nbLinesInAlignFilesInparallel
49
50
51 def _launchHmmpfamAndParseInParallel(self):
52 user = self._config.get("db_config", "user")
53 host = self._config.get("db_config", "host")
54 passwd = self._config.get("db_config", "passwd")
55 dbName = self._config.get("db_config", "dbname")
56 jobDb = RepetJob(user, host, passwd, dbName)
57 params = {"param": "--informat " + self._config.get("profil_search", "InputFormat") + " -E " + self._config.get("profil_search", "Evalue"), "profilDB": self._config.get("profil_search", "ProfilDatabank"), "outputDir": self._config.get("hmmer_config", "result_dir"), "query": self._config.get("hmmer_config", "query_dir"), "job_table": "jobs", "queue": self._config.get("sge_config", "queue"), "groupid": self._config.get("sge_config", "groupid"), "tmpDir": self._config.get("hmmer_config", "tmp_dir"), "scriptToLaunch": self._config.get("parse_config", "scriptToLaunch"), "cDir" : "."}
58 self.hmmpfamAndParse2align = HmmpfamAndParse2alignLauncher(jobDb, params)
59 self.hmmpfamAndParse2align.run()
60
61
62 def _splitInputFile(self, dbSplit):
63 dbSplit.setInFileName(self._inputFile)
64 dbSplit.setIsNewDir(True)
65 dbSplit.setNbSequences(3)
66 dbSplit.run()
67
68 def testAcceptanceHmmpfamAndParse2alignHasRunInparallel (self):
69
70 dbSplit = commons.pyRepetUnit.dbSplit.LauncherDbSplit.LauncherDbSplit()
71 self._splitInputFile(dbSplit)
72 self._launchHmmpfamAndParseInParallel()
73 fileUtils = FileUtils()
74 nbLinesInAlignFilesInparallel = self._countNblinesInAllResultsFilesInParallel(fileUtils)
75 dbSplit.clean()
76 self._launchHmmpfamAndParseNotInParallel()
77 nbLinesInAlignFileNotInparallel = fileUtils.countLinesInAFile(self._outputFileParsingNotInparallel)
78 self.assertEqual(nbLinesInAlignFilesInparallel,nbLinesInAlignFileNotInparallel)
79 self._clean()
80
81 def _clean(self):
82
83 dirToClean = self._config.get( "hmmer_config", "tmp_dir" )
84 os.system("rm " + dirToClean + "/*")
85 dirToClean = self._config.get( "hmmer_config", "result_dir" )
86 os.system("rm " + dirToClean + "/*")
87 os.system("rm HmmpfamJob*.e*")
88 os.system("rm HmmpfamJob*.o*")
89 os.remove(self._outputFileNotInparallel)
90 os.remove(self._outputFileParsingNotInparallel)
91
92
93 if __name__ == "__main__":
94 unittest.main()