CRAB/python/DataDiscovery.py

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


# #######################################
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):

        #       Attributes
        self.datasetPath = datasetPath
        self.cfg_params = cfg_params

        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
        dbs_url="http://cmsdbsprod.cern.ch/cms_dbs_prod_global/servlet/DBSServlet"
        if (self.cfg_params.has_key('CMSSW.dbs_url')):
            dbs_url=self.cfg_params['CMSSW.dbs_url']

        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'])

        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
        if (self.cfg_params.has_key('CMSSW.runselection')):
            runselection = parseRange2(self.cfg_params['CMSSW.runselection'])

        useParent = self.cfg_params.get('CMSSW.use_parent',False)
    
        allowedRetriveValue = [
                        'retrive_child', 
                        'retrive_block',
                        'retrive_lumi',
                        'retrive_run'
                        ]
        if useParent:  allowedRetriveValue.append('retrive_parent') 
        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)) 
        api = DBSAPI.dbsApi.DbsApi(args)
        try:
            if len(runselection) <= 0 :
                files = api.listFiles(path=self.datasetPath,retriveList=allowedRetriveValue)
            else :
                files=[]
                for arun in runselection:
                    try:
                        filesinrun = api.listFiles(path=self.datasetPath,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)

        # parse files and fill arrays
        for file in files :
            parList = []
            filename = file['LogicalFileName']
            # asked retry the list of parent for the given child 
            if useParent: parList = [x['LogicalFileName'] for x in file['ParentList']] 
            self.parent[filename] = parList 
            if filename.find('.dat') < 0 :
                fileblock = file['Block']['Name']
                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

        for block in self.eventsPerBlock.keys() :
            common.logger.debug(6,"DBSInfo: total nevts %i in block %s "%(self.eventsPerBlock[block],block))

        if len(self.eventsPerBlock) <= 0:
            raise NotExistingDatasetError(("\nNo data for %s in DBS\nPlease check"
                                            + " dataset path variables in crab.cfg")
                                            % self.datasetPath)


# #################################################
    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        

########################################################################
Revision:	1.21
Committed:	Thu Jun 5 07:13:31 2008 UTC (16 years, 10 months ago) by spiga
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_2_2_2_pre4, CRAB_2_2_2_pre3
Changes since 1.20:	+29 -2 lines
Log Message:	fully added support for two tier inputs
#	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	afanfani	1.1
8	afanfani	1.3
9	slacapra	1.18	# #######################################
10			class DBSError(exceptions.Exception):
11			def __init__(self, errorName, errorMessage):
12			args='\nERROR DBS %s : %s \n'%(errorName,errorMessage)
13			exceptions.Exception.__init__(self, args)
14			pass
15
16			def getErrorMessage(self):
17			""" Return error message """
18			return "%s" % (self.args)
19
20			# #######################################
21			class DBSInvalidDataTierError(exceptions.Exception):
22			def __init__(self, errorName, errorMessage):
23			args='\nERROR DBS %s : %s \n'%(errorName,errorMessage)
24			exceptions.Exception.__init__(self, args)
25			pass
26
27			def getErrorMessage(self):
28			""" Return error message """
29			return "%s" % (self.args)
30
31			# #######################################
32			class DBSInfoError:
33			def __init__(self, url):
34			print '\nERROR accessing DBS url : '+url+'\n'
35			pass
36
37	afanfani	1.1	# ####################################
38	afanfani	1.3	class DataDiscoveryError(exceptions.Exception):
39	slacapra	1.7	def __init__(self, errorMessage):
40	gutsche	1.15	self.args=errorMessage
41	slacapra	1.14	exceptions.Exception.__init__(self, self.args)
42	slacapra	1.7	pass
43
44			def getErrorMessage(self):
45			""" Return exception error """
46			return "%s" % (self.args)
47	afanfani	1.3
48	afanfani	1.1	# ####################################
49	afanfani	1.3	class NotExistingDatasetError(exceptions.Exception):
50	slacapra	1.7	def __init__(self, errorMessage):
51	gutsche	1.15	self.args=errorMessage
52	slacapra	1.14	exceptions.Exception.__init__(self, self.args)
53	slacapra	1.7	pass
54
55			def getErrorMessage(self):
56			""" Return exception error """
57			return "%s" % (self.args)
58	afanfani	1.1
59			# ####################################
60	afanfani	1.3	class NoDataTierinProvenanceError(exceptions.Exception):
61	slacapra	1.7	def __init__(self, errorMessage):
62	gutsche	1.15	self.args=errorMessage
63	slacapra	1.14	exceptions.Exception.__init__(self, self.args)
64	slacapra	1.7	pass
65
66			def getErrorMessage(self):
67			""" Return exception error """
68			return "%s" % (self.args)
69	afanfani	1.1
70			# ####################################
71			# class to find and extact info from published data
72			class DataDiscovery:
73	gutsche	1.15	def __init__(self, datasetPath, cfg_params):
74	afanfani	1.1
75	slacapra	1.18	# Attributes
76	slacapra	1.11	self.datasetPath = datasetPath
77	afanfani	1.1	self.cfg_params = cfg_params
78
79	slacapra	1.11	self.eventsPerBlock = {} # DBS output: map fileblocks-events for collection
80			self.eventsPerFile = {} # DBS output: map files-events
81	slacapra	1.18	self.blocksinfo = {} # DBS output: map fileblocks-files
82			self.maxEvents = 0 # DBS output: max events
83	spiga	1.21	self.parent = {} # DBS output: max events
84	afanfani	1.1
85			# ####################################
86			def fetchDBSInfo(self):
87			"""
88			Contact DBS
89			"""
90
91	slacapra	1.11	## get DBS URL
92	slacapra	1.19	dbs_url="http://cmsdbsprod.cern.ch/cms_dbs_prod_global/servlet/DBSServlet"
93			if (self.cfg_params.has_key('CMSSW.dbs_url')):
94	slacapra	1.11	dbs_url=self.cfg_params['CMSSW.dbs_url']
95	afanfani	1.3
96	slacapra	1.18	common.logger.debug(3,"Accessing DBS at: "+dbs_url)
97
98			## check if runs are selected
99	slacapra	1.19	runselection = []
100			if (self.cfg_params.has_key('CMSSW.runselection')):
101	slacapra	1.18	runselection = parseRange2(self.cfg_params['CMSSW.runselection'])
102
103	afanfani	1.20	common.logger.debug(6,"runselection is: %s"%runselection)
104	slacapra	1.18	## service API
105			args = {}
106			args['url'] = dbs_url
107			args['level'] = 'CRITICAL'
108
109	spiga	1.21	## check if has been requested to use the parent info
110			if (self.cfg_params.has_key('CMSSW.runselection')):
111			runselection = parseRange2(self.cfg_params['CMSSW.runselection'])
112
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))
124	slacapra	1.18	api = DBSAPI.dbsApi.DbsApi(args)
125	afanfani	1.5	try:
126	slacapra	1.18	if len(runselection) <= 0 :
127	spiga	1.21	files = api.listFiles(path=self.datasetPath,retriveList=allowedRetriveValue)
128	slacapra	1.18	else :
129	afanfani	1.20	files=[]
130			for arun in runselection:
131			try:
132	spiga	1.21	filesinrun = api.listFiles(path=self.datasetPath,retriveList=allowedRetriveValue,runNumber=arun)
133	afanfani	1.20	files.extend(filesinrun)
134			except:
135			msg="WARNING: problem extracting info from DBS for run %s "%arun
136			common.logger.message(msg)
137			pass
138
139	slacapra	1.18	except DbsBadRequest, msg:
140	slacapra	1.11	raise DataDiscoveryError(msg)
141	slacapra	1.13	except DBSError, msg:
142			raise DataDiscoveryError(msg)
143	slacapra	1.11
144	slacapra	1.18	# parse files and fill arrays
145			for file in files :
146	spiga	1.21	parList = []
147	slacapra	1.18	filename = file['LogicalFileName']
148	spiga	1.21	# 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	slacapra	1.18	if filename.find('.dat') < 0 :
152			fileblock = file['Block']['Name']
153			events = file['NumberOfEvents']
154	afanfani	1.20	# 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	slacapra	1.18	else :
166	afanfani	1.20	self.blocksinfo[fileblock] = [filename]
167	slacapra	1.18
168	afanfani	1.20	# total number of events
169			self.maxEvents += events
170	slacapra	1.11
171	slacapra	1.18	for block in self.eventsPerBlock.keys() :
172			common.logger.debug(6,"DBSInfo: total nevts %i in block %s "%(self.eventsPerBlock[block],block))
173	slacapra	1.11
174			if len(self.eventsPerBlock) <= 0:
175	slacapra	1.18	raise NotExistingDatasetError(("\nNo data for %s in DBS\nPlease check"
176	slacapra	1.11	+ " dataset path variables in crab.cfg")
177	slacapra	1.18	% self.datasetPath)
178	afanfani	1.1
179
180			# #################################################
181			def getMaxEvents(self):
182			"""
183	slacapra	1.11	max events
184	afanfani	1.1	"""
185	slacapra	1.18	return self.maxEvents
186	afanfani	1.1
187			# #################################################
188	slacapra	1.11	def getEventsPerBlock(self):
189	afanfani	1.1	"""
190	slacapra	1.11	list the event collections structure by fileblock
191	afanfani	1.1	"""
192	slacapra	1.11	return self.eventsPerBlock
193	afanfani	1.1
194			# #################################################
195	slacapra	1.11	def getEventsPerFile(self):
196	afanfani	1.1	"""
197	slacapra	1.11	list the event collections structure by file
198	afanfani	1.1	"""
199	slacapra	1.11	return self.eventsPerFile
200	afanfani	1.1
201			# #################################################
202	slacapra	1.11	def getFiles(self):
203	afanfani	1.1	"""
204	slacapra	1.11	return files grouped by fileblock
205	afanfani	1.1	"""
206	slacapra	1.11	return self.blocksinfo
207	afanfani	1.1
208	spiga	1.21	# #################################################
209			def getParent(self):
210			"""
211			return parent grouped by file
212			"""
213			return self.parent
214
215	afanfani	1.1	########################################################################