edmIntegrityCheck.IntegrityCheck Class Reference

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

Constructor & Destructor Documentation

__init__()

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

Definition at line 76 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

getParseNumberOfEvents()

def edmIntegrityCheck.IntegrityCheck.getParseNumberOfEvents	(		self,
			output
	)

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

Definition at line 314 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

listFiles()

def edmIntegrityCheck.IntegrityCheck.listFiles	(		self,
			dir
	)

Recursively list a file or directory on castor

Definition at line 295 of file edmIntegrityCheck.py.

References cmsswPreprocessor.CmsswPreprocessor.options, DTCalibrationWorker.DTCalibrationWorker.options, DTWorkflow.DTWorkflow.options, TestProcess.TestProcess.options, confdb.HLTProcess.options, and edmIntegrityCheck.IntegrityCheck.options.

Referenced by edmIntegrityCheck.IntegrityCheck.listRootFiles().

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

listRootFiles()

def edmIntegrityCheck.IntegrityCheck.listRootFiles	(		self,
			dir
	)

filter out filenames so that they only contain root files

Definition at line 299 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')]

query()

def edmIntegrityCheck.IntegrityCheck.query

(

self

)

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

Definition at line 91 of file edmIntegrityCheck.py.

References edmIntegrityCheck.IntegrityCheck.dataset, upgradeWorkflowComponents.UpgradeFragment.dataset, DiLeptonHelp::Counts.eventsTotal, edmIntegrityCheck.IntegrityCheck.eventsTotal, StraightTrackAlignment.eventsTotal, cmsswPreprocessor.CmsswPreprocessor.options, DTCalibrationWorker.DTCalibrationWorker.options, DTWorkflow.DTWorkflow.options, TestProcess.TestProcess.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)
    

report()

def edmIntegrityCheck.IntegrityCheck.report

(

self

)

Definition at line 213 of file edmIntegrityCheck.py.

References pat::GenericDuplicateRemover< Comparator, Arbitrator >.duplicates(), LumiList.LumiList.duplicates, edmIntegrityCheck.IntegrityCheck.eventsSeen, DiLeptonHelp::Counts.eventsTotal, edmIntegrityCheck.IntegrityCheck.eventsTotal, StraightTrackAlignment.eventsTotal, reco::HitPattern.int ::test::TestHitPattern::test(), edm::RunningAverage.int ::test_average::running_average::test(), mps_monitormerge.items, join(), cmsswPreprocessor.CmsswPreprocessor.options, DTCalibrationWorker.DTCalibrationWorker.options, DTWorkflow.DTWorkflow.options, TestProcess.TestProcess.options, confdb.HLTProcess.options, edmIntegrityCheck.IntegrityCheck.options, print(), str, value_test.ValueTestCase.test(), eventstfile_test.EventsTFileTestCase.test(), pat::Flags.test(), reco::PFBlock::Link.test, helper::Parser.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 self.test_result.items():
            print('Directory: %s' % dirname)
            for name, status in files.items():
                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]))
        

sortByBaseDir()

def edmIntegrityCheck.IntegrityCheck.sortByBaseDir	(		self,
			files
	)

Sort files into directories

Definition at line 303 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

stageHost()

def edmIntegrityCheck.IntegrityCheck.stageHost

(

self

)

Returns the CASTOR instance to use

Definition at line 291 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')

stripDuplicates()

def edmIntegrityCheck.IntegrityCheck.stripDuplicates

(

self

)

Definition at line 107 of file edmIntegrityCheck.py.

References mps_setup.append, edmIntegrityCheck.int, mps_monitormerge.items, SiStripPI.max, SiStripPI.min, FastTimerService_cff.range, and edmIntegrityCheck.IntegrityCheck.test_result.

Referenced by edmIntegrityCheck.IntegrityCheck.test().

    def stripDuplicates(self):
        
        import re
        
        filemask = {}
        for dirname, files in self.test_result.items():
            for name, status in files.items():
                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
    

structured()

def edmIntegrityCheck.IntegrityCheck.structured

(

self

)

Definition at line 238 of file edmIntegrityCheck.py.

References mps_setup.append, edmIntegrityCheck.IntegrityCheck.dataset, upgradeWorkflowComponents.UpgradeFragment.dataset, pat::GenericDuplicateRemover< Comparator, Arbitrator >.duplicates(), LumiList.LumiList.duplicates, edmIntegrityCheck.IntegrityCheck.eventsSeen, DiLeptonHelp::Counts.eventsTotal, edmIntegrityCheck.IntegrityCheck.eventsTotal, StraightTrackAlignment.eventsTotal, reco::HitPattern.int ::test::TestHitPattern::test(), edm::RunningAverage.int ::test_average::running_average::test(), mps_monitormerge.items, cmsswPreprocessor.CmsswPreprocessor.options, DTCalibrationWorker.DTCalibrationWorker.options, DTWorkflow.DTWorkflow.options, TestProcess.TestProcess.options, confdb.HLTProcess.options, edmIntegrityCheck.IntegrityCheck.options, edmIntegrityCheck.IntegrityCheck.stageHost(), value_test.ValueTestCase.test(), eventstfile_test.EventsTFileTestCase.test(), pat::Flags.test(), reco::PFBlock::Link.test, helper::Parser.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 self.test_result.items():
            report['PathList'].append(dirname)
            for name, status in files.items():
                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
    

test()

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

Definition at line 161 of file edmIntegrityCheck.py.

References Book.directory, edmIntegrityCheck.IntegrityCheck.directory, pat::GenericDuplicateRemover< Comparator, Arbitrator >.duplicates(), LumiList.LumiList.duplicates, edmIntegrityCheck.IntegrityCheck.eventsSeen, mps_monitormerge.items, edmIntegrityCheck.IntegrityCheck.listRootFiles(), cmsswPreprocessor.CmsswPreprocessor.options, DTCalibrationWorker.DTCalibrationWorker.options, DTWorkflow.DTWorkflow.options, TestProcess.TestProcess.options, confdb.HLTProcess.options, edmIntegrityCheck.IntegrityCheck.options, print(), DbQuery.query, edmIntegrityCheck.IntegrityCheck.query(), upload_popcon.HTTP.query(), confdbOfflineConverter.OfflineConverter.query(), cond::OMSService.query(), uploadConditions.HTTP.query(), edmIntegrityCheck.IntegrityCheck.sortByBaseDir(), edmIntegrityCheck.IntegrityCheck.stripDuplicates(), edmIntegrityCheck.IntegrityCheck.test_result, and edmIntegrityCheck.IntegrityCheck.testFileTimeOut().

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'].items():
                prev_results[name] = status
        
        filesToTest = self.sortByBaseDir(self.listRootFiles(self.directory))
        for dir, filelist in filesToTest.items():
            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
    

testFile()

def edmIntegrityCheck.IntegrityCheck.testFile	(		self,
			lfn
	)

Definition at line 324 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))
    

testFileTimeOut()

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

Definition at line 330 of file edmIntegrityCheck.py.

References 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)

Member Data Documentation

dataset

edmIntegrityCheck.IntegrityCheck.dataset

Definition at line 80 of file edmIntegrityCheck.py.

Referenced by edmIntegrityCheck.IntegrityCheck.query(), production_tasks.BaseDataset.query(), production_tasks.CheckDatasetExists.run(), production_tasks.BaseDataset.run(), production_tasks.GenerateMask.run(), production_tasks.CreateJobDirectory.run(), production_tasks.SourceCFG.run(), production_tasks.FullCFG.run(), production_tasks.WriteToDatasets.run(), production_tasks.RunCMSBatch.run(), production_tasks.MonitorJobs.run(), production_tasks.CleanJobFiles.run(), and edmIntegrityCheck.IntegrityCheck.structured().

directory

edmIntegrityCheck.IntegrityCheck.directory

Definition at line 83 of file edmIntegrityCheck.py.

Referenced by utils.unpickler.__init__(), dqm_interfaces.DirFetcher.run(), dqm_interfaces.DirWalkerDB.run(), utils.unpickler.run(), edmIntegrityCheck.IntegrityCheck.test(), and dqm_interfaces.DirWalkerFile.walk().

eventsSeen

edmIntegrityCheck.IntegrityCheck.eventsSeen

Definition at line 87 of file edmIntegrityCheck.py.

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

eventsTotal

edmIntegrityCheck.IntegrityCheck.eventsTotal

Definition at line 86 of file edmIntegrityCheck.py.

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

options

edmIntegrityCheck.IntegrityCheck.options

Definition at line 81 of file edmIntegrityCheck.py.

Referenced by betterConfigParser.BetterConfigParser.__updateDict(), submitPVValidationJobs.BetterConfigParser.__updateDict(), CrabHelper.CrabHelper.check_crabtask(), betterConfigParser.BetterConfigParser.checkInput(), CrabHelper.CrabHelper.crab_config_filename(), CrabHelper.CrabHelper.crab_config_filepath(), CrabHelper.CrabHelper.crab_taskname(), CrabHelper.CrabHelper.fill_options_from_crab_config(), edmIntegrityCheck.IntegrityCheck.listFiles(), production_tasks.MonitorJobs.monitor(), edmIntegrityCheck.IntegrityCheck.query(), production_tasks.BaseDataset.query(), edmIntegrityCheck.IntegrityCheck.report(), production_tasks.CheckDatasetExists.run(), production_tasks.BaseDataset.run(), production_tasks.GenerateMask.run(), production_tasks.CreateJobDirectory.run(), production_tasks.SourceCFG.run(), production_tasks.FullCFG.run(), production_tasks.WriteToDatasets.run(), production_tasks.RunCMSBatch.run(), production_tasks.MonitorJobs.run(), production_tasks.CheckJobStatus.run(), production_tasks.WriteJobReport.run(), production_tasks.CleanJobFiles.run(), edmIntegrityCheck.IntegrityCheck.structured(), CrabHelper.CrabHelper.submit_crab_task(), edmIntegrityCheck.IntegrityCheck.test(), and CrabHelper.CrabHelper.write_crabConfig().

test_result

edmIntegrityCheck.IntegrityCheck.test_result

Definition at line 89 of file edmIntegrityCheck.py.

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

topdir

edmIntegrityCheck.IntegrityCheck.topdir

Definition at line 82 of file edmIntegrityCheck.py.

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