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.3 by afanfani, Sun Jan 29 01:46:08 2006 UTC vs.
Revision 1.19 by slacapra, Fri Jan 4 17:30:56 2008 UTC

# Line 1 | Line 1
1 < #!/usr/bin/env python2
2 < import sys, os, string, re
3 < from DBSInfo import *
4 <
1 > #!/usr/bin/env python
2 > import exceptions
3 > import DBSAPI.dbsApi
4 > from DBSAPI.dbsApiException import *
5 > import common
6 > from crab_util import *
7 >
8 >
9 > # #######################################
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   # ####################################
38   class DataDiscoveryError(exceptions.Exception):
39 <  def __init__(self, errorMessage):
40 <   args=errorMessage
41 <   exceptions.Exception.__init__(self, args)
42 <   pass
43 <
44 <  def getErrorMessage(self):
45 <   """ Return exception error """
46 <   return "%s" % (self.args)
39 >    def __init__(self, errorMessage):
40 >        self.args=errorMessage
41 >        exceptions.Exception.__init__(self, self.args)
42 >        pass
43 >
44 >    def getErrorMessage(self):
45 >        """ Return exception error """
46 >        return "%s" % (self.args)
47  
48   # ####################################
49   class NotExistingDatasetError(exceptions.Exception):
50 <  def __init__(self, errorMessage):
51 <   args=errorMessage
52 <   exceptions.Exception.__init__(self, args)
53 <   pass
54 <
55 <  def getErrorMessage(self):
56 <   """ Return exception error """
57 <   return "%s" % (self.args)
50 >    def __init__(self, errorMessage):
51 >        self.args=errorMessage
52 >        exceptions.Exception.__init__(self, self.args)
53 >        pass
54 >
55 >    def getErrorMessage(self):
56 >        """ Return exception error """
57 >        return "%s" % (self.args)
58  
59   # ####################################
60   class NoDataTierinProvenanceError(exceptions.Exception):
61 <  def __init__(self, errorMessage):
62 <   args=errorMessage
63 <   exceptions.Exception.__init__(self, args)
64 <   pass
65 <
66 <  def getErrorMessage(self):
67 <   """ Return exception error """
68 <   return "%s" % (self.args)
61 >    def __init__(self, errorMessage):
62 >        self.args=errorMessage
63 >        exceptions.Exception.__init__(self, self.args)
64 >        pass
65 >
66 >    def getErrorMessage(self):
67 >        """ Return exception error """
68 >        return "%s" % (self.args)
69  
70   # ####################################
71   # class to find and extact info from published data
72   class DataDiscovery:
73 <    def __init__(self, owner, dataset, dataTiers, cfg_params):
73 >    def __init__(self, datasetPath, cfg_params):
74  
75 < #       Attributes
76 <        self.dbsdataset='/'+dataset+'/datatier/'+owner
46 <        self.dataTiers = dataTiers
75 >        #       Attributes
76 >        self.datasetPath = datasetPath
77          self.cfg_params = cfg_params
78  
79 <        self.dbspaths= []     # DBS output: list of dbspaths for all data
80 <        self.allblocks = []   # DBS output: list of map fileblocks-totevts for all dataset-owners
81 <        self.blocksinfo = {}  # DBS output: map fileblocks-totevts for the primary block, used internally to this class
82 < #DBS output: max events computed by method getMaxEvents
79 >        self.eventsPerBlock = {}  # DBS output: map fileblocks-events for collection
80 >        self.eventsPerFile = {}   # DBS output: map files-events
81 >        self.blocksinfo = {}      # DBS output: map fileblocks-files
82 >        self.maxEvents = 0        # DBS output: max events
83  
84   # ####################################
85      def fetchDBSInfo(self):
86          """
87          Contact DBS
88          """
59        parents = []
60        parentsblocksinfo = {}
89  
90 <        ## add the PU among the required data tiers if the Digi are requested
91 <        if (self.dataTiers.count('Digi')>0) & (self.dataTiers.count('PU')<=0) :
92 <          self.dataTiers.append('PU')
90 >        ## get DBS URL
91 >        dbs_url="http://cmsdbsprod.cern.ch/cms_dbs_prod_global/servlet/DBSServlet"
92 >        if (self.cfg_params.has_key('CMSSW.dbs_url')):
93 >            dbs_url=self.cfg_params['CMSSW.dbs_url']
94 >
95 >        common.logger.debug(3,"Accessing DBS at: "+dbs_url)
96 >
97 >        ## check if runs are selected
98 >        runselection = []
99 >        if (self.cfg_params.has_key('CMSSW.runselection')):
100 >            runselection = parseRange2(self.cfg_params['CMSSW.runselection'])
101 >
102 >        ## service API
103 >        args = {}
104 >        args['url']     = dbs_url
105 >        args['level']   = 'CRITICAL'
106  
107 <        ## get info about the requested dataset
67 <        dbs=DBSInfo(self.dbsdataset,self.dataTiers)
107 >        api = DBSAPI.dbsApi.DbsApi(args)
108          try:
109 <          self.blocksinfo=dbs.getDatasetContents()
110 <        except DBSError, ex:
111 <          raise DataDiscoveryError(ex.getErrorMessage())
112 <        
113 <        if len(self.blocksinfo)<=0:
114 <         msg="\nERROR Data %s do not exist in DBS! \n Check the dataset/owner variables in crab.cfg !"%self.dbsdataset
115 <         raise NotExistingDatasetError(msg)
116 <
117 <        currentdatatier=string.split(self.blocksinfo.keys()[0],'/')[2]
118 <        fakedatatier=string.split(self.dbsdataset,'/')[2]
119 <        currentdbsdataset=string.replace(self.dbsdataset, fakedatatier, currentdatatier)  
120 <
121 <        self.dbspaths.append(currentdbsdataset)    # add the requested dbspath
122 <
123 <        ## get info about the parents
124 <        try:
125 <          parents=dbs.getDatasetProvenance()
126 <        except DBSInvalidDataTierError, ex:
127 <          msg=ex.getErrorMessage()+' \n Check the data_tier variable in crab.cfg !\n'
128 <          raise DataDiscoveryError(msg)
129 <        except DBSError, ex:
130 <          raise DataDiscoveryError(ex.getErrorMessage())
131 <
132 <        ## check that the user asks for parent Data Tier really existing in the DBS provenance
133 <        self.checkParentDataTier(parents, self.dataTiers, currentdbsdataset)
134 <
135 <        ## for each parent get the corresponding fileblocks
136 <        for aparent in parents:
137 <           ## fill a list of dbspaths
138 <           parentdbsdataset=aparent.getDatasetPath()
139 <           self.dbspaths.append(parentdbsdataset)
140 <           pdbs=DBSInfo(parentdbsdataset,[])
141 <           try:
142 <             parentsblocksinfo=pdbs.getDatasetContents()
143 <           except DBSError, ex:
144 <            raise DataDiscoveryError(ex.getErrorMessage())
145 <
146 <           self.allblocks.append(parentsblocksinfo.keys()) # add parent fileblocksinfo
147 <
148 <        ## all the required blocks
149 <        self.allblocks.append(self.blocksinfo.keys()) # add also the current fileblocksinfo
150 <
151 <
152 < # #################################################
153 <    def checkParentDataTier(self, parents, user_datatiers, currentdbsdataset ):
154 <        """
155 <         check that the data tiers requested by the user really exists in the provenance of the given dataset
156 <        """
157 <
158 <        current_datatier=string.split(currentdbsdataset,'/')[2]
159 <
160 <        parent_datatypes=[]
121 <        for aparent in parents:
122 <          parent_datatypes.append(aparent.getDataType())
123 <
124 <        for datatier in user_datatiers:
125 <          if parent_datatypes.count(datatier)<=0:
126 <             # the current datatier is not supposed to be in the provenance
127 <             if not (datatier == current_datatier):  
128 <              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 !"%(currentdbsdataset,datatier)
129 <              raise  NoDataTierinProvenanceError(msg)
109 >            if len(runselection) <= 0 :
110 >                files = api.listDatasetFiles(self.datasetPath)
111 >            else :
112 >                files = api.listFiles(path=self.datasetPath, details=True)
113 >        except DbsBadRequest, msg:
114 >            raise DataDiscoveryError(msg)
115 >        except DBSError, msg:
116 >            raise DataDiscoveryError(msg)
117 >
118 >        # parse files and fill arrays
119 >        for file in files :
120 >            filename = file['LogicalFileName']
121 >            if filename.find('.dat') < 0 :
122 >                fileblock = file['Block']['Name']
123 >                events    = file['NumberOfEvents']
124 >                continue_flag = 0
125 >                if len(runselection) > 0 :
126 >                    runslist = file['RunsList']
127 >                    for run in runslist :
128 >                        runnumber = run['RunNumber']
129 >                        for selected_run in runselection :
130 >                            if runnumber == selected_run :
131 >                                continue_flag = 1
132 >                else :
133 >                    continue_flag = 1
134 >
135 >                if continue_flag == 1 :
136 >                    # number of events per block
137 >                    if fileblock in self.eventsPerBlock.keys() :
138 >                        self.eventsPerBlock[fileblock] += events
139 >                    else :
140 >                        self.eventsPerBlock[fileblock] = events
141 >
142 >                    # number of events per file
143 >                    self.eventsPerFile[filename] = events
144 >
145 >                    # number of events per block
146 >                    if fileblock in self.blocksinfo.keys() :
147 >                        self.blocksinfo[fileblock].append(filename)
148 >                    else :
149 >                        self.blocksinfo[fileblock] = [filename]
150 >
151 >                    # total number of events
152 >                    self.maxEvents += events
153 >
154 >        for block in self.eventsPerBlock.keys() :
155 >            common.logger.debug(6,"DBSInfo: total nevts %i in block %s "%(self.eventsPerBlock[block],block))
156 >
157 >        if len(self.eventsPerBlock) <= 0:
158 >            raise NotExistingDatasetError(("\nNo data for %s in DBS\nPlease check"
159 >                                            + " dataset path variables in crab.cfg")
160 >                                            % self.datasetPath)
161  
162  
163   # #################################################
164      def getMaxEvents(self):
165          """
166 <         max events of the primary dataset-owner
166 >        max events
167          """
168 <        ## loop over the fileblocks of the primary dataset-owner
138 <        nevts=0      
139 <        for blockevts in self.blocksinfo.values():
140 <          nevts=nevts+blockevts
141 <
142 <        return nevts
168 >        return self.maxEvents
169  
170   # #################################################
171 <    def getDBSPaths(self):
171 >    def getEventsPerBlock(self):
172          """
173 <         list the DBSpaths for all required data
173 >        list the event collections structure by fileblock
174          """
175 <        return self.dbspaths
175 >        return self.eventsPerBlock
176  
177   # #################################################
178 <    def getEVC(self):
178 >    def getEventsPerFile(self):
179          """
180 <         list the event collections structure by fileblock
180 >        list the event collections structure by file
181          """
182 <        print "To be used by a more complex job splitting... TODO later... "
157 <        print "it requires changes in what's returned by DBSInfo.getDatasetContents and then fetchDBSInfo"
182 >        return self.eventsPerFile
183  
184   # #################################################
185 <    def getFileBlocks(self):
185 >    def getFiles(self):
186          """
187 <         fileblocks for all required dataset-owners
187 >        return files grouped by fileblock
188          """
189 <        return self.allblocks        
189 >        return self.blocksinfo        
190  
191   ########################################################################
167
168

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines