CRAB/python/DataDiscovery.py

#!/usr/bin/env python
import exceptions
import DBSAPI.dbsApi
from DBSAPI.dbsApiException import * 
import common
from crab_util import *
import os 


# #######################################
class DBSError(exceptions.Exception):
    def __init__(self, errorName, errorMessage):
        args='\nERROR DBS %s : %s \n'%(errorName,errorMessage)
        exceptions.Exception.__init__(self, args)
        pass
    
    def getErrorMessage(self):
        """ Return error message """
        return "%s" % (self.args)

# #######################################
class DBSInvalidDataTierError(exceptions.Exception):
    def __init__(self, errorName, errorMessage):
        args='\nERROR DBS %s : %s \n'%(errorName,errorMessage)
        exceptions.Exception.__init__(self, args)
        pass
    
    def getErrorMessage(self):
        """ Return error message """
        return "%s" % (self.args)

# #######################################
class DBSInfoError:
    def __init__(self, url):
        print '\nERROR accessing DBS url : '+url+'\n'
        pass

# ####################################
class DataDiscoveryError(exceptions.Exception):
    def __init__(self, errorMessage):
        self.args=errorMessage
        exceptions.Exception.__init__(self, self.args)
        pass

    def getErrorMessage(self):
        """ Return exception error """
        return "%s" % (self.args)

# ####################################
class NotExistingDatasetError(exceptions.Exception):
    def __init__(self, errorMessage):
        self.args=errorMessage
        exceptions.Exception.__init__(self, self.args)
        pass

    def getErrorMessage(self):
        """ Return exception error """
        return "%s" % (self.args)

# ####################################
class NoDataTierinProvenanceError(exceptions.Exception):
    def __init__(self, errorMessage):
        self.args=errorMessage
        exceptions.Exception.__init__(self, self.args)
        pass

    def getErrorMessage(self):
        """ Return exception error """
        return "%s" % (self.args)

# ####################################
# class to find and extact info from published data
class DataDiscovery:
    def __init__(self, datasetPath, cfg_params, skipAnBlocks):

        #       Attributes
        self.datasetPath = datasetPath
        self.cfg_params = cfg_params
        self.skipBlocks = skipAnBlocks

        self.eventsPerBlock = {}  # DBS output: map fileblocks-events for collection
        self.eventsPerFile = {}   # DBS output: map files-events
        self.blocksinfo = {}      # DBS output: map fileblocks-files 
        self.maxEvents = 0        # DBS output: max events
        self.parent = {}       # DBS output: max events

# ####################################
    def fetchDBSInfo(self):
        """
        Contact DBS
        """
        ## get DBS URL
        global_url="http://cmsdbsprod.cern.ch/cms_dbs_prod_global/servlet/DBSServlet"
        caf_url = "http://cmsdbsprod.cern.ch/cms_dbs_caf_analysis_01/servlet/DBSServlet"
        dbs_url_map  =   {'glite':    global_url,
                          'glitecoll':global_url,\
                          'condor':   global_url,\
                          'condor_g': global_url,\
                          'glidein':  global_url,\
                          'lsf':      global_url,\
                          'caf':      caf_url,\
                          'sge':      global_url
                          }

        dbs_url_default = dbs_url_map[(common.scheduler.name()).lower()]
        dbs_url=  self.cfg_params.get('CMSSW.dbs_url', dbs_url_default)
        common.logger.debug(3,"Accessing DBS at: "+dbs_url)

        ## check if runs are selected
        runselection = []
        if (self.cfg_params.has_key('CMSSW.runselection')):
            runselection = parseRange2(self.cfg_params['CMSSW.runselection'])


        self.splitByRun = int(self.cfg_params.get('CMSSW.split_by_run', 0))
          
        self.ads = int(self.cfg_params.get('CMSSW.ads', 0))

        common.logger.debug(6,"runselection is: %s"%runselection)
        ## service API
        args = {}
        args['url']     = dbs_url
        args['level']   = 'CRITICAL'

        ## check if has been requested to use the parent info
        useparent = int(self.cfg_params.get('CMSSW.use_parent',0))

        ## check if has been asked for a non default file to store/read analyzed fileBlocks   
        defaultName = common.work_space.shareDir()+'AnalyzedBlocks.txt'  
        fileBlocks_FileName = os.path.abspath(self.cfg_params.get('CMSSW.fileblocks_file',defaultName))
 
        api = DBSAPI.dbsApi.DbsApi(args)

        self.files = self.queryDbs(api,path=self.datasetPath,runselection=runselection,useParent=useparent)

        anFileBlocks = []
        if self.skipBlocks: anFileBlocks = readTXTfile(self, fileBlocks_FileName) 

        # parse files and fill arrays
        for file in self.files :
            parList = []
            # skip already analyzed blocks
            fileblock = file['Block']['Name']
            if fileblock not in anFileBlocks :
                filename = file['LogicalFileName']
                # asked retry the list of parent for the given child 
                if useparent==1: parList = [x['LogicalFileName'] for x in file['ParentList']] 
                self.parent[filename] = parList 
                if filename.find('.dat') < 0 :
                    events    = file['NumberOfEvents']
                    # number of events per block
                    if fileblock in self.eventsPerBlock.keys() :
                        self.eventsPerBlock[fileblock] += events
                    else :
                        self.eventsPerBlock[fileblock] = events
                    # number of events per file
                    self.eventsPerFile[filename] = events
            
                    # number of events per block
                    if fileblock in self.blocksinfo.keys() :
                        self.blocksinfo[fileblock].append(filename)
                    else :
                        self.blocksinfo[fileblock] = [filename]
            
                    # total number of events
                    self.maxEvents += events
        if  self.skipBlocks and len(self.eventsPerBlock.keys()) == 0:
            msg = "No new fileblocks available for dataset: "+str(self.datasetPath)
            raise  CrabException(msg)    

        saveFblocks='' 
        for block in self.eventsPerBlock.keys() :
            saveFblocks += str(block)+'\n' 
            common.logger.debug(6,"DBSInfo: total nevts %i in block %s "%(self.eventsPerBlock[block],block))
        writeTXTfile(self, fileBlocks_FileName , saveFblocks) 
                      
        if len(self.eventsPerBlock) <= 0:
            raise NotExistingDatasetError(("\nNo data for %s in DBS\nPlease check"
                                            + " dataset path variables in crab.cfg")
                                            % self.datasetPath)


###########################

    def queryDbs(self,api,path=None,runselection=None,useParent=None):
 
        allowedRetriveValue = [#'retrive_parent', 
                               'retrive_block',
                               #'retrive_lumi',
                               'retrive_run'
                               ]
        try:
            if len(runselection) <=0 :
                if useParent==1 or self.splitByRun==1 :
                    if self.ads==1 :           
                        files = api.listFiles(analysisDataset=path, retriveList=allowedRetriveValue) 
                    else :
                        files = api.listFiles(path=path, retriveList=allowedRetriveValue) 
                    common.logger.debug(5,"Set of input parameters used for DBS query : \n"+str(allowedRetriveValue)) 
                    common.logger.write("Set of input parameters used for DBS query : \n"+str(allowedRetriveValue)) 
                else:
                    files = api.listDatasetFiles(self.datasetPath)
            else :
                files=[]
                for arun in runselection:
                    try:
                        if self.ads==1 : filesinrun = api.listFiles(analysisDataset=path,retriveList=allowedRetriveValue,runNumber=arun)
                        else: filesinrun = api.listFiles(path=path,retriveList=allowedRetriveValue,runNumber=arun)
                        files.extend(filesinrun)
                    except:
                        msg="WARNING: problem extracting info from DBS for run %s "%arun
                        common.logger.message(msg)
                        pass

        except DbsBadRequest, msg:
            raise DataDiscoveryError(msg)
        except DBSError, msg:
            raise DataDiscoveryError(msg)

        return files

# #################################################
    def getMaxEvents(self):
        """
        max events 
        """
        return self.maxEvents

# #################################################
    def getEventsPerBlock(self):
        """
        list the event collections structure by fileblock 
        """
        return self.eventsPerBlock

# #################################################
    def getEventsPerFile(self):
        """
        list the event collections structure by file 
        """
        return self.eventsPerFile

# #################################################
    def getFiles(self):
        """
        return files grouped by fileblock 
        """
        return self.blocksinfo        

# #################################################
    def getParent(self):
        """
        return parent grouped by file 
        """
        return self.parent        

# #################################################
    def getListFiles(self):
        """
        return parent grouped by file 
        """
        return self.files        

########################################################################
Revision:	1.28
Committed:	Fri Feb 6 18:14:58 2009 UTC (16 years, 2 months ago) by spiga
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_2_5_0, CRAB_2_5_0_pre7, CRAB_2_5_0_pre6, CRAB_2_5_0_pre5, CRAB_2_5_0_pre4, CRAB_2_5_0_pre3, CRAB_2_5_0_pre2
Changes since 1.27:	+0 -1 lines
Log Message:	removed a print
#	User	Rev	Content
1	gutsche	1.6	#!/usr/bin/env python
2	slacapra	1.18	import exceptions
3			import DBSAPI.dbsApi
4			from DBSAPI.dbsApiException import *
5			import common
6			from crab_util import *
7	spiga	1.22	import os
8	afanfani	1.1
9	afanfani	1.3
10	slacapra	1.18	# #######################################
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	afanfani	1.1	# ####################################
39	afanfani	1.3	class DataDiscoveryError(exceptions.Exception):
40	slacapra	1.7	def __init__(self, errorMessage):
41	gutsche	1.15	self.args=errorMessage
42	slacapra	1.14	exceptions.Exception.__init__(self, self.args)
43	slacapra	1.7	pass
44
45			def getErrorMessage(self):
46			""" Return exception error """
47			return "%s" % (self.args)
48	afanfani	1.3
49	afanfani	1.1	# ####################################
50	afanfani	1.3	class NotExistingDatasetError(exceptions.Exception):
51	slacapra	1.7	def __init__(self, errorMessage):
52	gutsche	1.15	self.args=errorMessage
53	slacapra	1.14	exceptions.Exception.__init__(self, self.args)
54	slacapra	1.7	pass
55
56			def getErrorMessage(self):
57			""" Return exception error """
58			return "%s" % (self.args)
59	afanfani	1.1
60			# ####################################
61	afanfani	1.3	class NoDataTierinProvenanceError(exceptions.Exception):
62	slacapra	1.7	def __init__(self, errorMessage):
63	gutsche	1.15	self.args=errorMessage
64	slacapra	1.14	exceptions.Exception.__init__(self, self.args)
65	slacapra	1.7	pass
66
67			def getErrorMessage(self):
68			""" Return exception error """
69			return "%s" % (self.args)
70	afanfani	1.1
71			# ####################################
72			# class to find and extact info from published data
73			class DataDiscovery:
74	spiga	1.22	def __init__(self, datasetPath, cfg_params, skipAnBlocks):
75	afanfani	1.1
76	slacapra	1.18	# Attributes
77	slacapra	1.11	self.datasetPath = datasetPath
78	afanfani	1.1	self.cfg_params = cfg_params
79	spiga	1.22	self.skipBlocks = skipAnBlocks
80	afanfani	1.1
81	slacapra	1.11	self.eventsPerBlock = {} # DBS output: map fileblocks-events for collection
82			self.eventsPerFile = {} # DBS output: map files-events
83	slacapra	1.18	self.blocksinfo = {} # DBS output: map fileblocks-files
84			self.maxEvents = 0 # DBS output: max events
85	spiga	1.21	self.parent = {} # DBS output: max events
86	afanfani	1.1
87			# ####################################
88			def fetchDBSInfo(self):
89			"""
90			Contact DBS
91			"""
92	slacapra	1.11	## get DBS URL
93	spiga	1.25	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	afanfani	1.3
105	spiga	1.25	dbs_url_default = dbs_url_map[(common.scheduler.name()).lower()]
106			dbs_url= self.cfg_params.get('CMSSW.dbs_url', dbs_url_default)
107	slacapra	1.18	common.logger.debug(3,"Accessing DBS at: "+dbs_url)
108
109			## check if runs are selected
110	slacapra	1.19	runselection = []
111			if (self.cfg_params.has_key('CMSSW.runselection')):
112	slacapra	1.18	runselection = parseRange2(self.cfg_params['CMSSW.runselection'])
113
114	spiga	1.26
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	afanfani	1.20	common.logger.debug(6,"runselection is: %s"%runselection)
120	slacapra	1.18	## service API
121			args = {}
122			args['url'] = dbs_url
123			args['level'] = 'CRITICAL'
124
125	spiga	1.21	## check if has been requested to use the parent info
126	spiga	1.26	useparent = int(self.cfg_params.get('CMSSW.use_parent',0))
127	spiga	1.21
128	spiga	1.22	## 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	slacapra	1.18	api = DBSAPI.dbsApi.DbsApi(args)
133	afanfani	1.20
134	spiga	1.27	self.files = self.queryDbs(api,path=self.datasetPath,runselection=runselection,useParent=useparent)
135	slacapra	1.11
136	spiga	1.22	anFileBlocks = []
137			if self.skipBlocks: anFileBlocks = readTXTfile(self, fileBlocks_FileName)
138
139	slacapra	1.18	# parse files and fill arrays
140	spiga	1.26	for file in self.files :
141	spiga	1.21	parList = []
142	spiga	1.22	# 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	spiga	1.26	if useparent==1: parList = [x['LogicalFileName'] for x in file['ParentList']]
148	spiga	1.22	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	slacapra	1.11
171	spiga	1.22	saveFblocks=''
172	slacapra	1.18	for block in self.eventsPerBlock.keys() :
173	spiga	1.22	saveFblocks += str(block)+'\n'
174	slacapra	1.18	common.logger.debug(6,"DBSInfo: total nevts %i in block %s "%(self.eventsPerBlock[block],block))
175	spiga	1.22	writeTXTfile(self, fileBlocks_FileName , saveFblocks)
176
177	slacapra	1.11	if len(self.eventsPerBlock) <= 0:
178	slacapra	1.18	raise NotExistingDatasetError(("\nNo data for %s in DBS\nPlease check"
179	slacapra	1.11	+ " dataset path variables in crab.cfg")
180	slacapra	1.18	% self.datasetPath)
181	afanfani	1.1
182
183	spiga	1.26	###########################
184
185	spiga	1.27	def queryDbs(self,api,path=None,runselection=None,useParent=None):
186	spiga	1.26
187			allowedRetriveValue = [#'retrive_parent',
188			'retrive_block',
189			#'retrive_lumi',
190			'retrive_run'
191			]
192			try:
193	spiga	1.27	if len(runselection) <=0 :
194	spiga	1.26	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			files = api.listDatasetFiles(self.datasetPath)
203			else :
204			files=[]
205			for arun in runselection:
206			try:
207			if self.ads==1 : filesinrun = api.listFiles(analysisDataset=path,retriveList=allowedRetriveValue,runNumber=arun)
208			else: filesinrun = api.listFiles(path=path,retriveList=allowedRetriveValue,runNumber=arun)
209			files.extend(filesinrun)
210			except:
211			msg="WARNING: problem extracting info from DBS for run %s "%arun
212			common.logger.message(msg)
213			pass
214
215			except DbsBadRequest, msg:
216			raise DataDiscoveryError(msg)
217			except DBSError, msg:
218			raise DataDiscoveryError(msg)
219
220			return files
221
222	afanfani	1.1	# #################################################
223			def getMaxEvents(self):
224			"""
225	slacapra	1.11	max events
226	afanfani	1.1	"""
227	slacapra	1.18	return self.maxEvents
228	afanfani	1.1
229			# #################################################
230	slacapra	1.11	def getEventsPerBlock(self):
231	afanfani	1.1	"""
232	slacapra	1.11	list the event collections structure by fileblock
233	afanfani	1.1	"""
234	slacapra	1.11	return self.eventsPerBlock
235	afanfani	1.1
236			# #################################################
237	slacapra	1.11	def getEventsPerFile(self):
238	afanfani	1.1	"""
239	slacapra	1.11	list the event collections structure by file
240	afanfani	1.1	"""
241	slacapra	1.11	return self.eventsPerFile
242	afanfani	1.1
243			# #################################################
244	slacapra	1.11	def getFiles(self):
245	afanfani	1.1	"""
246	slacapra	1.11	return files grouped by fileblock
247	afanfani	1.1	"""
248	slacapra	1.11	return self.blocksinfo
249	afanfani	1.1
250	spiga	1.21	# #################################################
251			def getParent(self):
252			"""
253			return parent grouped by file
254			"""
255			return self.parent
256
257	spiga	1.26	# #################################################
258			def getListFiles(self):
259			"""
260			return parent grouped by file
261			"""
262			return self.files
263
264	afanfani	1.1	########################################################################