Mercurial > repos > yufei-luo > s_mart

comparison commons/core/checker/CheckerUtils.py @ 6:769e306b7933

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Change the repository level.
author yufei-luo
date Fri, 18 Jan 2013 04:54:14 -0500
parents
children
comparison
equal deleted inserted replaced
5:ea3082881bf8 6:769e306b7933
1 # Copyright INRA (Institut National de la Recherche Agronomique)
2 # http://www.inra.fr
3 # http://urgi.versailles.inra.fr
4 #
5 # This software is governed by the CeCILL license under French law and
6 # abiding by the rules of distribution of free software. You can use,
7 # modify and/ or redistribute the software under the terms of the CeCILL
8 # license as circulated by CEA, CNRS and INRIA at the following URL
9 # "http://www.cecill.info".
10 #
11 # As a counterpart to the access to the source code and rights to copy,
12 # modify and redistribute granted by the license, users are provided only
13 # with a limited warranty and the software's author, the holder of the
14 # economic rights, and the successive licensors have only limited
15 # liability.
16 #
17 # In this respect, the user's attention is drawn to the risks associated
18 # with loading, using, modifying and/or developing or reproducing the
19 # software by the user in light of its specific status of free software,
20 # that may mean that it is complicated to manipulate, and that also
21 # therefore means that it is reserved for developers and experienced
22 # professionals having in-depth computer knowledge. Users are therefore
23 # encouraged to load and test the software's suitability as regards their
24 # requirements in conditions enabling the security of their systems and/or
25 # data to be ensured and, more generally, to use and operate it in the
26 # same conditions as regards security.
27 #
28 # The fact that you are presently reading this means that you have had
29 # knowledge of the CeCILL license and that you accept its terms.
30
31
32 import os
33 import sys
34 import re
35 import glob
36 import ConfigParser
37 from ConfigParser import NoOptionError
38 from ConfigParser import NoSectionError
39 from commons.core.checker.CheckerException import CheckerException
40
41
42 ## A set of static methods used to perform checks.
43 #
44 #
45 class CheckerUtils( object ):
46
47 ## Check if blastName param is in ["blastn", "blastp", "blastx", "tblastn", "tblastx"]
48 #
49 # @param blastName name to check
50 # @return True if name is in list False otherwise
51 #
52 def isBlastNameNotInBlastValues( blastName ):
53 blastValuesSet = set( ["blastn", "blastp", "blastx", "tblastn", "tblastx"] )
54 blastNameSet = set( [ blastName ] )
55 return not blastNameSet.issubset( blastValuesSet )
56
57 isBlastNameNotInBlastValues = staticmethod( isBlastNameNotInBlastValues )
58
59
60 ## Check if param is NOT "TRUE" and NOT false "FALSE"
61 #
62 # @param param str to check
63 # @return True if param is not eq to "TRUE" AND not eq to "FALSE", false otherwise
64 #
65 def isNotTRUEisNotFALSE( param ):
66 return param != "TRUE" and param != "FALSE"
67
68 isNotTRUEisNotFALSE = staticmethod( isNotTRUEisNotFALSE )
69
70
71 ## Check if resource (file or dir) do NOT exists
72 #
73 # @param resource file or dir to check
74 # @return True if resource exists False otherwise
75 #
76 def isRessourceNotExits( resource ):
77 return not os.path.exists( resource )
78
79 isRessourceNotExits = staticmethod( isRessourceNotExits )
80
81
82 ## Check a specific E-value format: de-dd
83 #
84 # @param param E-value to check
85 # @return True if format is de-dd False otherwise
86 #
87 def isNotAeValueWithOneDigit2DecimalsAtLeast( param ):
88 # \d\d stands for 2 digits and more ???
89 return not re.match( "\de\-\d\d", param )
90
91 isNotAeValueWithOneDigit2DecimalsAtLeast = staticmethod( isNotAeValueWithOneDigit2DecimalsAtLeast )
92
93
94 ## Check a number format
95 #
96 # @param param value to check
97 # @return True if param is a number (d+) False otherwise
98 #
99 def isNotANumber( param ):
100 return not re.match( "\d+", param )
101
102 isNotANumber = staticmethod( isNotANumber )
103
104
105 ## Check if an executable is in the user's PATH
106 #
107 # @param exeName name of the executable
108 # @return True if executable in user's PATH, False otherwise
109 #
110 def isExecutableInUserPath( exeName ):
111 dirPathList = os.environ["PATH"].split(":")
112 for dirPath in dirPathList:
113 if os.path.isdir( dirPath ):
114 try:
115 binPathList = glob.glob( dirPath + "/*" )
116 except OSError, e:
117 continue
118 for binPath in binPathList:
119 bin = os.path.basename( binPath )
120 if bin == exeName:
121 return True
122 return False
123
124 isExecutableInUserPath = staticmethod( isExecutableInUserPath )
125
126
127 ## Return the full path of a given executable
128 #
129 def getFullPathFromExecutable( exeName ):
130 lDirFromUserPath = os.environ["PATH"].split(":")
131 for dir in lDirFromUserPath:
132 if os.path.isdir( dir ):
133 try:
134 lExecutables = glob.glob( "%s/*" % ( dir ) )
135 except OSError, e:
136 continue
137 for exe in lExecutables:
138 path, exe = os.path.split( exe )
139 if exe == exeName:
140 return path
141 return ""
142
143 getFullPathFromExecutable = staticmethod( getFullPathFromExecutable )
144
145
146 #TODO: to remove ?
147 ## Check if a queue Name is valid. Warning: Only with the queue manager SGE
148 #
149 # @param fullQueueName name of the queue to test (with or without parameters)
150 # @return True if queue name is valid, False otherwise
151 #
152 def isQueueNameValid( fullQueueName ):
153 queueName = fullQueueName.split()[0]
154 if queueName == "none":
155 return True
156 queueFile = "queueName.txt"
157 if not CheckerUtils.isExecutableInUserPath( "qconf" ):
158 msg = "executable 'qconf' can't be found"
159 sys.stderr.write( "%s\n" % ( msg ) )
160 return False
161 cmd = "qconf -sql > " + queueFile
162 os.system( cmd )
163 queueFileHandler = open( queueFile, "r" )
164 lQueueNames = queueFileHandler.readlines()
165 queueFileHandler.close()
166 os.remove( queueFile )
167 queueNameValid = False
168 for qName in lQueueNames:
169 qName = qName.strip()
170 if qName == queueName:
171 queueNameValid = True
172 break
173 return queueNameValid
174
175 isQueueNameValid = staticmethod( isQueueNameValid )
176
177
178 ## Check if a string length is lower or equal than 15
179 #
180 # @param strName any string
181 # @return True if string length is <= 15, False otherwise
182 #
183 def isMax15Char( strName ):
184 return (len(strName) <= 15 )
185
186 isMax15Char = staticmethod( isMax15Char )
187
188
189 ## Check if a string is made with only alphanumeric or underscore character
190 #
191 # @param strName any string
192 # @return True if string is with alphanumeric or underscore, False otherwise
193 #
194 def isCharAlphanumOrUnderscore( strName ):
195 # authorized ALPHABET [a-z,A-Z,0-9,_]
196 p = re.compile('\W')
197 errList=p.findall(strName)
198 if len( errList ) > 0 :
199 return False
200 else:
201 return True
202
203 isCharAlphanumOrUnderscore = staticmethod( isCharAlphanumOrUnderscore )
204
205
206 ## Check if sectionName is in the configuration file
207 #
208 # @param config filehandle of configuration file
209 # @param sectionName string of section name to check
210 # @exception NoSectionError: if section not found raise a NoSectionError
211 #
212 def checkSectionInConfigFile( config, sectionName ):
213 if not (config.has_section(sectionName)):
214 raise NoSectionError(sectionName)
215
216 checkSectionInConfigFile = staticmethod( checkSectionInConfigFile )
217
218
219 ## Check if an option is in a specified section in the configuration file
220 #
221 # @param config filehandle of configuration file
222 # @param sectionName string of section name
223 # @param optionName string of option name to check
224 # @exception NoOptionError: if option not found raise a NoOptionError
225 #
226 def checkOptionInSectionInConfigFile( config, sectionName, optionName ):
227 config.get( sectionName, optionName )
228
229 checkOptionInSectionInConfigFile = staticmethod( checkOptionInSectionInConfigFile )
230
231
232 ## Check version number coherency between configFile and CHANGELOG
233 #
234 # @param config ConfigParser Instance of configuration file
235 # @param changeLogFileHandle CHANGELOG file handle
236 # @exception NoOptionError: if option not found raise a NoOptionError
237 #
238 def checkConfigVersion( changeLogFileHandle, config ):
239 line = changeLogFileHandle.readline()
240 while not line.startswith("REPET release "):
241 line = changeLogFileHandle.readline()
242 numVersionChangeLog = line.split()[2]
243
244 numVersionConfig = config.get("repet_env", "repet_version")
245
246 if not numVersionChangeLog == numVersionConfig:
247 message = "*** Error: wrong config file version. Expected version num is " + numVersionChangeLog + " but actual in config file is " + numVersionConfig
248 raise CheckerException(message)
249
250 checkConfigVersion = staticmethod( checkConfigVersion )
251
252
253 ## Get version number from CHANGELOG
254 #
255 # @param changeLogFile CHANGELOG file name
256 #
257 def getVersionFromChangelogFile(changeLogFileName):
258 with open(changeLogFileName) as changeLogFileHandle:
259 line = changeLogFileHandle.readline()
260 while not line.startswith("REPET release "):
261 line = changeLogFileHandle.readline()
262 numVersionChangeLog = line.split()[2]
263 return numVersionChangeLog
264
265
266 getVersionFromChangelogFile = staticmethod( getVersionFromChangelogFile )
267
268
269 ## Check if headers of an input file contain only alpha numeric characters and "_ : . -"
270 #
271 # @param fileHandler file handle
272 # @exception CheckerException if bad header raise a CheckerException
273 #
274 def checkHeaders( fileHandler ):
275 lHeaders = CheckerUtils._getHeaderFromFastaFile(fileHandler)
276 p = re.compile('[^a-zA-Z0-9_:\.\-]', re.IGNORECASE)
277 lWrongHeaders = []
278 for header in lHeaders:
279 errList=p.findall(header)
280 if len( errList ) > 0 :
281 lWrongHeaders.append(header)
282 if lWrongHeaders != []:
283 exception = CheckerException()
284 exception.setMessages(lWrongHeaders)
285 raise exception
286
287 checkHeaders = staticmethod( checkHeaders )
288
289
290 def _getHeaderFromFastaFile( inFile ):
291 lHeaders = []
292 while True:
293 line = inFile.readline()
294 if line == "":
295 break
296 if line[0] == ">":
297 lHeaders.append( line[1:-1] )
298 return lHeaders
299
300 _getHeaderFromFastaFile = staticmethod( _getHeaderFromFastaFile )
301
302
303 ## Return True if an option is in a specified section in the configuration file, False otherwise
304 #
305 # @param config handler of configuration file
306 # @param sectionName string of section name
307 # @param optionName string of option name to check
308 #
309 def isOptionInSectionInConfig( configHandler, section, option ):
310 try:
311 CheckerUtils.checkOptionInSectionInConfigFile( configHandler, section, option )
312 except NoOptionError:
313 return False
314 return True
315
316 isOptionInSectionInConfig = staticmethod( isOptionInSectionInConfig )