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
    rc = checkFile(fileName = fname, the_type = 'event', requireTree = True)
    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))