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.29 by edelmann, Tue Mar 17 14:36:44 2009 UTC

# Line 4 | Line 4 | import DBSAPI.dbsApi
4   from DBSAPI.dbsApiException import *
5   import common
6   from crab_util import *
7 + import os
8  
9  
10   # #######################################
# Line 70 | Line 71 | class NoDataTierinProvenanceError(except
71   # ####################################
72   # class to find and extact info from published data
73   class DataDiscovery:
74 <    def __init__(self, datasetPath, cfg_params):
74 >    def __init__(self, datasetPath, cfg_params, skipAnBlocks):
75  
76          #       Attributes
77          self.datasetPath = datasetPath
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
# Line 87 | Line 89 | class DataDiscovery:
89          """
90          Contact DBS
91          """
90 –
92          ## get DBS URL
93 <        dbs_url="http://cmsdbsprod.cern.ch/cms_dbs_prod_global/servlet/DBSServlet"
94 <        if (self.cfg_params.has_key('CMSSW.dbs_url')):
95 <            dbs_url=self.cfg_params['CMSSW.dbs_url']
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 >                          'arc':      global_url
104 >                          }
105  
106 +        dbs_url_default = dbs_url_map[(common.scheduler.name()).lower()]
107 +        dbs_url=  self.cfg_params.get('CMSSW.dbs_url', dbs_url_default)
108          common.logger.debug(3,"Accessing DBS at: "+dbs_url)
109  
110          ## check if runs are selected
# Line 100 | Line 112 | class DataDiscovery:
112          if (self.cfg_params.has_key('CMSSW.runselection')):
113              runselection = parseRange2(self.cfg_params['CMSSW.runselection'])
114  
115 +
116 +        self.splitByRun = int(self.cfg_params.get('CMSSW.split_by_run', 0))
117 +          
118 +        self.ads = int(self.cfg_params.get('CMSSW.ads', 0))
119 +
120          common.logger.debug(6,"runselection is: %s"%runselection)
121          ## service API
122          args = {}
# Line 107 | Line 124 | class DataDiscovery:
124          args['level']   = 'CRITICAL'
125  
126          ## check if has been requested to use the parent info
127 <        if (self.cfg_params.has_key('CMSSW.runselection')):
111 <            runselection = parseRange2(self.cfg_params['CMSSW.runselection'])
127 >        useparent = int(self.cfg_params.get('CMSSW.use_parent',0))
128  
129 <        useParent = self.cfg_params.get('CMSSW.use_parent',False)
130 <    
131 <        allowedRetriveValue = [
132 <                        '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))
129 >        ## check if has been asked for a non default file to store/read analyzed fileBlocks  
130 >        defaultName = common.work_space.shareDir()+'AnalyzedBlocks.txt'  
131 >        fileBlocks_FileName = os.path.abspath(self.cfg_params.get('CMSSW.fileblocks_file',defaultName))
132 >
133          api = DBSAPI.dbsApi.DbsApi(args)
134 +
135 +        self.files = self.queryDbs(api,path=self.datasetPath,runselection=runselection,useParent=useparent)
136 +
137 +        anFileBlocks = []
138 +        if self.skipBlocks: anFileBlocks = readTXTfile(self, fileBlocks_FileName)
139 +
140 +        # parse files and fill arrays
141 +        for file in self.files :
142 +            parList = []
143 +            # skip already analyzed blocks
144 +            fileblock = file['Block']['Name']
145 +            if fileblock not in anFileBlocks :
146 +                filename = file['LogicalFileName']
147 +                # asked retry the list of parent for the given child
148 +                if useparent==1: parList = [x['LogicalFileName'] for x in file['ParentList']]
149 +                self.parent[filename] = parList
150 +                if filename.find('.dat') < 0 :
151 +                    events    = file['NumberOfEvents']
152 +                    # number of events per block
153 +                    if fileblock in self.eventsPerBlock.keys() :
154 +                        self.eventsPerBlock[fileblock] += events
155 +                    else :
156 +                        self.eventsPerBlock[fileblock] = events
157 +                    # number of events per file
158 +                    self.eventsPerFile[filename] = events
159 +            
160 +                    # number of events per block
161 +                    if fileblock in self.blocksinfo.keys() :
162 +                        self.blocksinfo[fileblock].append(filename)
163 +                    else :
164 +                        self.blocksinfo[fileblock] = [filename]
165 +            
166 +                    # total number of events
167 +                    self.maxEvents += events
168 +        if  self.skipBlocks and len(self.eventsPerBlock.keys()) == 0:
169 +            msg = "No new fileblocks available for dataset: "+str(self.datasetPath)
170 +            raise  CrabException(msg)    
171 +
172 +        saveFblocks=''
173 +        for block in self.eventsPerBlock.keys() :
174 +            saveFblocks += str(block)+'\n'
175 +            common.logger.debug(6,"DBSInfo: total nevts %i in block %s "%(self.eventsPerBlock[block],block))
176 +        writeTXTfile(self, fileBlocks_FileName , saveFblocks)
177 +                      
178 +        if len(self.eventsPerBlock) <= 0:
179 +            raise NotExistingDatasetError(("\nNo data for %s in DBS\nPlease check"
180 +                                            + " dataset path variables in crab.cfg")
181 +                                            % self.datasetPath)
182 +
183 +
184 + ###########################
185 +
186 +    def queryDbs(self,api,path=None,runselection=None,useParent=None):
187 +
188 +        allowedRetriveValue = [#'retrive_parent',
189 +                               'retrive_block',
190 +                               #'retrive_lumi',
191 +                               'retrive_run'
192 +                               ]
193          try:
194 <            if len(runselection) <= 0 :
195 <                files = api.listFiles(path=self.datasetPath,retriveList=allowedRetriveValue)
194 >            if len(runselection) <=0 :
195 >                if useParent==1 or self.splitByRun==1 :
196 >                    if self.ads==1 :          
197 >                        files = api.listFiles(analysisDataset=path, retriveList=allowedRetriveValue)
198 >                    else :
199 >                        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))
202 >                else:
203 >                    files = api.listDatasetFiles(self.datasetPath)
204              else :
205                  files=[]
206                  for arun in runselection:
207                      try:
208 <                        filesinrun = api.listFiles(path=self.datasetPath,retriveList=allowedRetriveValue,runNumber=arun)
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
# Line 141 | Line 218 | class DataDiscovery:
218          except DBSError, msg:
219              raise DataDiscoveryError(msg)
220  
221 <        # 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)
178 <
221 >        return files
222  
223   # #################################################
224      def getMaxEvents(self):
# Line 212 | Line 255 | class DataDiscovery:
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