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.27 by spiga, Wed Feb 4 15:24:41 2009 UTC vs.
Revision 1.46 by spiga, Wed Jun 2 13:55:14 2010 UTC

# Line 1 | Line 1
1   #!/usr/bin/env python
2 +
3 + __revision__ = "$Id$"
4 + __version__ = "$Revision$"
5 +
6   import exceptions
7   import DBSAPI.dbsApi
8 < from DBSAPI.dbsApiException import *
8 > from DBSAPI.dbsApiException import *
9   import common
10   from crab_util import *
11 < import os
11 > from LumiList import LumiList
12 > import os
13 >
14  
15  
10 # #######################################
16   class DBSError(exceptions.Exception):
17      def __init__(self, errorName, errorMessage):
18          args='\nERROR DBS %s : %s \n'%(errorName,errorMessage)
19          exceptions.Exception.__init__(self, args)
20          pass
21 <    
21 >
22      def getErrorMessage(self):
23          """ Return error message """
24          return "%s" % (self.args)
25  
26 < # #######################################
26 >
27 >
28   class DBSInvalidDataTierError(exceptions.Exception):
29      def __init__(self, errorName, errorMessage):
30          args='\nERROR DBS %s : %s \n'%(errorName,errorMessage)
31          exceptions.Exception.__init__(self, args)
32          pass
33 <    
33 >
34      def getErrorMessage(self):
35          """ Return error message """
36          return "%s" % (self.args)
37  
38 < # #######################################
38 >
39 >
40   class DBSInfoError:
41      def __init__(self, url):
42          print '\nERROR accessing DBS url : '+url+'\n'
43          pass
44  
45 < # ####################################
45 >
46 >
47   class DataDiscoveryError(exceptions.Exception):
48      def __init__(self, errorMessage):
49          self.args=errorMessage
# Line 46 | Line 54 | class DataDiscoveryError(exceptions.Exce
54          """ Return exception error """
55          return "%s" % (self.args)
56  
57 < # ####################################
57 >
58 >
59   class NotExistingDatasetError(exceptions.Exception):
60      def __init__(self, errorMessage):
61          self.args=errorMessage
# Line 57 | Line 66 | class NotExistingDatasetError(exceptions
66          """ Return exception error """
67          return "%s" % (self.args)
68  
69 < # ####################################
69 >
70 >
71   class NoDataTierinProvenanceError(exceptions.Exception):
72      def __init__(self, errorMessage):
73          self.args=errorMessage
# Line 68 | Line 78 | class NoDataTierinProvenanceError(except
78          """ Return exception error """
79          return "%s" % (self.args)
80  
81 < # ####################################
82 < # class to find and extact info from published data
81 >
82 >
83   class DataDiscovery:
84 +    """
85 +    Class to find and extact info from published data
86 +    """
87      def __init__(self, datasetPath, cfg_params, skipAnBlocks):
88  
89          #       Attributes
90          self.datasetPath = datasetPath
91 +        # Analysis dataset is primary/processed/tier/definition
92 +        self.ads = len(self.datasetPath.split("/")) > 4
93          self.cfg_params = cfg_params
94          self.skipBlocks = skipAnBlocks
95  
96          self.eventsPerBlock = {}  # DBS output: map fileblocks-events for collection
97          self.eventsPerFile = {}   # DBS output: map files-events
98 <        self.blocksinfo = {}      # DBS output: map fileblocks-files
98 > #         self.lumisPerBlock = {}   # DBS output: number of lumis in each block
99 > #         self.lumisPerFile = {}    # DBS output: number of lumis in each file
100 >        self.blocksinfo = {}      # DBS output: map fileblocks-files
101          self.maxEvents = 0        # DBS output: max events
102 <        self.parent = {}       # DBS output: max events
102 >        self.maxLumis = 0         # DBS output: total number of lumis
103 >        self.parent = {}          # DBS output: parents of each file
104 >        self.lumis = {}           # DBS output: lumis in each file
105 >        self.lumiMask = None
106 >        self.splitByLumi = False
107  
87 # ####################################
108      def fetchDBSInfo(self):
109          """
110          Contact DBS
111          """
112          ## get DBS URL
113          global_url="http://cmsdbsprod.cern.ch/cms_dbs_prod_global/servlet/DBSServlet"
114 <        caf_url = "http://cmsdbsprod.cern.ch/cms_dbs_caf_analysis_01/servlet/DBSServlet"
115 <        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)
114 >        dbs_url=  self.cfg_params.get('CMSSW.dbs_url', global_url)
115 >        common.logger.info("Accessing DBS at: "+dbs_url)
116  
117          ## check if runs are selected
118          runselection = []
119          if (self.cfg_params.has_key('CMSSW.runselection')):
120              runselection = parseRange2(self.cfg_params['CMSSW.runselection'])
121  
122 +        ## check if various lumi parameters are set
123 +        self.lumiMask = self.cfg_params.get('CMSSW.lumi_mask',None)
124 +        self.lumiParams = self.cfg_params.get('CMSSW.total_number_of_lumis',None) or \
125 +                          self.cfg_params.get('CMSSW.lumis_per_job',None)
126 +
127 +        lumiList = None
128 +        if self.lumiMask:
129 +            lumiList = LumiList(filename=self.lumiMask)
130 +        if runselection:
131 +            runList = LumiList(runs = runselection)
132  
133          self.splitByRun = int(self.cfg_params.get('CMSSW.split_by_run', 0))
134 <          
135 <        self.ads = int(self.cfg_params.get('CMSSW.ads', 0))
134 >        common.logger.log(10-1,"runselection is: %s"%runselection)
135 >
136 >        if not self.splitByRun:
137 >            self.splitByLumi = self.lumiMask or self.lumiParams or self.ads
138  
119        common.logger.debug(6,"runselection is: %s"%runselection)
139          ## service API
140          args = {}
141          args['url']     = dbs_url
# Line 125 | Line 144 | class DataDiscovery:
144          ## check if has been requested to use the parent info
145          useparent = int(self.cfg_params.get('CMSSW.use_parent',0))
146  
147 <        ## check if has been asked for a non default file to store/read analyzed fileBlocks  
148 <        defaultName = common.work_space.shareDir()+'AnalyzedBlocks.txt'  
147 >        ## check if has been asked for a non default file to store/read analyzed fileBlocks
148 >        defaultName = common.work_space.shareDir()+'AnalyzedBlocks.txt'
149          fileBlocks_FileName = os.path.abspath(self.cfg_params.get('CMSSW.fileblocks_file',defaultName))
131
132        api = DBSAPI.dbsApi.DbsApi(args)
150  
151 +        api = DBSAPI.dbsApi.DbsApi(args)
152          self.files = self.queryDbs(api,path=self.datasetPath,runselection=runselection,useParent=useparent)
153  
154 +        # Check to see what the dataset is
155 +        pdsName = self.datasetPath.split("/")[1]
156 +        primDSs = api.listPrimaryDatasets(pdsName)
157 +        dataType = primDSs[0]['Type']
158 +        common.logger.debug("Datatype is %s" % dataType)
159 +        if dataType == 'data' and not (self.splitByRun or self.splitByLumi):
160 +            msg = 'Data must be split by lumi or by run. ' \
161 +                  'Please see crab -help for the correct settings'
162 +            raise  CrabException(msg)
163 +
164 +
165 +
166          anFileBlocks = []
167 <        if self.skipBlocks: anFileBlocks = readTXTfile(self, fileBlocks_FileName)
167 >        if self.skipBlocks: anFileBlocks = readTXTfile(self, fileBlocks_FileName)
168  
169          # parse files and fill arrays
170          for file in self.files :
171 <            parList = []
171 >            parList  = []
172 >            fileLumis = [] # List of tuples
173              # skip already analyzed blocks
174              fileblock = file['Block']['Name']
175              if fileblock not in anFileBlocks :
176                  filename = file['LogicalFileName']
177 <                # asked retry the list of parent for the given child
178 <                if useparent==1: parList = [x['LogicalFileName'] for x in file['ParentList']]
179 <                self.parent[filename] = parList
177 >                # asked retry the list of parent for the given child
178 >                if useparent==1:
179 >                    parList = [x['LogicalFileName'] for x in file['ParentList']]
180 >                if self.splitByLumi:
181 >                    fileLumis = [ (x['RunNumber'], x['LumiSectionNumber'])
182 >                                 for x in file['LumiList'] ]
183 >                self.parent[filename] = parList
184 >                # For LumiMask, intersection of two lists.
185 >                if self.lumiMask and runselection:
186 >                    self.lumis[filename] = runList.filterLumis(lumiList.filterLumis(fileLumis))
187 >                elif runselection:
188 >                    self.lumis[filename] = runList.filterLumis(fileLumis)
189 >                elif self.lumiMask:
190 >                    self.lumis[filename] = lumiList.filterLumis(fileLumis)
191 >                else:
192 >                    self.lumis[filename] = fileLumis
193                  if filename.find('.dat') < 0 :
194                      events    = file['NumberOfEvents']
195 <                    # number of events per block
195 >                    # Count number of events and lumis per block
196                      if fileblock in self.eventsPerBlock.keys() :
197                          self.eventsPerBlock[fileblock] += events
198                      else :
199                          self.eventsPerBlock[fileblock] = events
200 <                    # number of events per file
200 >                    # Number of events per file
201                      self.eventsPerFile[filename] = events
202 <            
203 <                    # number of events per block
202 >
203 >                    # List of files per block
204                      if fileblock in self.blocksinfo.keys() :
205                          self.blocksinfo[fileblock].append(filename)
206                      else :
207                          self.blocksinfo[fileblock] = [filename]
208 <            
208 >
209                      # total number of events
210                      self.maxEvents += events
211 +                    self.maxLumis  += len(self.lumis[filename])
212 +
213          if  self.skipBlocks and len(self.eventsPerBlock.keys()) == 0:
214              msg = "No new fileblocks available for dataset: "+str(self.datasetPath)
215 <            raise  CrabException(msg)    
215 >            raise  CrabException(msg)
216 >
217  
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                      
218          if len(self.eventsPerBlock) <= 0:
219              raise NotExistingDatasetError(("\nNo data for %s in DBS\nPlease check"
220                                              + " dataset path variables in crab.cfg")
221                                              % self.datasetPath)
222  
223  
183 ###########################
184
224      def queryDbs(self,api,path=None,runselection=None,useParent=None):
225 <
226 <        allowedRetriveValue = [#'retrive_parent',
227 <                               'retrive_block',
228 <                               #'retrive_lumi',
229 <                               'retrive_run'
230 <                               ]
225 >
226 >        allowedRetriveValue = ['retrive_block', 'retrive_run']
227 >        if self.ads or self.lumiMask or self.lumiParams:
228 >            allowedRetriveValue.append('retrive_lumi')
229 >        if useParent == 1: allowedRetriveValue.append('retrive_parent')
230 >        common.logger.debug("Set of input parameters used for DBS query: %s" % allowedRetriveValue)
231          try:
232 <            if len(runselection) <=0 :
233 <                if useParent==1 or self.splitByRun==1 :
234 <                    if self.ads==1 :          
235 <                        files = api.listFiles(analysisDataset=path, retriveList=allowedRetriveValue)
232 >            if len(runselection) <=0 or self.splitByLumi:
233 >                if useParent==1 or self.splitByRun==1 or self.splitByLumi:
234 >                    if self.ads:
235 >                        files = api.listFiles(analysisDataset=path, retriveList=allowedRetriveValue)
236                      else :
237 <                        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))
237 >                        files = api.listFiles(path=path, retriveList=allowedRetriveValue)
238                  else:
202                    print 'MALE2'
239                      files = api.listDatasetFiles(self.datasetPath)
240              else :
241                  files=[]
242                  for arun in runselection:
243                      try:
244 <                        if self.ads==1 : filesinrun = api.listFiles(analysisDataset=path,retriveList=allowedRetriveValue,runNumber=arun)
245 <                        else: filesinrun = api.listFiles(path=path,retriveList=allowedRetriveValue,runNumber=arun)
244 >                        if self.ads:
245 >                            filesinrun = api.listFiles(analysisDataset=path,retriveList=allowedRetriveValue,runNumber=arun)
246 >                        else:
247 >                            filesinrun = api.listFiles(path=path,retriveList=allowedRetriveValue,runNumber=arun)
248                          files.extend(filesinrun)
249                      except:
250                          msg="WARNING: problem extracting info from DBS for run %s "%arun
251 <                        common.logger.message(msg)
251 >                        common.logger.info(msg)
252                          pass
253  
254          except DbsBadRequest, msg:
# Line 220 | Line 258 | class DataDiscovery:
258  
259          return files
260  
261 < # #################################################
261 >
262      def getMaxEvents(self):
263          """
264 <        max events
264 >        max events
265          """
266          return self.maxEvents
267  
268 < # #################################################
268 >
269 >    def getMaxLumis(self):
270 >        """
271 >        Return the number of lumis in the dataset
272 >        """
273 >        return self.maxLumis
274 >
275 >
276      def getEventsPerBlock(self):
277          """
278 <        list the event collections structure by fileblock
278 >        list the event collections structure by fileblock
279          """
280          return self.eventsPerBlock
281  
282 < # #################################################
282 >
283      def getEventsPerFile(self):
284          """
285 <        list the event collections structure by file
285 >        list the event collections structure by file
286          """
287          return self.eventsPerFile
288  
289 < # #################################################
289 >
290      def getFiles(self):
291          """
292 <        return files grouped by fileblock
292 >        return files grouped by fileblock
293          """
294 <        return self.blocksinfo        
294 >        return self.blocksinfo
295 >
296  
251 # #################################################
297      def getParent(self):
298          """
299 <        return parent grouped by file
299 >        return parent grouped by file
300          """
301 <        return self.parent        
301 >        return self.parent
302  
303 < # #################################################
304 <    def getListFiles(self):
303 >
304 >    def getLumis(self):
305          """
306 <        return parent grouped by file
306 >        return lumi sections grouped by file
307          """
308 <        return self.files        
308 >        return self.lumis
309 >
310  
311 < ########################################################################
311 >    def getListFiles(self):
312 >        """
313 >        return parent grouped by file
314 >        """
315 >        return self.files

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines