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.8 by spiga, Thu Jul 20 11:59:09 2006 UTC vs.
Revision 1.25 by spiga, Thu Jul 3 23:02:38 2008 UTC

# Line 1 | Line 1
1   #!/usr/bin/env python
2 < import sys, os, string, re
3 < from DBSInfo import *
2 > import exceptions
3 > import DBSAPI.dbsApi
4 > from DBSAPI.dbsApiException import *
5 > import common
6 > from crab_util import *
7 > import os
8  
9  
10 + # #######################################
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   # ####################################
39   class DataDiscoveryError(exceptions.Exception):
40      def __init__(self, errorMessage):
41 <        args=errorMessage
42 <        exceptions.Exception.__init__(self, args)
41 >        self.args=errorMessage
42 >        exceptions.Exception.__init__(self, self.args)
43          pass
44  
45      def getErrorMessage(self):
# Line 17 | Line 49 | class DataDiscoveryError(exceptions.Exce
49   # ####################################
50   class NotExistingDatasetError(exceptions.Exception):
51      def __init__(self, errorMessage):
52 <        args=errorMessage
53 <        exceptions.Exception.__init__(self, args)
52 >        self.args=errorMessage
53 >        exceptions.Exception.__init__(self, self.args)
54          pass
55  
56      def getErrorMessage(self):
# Line 28 | Line 60 | class NotExistingDatasetError(exceptions
60   # ####################################
61   class NoDataTierinProvenanceError(exceptions.Exception):
62      def __init__(self, errorMessage):
63 <        args=errorMessage
64 <        exceptions.Exception.__init__(self, args)
63 >        self.args=errorMessage
64 >        exceptions.Exception.__init__(self, self.args)
65          pass
66  
67      def getErrorMessage(self):
# Line 39 | Line 71 | class NoDataTierinProvenanceError(except
71   # ####################################
72   # class to find and extact info from published data
73   class DataDiscovery:
74 <    def __init__(self, owner, dataset, dataTiers, cfg_params):
74 >    def __init__(self, datasetPath, cfg_params, skipAnBlocks):
75  
76 < #       Attributes
77 <        self.owner = owner
46 <        self.dataset = dataset
47 <        self.dataTiers = dataTiers
76 >        #       Attributes
77 >        self.datasetPath = datasetPath
78          self.cfg_params = cfg_params
79 +        self.skipBlocks = skipAnBlocks
80  
81 <        self.dbspaths= []     # DBS output: list of dbspaths for all data
82 <        self.allblocks = []   # DBS output: list of map fileblocks-totevts for all dataset-owners
83 <        self.blocksinfo = {}  # DBS output: map fileblocks-totevts for the primary block, used internally to this class
84 < #DBS output: max events computed by method getMaxEvents
81 >        self.eventsPerBlock = {}  # DBS output: map fileblocks-events for collection
82 >        self.eventsPerFile = {}   # DBS output: map files-events
83 >        self.blocksinfo = {}      # DBS output: map fileblocks-files
84 >        self.maxEvents = 0        # DBS output: max events
85 >        self.parent = {}       # DBS output: max events
86  
87   # ####################################
88      def fetchDBSInfo(self):
89          """
90          Contact DBS
91          """
92 <
93 <        ## add the PU among the required data tiers if the Digi are requested
94 <        if (self.dataTiers.count('Digi')>0) & (self.dataTiers.count('PU')<=0) :
95 <            self.dataTiers.append('PU')
96 <
97 <        ## get info about the requested dataset
98 <        dbs=DBSInfo()
92 >        ## get 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 >                          }
104 >
105 >        dbs_url_default = dbs_url_map[(common.scheduler.name()).lower()]
106 >        dbs_url=  self.cfg_params.get('CMSSW.dbs_url', dbs_url_default)
107 >        common.logger.debug(3,"Accessing DBS at: "+dbs_url)
108 >
109 >        ## check if runs are selected
110 >        runselection = []
111 >        if (self.cfg_params.has_key('CMSSW.runselection')):
112 >            runselection = parseRange2(self.cfg_params['CMSSW.runselection'])
113 >
114 >        common.logger.debug(6,"runselection is: %s"%runselection)
115 >        ## service API
116 >        args = {}
117 >        args['url']     = dbs_url
118 >        args['level']   = 'CRITICAL'
119 >
120 >        ## check if has been requested to use the parent info
121 >        useParent = self.cfg_params.get('CMSSW.use_parent',False)
122 >
123 >        ## check if has been asked for a non default file to store/read analyzed fileBlocks  
124 >        defaultName = common.work_space.shareDir()+'AnalyzedBlocks.txt'  
125 >        fileBlocks_FileName = os.path.abspath(self.cfg_params.get('CMSSW.fileblocks_file',defaultName))
126 >
127 >        api = DBSAPI.dbsApi.DbsApi(args)
128 >        allowedRetriveValue = ['retrive_parent',
129 >                               'retrive_block',
130 >                               'retrive_lumi',
131 >                               'retrive_run'
132 >                               ]
133          try:
134 <            self.datasets = dbs.getMatchingDatasets(self.owner, self.dataset)
135 <        except DBSError, ex:
136 <            raise DataDiscoveryError(ex.getErrorMessage())
137 <        if len(self.datasets) == 0:
138 <            raise DataDiscoveryError("Owner=%s, Dataset=%s unknown to DBS" % (self.owner, self.dataset))
139 <        if len(self.datasets) > 1:
140 <            raise DataDiscoveryError("Owner=%s, Dataset=%s is ambiguous" % (self.owner, self.dataset))
141 <        try:
142 <            self.dbsdataset = self.datasets[0].get('datasetPathName')
143 <            self.blocksinfo = dbs.getDatasetContents(self.dbsdataset)
144 <            self.allblocks.append (self.blocksinfo.keys ()) # add also the current fileblocksinfo
145 <            self.dbspaths.append(self.dbsdataset)
146 <        except DBSError, ex:
147 <            raise DataDiscoveryError(ex.getErrorMessage())
148 <        
149 <        if len(self.blocksinfo)<=0:
150 <            msg="\nERROR Data for %s do not exist in DBS! \n Check the dataset/owner variables in crab.cfg !"%self.dbsdataset
85 <            raise NotExistingDatasetError(msg)
86 <
134 >            if len(runselection) <= 0 :
135 >                if useParent:
136 >                    files = api.listFiles(path=self.datasetPath, retriveList=allowedRetriveValue)
137 >                    common.logger.debug(5,"Set of input parameters used for DBS query : \n"+str(allowedRetriveValue))
138 >                    common.logger.write("Set of input parameters used for DBS query : \n"+str(allowedRetriveValue))
139 >                else:
140 >                    files = api.listDatasetFiles(self.datasetPath)
141 >            else :
142 >                files=[]
143 >                for arun in runselection:
144 >                    try:
145 >                        filesinrun = api.listFiles(path=self.datasetPath,retriveList=allowedRetriveValue,runNumber=arun)
146 >                        files.extend(filesinrun)
147 >                    except:
148 >                        msg="WARNING: problem extracting info from DBS for run %s "%arun
149 >                        common.logger.message(msg)
150 >                        pass
151  
152 <        ## get info about the parents
153 <        try:
154 <            parents=dbs.getDatasetProvenance(self.dbsdataset, self.dataTiers)
91 <        except DBSInvalidDataTierError, ex:
92 <            msg=ex.getErrorMessage()+' \n Check the data_tier variable in crab.cfg !\n'
152 >        except DbsBadRequest, msg:
153 >            raise DataDiscoveryError(msg)
154 >        except DBSError, msg:
155              raise DataDiscoveryError(msg)
94        except DBSError, ex:
95            raise DataDiscoveryError(ex.getErrorMessage())
156  
157 <        ## check that the user asks for parent Data Tier really existing in the DBS provenance
158 <        self.checkParentDataTier(parents, self.dataTiers)
157 >        anFileBlocks = []
158 >        if self.skipBlocks: anFileBlocks = readTXTfile(self, fileBlocks_FileName)
159 >
160 >        # parse files and fill arrays
161 >        for file in files :
162 >            parList = []
163 >            # skip already analyzed blocks
164 >            fileblock = file['Block']['Name']
165 >            if fileblock not in anFileBlocks :
166 >                filename = file['LogicalFileName']
167 >                # asked retry the list of parent for the given child
168 >                if useParent: parList = [x['LogicalFileName'] for x in file['ParentList']]
169 >                self.parent[filename] = parList
170 >                if filename.find('.dat') < 0 :
171 >                    events    = file['NumberOfEvents']
172 >                    # number of events per block
173 >                    if fileblock in self.eventsPerBlock.keys() :
174 >                        self.eventsPerBlock[fileblock] += events
175 >                    else :
176 >                        self.eventsPerBlock[fileblock] = events
177 >                    # number of events per file
178 >                    self.eventsPerFile[filename] = events
179 >            
180 >                    # number of events per block
181 >                    if fileblock in self.blocksinfo.keys() :
182 >                        self.blocksinfo[fileblock].append(filename)
183 >                    else :
184 >                        self.blocksinfo[fileblock] = [filename]
185 >            
186 >                    # total number of events
187 >                    self.maxEvents += events
188 >        if  self.skipBlocks and len(self.eventsPerBlock.keys()) == 0:
189 >            msg = "No new fileblocks available for dataset: "+str(self.datasetPath)
190 >            raise  CrabException(msg)    
191 >
192 >        saveFblocks=''
193 >        for block in self.eventsPerBlock.keys() :
194 >            saveFblocks += str(block)+'\n'
195 >            common.logger.debug(6,"DBSInfo: total nevts %i in block %s "%(self.eventsPerBlock[block],block))
196 >        writeTXTfile(self, fileBlocks_FileName , saveFblocks)
197 >                      
198 >        if len(self.eventsPerBlock) <= 0:
199 >            raise NotExistingDatasetError(("\nNo data for %s in DBS\nPlease check"
200 >                                            + " dataset path variables in crab.cfg")
201 >                                            % self.datasetPath)
202  
100        ## for each parent get the corresponding fileblocks
101        try:
102            for p in parents:
103                ## fill a list of dbspaths
104                parentPath = p.get('parent').get('datasetPathName')
105                self.dbspaths.append (parentPath)
106                parentBlocks = dbs.getDatasetContents (parentPath)
107                self.allblocks.append (parentBlocks.keys ())  # add parent fileblocksinfo
108        except DBSError, ex:
109                raise DataDiscoveryError(ex.getErrorMessage())
203  
204   # #################################################
205 <    def checkParentDataTier(self, parents, dataTiers):
205 >    def getMaxEvents(self):
206          """
207 <        check that the data tiers requested by the user really exists in the provenance of the given dataset
207 >        max events
208          """
209 <        startType = string.split(self.dbsdataset,'/')[2]
117 <        # for example 'type' is PU and 'dataTier' is Hit
118 <        parentTypes = map(lambda p: p.get('type'), parents)
119 <        for tier in dataTiers:
120 <            if parentTypes.count(tier) <= 0 and tier != startType:
121 <                msg="\nERROR Data %s not published in DBS with asked data tiers : the data tier not found is %s !\n  Check the data_tier variable in crab.cfg !"%(self.dbsdataset,tier)
122 <                raise  NoDataTierinProvenanceError(msg)
123 <
209 >        return self.maxEvents
210  
211   # #################################################
212 <    def getMaxEvents(self):
212 >    def getEventsPerBlock(self):
213          """
214 <        max events of the primary dataset-owner
214 >        list the event collections structure by fileblock
215          """
216 <        ## loop over the fileblocks of the primary dataset-owner
131 <        nevts=0      
132 <        for blockevts in self.blocksinfo.values():
133 <            nevts=nevts+blockevts
134 <
135 <        return nevts
216 >        return self.eventsPerBlock
217  
218   # #################################################
219 <    def getDBSPaths(self):
219 >    def getEventsPerFile(self):
220          """
221 <        list the DBSpaths for all required data
221 >        list the event collections structure by file
222          """
223 <        return self.dbspaths
223 >        return self.eventsPerFile
224  
225   # #################################################
226 <    def getEVC(self):
226 >    def getFiles(self):
227          """
228 <        list the event collections structure by fileblock
228 >        return files grouped by fileblock
229          """
230 <        print "To be used by a more complex job splitting... TODO later... "
150 <        print "it requires changes in what's returned by DBSInfo.getDatasetContents and then fetchDBSInfo"
230 >        return self.blocksinfo        
231  
232   # #################################################
233 <    def getFileBlocks(self):
233 >    def getParent(self):
234          """
235 <        fileblocks for all required dataset-owners
235 >        return parent grouped by file
236          """
237 <        return self.allblocks        
237 >        return self.parent        
238  
239   ########################################################################

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines