File: Alignment.py

package info (click to toggle)
mobyle 1.5.5%2Bdfsg-6
  • links: PTS, VCS
  • area: main
  • in suites: buster
  • size: 8,288 kB
  • sloc: python: 22,709; makefile: 35; sh: 33; ansic: 10; xml: 6
file content (109 lines) | stat: -rw-r--r-- 5,230 bytes parent folder | download | duplicates (4)
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