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

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines