ViewVC Help
View File | Revision Log | Show Annotations | Root Listing
root/cvsroot/COMP/CRAB/python/DataDiscovery.py
(Generate patch)

Comparing COMP/CRAB/python/DataDiscovery.py (file contents):
Revision 1.13 by slacapra, Mon Dec 18 12:28:06 2006 UTC vs.
Revision 1.27 by spiga, Wed Feb 4 15:24:41 2009 UTC

# Line 1 | Line 1
1   #!/usr/bin/env python
2 < import sys, os, string, re
3 < from DBSInfo import *
2 > import exceptions
3 > import DBSAPI.dbsApi
4 > from DBSAPI.dbsApiException import *
5 > import common
6 > from crab_util import *
7 > import os
8  
9  
10 + # #######################################
11 + class DBSError(exceptions.Exception):
12 +    def __init__(self, errorName, errorMessage):
13 +        args='\nERROR DBS %s : %s \n'%(errorName,errorMessage)
14 +        exceptions.Exception.__init__(self, args)
15 +        pass
16 +    
17 +    def getErrorMessage(self):
18 +        """ Return error message """
19 +        return "%s" % (self.args)
20 +
21 + # #######################################
22 + class DBSInvalidDataTierError(exceptions.Exception):
23 +    def __init__(self, errorName, errorMessage):
24 +        args='\nERROR DBS %s : %s \n'%(errorName,errorMessage)
25 +        exceptions.Exception.__init__(self, args)
26 +        pass
27 +    
28 +    def getErrorMessage(self):
29 +        """ Return error message """
30 +        return "%s" % (self.args)
31 +
32 + # #######################################
33 + class DBSInfoError:
34 +    def __init__(self, url):
35 +        print '\nERROR accessing DBS url : '+url+'\n'
36 +        pass
37 +
38   # ####################################
39   class DataDiscoveryError(exceptions.Exception):
40      def __init__(self, errorMessage):
41 <        args=errorMessage
42 <        exceptions.Exception.__init__(self, args)
41 >        self.args=errorMessage
42 >        exceptions.Exception.__init__(self, self.args)
43          pass
44  
45      def getErrorMessage(self):
# Line 17 | Line 49 | class DataDiscoveryError(exceptions.Exce
49   # ####################################
50   class NotExistingDatasetError(exceptions.Exception):
51      def __init__(self, errorMessage):
52 <        args=errorMessage
53 <        exceptions.Exception.__init__(self, args)
52 >        self.args=errorMessage
53 >        exceptions.Exception.__init__(self, self.args)
54          pass
55  
56      def getErrorMessage(self):
# Line 28 | Line 60 | class NotExistingDatasetError(exceptions
60   # ####################################
61   class NoDataTierinProvenanceError(exceptions.Exception):
62      def __init__(self, errorMessage):
63 <        args=errorMessage
64 <        exceptions.Exception.__init__(self, args)
63 >        self.args=errorMessage
64 >        exceptions.Exception.__init__(self, self.args)
65          pass
66  
67      def getErrorMessage(self):
# Line 39 | Line 71 | class NoDataTierinProvenanceError(except
71   # ####################################
72   # class to find and extact info from published data
73   class DataDiscovery:
74 <    def __init__(self, datasetPath, dataTiers, cfg_params):
74 >    def __init__(self, datasetPath, cfg_params, skipAnBlocks):
75  
76 < #       Attributes
76 >        #       Attributes
77          self.datasetPath = datasetPath
46        self.dataTiers = dataTiers
78          self.cfg_params = cfg_params
79 +        self.skipBlocks = skipAnBlocks
80  
81          self.eventsPerBlock = {}  # DBS output: map fileblocks-events for collection
82          self.eventsPerFile = {}   # DBS output: map files-events
83 <        self.blocksinfo = {}  # DBS output: map fileblocks-files
84 < #DBS output: max events computed by method getMaxEvents
83 >        self.blocksinfo = {}      # DBS output: map fileblocks-files
84 >        self.maxEvents = 0        # DBS output: max events
85 >        self.parent = {}       # DBS output: max events
86  
87   # ####################################
88      def fetchDBSInfo(self):
89          """
90          Contact DBS
91          """
59
92          ## get DBS URL
93 <        try:
94 <            dbs_url=self.cfg_params['CMSSW.dbs_url']
95 <        except KeyError:
96 <            dbs_url="http://cmsdoc.cern.ch/cms/test/aprom/DBS/CGIServer/prodquery"
97 <
98 <        ## get info about the requested dataset
99 <        try:
100 <            dbs_instance=self.cfg_params['CMSSW.dbs_instance']
101 <        except KeyError:
102 <            dbs_instance="MCGlobal/Writer"
93 >        global_url="http://cmsdbsprod.cern.ch/cms_dbs_prod_global/servlet/DBSServlet"
94 >        caf_url = "http://cmsdbsprod.cern.ch/cms_dbs_caf_analysis_01/servlet/DBSServlet"
95 >        dbs_url_map  =   {'glite':    global_url,
96 >                          'glitecoll':global_url,\
97 >                          'condor':   global_url,\
98 >                          'condor_g': global_url,\
99 >                          'glidein':  global_url,\
100 >                          'lsf':      global_url,\
101 >                          'caf':      caf_url,\
102 >                          'sge':      global_url
103 >                          }
104 >
105 >        dbs_url_default = dbs_url_map[(common.scheduler.name()).lower()]
106 >        dbs_url=  self.cfg_params.get('CMSSW.dbs_url', dbs_url_default)
107 >        common.logger.debug(3,"Accessing DBS at: "+dbs_url)
108 >
109 >        ## check if runs are selected
110 >        runselection = []
111 >        if (self.cfg_params.has_key('CMSSW.runselection')):
112 >            runselection = parseRange2(self.cfg_params['CMSSW.runselection'])
113 >
114 >
115 >        self.splitByRun = int(self.cfg_params.get('CMSSW.split_by_run', 0))
116 >          
117 >        self.ads = int(self.cfg_params.get('CMSSW.ads', 0))
118 >
119 >        common.logger.debug(6,"runselection is: %s"%runselection)
120 >        ## service API
121 >        args = {}
122 >        args['url']     = dbs_url
123 >        args['level']   = 'CRITICAL'
124 >
125 >        ## check if has been requested to use the parent info
126 >        useparent = int(self.cfg_params.get('CMSSW.use_parent',0))
127 >
128 >        ## check if has been asked for a non default file to store/read analyzed fileBlocks  
129 >        defaultName = common.work_space.shareDir()+'AnalyzedBlocks.txt'  
130 >        fileBlocks_FileName = os.path.abspath(self.cfg_params.get('CMSSW.fileblocks_file',defaultName))
131  
132 <        dbs = DBSInfo(dbs_url, dbs_instance)
73 <        try:
74 <            self.datasets = dbs.getMatchingDatasets(self.datasetPath)
75 <        except dbsCgiApi.DbsCgiExecutionError, msg:
76 <            raise DataDiscoveryError(msg)
77 <        except DBSError, msg:
78 <            raise DataDiscoveryError(msg)
132 >        api = DBSAPI.dbsApi.DbsApi(args)
133  
134 <        if len(self.datasets) == 0:
81 <            raise DataDiscoveryError("DatasetPath=%s unknown to DBS" %self.datasetPath)
82 <        if len(self.datasets) > 1:
83 <            raise DataDiscoveryError("DatasetPath=%s is ambiguous" %self.datasetPath)
134 >        self.files = self.queryDbs(api,path=self.datasetPath,runselection=runselection,useParent=useparent)
135  
136 <        try:
137 <            self.dbsdataset = self.datasets[0].get('datasetPathName')
136 >        anFileBlocks = []
137 >        if self.skipBlocks: anFileBlocks = readTXTfile(self, fileBlocks_FileName)
138  
139 <            self.eventsPerBlock = dbs.getEventsPerBlock(self.dbsdataset)
140 <            self.blocksinfo = dbs.getDatasetFileBlocks(self.dbsdataset)
141 <            self.eventsPerFile = dbs.getEventsPerFile(self.dbsdataset)
142 <        except DBSError, ex:
143 <            raise DataDiscoveryError(ex.getErrorMessage())
144 <        
139 >        # parse files and fill arrays
140 >        for file in self.files :
141 >            parList = []
142 >            # skip already analyzed blocks
143 >            fileblock = file['Block']['Name']
144 >            if fileblock not in anFileBlocks :
145 >                filename = file['LogicalFileName']
146 >                # asked retry the list of parent for the given child
147 >                if useparent==1: parList = [x['LogicalFileName'] for x in file['ParentList']]
148 >                self.parent[filename] = parList
149 >                if filename.find('.dat') < 0 :
150 >                    events    = file['NumberOfEvents']
151 >                    # number of events per block
152 >                    if fileblock in self.eventsPerBlock.keys() :
153 >                        self.eventsPerBlock[fileblock] += events
154 >                    else :
155 >                        self.eventsPerBlock[fileblock] = events
156 >                    # number of events per file
157 >                    self.eventsPerFile[filename] = events
158 >            
159 >                    # number of events per block
160 >                    if fileblock in self.blocksinfo.keys() :
161 >                        self.blocksinfo[fileblock].append(filename)
162 >                    else :
163 >                        self.blocksinfo[fileblock] = [filename]
164 >            
165 >                    # total number of events
166 >                    self.maxEvents += events
167 >        if  self.skipBlocks and len(self.eventsPerBlock.keys()) == 0:
168 >            msg = "No new fileblocks available for dataset: "+str(self.datasetPath)
169 >            raise  CrabException(msg)    
170 >
171 >        saveFblocks=''
172 >        for block in self.eventsPerBlock.keys() :
173 >            saveFblocks += str(block)+'\n'
174 >            common.logger.debug(6,"DBSInfo: total nevts %i in block %s "%(self.eventsPerBlock[block],block))
175 >        writeTXTfile(self, fileBlocks_FileName , saveFblocks)
176 >                      
177          if len(self.eventsPerBlock) <= 0:
178 <            raise NotExistingDatasetError (("\nNo data for %s in DBS\nPlease check"
178 >            raise NotExistingDatasetError(("\nNo data for %s in DBS\nPlease check"
179                                              + " dataset path variables in crab.cfg")
180 <                                            % self.dbsdataset)
180 >                                            % self.datasetPath)
181  
182  
183 + ###########################
184 +
185 +    def queryDbs(self,api,path=None,runselection=None,useParent=None):
186 +
187 +        allowedRetriveValue = [#'retrive_parent',
188 +                               'retrive_block',
189 +                               #'retrive_lumi',
190 +                               'retrive_run'
191 +                               ]
192 +        try:
193 +            if len(runselection) <=0 :
194 +                if useParent==1 or self.splitByRun==1 :
195 +                    if self.ads==1 :          
196 +                        files = api.listFiles(analysisDataset=path, retriveList=allowedRetriveValue)
197 +                    else :
198 +                        files = api.listFiles(path=path, retriveList=allowedRetriveValue)
199 +                    common.logger.debug(5,"Set of input parameters used for DBS query : \n"+str(allowedRetriveValue))
200 +                    common.logger.write("Set of input parameters used for DBS query : \n"+str(allowedRetriveValue))
201 +                else:
202 +                    print 'MALE2'
203 +                    files = api.listDatasetFiles(self.datasetPath)
204 +            else :
205 +                files=[]
206 +                for arun in runselection:
207 +                    try:
208 +                        if self.ads==1 : filesinrun = api.listFiles(analysisDataset=path,retriveList=allowedRetriveValue,runNumber=arun)
209 +                        else: filesinrun = api.listFiles(path=path,retriveList=allowedRetriveValue,runNumber=arun)
210 +                        files.extend(filesinrun)
211 +                    except:
212 +                        msg="WARNING: problem extracting info from DBS for run %s "%arun
213 +                        common.logger.message(msg)
214 +                        pass
215 +
216 +        except DbsBadRequest, msg:
217 +            raise DataDiscoveryError(msg)
218 +        except DBSError, msg:
219 +            raise DataDiscoveryError(msg)
220 +
221 +        return files
222 +
223   # #################################################
224      def getMaxEvents(self):
225          """
226          max events
227          """
228 <        ## loop over the event collections
106 <        nevts=0      
107 <        for evc_evts in self.eventsPerBlock.values():
108 <            nevts=nevts+evc_evts
109 <
110 <        return nevts
228 >        return self.maxEvents
229  
230   # #################################################
231      def getEventsPerBlock(self):
# Line 130 | Line 248 | class DataDiscovery:
248          """
249          return self.blocksinfo        
250  
251 + # #################################################
252 +    def getParent(self):
253 +        """
254 +        return parent grouped by file
255 +        """
256 +        return self.parent        
257 +
258 + # #################################################
259 +    def getListFiles(self):
260 +        """
261 +        return parent grouped by file
262 +        """
263 +        return self.files        
264 +
265   ########################################################################

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines