def createPoolFileCatalog(file_dictionary, lfns, pfc_name="PoolFileCatalog.xml", forceLogical=False):
    """
    Create the PoolFileCatalog.xml
    file_dictionary = { guid1 : sfn1, ... }
    Adapted from R. Walker's code
    Note: the logical tag can cause problems in some cases (for the payload).
    """

    outxml = ''
    if len(file_dictionary) == 0:
        tolog('No input files so no PFC created')
    else:
        dom = minidom.getDOMImplementation()
        doctype = dom.createDocumentType("POOLFILECATALOG","","InMemory")
        doc = dom.createDocument(None, "POOLFILECATALOG", doctype)
        root = doc.documentElement
        doc.appendChild(root)

        # Prepare plain text as can`t trust minidom on python <2.3
        pfc_text = '<?xml version="1.0" ?>\n'
        pfc_text += '<!-- Edited By the PanDA Pilot -->\n'
        pfc_text += '<!DOCTYPE POOLFILECATALOG  SYSTEM "InMemory">\n'
        pfc_text += '<POOLFILECATALOG>\n'

        # Strip .N because stagein makes soft link, and params have no .N
        for guid in file_dictionary.keys():
            sfn = file_dictionary[guid]
            ftype='ROOT_All'

            _file = doc.createElement("File")
            _file.setAttribute('ID', guid)
            root.appendChild(_file)

            # physical element - file in local directory without .N extension
            physical = doc.createElement("physical")
            _file.appendChild(physical)

            pfn = doc.createElement('pfn')
            pfn.setAttribute('filetype', ftype)
            pfn.setAttribute('name', sfn)
            physical.appendChild(pfn)

            # forceLogical is set for TURL based PFCs. In this case, the LFN must not contain any legacy __DQ2-parts
            if forceLogical:
                logical = doc.createElement('logical')
                logical.setAttribute('name', os.path.basename(sfn))
                _file.appendChild(logical)

                # remove any legacy __DQ2 substring from the LFN if necessary
                _lfn = getLFN(sfn, lfns) #os.path.basename(sfn)
                if "__DQ2" in _lfn:
                    _lfn = stripDQ2FromLFN(_lfn)

                # remove any rucio :-separator if present
                if ":" in _lfn:
                    _lfn = _lfn.split(":")[1]

                pfc_text += '  <File ID="%s">\n    <physical>\n      <pfn filetype="%s" name="%s"/>\n    </physical>\n    <logical>\n      <lfn name="%s"/>\n    </logical>\n  </File>\n' % (guid, ftype, sfn, _lfn)

            else:
                logical = doc.createElement('logical')
                _file.appendChild(logical)
                pfc_text += '  <File ID="%s">\n    <physical>\n      <pfn filetype="%s" name="%s"/>\n    </physical>\n    <logical/>\n  </File>\n' %\
                            (guid, ftype, sfn)

        pfc_text += '</POOLFILECATALOG>\n'
        # tolog(str(doc.toxml()))

        # add escape character for & (needed for google turls)
        if '&' in pfc_text:
            pfc_text = pfc_text.replace('&',  '&#038;')

        tolog(pfc_text)

        try:
            tolog("Writing XML to %s" % (pfc_name))
            f = open(pfc_name, 'w')
            f.write(pfc_text)
        except Exception, e:
            tolog("!!WARNING!!2999!! Could not create XML file: %s" % (e))
        else:
            tolog("Created PFC XML")
            f.close()

        outxml = pfc_text

    return outxml def tolog(msg, tofile=True, label='INFO', essential=False):
    """ Write date+msg to pilot log and to stdout """

    try:
        import inspect

        MAXLENGTH = 12
        # getting the name of the module that is invoking tolog() and adjust the length
        try:
            module_name = os.path.basename(inspect.stack()[1][1])
        except Exception, e:
            module_name = "unknown"
            #print "Exception caught by tolog(): ", e,
        module_name_cut = module_name[0:MAXLENGTH].ljust(MAXLENGTH)
        msg = "%i|%s| %s" % (os.getpid(),module_name_cut, msg)

        t = timeStampUTC(format='%Y-%m-%d %H:%M:%S')
        if tofile:
            appendToLog("%s|%s\n" % (t, msg))

        # remove backquotes from the msg since they cause problems with batch submission of pilot
        # (might be present in error messages from the OS)
        msg = msg.replace("`","'")
        msg = msg.replace('"','\\"')
        print "%s| %s" % (t, msg)

        # write any FAILED messages to stderr
        if "!!FAILED!!" in msg:
            try:
                print >> sys.stderr, "%s| %s" % (t, msg)
            except:
                print "Failed to print to sys.stderr: %s" % (t, msg)
    except:
        print "!!WARNING!!4000!! %s" % traceback.format_exc() def addToSkipped(lfn, guid):
    """ add metadata for skipped file """

    ec = 0
    try:
        # append to skipped.xml file
        fd = open("skipped.xml", "a")
    except Exception, e:
        tolog("!!WARNING!!2999!! Exception caught: %s" % e)
        ec = -1
    else:
        fd.write('  <File ID="%s">\n' % (guid))
        fd.write("    <logical>\n")
        fd.write('      <lfn name="%s"/>\n' % (lfn))
        fd.write("    </logical>\n")
        fd.write("  </File>\n")
        fd.close()
    return ec def removeDuplicates(seq):
    """ Order preserving duplicate removal """

    checked = []
    for entry in seq:
        if entry not in checked:
            checked.append(entry)
    return checked def dumpOrderedItems(l):
    """ dump list l """

    _i = 0
    for item in l:
        _i += 1
        if item == "":
            tolog("%d. <empty>" % (_i))
        else:
            tolog("%d. %s" % (_i, item)) def hasBeenTransferred(fields):
    """ determine whether files were successfully transferred """

    status = False
    s = 0
    # the fields will all be empty if no files were transferred
    for field in fields:
        s += len(field)
    if s > 0:
        status = True
    return status def getLFN(pfn, lfns):
    """ Identify the LFN from the list of LFNs corresponding to a PFN """
    # Note: important since the basename of the PFN can contain additional characters,
    # e.g. PFN = /../data15_cos.00259101.physics_IDCosmic.merge.RAW._lb0116._SFO-ALL._0001.1_1427497374
    # but LFN = data15_cos.00259101.physics_IDCosmic.merge.RAW._lb0116._SFO-ALL._0001.1

    lfn = ""
    for _lfn in lfns:
        _basename = os.path.basename(pfn)
        if _lfn in _basename:
#        if _basename.endswith(_lfn):
            # Handle scopes in case they are present
            if ":" in _basename:
                l = _basename.split(":")
                lfn = l[1]
            else:
                lfn = _lfn

    if lfn == "":
        tolog("!!WARNING!!2323!! Correct LFN could not be identified: pfn=%s, lfns=%s (assume basename of PFN)" % (pfn, str(lfns)))
        lfn = os.path.basename(pfn)

    return lfn def makeTransRegReport(all_transferred, some_transferred, latereg, nr_transferred, nr_files, fail, ec, ret, fields):
    """ make the transfer and registration report """

    error = PilotErrors()

    tolog("")
    tolog("..Transfer and registration report.........................................................................")
    tolog(". Mover has finished")

    if all_transferred and not latereg:
        if nr_files > 1:
            tolog(". All (%d) files have been transferred and registered" % (nr_files))
        else:
            tolog(". The single file has been transferred and registered")
    elif all_transferred and latereg:
        if nr_files > 1:
            tolog(". All (%d) files have been transferred but not registered" % (nr_files))
            tolog(". The files will be registrered by a later pilot if job recovery is supported,")
        else:
            tolog(". The single file has been transferred but not registered")
            tolog(". The file will be registrered by a later pilot if job recovery is supported,")
        tolog(". otherwise this job will fail")
    elif some_transferred and latereg:
        tolog(". Some files (%d/%d) were transferred but no file was registered" % (nr_transferred, nr_files))
        tolog(". The remaining files will be transferred and registrered by a later pilot if job recovery is supported,")
        tolog(". otherwise this job will fail")
    elif some_transferred and not latereg:
        tolog(". Some files (%d/%d) were transferred and registered" % (nr_transferred, nr_files))
        tolog(". The remaining files will be transferred and registrered by a later pilot if job recovery is supported,")
        tolog(". otherwise this job will fail")
    elif not some_transferred:
        tolog(". No files (%d/%d) were transferred or registered" % (nr_transferred, nr_files))
        if nr_files > 1:
            tolog(". The files will be transferred and registrered by a later pilot if job recovery is supported,")
        else:
            tolog(". The file will be transferred and registrered by a later pilot if job recovery is supported,")
            tolog(". otherwise this job will fail")
    else:
        tolog(". Mover has finished")

    if fail != 0:
        tolog(". File transfer exit code                       : (%d, %s)" % (fail, error.getErrorStr(fail)))
    else:
        tolog(". File transfer exit code                       : (%d, <no error>)" % (fail))

    if some_transferred:
        tolog(". File registration return values               : (%d, %s, %s)" %\
              (ec, error.getErrorStr(ec), str(ret)))

    tolog(". Put function will return fields               : %s" % str(fields))
    tolog(". Transfer and registration report produced at  : %s" % timeStamp())
    tolog("...........................................................................................................")
    tolog("") def readpar(parameter, alt=False, version=0, queuename=None):
    """ Read 'parameter' from queuedata via SiteInformation class """

    from SiteInformation import SiteInformation
    si = SiteInformation()

    return si.readpar(parameter, alt=alt, version=version, queuename=queuename) def getMaxInputSize(MB=False):
    """ Return a proper maxinputsize value """

    _maxinputsize = readpar('maxwdir') # normally 14336+2000 MB
    MAX_INPUT_FILESIZES = 14*1024*1024*1024 # 14 GB, 14336 MB (pilot default)
    MAX_INPUT_FILESIZES_MB = 14*1024 # 14336 MB (pilot default)
    if _maxinputsize != "":
        try:
            if MB: # convert to MB int
                _maxinputsize = int(_maxinputsize) # MB
            else: # convert to B int
                _maxinputsize = int(_maxinputsize)*1024*1024 # MB -> B
        except Exception, e:
            tolog("!!WARNING!!2999!! schedconfig.maxinputsize: %s" % e)
            if MB:
                _maxinputsize = MAX_INPUT_FILESIZES_MB
            else:
                _maxinputsize = MAX_INPUT_FILESIZES
    else:
        if MB:
            _maxinputsize = MAX_INPUT_FILESIZES_MB
        else:
            _maxinputsize = MAX_INPUT_FILESIZES

    if MB:
        tolog("Max input size = %d MB (pilot default)" % (_maxinputsize))
    else:
        tolog("Max input size = %d B (pilot default)" % (_maxinputsize))

    return _maxinputsize def headPilotErrorDiag(pilotErrorDiag, size=256):
    """ Return the first n characters of pilotErrorDiag """

    try:
        return pilotErrorDiag[:size]
    except:
        return pilotErrorDiag def getCopysetup(mode="get"):
    """ extract a verified copysetup[in] script from queuedata """

    copysetup_tmp = readpar('copysetup')

    if mode == "get":
        _copysetup = readpar('copysetupin')
        if _copysetup == "":
            # not set, use same copysetup for stage-in as for stage-out
            _copysetup = copysetup_tmp
    else:
        _copysetup = copysetup_tmp

    # copysetup can contain ^-signs for remote i/o sites
    if _copysetup.find('^') > -1:
        copysetup = _copysetup.split('^')[0]
    else:
        copysetup = copysetup_tmp

    # make sure that the script exists
    if copysetup == "":
        tolog("No copysetup found")
    elif not os.path.exists(copysetup) and '^' not in copysetup:
        tolog("!!WARNING!!2998!! copysetup does not exist: %s (reset to empty string)" % (copysetup))
        copysetup = ""
    elif '^' in copysetup:
        tolog("No path in copysetup (%s, reset to empty string)" % (copysetup))
        copysetup = ""
    else:
        tolog("Extracted copysetup: %s" % (copysetup))

    return copysetup def getCopyprefixLists(copyprefix):
    """ Get the copyprefix lists """

    pfrom, pto = getCopyprefixFromTo(copyprefix)
    if "," in pfrom:
        pfroms = pfrom.split(",")
    else:
        pfroms = [pfrom]
    if "," in pto:
        ptos = pto.split(",")
    else:
        ptos = [pto]

    return pfroms, ptos def getExperiment(experiment):
    """ Return a reference to an experiment class """

    from ExperimentFactory import ExperimentFactory
    factory = ExperimentFactory()
    _exp = None

    try:
        experimentClass = factory.newExperiment(experiment)
    except Exception, e:
        tolog("!!WARNING!!1114!! Experiment factory threw an exception: %s" % (e))
    else:
        _exp = experimentClass()

    return _exp def getSiteInformation(experiment):
    """ Return a reference to a site information class """

    # The SiteInformationFactory ensures that the returned object is a Singleton
    # Usage:
    #    _exp = getSiteInformation(readpar('experiment')) # or from pilot option
    #    if _exp:
    #        _exp.somemethod("Hello")
    #    else:
    #        tolog("!!WARNING!!1111!! Failed to instantiate experiment class")

    from SiteInformationFactory import SiteInformationFactory
    factory = SiteInformationFactory()
    _exp = None

    try:
        siteInformationClass = factory.newSiteInformation(experiment)
    except Exception, e:
        tolog("!!WARNING!!1114!! SiteInformation factory threw an exception: %s" % (e))
    else:
        _exp = siteInformationClass()
        tolog("getSiteInformation: got experiment=%s" % (_exp.getExperiment()))

    return _exp def stripDQ2FromLFN(lfn):
    """ Remove any legacy __DQ2 part of an LFN """
    # E.g. LFN = AOD.505307._000001.pool.root.9__DQ2-1315236060
    # -> AOD.505307._000001.pool.root.9
    # This method assumes that the LFN contains the legacy __DQ2-<nr> substring

    pattern = "(\s*)\_\_DQ2\-[0-9]+"

    found = re.search(pattern, lfn)
    if found:
        try:
            __DQ2 = found.group(0)
        except Exception, e:
            tolog("!!WARNING!!1112!! Failed to identify legacy __DQ2 substring: %s" % (e))
        else:
            lfn = lfn.replace(__DQ2, "")

    return lfn def extractPattern(source, pattern):
    """ Extract 'pattern' from 'source' using a regular expression """

    # Examples
    # surl = "srm://dcache-se-atlas.desy.de/pnfs/desy.de/atlas/dq2/atlasdatadisk/rucio/mc12_8TeV/3e/51/NTUP_SMWZ.01355318._000001.root.1"
    # extractPattern(surl, r'\/rucio\/.+(\/[a-zA-Z0-9]{2}\/[a-zA-Z0-9]{2}\/)') -> "/3e/51/"
    # extractPattern(surl, r'\/rucio\/(.+)\/[a-zA-Z0-9]{2}\/[a-zA-Z0-9]{2}\/') -> "mc12_8TeV"

    extracted = ""
    _pattern = re.compile(pattern)
    _extracted = re.findall(_pattern, source)
    if _extracted != []:
        extracted = _extracted[0]

    return extracted def dumpFile(filename, topilotlog=False):
    """ dump a given file to stdout or to pilotlog """

    if os.path.exists(filename):
        tolog("Dumping file: %s (size: %d)" % (filename, os.path.getsize(filename)))
        try:
            f = open(filename, "r")
        except IOError, e:
            tolog("!!WARNING!!4000!! Exception caught: %s" % (e))
        else:
            i = 0
            for line in f.readlines():
                i += 1
                line = line.rstrip()
                if topilotlog:
                    tolog("%s" % (line))
                else:
                    print "%s" % (line)
            f.close()
            tolog("Dumped %d lines from file %s" % (i, filename))
    else:
        tolog("!!WARNING!!4000!! %s does not exist" % (filename)) def updateInputFileWithTURLs(jobPars, LFN_to_TURL_dictionary):
    """ Update the @poo input file list with TURLs """

    status = False

    # First try to get the @poo filename (which actually contains the full local path to the file)
    filename = getPooFilenameFromJobPars(jobPars)
    if filename != "":
        if os.path.exists(filename):
            try:
                f = open(filename, "r")
            except IOError, e:
                tolog("!!WARNING!!2997!! Caught exception: %s" % (e))
            else:
                # Read file
                lines = f.readlines()
                f.close()

                # Process file
                turls = []
                header = lines[0]
                for lfn in lines:
                    # Note: the 'lfn' is actually a local file path, but will end with an \n
                    lfn = os.path.basename(lfn)
                    try:
                        # Try to get the corresponding dictionary entry (assume there is a corresponding entry in the actual TURL dictionary)
                        if lfn.endswith('\n'):
                            lfn = lfn[:-1]
                        turl = LFN_to_TURL_dictionary[lfn]
                    except:
                        pass
                    else:
                        turls.append(turl)

                if turls != []:
                    # Overwrite the old file with updated TURL info
                    try:
                        f = open(filename, "w")
                    except IOError, e:
                        tolog("!!WARNING!!2997!! Caught exception: %s" % (e))
                    else:
                        # Write file
                        f.write(header)
                        for turl in turls:
                            f.write(turl + "\n")
                        # Process file
                        f.close()
                        status = True
                else:
                    tolog("!!WARNING!!2998!! Failed to extract TURLs (empty TURL list)")
        else:
            tolog("!!WARNING!!2342!! File not found: %s" % (filename))
    else:
        tolog("!!WARNING!!2343!! Found no @input filename in jobPars: %s" % (jobPars)) def getPilotVersion(initdir):
    """ Load the pilot version string from file VERSION """

    version = "PICARD"
    try:
        f = open(os.path.join(initdir, "PILOTVERSION"), "r")
    except Exception, e:
        print "!!WARNING!!0000!! Could not read pilot version from file: %s" % (e )
    else:
        _version = f.read().strip()
        # remove trailing \n if present
        if "\n" in _version:
            _version = _version.replace("\n", "")
        # trivial check
        pattern = re.compile(r"[A-Z]+ [A-Za-z0-9.]+")
        v = re.findall(pattern, _version)
        if v == []:
            print "!!WARNING!!0000!! Not a valid version format: %s" % (version)
        else:
            version = _version

    return version def getExtension(alternative='pickle'):
    """ get the file extension (json or whatever 'alternative' is set to, pickle by default) """

    try:
        from json import load
    except:
        extension = alternative
    else:
        extension = "json"

    return extension def getTracingReportFilename():
    """ Return the name of the tracing report JSON file """

    return "tracing_report.json" def readJSON(file_name):
    """ Read a dictionary from a JSON file """

    dictionary = {}
    from json import load
    f = openFile(file_name, 'r')
    if f:
        # Read the dictionary
        try:
            dictionary = load(f)
        except Exception, e:
            tolog("!!WARNING!!2332!! Failed to read dictionary from file %s: %s" % (file_name, e))
        else:
            f.close()

    return dictionary def getHashedBucketEndpoint(endpoint, file_name):
    """ Return a hashed bucket endpoint """
    # Example:
    # endpoint = "atlas_logs", file_name = "log.tgz"
    # -> hash = "07" and hashed_endpoint = "atlas_logs_07"

#    return endpoint + "_" + getHash(file_name, 2)
    return endpoint def getDirectAccess(analyjob=False):
    """ Should direct i/o be used, and which type of direct i/o """

    directInLAN = useDirectAccessLAN()
    directInWAN = useDirectAccessWAN()
    directInType = 'None'

    directIn = False
    if directInLAN:
        directInType = 'LAN'
        directIn = True
    if directInWAN:
    # if (directInWAN and not analyjob) or (directInWAN and directInLAN and analyjob):
        directInType = 'WAN' # Overrides LAN if both booleans are set to True
        directIn = True

    return directIn, directInType def useDirectAccessWAN():
    """ Should direct i/o be used over WAN? """

    return _useDirectAccess(LAN=False, WAN=True) def updateCopysetups(cmd3, transferType=None, useCT=None, directIn=None, useFileStager=None):
    """ Update the relevant copysetup fields for remote I/O or file stager """

    si = SiteInformation()

    _copysetupin = readpar('copysetupin')
    _copysetup = readpar('copysetup')

    if _copysetupin != "":
        si.updateCopysetup(cmd3, 'copysetupin', _copysetupin, transferType=transferType, useCT=useCT, directIn=directIn, useFileStager=useFileStager)
    else:
        si.updateCopysetup(cmd3, 'copysetup', _copysetup, transferType=transferType, useCT=useCT, directIn=directIn, useFileStager=useFileStager) class PilotErrors:
    """ Pilot error handling """

    # error codes
    ERR_UNKNOWNERROR = 0
    ERR_GENERALERROR = 1008
    ERR_DIRECTIOFILE = 1009 # harmless, just means that copy-to-scratch was skipped in favor or direct i/o access
    ERR_GETDATAEXC = 1097
    ERR_NOLOCALSPACE = 1098
    ERR_STAGEINFAILED = 1099
    ERR_REPNOTFOUND = 1100
    ERR_LRCREGCONNREF = 1101
    ERR_NOSUCHFILE = 1103
    ERR_USERDIRTOOLARGE = 1104
    ERR_LFCADDCSUMFAILED = 1105
    ERR_STDOUTTOOBIG = 1106
    ERR_MISSDBREL = 1107
    ERR_FAILEDLCGREG = 1108
    ERR_CMTCONFIG = 1109
    ERR_SETUPFAILURE = 1110
    ERR_RUNJOBEXC = 1111
    ERR_PILOTEXC = 1112
    ERR_GETLFCIMPORT = 1113
    ERR_PUTLFCIMPORT = 1114
    ERR_NFSSQLITE = 1115
    ERR_QUEUEDATA = 1116
    ERR_QUEUEDATANOTOK = 1117
    ERR_CURLSPACE = 1118
    ERR_DDMSPACE = 1119
    ERR_NOSTMATCHDEST = 1120 # not used
    ERR_NOLFCSFN = 1122
    ERR_MISSINGGUID = 1123
    ERR_OUTPUTFILETOOLARGE = 1124
    ERR_NOPFC = 1130
    ERR_PUTFUNCNOCALL = 1131
    ERR_LRCREG = 1132
    ERR_NOSTORAGE = 1133
    ERR_MKDIR = 1134
    ERR_FAILEDSIZELOCAL = 1135
    ERR_FAILEDMD5LOCAL = 1136
    ERR_STAGEOUTFAILED = 1137
    ERR_FAILEDSIZE = 1138
    ERR_PUTWRONGSIZE = 1139
    ERR_FAILEDMD5 = 1140
    ERR_PUTMD5MISMATCH = 1141
    ERR_CHMODTRF = 1143
    ERR_PANDAKILL = 1144
    ERR_GETMD5MISMATCH = 1145
    ERR_DYNTRFINST = 1146
    ERR_FAILEDRM = 1148
    ERR_TRFDOWNLOAD = 1149
    ERR_LOOPINGJOB = 1150
    ERR_GETTIMEOUT = 1151
    ERR_PUTTIMEOUT = 1152
    ERR_LOSTJOBNOTFINISHED = 1153
    ERR_LOSTJOBLOGREG = 1154
    ERR_LOSTJOBFILETRANSFER = 1155
    ERR_LOSTJOBRECOVERY = 1156
    ERR_LOSTJOBMAXEDOUT = 1158
    ERR_LOSTJOBPFC = 1159
    ERR_LRCREGSTRSIZE = 1160
    ERR_LOSTJOBXML = 1161
    ERR_LRCREGDUP = 1162
    ERR_NOPROXY = 1163
    ERR_MISSINGLOCALFILE = 1164
    ERR_MISSINGOUTPUTFILE = 1165
    ERR_SIGPIPE = 1166
    ERR_MISSFILEXML = 1167
    ERR_SIZETOOLARGE = 1168
    ERR_FAILEDLFCREG = 1169
    ERR_FAILEDADLOCAL = 1170
    ERR_GETADMISMATCH = 1171
    ERR_PUTADMISMATCH = 1172
    ERR_PANDAMOVERFILENOTCACHED = 1173
    ERR_PANDAMOVERTRANSFER = 1174
    ERR_GETWRONGSIZE = 1175
    ERR_NOCHILDPROCESSES = 1176
    ERR_NOVOMSPROXY = 1177
    ERR_NOSTAGEDFILES = 1178
    ERR_FAILEDLFCGETREPS = 1179
    ERR_GETGLOBUSSYSERR = 1180
    ERR_PUTGLOBUSSYSERR = 1181
    ERR_FAILEDLFCGETREP = 1182
    ERR_GUIDSEXISTSINLRC = 1183
    ERR_MISSINGPFC = 1184
    ERR_NOSOFTWAREDIR = 1186
    ERR_NOPAYLOADMETADATA = 1187
    ERR_LCGGETTURLS = 1188
    ERR_LCGGETTURLSTIMEOUT = 1189
    ERR_LFNTOOLONG = 1190
    ERR_ZEROFILESIZE = 1191
    ERR_DBRELNOTYETTRANSFERRED = 1192
    ERR_SEPROBLEM = 1193
    ERR_NOFILEVERIFICATION = 1194
    ERR_COMMANDTIMEOUT = 1195
    ERR_GETFAILEDTOMOUNTNFS4 = 1196
    ERR_GETPNFSSYSTEMERROR = 1197
    ERR_MKDIRWORKDIR = 1199
    ERR_KILLSIGNAL = 1200
    ERR_SIGTERM = 1201
    ERR_SIGQUIT = 1202
    ERR_SIGSEGV = 1203
    ERR_SIGXCPU = 1204
#    ERR_USERKILL = 1205 # not used by pilot
    ERR_SIGBUS = 1206
    ERR_SIGUSR1 = 1207
    ERR_NOPAYLOADOUTPUT = 1210
    ERR_MISSINGINSTALLATION = 1211
    ERR_PAYLOADOUTOFMEMORY = 1212
    ERR_REACHEDMAXTIME = 1213
    ERR_DAFSNOTALLOWED = 1214
    ERR_NOTCPCONNECTION = 1215
    ERR_NOPILOTTCPSERVER = 1216
    ERR_CORECOUNTMISMATCH = 1217
    ERR_RUNEVENTEXC = 1218
    ERR_UUIDGEN = 1219
    ERR_UNKNOWN = 1220
    ERR_FILEEXIST = 1221
    ERR_GETKEYPAIR = 1222
    ERR_BADALLOC = 1223
    ERR_ESRECOVERABLE = 1224
    ERR_ESMERGERECOVERABLE = 1225
    ERR_GLEXEC = 1226
    ERR_ESATHENAMPDIED = 1227
    ERR_ESFATAL = 1228
    ERR_TEFATAL = 1229
    ERR_TEBADURL = 1230
    ERR_TEINVALIDGUID = 1231
    ERR_TEWRONGGUID = 1232
    ERR_TEHOSTNAME = 1233
    ERR_EXECUTEDCLONEJOB = 1234
    ERR_PAYLOADEXCEEDMAXMEM = 1235
    ERR_FAILEDBYSERVER = 1236
    ERR_ESKILLEDBYSERVER = 1237
    ERR_NOEVENTS = 1238
    ERR_OVERSUBSCRIBEDEVENTS = 1239
    ERR_ESMESSAGESERVER = 1240
    ERR_ESOBJECTSTORESETUP = 1241
    ERR_CHKSUMNOTSUP = 1242
    ERR_ESPREFETCHERDIED = 1243
    ERR_NORELEASEFOUND = 1244
    ERR_TOOFEWEVENTS = 1245
    ERR_NOUSERTARBALL = 1246
    ERR_BADXML = 1247
    ERR_WRONGARCHITECTURE = 1248
    ERR_RUNGENFAILURE = 1249

    # internal error codes
    ERR_DDMREG = 1
    ERR_FILEONTAPE = 2

    pilotError = {
        ERR_UNKNOWNERROR : "",
        ERR_GENERALERROR : "General pilot error, consult batch log",
        ERR_GETDATAEXC : "Get function can not be called for staging input file",
        ERR_NOLOCALSPACE : "No space left on local disk",
        ERR_STAGEINFAILED : "Get error: Staging input file failed",
        ERR_REPNOTFOUND : "Get error: Replica not found",
        ERR_LRCREGCONNREF : "LRC registration error: Connection refused",
        # 1102 : "Expected output file does not exist", # not used, see ERR_MISSINGOUTPUTFILE below
        ERR_NOSUCHFILE : "No such file or directory",
        ERR_USERDIRTOOLARGE : "User work directory too large",
        ERR_LFCADDCSUMFAILED : "Put error: Failed to add file size and checksum to LFC",
        ERR_STDOUTTOOBIG : "Payload stdout file too big",
        ERR_MISSDBREL : "Get error: Missing DBRelease file",
        ERR_FAILEDLCGREG : "Put error: LCG registration failed",
        ERR_CMTCONFIG : "Required CMTCONFIG incompatible with WN",
        ERR_SETUPFAILURE : "Failed during setup",
        ERR_RUNJOBEXC : "Exception caught by RunJob*",
        ERR_PILOTEXC : "Exception caught by pilot",
        ERR_GETLFCIMPORT : "Get error: Failed to import LFC python module",
        ERR_PUTLFCIMPORT : "Put error: Failed to import LFC python module",
        ERR_NFSSQLITE : "NFS SQLite locking problems",
        ERR_QUEUEDATA : "Pilot could not download queuedata",
        ERR_QUEUEDATANOTOK : "Pilot found non-valid queuedata",
        ERR_CURLSPACE : "Pilot could not curl space report",
        ERR_DDMSPACE : "Pilot aborted due to DDM space shortage",
        ERR_NOSTMATCHDEST : "Space token descriptor does not match destination path",
        # 1121 : "Can not read the xml file for registering output files to dispatcher", # not used
        ERR_NOLFCSFN : "Bad replica entry returned by lfc_getreplicas(): SFN not set in LFC for this guid",
        ERR_MISSINGGUID : "Missing guid in output file list",
        ERR_OUTPUTFILETOOLARGE : "Output file too large",
        ERR_NOPFC : "Get error: Failed to get PoolFileCatalog",
        ERR_PUTFUNCNOCALL : "Put function can not be called for staging out",
        ERR_LRCREG : "LRC registration error (consult log file)",
        ERR_NOSTORAGE : "Put error: Fetching default storage URL failed",
        ERR_MKDIR : "Put error: Error in mkdir on localSE, not allowed or no available space",
        ERR_FAILEDSIZELOCAL : "Could not get file size in job workdir",
        ERR_FAILEDMD5LOCAL : "Error running md5sum on the file in job workdir",
        ERR_STAGEOUTFAILED : "Put error: Error in copying the file from job workdir to localSE",
        ERR_FAILEDSIZE : "Put error: could not get the file size on localSE",
        ERR_PUTWRONGSIZE : "Put error: Problem with copying from job workdir to local SE: size mismatch",
        ERR_FAILEDMD5 : "Put error: Error running md5sum on the file on local SE",
        ERR_PUTMD5MISMATCH : "Put error: Problem with copying from job workdir to local SE: md5sum mismatch",
        # 1142 : "Put error: failed to register the file on local SE", # not used
        ERR_CHMODTRF : "Failed to chmod trf",
        ERR_PANDAKILL : "This job was killed by panda server",
        ERR_GETMD5MISMATCH : "Get error: md5sum mismatch on input file",
        ERR_DYNTRFINST : "Trf installation dir does not exist and could not be installed",
        # 1147 : "Put error: dccp returned readOnly",  # not used
        ERR_FAILEDRM : "Put error: Failed to remove readOnly file in dCache",
        ERR_TRFDOWNLOAD : "wget command failed to download trf",
        ERR_LOOPINGJOB : "Looping job killed by pilot",
        ERR_GETTIMEOUT : "Get error: Input file staging timed out",
        ERR_PUTTIMEOUT : "Put error: File copy timed out",
        ERR_LOSTJOBNOTFINISHED : "Lost job was not finished",
        ERR_LOSTJOBLOGREG : "Failed to register log file",
        ERR_LOSTJOBFILETRANSFER : "Failed to move output files for lost job",
        ERR_LOSTJOBRECOVERY : "Pilot could not recover job",
        # 1157 : "Could not create log file", # not used
        ERR_LOSTJOBMAXEDOUT : "Reached maximum number of recovery attempts",
        ERR_LOSTJOBPFC : "Job recovery could not read PoolFileCatalog.xml file (guids lost)",
        ERR_LRCREGSTRSIZE : "LRC registration error: file name string size exceeded limit of 250",
        ERR_LOSTJOBXML : "Job recovery could not generate xml for remaining output files",
        ERR_LRCREGDUP : "LRC registration error: Non-unique LFN",
        ERR_NOPROXY : "Grid proxy not valid",
        ERR_MISSINGLOCALFILE : "Get error: Local input file missing",
        ERR_MISSINGOUTPUTFILE : "Put error: Local output file missing",
        ERR_SIGPIPE : "Put error: File copy broken by SIGPIPE",
        ERR_MISSFILEXML : "Get error: Input file missing in PoolFileCatalog.xml",
        ERR_SIZETOOLARGE : "Get error: Total file size too large",
        ERR_FAILEDLFCREG : "Put error: File registration failed",
        ERR_FAILEDADLOCAL : "Error running adler32 on the file in job workdir",
        ERR_GETADMISMATCH : "Get error: adler32 mismatch on input file",
        ERR_PUTADMISMATCH : "Put error: adler32 mismatch on output file",
        ERR_PANDAMOVERFILENOTCACHED : "PandaMover staging error: File is not cached",
        ERR_PANDAMOVERTRANSFER : "PandaMover transfer failure",
        ERR_GETWRONGSIZE : "Get error: Problem with copying from local SE to job workdir: size mismatch",
        ERR_NOCHILDPROCESSES : "Pilot has no child processes (job wrapper has either crashed or did not send final status)",
        ERR_NOVOMSPROXY : "Voms proxy not valid",
        ERR_NOSTAGEDFILES : "Get error: No input files are staged",
        ERR_FAILEDLFCGETREPS : "Get error: Failed to get replicas",
        ERR_GETGLOBUSSYSERR : "Get error: Globus system error",
        ERR_PUTGLOBUSSYSERR : "Put error: Globus system error",
        ERR_FAILEDLFCGETREP : "Get error: Failed to get replica",
        ERR_GUIDSEXISTSINLRC : "LRC registration error: Guid-metadata entry already exists",
        ERR_MISSINGPFC : "Put error: PoolFileCatalog could not be found in workdir",
        # 1185 : "Put error: Error running adler32 on the file in job workdir", # not used
        ERR_NOSOFTWAREDIR : "Software directory does not exist",
        ERR_NOPAYLOADMETADATA : "Payload metadata is not available",
        ERR_LCGGETTURLS : "lcg-getturls failed",
        ERR_LCGGETTURLSTIMEOUT : "lcg-getturls was timed-out",
        ERR_LFNTOOLONG : "LFN too long (exceeding limit of 150 characters)",
        ERR_ZEROFILESIZE : "Illegal zero file size",
        ERR_DBRELNOTYETTRANSFERRED : "DBRelease file has not been transferred yet",
        ERR_NOFILEVERIFICATION : "File verification failed",
        ERR_COMMANDTIMEOUT : "Command timed out",
        ERR_GETFAILEDTOMOUNTNFS4 : "Get error: Failed to mount NSF4",
        ERR_GETPNFSSYSTEMERROR : "Get error: PNFS system error",
        # 1198 : "Can not check the child process status from the heartbeat process", # not used
        ERR_MKDIRWORKDIR : "Could not create directory",
        ERR_KILLSIGNAL : "Job terminated by unknown kill signal",
        ERR_SIGTERM : "Job killed by signal: SIGTERM",
        ERR_SIGQUIT : "Job killed by signal: SIGQUIT",
        ERR_SIGSEGV : "Job killed by signal: SIGSEGV",
        ERR_SIGXCPU : "Job killed by signal: SIGXCPU",
        ERR_SIGUSR1 : "Job killed by signal: SIGUSR1",
        ERR_SIGBUS : "Job killed by signal: SIGBUS",
        ERR_NOPAYLOADOUTPUT : "No payload output",
        ERR_MISSINGINSTALLATION : "Missing installation",
        ERR_PAYLOADOUTOFMEMORY : "Payload ran out of memory",
        ERR_REACHEDMAXTIME : "Reached batch system time limit",
        ERR_DAFSNOTALLOWED : "Site does not allow requested direct access or file stager",
        ERR_NOTCPCONNECTION : "Failed to open TCP connection to localhost (worker node network problem)",
        ERR_NOPILOTTCPSERVER : "Pilot TCP server has died",
        ERR_CORECOUNTMISMATCH : "Mismatch between core count in job and queue definition",
        ERR_RUNEVENTEXC : "Exception caught by runEvent",
        ERR_UUIDGEN : "uuidgen failed to produce a guid",
        ERR_UNKNOWN : "Job failed due to unknown reason (consult log file)",
        ERR_FILEEXIST : "File already exist",
        ERR_GETKEYPAIR : "Failed to get security key pair",
        ERR_BADALLOC : "TRF failed due to bad_alloc",
        ERR_ESMERGERECOVERABLE : "Recoverable Event Service Merge error",
        ERR_ESRECOVERABLE: "Recoverable Event Service error",
        ERR_GLEXEC: "gLExec related error",
        ERR_ESATHENAMPDIED: "AthenaMP ended Event Service job prematurely",
        ERR_ESFATAL: "Fatal Event Service error",
        ERR_TEFATAL: "Fatal Token Extractor error",
        ERR_TEHOSTNAME: "Token Extractor error: Host name could not be resolved",
        ERR_TEBADURL: "Token Extractor error: Bad URL",
        ERR_TEINVALIDGUID: "Token Extractor error: Invalid GUID length",
        ERR_TEWRONGGUID: "Token Extractor error: No tokens for this GUID",
        ERR_EXECUTEDCLONEJOB: "Already executed clone job",
        ERR_PAYLOADEXCEEDMAXMEM: "Payload exceeded maximum allowed memory",
        ERR_FAILEDBYSERVER: "Failed by server",
        ERR_ESKILLEDBYSERVER: "Event Service job killed by server",
        ERR_NOEVENTS: "Event Service no available events",
        ERR_OVERSUBSCRIBEDEVENTS: "Event Service over subscribed events",
        ERR_ESMESSAGESERVER: "Event service message server error",
        ERR_ESOBJECTSTORESETUP: "Event service objectstore setup",
        ERR_CHKSUMNOTSUP: "Mover error: query checksum is not supported",
        ERR_ESPREFETCHERDIED: "Prefetcher ended Event Service job prematurely",
        ERR_NORELEASEFOUND: "No release candidates found",
        ERR_TOOFEWEVENTS: "Too few events, less events than minimal requirement",
        ERR_NOUSERTARBALL: "User tarball cannot be downloaded from PanDA server",
        ERR_BADXML: "Badly formed XML",
        ERR_WRONGARCHITECTURE: "Job built on wrong architecture",
        ERR_RUNGENFAILURE: "RunGen failure (consult log file)",

    }

    getErrorCodes = [1097, 1099, 1100, 1103, 1107, 1113, 1130, 1145, 1151, 1164, 1167, 1168, 1171, 1175, 1178, 1179, 1180, 1182]
    putErrorCodes = [1101, 1114, 1122, 1131, 1132, 1133, 1134, 1135, 1136, 1137, 1138, 1140, 1141, 1152, 1154, 1155, 1181]
    recoverableErrorCodes = [0] + putErrorCodes

    # Error codes that will issue a Pilot-controlled resubmission
    PilotResubmissionErrorCodes = [1008, 1098, 1099, 1110, 1113, 1114, 1115, 1116, 1117, 1137, 1139, 1151, 1152, 1171, 1172, 1177, 1179, 1180, 1181, 1182, 1188, 1189, 1195, 1196, 1197, 1219]

    # Error codes used with FAX fail-over (only an error code in this list will allow FAX fail-over)
    PilotFAXErrorCodes = [1103] + PilotResubmissionErrorCodes

    def getPilotErrorDiag(self, code=0):
        """ Return text corresponding to error code """

        pilotErrorDiag = ""
        if code in self.pilotError.keys():
            pilotErrorDiag = self.pilotError[code]
        else:
            pilotErrorDiag = "Unknown pilot error code"
        return pilotErrorDiag

    def isGetErrorCode(self, code=0):
        """ Determine whether code is in the put error list or not """

        state = False
        if code in self.getErrorCodes:
            state = True

        return state

    def isPutErrorCode(self, code=0):
        """ Determine whether code is in the put error list or not """

        state = False
        if code in self.putErrorCodes:
            state = True

        return state

    @classmethod
    def isRecoverableErrorCode(self, code=0):
        """ Determine whether code is a recoverable error code or not """

        return code in self.recoverableErrorCodes

    def isPilotResubmissionErrorCode(self, code=0):
        """ Determine whether code issues a Pilot-controlled resubmission """

        state = False
        if code in self.PilotResubmissionErrorCodes:
            state = True

        return state

    def isPilotFAXErrorCode(self, code=0):
        """ Determine whether code allows for a FAX fail-over """

        state = False
        if code in self.PilotFAXErrorCodes:
            state = True

        return state

    @classmethod
    def getErrorStr(self, code):
        """
        Avoids exception if an error is not in the dictionary.
        An empty string is returned if the error is not in the dictionary.
        """
        return self.pilotError.get(code, '')

    def getErrorName(self, code):
        """ From the error code to get the error name"""
        for k in self.__class__.__dict__.keys():
            if self.__class__.__dict__[k] == code:
                return k
        return None class configSiteMover:
    COMMAND_MD5 = 'md5sum'
    COMMAND_AD = 'adler'
    PERMISSIONS_DIR = 0775
    PERMISSIONS_FILE = 0644
    ARCH_DEFAULT = ''
import os
import sys
import commands
import re
import urllib
import SiteMoverFarm
        import traceback
        import traceback
        import traceback
import functools
                    import traceback
        import traceback
        import lfc
        import hashlib
        import md5
        import socket
        import traceback
            import copy
    import Node
            import sys
from xml.dom import minidom
from time import time, sleep
from timed_command import timed_command
from pUtil import createPoolFileCatalog, tolog, addToSkipped, removeDuplicates, dumpOrderedItems,\
     hasBeenTransferred, getLFN, makeTransRegReport, readpar, getMaxInputSize, headPilotErrorDiag, getCopysetup,\
     getCopyprefixLists, getExperiment, getSiteInformation, stripDQ2FromLFN, extractPattern, dumpFile, updateInputFileWithTURLs, getPilotVersion
from FileHandling import getExtension, getTracingReportFilename, readJSON, getHashedBucketEndpoint, getDirectAccess, useDirectAccessWAN
from FileStateClient import updateFileState, dumpFileStates
from RunJobUtilities import updateCopysetups
from SysLog import sysLog, dumpSysLogTail
from PilotErrors import PilotErrors
from futil import *
from configSiteMover import config_sm
    from PilotErrors import PilotException
    from movers import JobMover
    from movers.trace_report import TraceReport
    from PilotErrors import PilotException
    from movers import JobMover
    from movers.trace_report import TraceReport
    from PilotErrors import PilotException
    from movers import JobMover
    from movers.trace_report import TraceReport
    from pUtil import getPilotVersion
    from DBReleaseHandler import DBReleaseHandler
                from json import load
                from pickle import load
        from shutil import move
    from SiteMover import SiteMover
    from movers import JobMover
    from movers.trace_report import TraceReport
    from SiteMover import SiteMover
    from SiteMover import SiteMover
            from json import dump
            from pickle import dump
    from SiteMover import SiteMover
    from SiteMover import SiteMover
            from random import shuffle
    from SiteMover import SiteMover
        from rucio.client import Client
        if special_log_transfer:
            do_stageout_func = mover.stageout_logfiles_os

        transferred_files, failed_transfers = do_stageout_func()
    except PilotException, e:
        error = e
    except Exception, e:
        tolog("ERROR: Mover put data failed [stageout]: exception caught: %s" % e)
        tolog(traceback.format_exc())
        error = PilotException('STAGEOUT FAILED, exception=%s' % e, code=PilotErrors.ERR_STAGEOUTFAILED, state='STAGEOUT_FAILED')

    if error:
        ## send trace
        mover.trace_report.update(clientState=error.state or 'STAGEOUT_FAILED', stateReason=error.message, timeEnd=time())
        mover.sendTrace(mover.trace_report)
        return error.code, error.message, [], "", 0, 0

    tolog("Mover put data finished")

    # prepare compatible output
    # keep track of which files have been copied

    fields = [''] * 7 # file info field used by job recovery in OLD compatible format

    #errors = []
    #for is_success, success_transfers, failed_transfers, exception in output:
    #    for fdata in success_transfers: # keep track of which files have been copied
    #        for i,v in enumerate(['surl', 'lfn', 'guid', 'filesize', 'checksum', 'farch', 'pfn']): # farch is not used
    #            value = fdata.get(v, '')
    #            if fields[i]:
    #                fields[i] += '+'
    #            fields[i] += '%s' % str(value)
    #    if exception:
    #        errors.append(str(exception))
    #    for err in failed_transfers:
    #        errors.append(str(err))

    files = job.outData if not log_transfer else job.logData
    if special_log_transfer:
        files = job.logSpecialData

    not_transferred = [e.lfn for e in files if e.status not in ['transferred']]
    if not_transferred:
        err_msg = 'STAGEOUT FAILED: not all output files have been copied: remain files=%s, errors=%s' % ('\n'.join(not_transferred), ';'.join([str(ee) for ee in failed_transfers]))
        tolog("Mover put data finished: error_msg=%s" % err_msg)
        return PilotErrors.ERR_STAGEOUTFAILED, err_msg, [], "", 0, 0

    return 0, "", fields, "", len(transferred_files), 0

# new mover implementation
def put_data_es(job, jobSite, stageoutTries, files, workDir=None, activity=None, pinitdir=""):
    """
        Do jobmover.stageout_outfiles or jobmover.stageout_logfiles (if log_transfer=True)
        or jobmover.stageout_logfiles_os (if special_log_transfer=True)
        :backward compatible return:  (rc, pilotErrorDiag, rf, "", filesNormalStageOut, filesAltStageOut)
    """

    tolog("Mover put data started [new implementation]")


    si = getSiteInformation(job.experiment)
    si.setQueueName(jobSite.computingElement) # WARNING: SiteInformation is singleton: may be used in other functions! FIX me later

    workDir = workDir or os.path.dirname(job.workdir)

    mover = JobMover(job, si, workDir=workDir, stageoutretry=stageoutTries)

    eventType = "put_es"

    rse = getRSE(job.ddmEndPointOut)  # at this point, get the RSE for the first file in the list, fileid=0
    client = getClientVersionString(pinitdir)
    mover.trace_report = TraceReport(pq=jobSite.sitename, localSite=rse, remoteSite=rse, dataset="", eventType=eventType, eventVersion=client)
    mover.trace_report.init(job)
    error = None
    storageId = None
    try:
        if not activity:
            activity = "es_events"

        file = files[0]
        if file.storageId and file.storageId != -1:
            storageId = file.storageId
            copytools = [('objectstore', {'setup': ''})]
        else:
            copytools = None
        transferred_files, failed_transfers = mover.stageout(activity=activity, files=files, copytools=copytools)
    except PilotException, e:
        error = e
    except Exception, e:
        tolog("ERROR: Mover put data failed [stageout]: exception caught: %s" % e)
        tolog(traceback.format_exc())
        error = PilotException('STAGEOUT FAILED, exception=%s' % e, code=PilotErrors.ERR_STAGEOUTFAILED, state='STAGEOUT_FAILED')

    if error:
        ## send trace
        mover.trace_report.update(clientState=error.state or 'STAGEOUT_FAILED', stateReason=error.message, timeEnd=time())
        mover.sendTrace(mover.trace_report)
        return error.code, error.message, None

    tolog("Mover put data finished")

    # prepare compatible output
    # keep track of which files have been copied

    not_transferred = [e.lfn for e in files if e.status not in ['transferred']]
    if not_transferred:
        err_msg = 'STAGEOUT FAILED: not all output files have been copied: remain files=%s, errors=%s' % ('\n'.join(not_transferred), ';'.join([str(ee) for ee in failed_transfers]))
        tolog("Mover put data finished: error_msg=%s" % err_msg)
        return PilotErrors.ERR_STAGEOUTFAILED, err_msg, None

    return 0, "", storageId


def getClientVersionString(pinitdir):
    """

    :param pinitdir:
    :return:
    """


    pilot_version = getPilotVersion(pinitdir)