HeppyCore/scripts/edmIntegrityCheck.py

0001 #!/usr/bin/env python3
0002
0003 from PhysicsTools.HeppyCore.utils.edmIntegrityCheck import PublishToFileSystem, IntegrityCheck
0004 import das
0005
0006 import copy, os
0007
0008 if __name__ == '__main__':
0009
0010     from optparse import OptionParser, OptionGroup
0011
0012     usage = """usage: %prog [options] /Sample/Name/On/Castor
0013
0014 e.g.: %prog -u wreece -p -w 'PFAOD_*.root' /MultiJet/Run2011A-05Aug2011-v1/AOD/V2
0015     """
0016     das = das.DASOptionParser(usage=usage)
0017     group = OptionGroup(das.parser,'edmIntegrityCheck Options','Options related to checking files on CASTOR')
0018
0019     group.add_option("-d", "--device", dest="device", default='cmst3',help="The storage device to write to, e.g. 'cmst3'")
0020     group.add_option("-n", "--name", dest="name", default=None,help='The name of the dataset in DAS. Will be guessed if not specified')
0021     group.add_option("-p", "--printout", dest="printout", default=False, action='store_true',help='Print a report to stdout')
0022     group.add_option("-r", "--recursive", dest="resursive", default=False, action='store_true',help='Walk the mass storage device recursively')
0023     group.add_option("-u", "--user", dest="user", default=os.environ['USER'],help='The username to use when looking at mass storage devices')
0024     group.add_option("-w", "--wildcard", dest="wildcard", default=None,help='A UNIX style wildcard to specify which files to check')
0025     group.add_option("--update", dest="update", default=False, action='store_true',help='Only update the status of corrupted files')
0026     group.add_option("-t","--timeout", dest="timeout", default=-1, type=int, help='Set a timeout on the edmFileUtil calls')
0027     group.add_option("--min-run", dest="min_run", default=-1, type=int, help='When querying DBS, require runs >= than this run')
0028     group.add_option("--max-run", dest="max_run", default=-1, type=int, help='When querying DBS, require runs <= than this run')
0029     group.add_option("--max_threads", dest="max_threads", default=None,help='The maximum number of threads to use')
0030     das.parser.add_option_group(group)
0031     (opts, datasets) = das.get_opt()
0032
0033     if len(datasets)==0:
0034         print(das.parser.print_help())
0035         print()
0036         print('need to provide a dataset in argument')
0037
0038     def work(d,op):
0039         tokens = d.split('%')
0040         if len(tokens) == 2:
0041             op.user = tokens[0]
0042             d = tokens[1]
0043
0044         check = IntegrityCheck(d,op)
0045         pub = PublishToFileSystem(check)
0046
0047         previous = None
0048         if op.update:
0049             previous = pub.get(check.directory)
0050
0051         check.test(previous = previous, timeout = op.timeout)
0052         if op.printout:
0053             check.report()
0054         report = check.structured()
0055         pub.publish(report)
0056
0057         return d
0058
0059     def callback(result):
0060         print('Checking thread done: ',str(result))
0061
0062     #submit the main work in a multi-threaded way
0063
0064     if len(datasets) == 1:
0065         d = datasets[0]
0066         work(d, copy.deepcopy(opts))
0067     else:
0068         import multiprocessing
0069         if opts.max_threads is not None and opts.max_threads:
0070             opts.max_threads = int(opts.max_threads)
0071         pool = multiprocessing.Pool(processes=opts.max_threads)
0072
0073         for d in datasets:
0074             pool.apply_async(work, args=(d,copy.deepcopy(opts)),callback=callback)
0075         pool.close()
0076         pool.join()