NTUPMerge_tf.py

Go to the documentation of this file.

#! /usr/bin/env python
 
# Copyright (C) 2002-2025 CERN for the benefit of the ATLAS collaboration
 
 
 
import sys
import time
 
# Setup core logging here
from PyJobTransforms.trfLogger import msg
msg.info('logging set in %s' % sys.argv[0])
 
from PyJobTransforms.transform import transform
from PyJobTransforms.trfArgs import addExtraDPDTypes
from PyJobTransforms.trfDecorators import stdTrfExceptionHandler, sigUsrStackTrace
from PATJobTransforms.PATTransformUtils import addNTUPMergeSubsteps, addPhysValidationMergeFiles
 
@stdTrfExceptionHandler
@sigUsrStackTrace
def main():
    
    msg.info('This is %s' % sys.argv[0])
    if sys.argv[1:] == []:
        msg.info("%s stopped at %s, no input parameters given" % (sys.argv[0], time.asctime()))
    
    trf = getTransform()
    trf.parseCmdLineArgs(sys.argv[1:])
    trf.execute()
    trf.generateReport()
 
    msg.info("%s stopped at %s, tf exit code %d" % (sys.argv[0], time.asctime(), trf.exitCode))
    sys.exit(trf.exitCode)
 
def getTransform():
    msg.debug("in getTransform...")
 
    # get the default executor list
    executorSet = set()
    # instantiate a transform with no steps
    trf = transform(executor = executorSet, description = 'ATLAS NTUPLE merge and post-processing transform')
    
    # add custom merge and post-processing 
    # steering parameters and get the 'args'
    addPhysValidationMergeFiles(trf.parser)
    args = trf.parser.parse_args(args=['--skipPostProcessing'])
 
    # get the modified executor
    mergeStepSet = set()
 
    # check if the user used custom parameters
    # add to the transform the merge and 
    # post-processing steps conditionally 
    # based on user's input
    addNTUPMergeSubsteps(mergeStepSet, skip_post_processing = args.skipPostProcessing)
    trf.appendToExecutorSet(list(mergeStepSet))
 
    # additional setup
    addExtraDPDTypes(trf.parser, transform=trf, NTUPMergerArgs = True)
    return trf
 
if __name__ == '__main__':
    main()