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.21 by spiga, Thu Jun 5 07:13:31 2008 UTC vs.
Revision 1.37.2.1 by ewv, Thu Jan 21 16:13:28 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
12 +
13  
14  
9 # #######################################
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 45 | 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 56 | 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 67 | 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 <    def __init__(self, datasetPath, cfg_params):
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 or len(self.datasetPath.split("/")) == 1
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  
85 # ####################################
106      def fetchDBSInfo(self):
107          """
108          Contact DBS
109          """
90
110          ## get DBS URL
111 <        dbs_url="http://cmsdbsprod.cern.ch/cms_dbs_prod_global/servlet/DBSServlet"
112 <        if (self.cfg_params.has_key('CMSSW.dbs_url')):
113 <            dbs_url=self.cfg_params['CMSSW.dbs_url']
114 <
115 <        common.logger.debug(3,"Accessing DBS at: "+dbs_url)
111 >        global_url="http://cmsdbsprod.cern.ch/cms_dbs_prod_global/servlet/DBSServlet"
112 >        caf_url = "http://cmsdbsprod.cern.ch/cms_dbs_caf_analysis_01/servlet/DBSServlet"
113 >        dbs_url_map  =   {'glite':    global_url,
114 >                          'glite_slc5':global_url,\
115 >                          'glitecoll':global_url,\
116 >                          'condor':   global_url,\
117 >                          'condor_g': global_url,\
118 >                          'glidein':  global_url,\
119 >                          'lsf':      global_url,\
120 >                          'caf':      caf_url,\
121 >                          'sge':      global_url,\
122 >                          'arc':      global_url,\
123 >                          'pbs':      global_url
124 >                          }
125 >
126 >        dbs_url_default = dbs_url_map[(common.scheduler.name()).lower()]
127 >        dbs_url=  self.cfg_params.get('CMSSW.dbs_url', dbs_url_default)
128 >        common.logger.info("Accessing DBS at: "+dbs_url)
129  
130          ## check if runs are selected
131          runselection = []
132          if (self.cfg_params.has_key('CMSSW.runselection')):
133              runselection = parseRange2(self.cfg_params['CMSSW.runselection'])
134  
135 <        common.logger.debug(6,"runselection is: %s"%runselection)
135 >
136 >        self.splitByRun = int(self.cfg_params.get('CMSSW.split_by_run', 0))
137 >
138 >        common.logger.log(10-1,"runselection is: %s"%runselection)
139          ## service API
140          args = {}
141          args['url']     = dbs_url
142          args['level']   = 'CRITICAL'
143 +        args['adshome']   = '$HOME/DBSADS'
144  
145          ## check if has been requested to use the parent info
146 <        if (self.cfg_params.has_key('CMSSW.runselection')):
147 <            runselection = parseRange2(self.cfg_params['CMSSW.runselection'])
146 >        useparent = int(self.cfg_params.get('CMSSW.use_parent',0))
147 >
148 >        ## check if has been asked for a non default file to store/read analyzed fileBlocks
149 >        defaultName = common.work_space.shareDir()+'AnalyzedBlocks.txt'
150 >        fileBlocks_FileName = os.path.abspath(self.cfg_params.get('CMSSW.fileblocks_file',defaultName))
151  
113        useParent = self.cfg_params.get('CMSSW.use_parent',False)
114    
115        allowedRetriveValue = [
116                        'retrive_child',
117                        'retrive_block',
118                        'retrive_lumi',
119                        'retrive_run'
120                        ]
121        if useParent:  allowedRetriveValue.append('retrive_parent')
122        common.logger.debug(5,"Set of input parameters used for DBS query : \n"+str(allowedRetriveValue))
123        common.logger.write("Set of input parameters used for DBS query : \n"+str(allowedRetriveValue))
152          api = DBSAPI.dbsApi.DbsApi(args)
153 +        self.files = self.queryDbs(api,path=self.datasetPath,runselection=runselection,useParent=useparent)
154 +
155 +        anFileBlocks = []
156 +        if self.skipBlocks: anFileBlocks = readTXTfile(self, fileBlocks_FileName)
157 +
158 +        # parse files and fill arrays
159 +        for file in self.files :
160 +            parList  = []
161 +            lumiList = [] # List of tuples
162 +            # skip already analyzed blocks
163 +            fileblock = file['Block']['Name']
164 +            if fileblock not in anFileBlocks :
165 +                filename = file['LogicalFileName']
166 +                # asked retry the list of parent for the given child
167 +                if useparent==1:
168 +                    parList = [x['LogicalFileName'] for x in file['ParentList']]
169 +                if self.ads:
170 +                    lumiList = [ (x['RunNumber'], x['LumiSectionNumber'])
171 +                                 for x in file['LumiList'] ]
172 +                self.parent[filename] = parList
173 +                self.lumis[filename] = lumiList
174 +                if filename.find('.dat') < 0 :
175 +                    events    = file['NumberOfEvents']
176 +                    # Count number of events and lumis per block
177 +                    if fileblock in self.eventsPerBlock.keys() :
178 +                        self.eventsPerBlock[fileblock] += events
179 +                    else :
180 +                        self.eventsPerBlock[fileblock] = events
181 +                    # Number of events per file
182 +                    self.eventsPerFile[filename] = events
183 +
184 +                    # List of files per block
185 +                    if fileblock in self.blocksinfo.keys() :
186 +                        self.blocksinfo[fileblock].append(filename)
187 +                    else :
188 +                        self.blocksinfo[fileblock] = [filename]
189 +
190 +                    # total number of events
191 +                    self.maxEvents += events
192 +                    self.maxLumis  += len(lumiList)
193 +
194 +        if  self.skipBlocks and len(self.eventsPerBlock.keys()) == 0:
195 +            msg = "No new fileblocks available for dataset: "+str(self.datasetPath)
196 +            raise  CrabException(msg)
197 +
198 +        saveFblocks=''
199 +        for block in self.eventsPerBlock.keys() :
200 +            saveFblocks += str(block)+'\n'
201 +            common.logger.log(10-1,"DBSInfo: total nevts %i in block %s "%(self.eventsPerBlock[block],block))
202 +        writeTXTfile(self, fileBlocks_FileName , saveFblocks)
203 +
204 +        if len(self.eventsPerBlock) <= 0:
205 +            raise NotExistingDatasetError(("\nNo data for %s in DBS\nPlease check"
206 +                                            + " dataset path variables in crab.cfg")
207 +                                            % self.datasetPath)
208 +
209 +
210 +    def queryDbs(self,api,path=None,runselection=None,useParent=None):
211 +
212 +        allowedRetriveValue = ['retrive_block', 'retrive_run']
213 +        if self.ads: allowedRetriveValue.append('retrive_lumi')
214 +        if useParent == 1: allowedRetriveValue.append('retrive_parent')
215 +        common.logger.debug("Set of input parameters used for DBS query: %s" % allowedRetriveValue)
216          try:
217 <            if len(runselection) <= 0 :
218 <                files = api.listFiles(path=self.datasetPath,retriveList=allowedRetriveValue)
217 >            if len(runselection) <=0 :
218 >                if useParent==1 or self.splitByRun==1 or self.ads:
219 >                    if self.ads:
220 >                        files = api.listFiles(analysisDataset=path, retriveList=allowedRetriveValue)
221 >                    else :
222 >                        files = api.listFiles(path=path, retriveList=allowedRetriveValue)
223 >                else:
224 >                    files = api.listDatasetFiles(self.datasetPath)
225              else :
226                  files=[]
227                  for arun in runselection:
228                      try:
229 <                        filesinrun = api.listFiles(path=self.datasetPath,retriveList=allowedRetriveValue,runNumber=arun)
229 >                        if self.ads:
230 >                            filesinrun = api.listFiles(analysisDataset=path,retriveList=allowedRetriveValue,runNumber=arun)
231 >                        else:
232 >                            filesinrun = api.listFiles(path=path,retriveList=allowedRetriveValue,runNumber=arun)
233                          files.extend(filesinrun)
234                      except:
235                          msg="WARNING: problem extracting info from DBS for run %s "%arun
236 <                        common.logger.message(msg)
236 >                        common.logger.info(msg)
237                          pass
238  
239          except DbsBadRequest, msg:
# Line 141 | Line 241 | class DataDiscovery:
241          except DBSError, msg:
242              raise DataDiscoveryError(msg)
243  
244 <        # parse files and fill arrays
145 <        for file in files :
146 <            parList = []
147 <            filename = file['LogicalFileName']
148 <            # asked retry the list of parent for the given child
149 <            if useParent: parList = [x['LogicalFileName'] for x in file['ParentList']]
150 <            self.parent[filename] = parList
151 <            if filename.find('.dat') < 0 :
152 <                fileblock = file['Block']['Name']
153 <                events    = file['NumberOfEvents']
154 <                # number of events per block
155 <                if fileblock in self.eventsPerBlock.keys() :
156 <                    self.eventsPerBlock[fileblock] += events
157 <                else :
158 <                    self.eventsPerBlock[fileblock] = events
159 <                # number of events per file
160 <                self.eventsPerFile[filename] = events
161 <
162 <                # number of events per block
163 <                if fileblock in self.blocksinfo.keys() :
164 <                    self.blocksinfo[fileblock].append(filename)
165 <                else :
166 <                    self.blocksinfo[fileblock] = [filename]
167 <
168 <                # total number of events
169 <                self.maxEvents += events
170 <
171 <        for block in self.eventsPerBlock.keys() :
172 <            common.logger.debug(6,"DBSInfo: total nevts %i in block %s "%(self.eventsPerBlock[block],block))
173 <
174 <        if len(self.eventsPerBlock) <= 0:
175 <            raise NotExistingDatasetError(("\nNo data for %s in DBS\nPlease check"
176 <                                            + " dataset path variables in crab.cfg")
177 <                                            % self.datasetPath)
244 >        return files
245  
246  
180 # #################################################
247      def getMaxEvents(self):
248          """
249 <        max events
249 >        max events
250          """
251          return self.maxEvents
252  
253 < # #################################################
253 >
254 >    def getMaxLumis(self):
255 >        """
256 >        Return the number of lumis in the dataset
257 >        """
258 >        return self.maxLumis
259 >
260 >
261      def getEventsPerBlock(self):
262          """
263 <        list the event collections structure by fileblock
263 >        list the event collections structure by fileblock
264          """
265          return self.eventsPerBlock
266  
267 < # #################################################
267 >
268      def getEventsPerFile(self):
269          """
270 <        list the event collections structure by file
270 >        list the event collections structure by file
271          """
272          return self.eventsPerFile
273  
274 < # #################################################
274 >
275      def getFiles(self):
276          """
277 <        return files grouped by fileblock
277 >        return files grouped by fileblock
278          """
279 <        return self.blocksinfo        
279 >        return self.blocksinfo
280 >
281  
208 # #################################################
282      def getParent(self):
283          """
284 <        return parent grouped by file
284 >        return parent grouped by file
285          """
286 <        return self.parent        
286 >        return self.parent
287  
288 < ########################################################################
288 >
289 >    def getLumis(self):
290 >        """
291 >        return lumi sections grouped by file
292 >        """
293 >        return self.lumis
294 >
295 >
296 >    def getListFiles(self):
297 >        """
298 >        return parent grouped by file
299 >        """
300 >        return self.files

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines