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 75 of file edmIntegrityCheck.py.

Constructor & Destructor Documentation

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

Definition at line 77 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 315 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 296 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 300 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 92 of file edmIntegrityCheck.py.

References edmIntegrityCheck.IntegrityCheck.dataset, genericValidation.GenericValidationData.dataset, edmIntegrityCheck.IntegrityCheck.eventsTotal, FileExportPlugin.FileExportPlugin.options, cmsswPreprocessor.CmsswPreprocessor.options, DTCalibrationWorker.DTCalibrationWorker.options, DTWorkflow.DTWorkflow.options, TestProcess.TestProcess.options, DOTExport.DotProducer.options, confdb.HLTProcess.options, edmIntegrityCheck.IntegrityCheck.options, and validateAlignments.ValidationJobMultiIOV.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 214 of file edmIntegrityCheck.py.

References pat::GenericDuplicateRemover< Comparator, Arbitrator >.duplicates(), LumiList.LumiList.duplicates, edmIntegrityCheck.IntegrityCheck.eventsSeen, edmIntegrityCheck.IntegrityCheck.eventsTotal, reco::HitPattern.int::test::TestHitPattern::test(), edm::RunningAverage.int::test_average::running_average::test(), join(), edm.print(), str, value_test.ValueTestCase.test(), eventstfile_test.EventsTFileTestCase.test(), pat::Flags.test(), helper::Parser.test(), reco::PFBlock::Link.test, pftools::CaloBox.test(), helper::ScannerBase.test(), MiniFloatConverter::ReduceMantissaToNbitsRounding.test, DiMuonHistograms.test, XMLProcessor.test(), edmIntegrityCheck.IntegrityCheck.test(), edm::test::TestProcessor.test(), DTTFBitArray< N >.test(), BitArray< N >.test(), TwoObjectVariable< LHS, lLHS, RHS, lRHS, Calculator >::getObject.test, cond::SmallWORMDict.test::SmallWORMDict::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 six.iteritems(self.test_result):
             print('Directory: %s' % dirname)
             for name, status in six.iteritems(files):
                 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 304 of file edmIntegrityCheck.py.

References mps_setup.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 292 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 108 of file edmIntegrityCheck.py.

References mps_setup.append, edmIntegrityCheck.int, list(), SiStripPI.max, min(), and edmIntegrityCheck.IntegrityCheck.test_result.

Referenced by edmIntegrityCheck.IntegrityCheck.test().

     def stripDuplicates(self):
         
         import re
         
         filemask = {}
         for dirname, files in six.iteritems(self.test_result):
             for name, status in six.iteritems(files):
                 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 range(mmin, mmax+1):
             if i in files:
                 duplicates = sorted(files[i])
 
                 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 239 of file edmIntegrityCheck.py.

References mps_setup.append, edmIntegrityCheck.IntegrityCheck.dataset, genericValidation.GenericValidationData.dataset, pat::GenericDuplicateRemover< Comparator, Arbitrator >.duplicates(), LumiList.LumiList.duplicates, edmIntegrityCheck.IntegrityCheck.eventsSeen, edmIntegrityCheck.IntegrityCheck.eventsTotal, reco::HitPattern.int::test::TestHitPattern::test(), edm::RunningAverage.int::test_average::running_average::test(), edmIntegrityCheck.IntegrityCheck.stageHost(), value_test.ValueTestCase.test(), eventstfile_test.EventsTFileTestCase.test(), pat::Flags.test(), helper::Parser.test(), reco::PFBlock::Link.test, pftools::CaloBox.test(), helper::ScannerBase.test(), MiniFloatConverter::ReduceMantissaToNbitsRounding.test, DiMuonHistograms.test, XMLProcessor.test(), edmIntegrityCheck.IntegrityCheck.test(), edm::test::TestProcessor.test(), DTTFBitArray< N >.test(), BitArray< N >.test(), TwoObjectVariable< LHS, lLHS, RHS, lRHS, Calculator >::getObject.test, cond::SmallWORMDict.test::SmallWORMDict::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 six.iteritems(self.test_result):
             report['PathList'].append(dirname)
             for name, status in six.iteritems(files):
                 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 162 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 six.iteritems(previous['Files']):
                 prev_results[name] = status
         
         filesToTest = self.sortByBaseDir(self.listRootFiles(self.directory))
         for dir, filelist in six.iteritems(filesToTest):
             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("Warning: The wildcard '%s' does not match any files in '%s'. Please check you are using quotes." % (self.options.wildcard,self.directory), file=sys.stderr)
 
             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), end=' ')
                     OK, num = prev_results[lfn]
                 else:
                     if self.options.printout:
                         print('[%i/%i]\t Checking %s...' % (count, len(filtered),fname), end=' ')
                     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 325 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 331 of file edmIntegrityCheck.py.

References edm.print(), 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("ERROR:\tedmFileUtil timed out for lfn '%s' (%d)" % (lfn,timeout), file=sys.stderr)
                 return (False,-1)
         if timeout > 0:
             return tf(lfn)
         else:
             return self.testFile(lfn)
 
 
 