Inheritance diagram for edmIntegrityCheck.IntegrityCheck:

Public Member Functions
def	__init__ (self, dataset, options)

def	getParseNumberOfEvents (self, output)

def	listFiles (self, dir)

def	listRootFiles (self, dir)

def	query (self)

def	report (self)

def	sortByBaseDir (self, files)

def	stageHost (self)

def	stripDuplicates (self)

def	structured (self)

def	test (self, previous=None, timeout=-1)

def	testFile (self, lfn)

def	testFileTimeOut (self, lfn, timeout)

Public Attributes
	dataset

	directory

	eventsSeen

	eventsTotal

	options

	test_result

	topdir

Detailed Description

Definition at line 71 of file edmIntegrityCheck.py.

Constructor & Destructor Documentation

def edmIntegrityCheck.IntegrityCheck.__init__	(	self,
		dataset,
		options
	)

Definition at line 73 of file edmIntegrityCheck.py.

     def __init__(self, dataset, options):
         if not dataset.startswith(os.sep):
             dataset = os.sep + dataset
 
         self.dataset = dataset
         self.options = options
         self.topdir = castortools.lfnToCastor( castorBaseDir(user=options.user) )
         self.directory = os.path.join(self.topdir, *self.dataset.split(os.sep))
         
         #event counters
         self.eventsTotal = -1
         self.eventsSeen = 0
         
         self.test_result = None
     

Member Function Documentation

def edmIntegrityCheck.IntegrityCheck.getParseNumberOfEvents	(	self,
		output
	)

Parse the output of edmFileUtil to get the number of events found

Definition at line 312 of file edmIntegrityCheck.py.

References edmIntegrityCheck.int.

Referenced by edmIntegrityCheck.IntegrityCheck.testFile().

     def getParseNumberOfEvents(self,output):
         """Parse the output of edmFileUtil to get the number of events found"""
         tokens = output.split(' ')
         result = -2
         try:
             result = int(tokens[-4])
         except ValueError:
             pass
         return result
 

def edmIntegrityCheck.IntegrityCheck.listFiles	(	self,
		dir
	)

Recursively list a file or directory on castor

Definition at line 293 of file edmIntegrityCheck.py.

Referenced by edmIntegrityCheck.IntegrityCheck.listRootFiles().

     def listFiles(self,dir):
         """Recursively list a file or directory on castor"""
         return castortools.listFiles(dir,self.options.resursive)
 

def edmIntegrityCheck.IntegrityCheck.listRootFiles	(	self,
		dir
	)

filter out filenames so that they only contain root files

Definition at line 297 of file edmIntegrityCheck.py.

References edmIntegrityCheck.IntegrityCheck.listFiles().

Referenced by edmIntegrityCheck.IntegrityCheck.test().

     def listRootFiles(self,dir):
         """filter out filenames so that they only contain root files"""
         return [f for f in self.listFiles(dir) if f.lower().endswith('.root')]
 

def edmIntegrityCheck.IntegrityCheck.query ( self )

Query DAS to find out how many events are in the dataset

Definition at line 88 of file edmIntegrityCheck.py.

References CalibratedPatElectronProducer.dataset, CalibratedElectronProducer.dataset, edmIntegrityCheck.IntegrityCheck.dataset, genericValidation.GenericValidationData.dataset, edmIntegrityCheck.IntegrityCheck.eventsTotal, FileExportPlugin.FileExportPlugin.options, cmsswPreprocessor.CmsswPreprocessor.options, DOTExport.DotProducer.options, eventsfwlite.Events.options, confdb.HLTProcess.options, and edmIntegrityCheck.IntegrityCheck.options.

Referenced by production_tasks.BaseDataset.run(), and edmIntegrityCheck.IntegrityCheck.test().

     def query(self):
         """Query DAS to find out how many events are in the dataset"""
         from production_tasks import BaseDataset
         base = BaseDataset(self.dataset, self.options.user, self.options)
 
         data = None
         output = base.run({})
         if 'Das' in output:
             self.options.name = output['Name']
             data = output['Das']
             
         if data is None:
             raise Exception("Dataset '%s' not found in Das. Please check." % self.dataset)
         #get the number of events in the dataset
         self.eventsTotal = CMSDataset.findPrimaryDatasetEntries(self.options.name, self.options.min_run, self.options.max_run)
     

def edmIntegrityCheck.IntegrityCheck.report ( self )

Definition at line 211 of file edmIntegrityCheck.py.

References pat::GenericDuplicateRemover< Comparator, Arbitrator >.duplicates(), LumiList.LumiList.duplicates, edmIntegrityCheck.IntegrityCheck.eventsSeen, edmIntegrityCheck.IntegrityCheck.eventsTotal, edm::RunningAverage.int::test::running_average::test(), join(), harvestTrackValidationPlots.str, value_test.ValueTestCase.test(), eventstfile_test.EventsTFileTestCase.test(), pat::Flags.test(), helper::Parser.test(), pftools::CaloBox.test(), reco::PFBlock::Link.test, helper::ScannerBase.test(), DiMuonHistograms.test, edmIntegrityCheck.IntegrityCheck.test(), XMLProcessor.test(), BitArray< N >.test(), DTTFBitArray< N >.test(), TwoObjectVariable< LHS, lLHS, RHS, lRHS, Calculator >::getObject.test, cond::SmallWORMDict.test::SmallWORMDict::test, reco::HitPattern.test::TestHitPattern::test, edmIntegrityCheck.IntegrityCheck.test_result, and edmIntegrityCheck.IntegrityCheck.topdir.

     def report(self):
         
         if self.test_result is None:
             self.test()
             
         print 'DBS Dataset name: %s' % self.options.name
         print 'Storage path: %s' % self.topdir
         
         for dirname, files in self.test_result.iteritems():
             print 'Directory: %s' % dirname
             for name, status in files.iteritems():
                 fname = os.path.join(dirname, name)
                 if not fname in self.duplicates:
                     print '\t\t %s: %s' % (name, str(status))
                 else:
                     print '\t\t %s: %s (Valid duplicate)' % (name, str(status))
         print 'Total entries in DBS: %i' % self.eventsTotal
         print 'Total entries in processed files: %i' % self.eventsSeen
         if self.eventsTotal>0:
             print 'Fraction of dataset processed: %f' % (self.eventsSeen/(1.*self.eventsTotal))
         else:
             print 'Total entries in DBS not determined' 
         if self.bad_jobs:
             print "Bad Crab Jobs: '%s'" % ','.join([str(j) for j in self.bad_jobs])
         

def edmIntegrityCheck.IntegrityCheck.sortByBaseDir	(	self,
		files
	)

Sort files into directories

Definition at line 301 of file edmIntegrityCheck.py.

References mps_alisetup.append.

Referenced by edmIntegrityCheck.IntegrityCheck.test().

     def sortByBaseDir(self,files):
         """Sort files into directories"""
         result = {}
         for f in files:
             dirname = os.path.dirname(f)
             filename = os.path.basename(f)
             if dirname not in result: result[dirname] = []
             result[dirname].append(filename)
         return result
 
 

def edmIntegrityCheck.IntegrityCheck.stageHost ( self )

Returns the CASTOR instance to use

Definition at line 289 of file edmIntegrityCheck.py.

Referenced by edmIntegrityCheck.IntegrityCheck.structured().

     def stageHost(self):
         """Returns the CASTOR instance to use"""
         return os.environ.get('STAGE_HOST','castorcms')
 

def edmIntegrityCheck.IntegrityCheck.stripDuplicates ( self )

Definition at line 104 of file edmIntegrityCheck.py.

References mps_alisetup.append, edmIntegrityCheck.int, list(), hpstanc_transforms.max, and min().

Referenced by edmIntegrityCheck.IntegrityCheck.test().

     def stripDuplicates(self):
         
         import re
         
         filemask = {}
         for dirname, files in self.test_result.iteritems():
             for name, status in files.iteritems():
                 fname = os.path.join(dirname, name)
                 filemask[fname] = status
         
         def isCrabFile(name):
             _, fname = os.path.split(name)
             base, _ = os.path.splitext(fname)
             return re.match(".*_\d+_\d+_\w+$", base) is not None, base
         def getCrabIndex(base):
             tokens = base.split('_')
             if len(tokens) > 2:
                 return (int(tokens[-3]), int(tokens[-2]))
             return None
             
         files = {}
         
         mmin = 1000000000
         mmax = -100000000
         for f in filemask:
             isCrab, base = isCrabFile(f)
             if isCrab:
                 index = getCrabIndex(base)
                 if index is not None:
                     jobid, retry = index
                     
                     mmin = min(mmin, jobid)
                     mmax = max(mmax, jobid)
                     if jobid in files and filemask[f][0]:
                         files[jobid].append((retry, f))
                     elif filemask[f][0]:
                         files[jobid] = [(retry, f)]
 
         good_duplicates = {}
         bad_jobs = set()
         sum_dup = 0
         for i in xrange(mmin, mmax+1):
             if i in files:
                 duplicates = files[i]
                 duplicates.sort()
 
                 fname = duplicates[-1][1]
                 if len(duplicates) > 1:
                     for d in duplicates[:-1]:
                         good_duplicates[d[1]] = filemask[d[1]][1]
                         sum_dup += good_duplicates[d[1]]
             else:
                 bad_jobs.add(i)
         return good_duplicates, sorted(list(bad_jobs)), sum_dup
     

def edmIntegrityCheck.IntegrityCheck.structured ( self )

Definition at line 236 of file edmIntegrityCheck.py.

References mps_alisetup.append, CalibratedPatElectronProducer.dataset, CalibratedElectronProducer.dataset, edmIntegrityCheck.IntegrityCheck.dataset, genericValidation.GenericValidationData.dataset, pat::GenericDuplicateRemover< Comparator, Arbitrator >.duplicates(), LumiList.LumiList.duplicates, edmIntegrityCheck.IntegrityCheck.eventsSeen, edmIntegrityCheck.IntegrityCheck.eventsTotal, edm::RunningAverage.int::test::running_average::test(), edmIntegrityCheck.IntegrityCheck.stageHost(), value_test.ValueTestCase.test(), eventstfile_test.EventsTFileTestCase.test(), pat::Flags.test(), helper::Parser.test(), pftools::CaloBox.test(), reco::PFBlock::Link.test, helper::ScannerBase.test(), DiMuonHistograms.test, edmIntegrityCheck.IntegrityCheck.test(), XMLProcessor.test(), DTTFBitArray< N >.test(), BitArray< N >.test(), TwoObjectVariable< LHS, lLHS, RHS, lRHS, Calculator >::getObject.test, cond::SmallWORMDict.test::SmallWORMDict::test, reco::HitPattern.test::TestHitPattern::test, edmIntegrityCheck.IntegrityCheck.test_result, and edmIntegrityCheck.IntegrityCheck.topdir.

     def structured(self):
         
         if self.test_result is None:
             self.test()
         
         totalGood = 0
         totalBad = 0
 
         report = {'data':{},
                   'ReportVersion':3,
                   'PrimaryDataset':self.options.name,
                   'Name':self.dataset,
                   'PhysicsGroup':'CMG',
                   'Status':'VALID',
                   'TierList':[],
                   'AlgoList':[],
                   'RunList':[],
                   'PathList':[],
                   'Topdir':self.topdir,
                   'StageHost':self.stageHost(),
                   'CreatedBy':self.options.user,
                   'DateCreated':datetime.datetime.now().strftime("%s"),
                   'Files':{}}
         
         for dirname, files in self.test_result.iteritems():
             report['PathList'].append(dirname)
             for name, status in files.iteritems():
                 fname = os.path.join(dirname, name)
                 report['Files'][fname] = status
                 if status[0]:
                     totalGood += 1
                 else:
                     totalBad += 1
                 
         report['PrimaryDatasetEntries'] = self.eventsTotal
         if self.eventsTotal>0:
             report['PrimaryDatasetFraction'] = (self.eventsSeen/(1.*self.eventsTotal))
         else:
             report['PrimaryDatasetFraction'] = -1.
         report['FilesEntries'] = self.eventsSeen
 
         report['FilesGood'] = totalGood
         report['FilesBad'] = totalBad
         report['FilesCount'] = totalGood + totalBad
         
         report['BadJobs'] = self.bad_jobs
         report['ValidDuplicates'] = self.duplicates
         
         report['MinRun'] = self.options.min_run
         report['MaxRun'] = self.options.max_run
 
         return report
     

def edmIntegrityCheck.IntegrityCheck.test	(	self,
		previous = `None`,
		timeout = `-1`
	)

Definition at line 159 of file edmIntegrityCheck.py.

Referenced by edmIntegrityCheck.IntegrityCheck.report(), and edmIntegrityCheck.IntegrityCheck.structured().

     def test(self, previous = None, timeout = -1):
         if not castortools.fileExists(self.directory):
             raise Exception("The top level directory '%s' for this dataset does not exist" % self.directory)
 
         self.query()
 
         test_results = {}
 
         #support updating to speed things up
         prev_results = {}
         if previous is not None:
             for name, status in previous['Files'].iteritems():
                 prev_results[name] = status
         
         filesToTest = self.sortByBaseDir(self.listRootFiles(self.directory))
         for dir, filelist in filesToTest.iteritems():
             filemask = {}
             #apply a UNIX wildcard if specified
             filtered = filelist
             if self.options.wildcard is not None:
                 filtered = fnmatch.filter(filelist, self.options.wildcard)
                 if not filtered:
                     print >> sys.stderr, "Warning: The wildcard '%s' does not match any files in '%s'. Please check you are using quotes." % (self.options.wildcard,self.directory)
 
             count = 0
             for ff in filtered:
                 fname = os.path.join(dir, ff)
                 lfn = castortools.castorToLFN(fname)
                 
                 #try to update from the previous result if available 
                 if lfn in prev_results and prev_results[lfn][0]:
                     if self.options.printout:
                         print '[%i/%i]\t Skipping %s...' % (count, len(filtered),fname),
                     OK, num = prev_results[lfn]
                 else:
                     if self.options.printout:
                         print '[%i/%i]\t Checking %s...' % (count, len(filtered),fname),
                     OK, num = self.testFileTimeOut(lfn, timeout)
 
                 filemask[ff] = (OK,num)
                 if self.options.printout:
                     print (OK, num)
                 if OK:
                     self.eventsSeen += num
                 count += 1
             test_results[castortools.castorToLFN(dir)] = filemask
         self.test_result = test_results
 
         self.duplicates, self.bad_jobs, sum_dup = self.stripDuplicates()
         #remove duplicate entries from the event count
         self.eventsSeen -= sum_dup
     

def edmIntegrityCheck.IntegrityCheck.testFile	(	self,
		lfn
	)

Definition at line 322 of file edmIntegrityCheck.py.

References communicate(), and edmIntegrityCheck.IntegrityCheck.getParseNumberOfEvents().

Referenced by edmIntegrityCheck.IntegrityCheck.testFileTimeOut().

     def testFile(self,lfn):
         stdout = subprocess.Popen(['edmFileUtil',lfn], stdout=subprocess.PIPE,stderr=subprocess.PIPE).communicate()[0]
         for error in ["Fatal Root Error","Could not open file","Not a valid collection"]:
             if error in stdout: return (False,-1)
         return (True,self.getParseNumberOfEvents(stdout))
     

def edmIntegrityCheck.IntegrityCheck.testFileTimeOut	(	self,
		lfn,
		timeout
	)

Definition at line 328 of file edmIntegrityCheck.py.

References AlignmentIORootBase.testFile(), edmIntegrityCheck.IntegrityCheck.testFile(), and timeout.timed_out().

Referenced by edmIntegrityCheck.IntegrityCheck.test().

     def testFileTimeOut(self,lfn, timeout):
         @timed_out(timeout)
         def tf(lfn):
             try:
                 return self.testFile(lfn)
             except TimedOutExc as e:
                 print >> sys.stderr, "ERROR:\tedmFileUtil timed out for lfn '%s' (%d)" % (lfn,timeout)
                 return (False,-1)
         if timeout > 0:
             return tf(lfn)
         else:
             return self.testFile(lfn)
 
 
 