1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109
|
########################################################################################
# #
# Author: Sandrine Larroude #
# Organization:'Biological Software and Databases' Group, Institut Pasteur, Paris. #
# Distributed under GPLv2 Licence. Please refer to the COPYING.LIB document. #
# #
########################################################################################
"""
Content the Mobyle Parameter types related to genomic bioloy
"""
from logging import getLogger
c_log = getLogger(__name__)
b_log = getLogger('Mobyle.builder')
from Mobyle.MobyleError import MobyleError , UnDefAttrError
from Mobyle.Classes.Core import AbstractTextDataType , safeMask
from Mobyle.ConfigManager import Config
_cfg = Config()
class AlignmentDataType( AbstractTextDataType ):
def detect( self , value ):
"""
detects the format of the sequence(s) contained in fileName
@param value: the src object and the filename in the src of the data to detect
@type value: tuple ( session/Job/MobyleJob instance , string filename )
@return: a tuple of the detection run information:
- the detected format,
- the detected items number,
- program name used for the detection.
"""
detected_mt = super( AlignmentDataType , self ).detect( value )
squizzDir = _cfg.format_detector_cache_path()
if squizzDir :
##############################################
# This part of is used to back up all #
# submitted sequences which are not detected #
# by squizz for further analysis #
##############################################
squizz_detector = None
for detector in _cfg.dataconverter( self.__class__.__name__[:-8] ):
if detector.program_name == 'squizz':
squizz_detector = detector
break
if squizz_detector is not None :
detected_data_format = detected_mt.getDataFormat()
from os import link
from os.path import join as os_join
if ( detected_data_format is None ) or ( detected_data_format in squizz_detector.detectedFormat() and not detected_mt.getFormatProgram() == 'squizz' ):
try:
#dump the data to further annalysis
link( os_join( value[0].getDir() , value[1] ) ,
os_join( squizzDir , "%s_%s_%s"%( self.__class__.__name__[:-8] ,
value[0].getKey() ,
value[1] ) )
)
except Exception , err :
c_log.error( "unable to link data in format_detector_cache_path : %s " % err )
return detected_mt
def validate( self , param ):
"""
@return: True if the value is valid, False otherwise
"""
value = param.getValue()
if param.isout():
if value is not None : #not possible for the user to modify an isout parameter
return False
else:
#####################################################
# #
# check if the Parameter have a secure filenames #
# #
#####################################################
try:
debug = param.getDebug()
if debug > 1:
b_log.debug( "check if the Parameter have a secure filename" )
#getFilenames returns a list of unix file mask, result of a code evaluation
#None is returned if there is no mask for a parameter.
masks = param.getFilenames( )
for mask in masks :
if mask is None:
continue
mySafeMask = safeMask( mask )
if debug > 1:
b_log.debug( "filename= %s safeMask = %s" % (mask, mySafeMask))
if not mySafeMask or mySafeMask != mask :
raise MobyleError , "have an unsecure filenames value before safeMask: %s , after safeMask: %s" % (mask, mySafeMask)
elif debug > 1:
b_log.debug( "filename = %s ...........OK" % mask )
except UnDefAttrError :
b_log.debug("no filenames")
else:
if value is None:
return True
else:
return True
|