trfFileValidationFunctions.py

Go to the documentation of this file.

# Copyright (C) 2002-2025 CERN for the benefit of the ATLAS collaboration
 

 
import logging
msg = logging.getLogger(__name__)
 
import PyJobTransforms.trfExceptions as trfExceptions
 

def returnIntegrityOfPOOLFile(fname):
    from PyJobTransforms.trfValidateRootFile import checkFile, msg as logger
    import multiprocessing
 
    if (level := msg.getEffectiveLevel()) < logger.getEffectiveLevel():
        logger.setLevel(level)
        msg.debug(f"Set logging level of {logger.name!r} to {logging.getLevelName(level)!r}")
 
    msg.debug(f"Current process: {multiprocessing.current_process().name}")
 
    rc = checkFile(fileName=fname, the_type='event', requireTree=False)
    if rc == 0:
        return (True, "integrity of {fileName} good".format(fileName = str(fname)))
    else:
        return (False, "integrity of {fileName} bad: return code: {integrityStatus}".format(fileName = str(fname), integrityStatus = rc))
 

def returnIntegrityOfNTUPFile(fname):
    from PyJobTransforms.trfValidateRootFile import checkFile
    rc = checkFile(fileName = fname, the_type = 'basket', requireTree = False)
    if rc == 0:
        return (True, "integrity of {fileName} good".format(fileName = str(fname)))
    else:
        return (False, "integrity of {fileName} bad: return code: {integrityStatus}".format(fileName = str(fname), integrityStatus = rc))
 

def returnIntegrityOfBSFile(fname):
    try:
        from PyJobTransforms.trfUtils import call
        rc = call(["AtlListBSEvents", "-c", fname],
            logger = msg,
            message = "Report by AtlListBSEvents: ",
            timeout = None
        )
    except trfExceptions.TransformTimeoutException:
        return False
    if rc == 0:
        return (True, "integrity of {fileName} good".format(fileName = str(fname)))
    else:
        return (False, "integrity of {fileName} bad: return code: {integrityStatus}".format(fileName = str(fname), integrityStatus = rc))
 

def returnIntegrityOfHISTFile(fname):
    rc = 0 # (default behaviour)
    if rc == 0:
        return (True, "integrity of {fileName} good".format(fileName = str(fname)))
    else:
        return (False, "integrity of {fileName} bad: return code: {integrityStatus}".format(fileName = str(fname), integrityStatus = rc))