pandaserver/taskbuffer/JediFileSpec.py

0001 """
0002 file specification for JEDI
0003
0004 """
0005
0006 import re
0007 import types
0008
0009 from pandaserver.taskbuffer.FileSpec import FileSpec as JobFileSpec
0010
0011
0012 class JediFileSpec(object):
0013     # attributes
0014     _attributes = (
0015         "jediTaskID",
0016         "datasetID",
0017         "fileID",
0018         "creationDate",
0019         "lastAttemptTime",
0020         "lfn",
0021         "GUID",
0022         "type",
0023         "status",
0024         "fsize",
0025         "checksum",
0026         "scope",
0027         "attemptNr",
0028         "maxAttempt",
0029         "nEvents",
0030         "keepTrack",
0031         "startEvent",
0032         "endEvent",
0033         "firstEvent",
0034         "boundaryID",
0035         "PandaID",
0036         "failedAttempt",
0037         "lumiBlockNr",
0038         "outPandaID",
0039         "maxFailure",
0040         "ramCount",
0041         "is_waiting",
0042         "proc_status",
0043     )
0044     # attributes which have 0 by default
0045     _zeroAttrs = ("fsize", "attemptNr", "failedAttempt", "ramCount")
0046     # mapping between sequence and attr
0047     _seqAttrMap = {"fileID": "ATLAS_PANDA.JEDI_DATASET_CONT_FILEID_SEQ.nextval"}
0048
0049     # constructor
0050     def __init__(self):
0051         # install attributes
0052         for attr in self._attributes:
0053             if attr in self._zeroAttrs:
0054                 object.__setattr__(self, attr, 0)
0055             else:
0056                 object.__setattr__(self, attr, None)
0057         # map of changed attributes
0058         object.__setattr__(self, "_changedAttrs", {})
0059         # locality
0060         object.__setattr__(self, "locality", {})
0061         # source name
0062         object.__setattr__(self, "sourceName", None)
0063
0064     # override __setattr__ to collecte the changed attributes
0065     def __setattr__(self, name, value):
0066         oldVal = getattr(self, name)
0067         object.__setattr__(self, name, value)
0068         newVal = getattr(self, name)
0069         # collect changed attributes
0070         if oldVal != newVal:
0071             self._changedAttrs[name] = value
0072
0073     # reset changed attribute list
0074     def resetChangedList(self):
0075         object.__setattr__(self, "_changedAttrs", {})
0076
0077     # return map of values
0078     def valuesMap(self, useSeq=False, onlyChanged=False):
0079         ret = {}
0080         for attr in self._attributes:
0081             # use sequence
0082             if useSeq and attr in self._seqAttrMap:
0083                 continue
0084             # only changed attributes
0085             if onlyChanged:
0086                 if attr not in self._changedAttrs:
0087                     continue
0088             val = getattr(self, attr)
0089             if val is None:
0090                 if attr in self._zeroAttrs:
0091                     val = 0
0092                 else:
0093                     val = None
0094             ret[f":{attr}"] = val
0095         return ret
0096
0097     # pack tuple into FileSpec
0098     def pack(self, values):
0099         for i in range(len(self._attributes)):
0100             attr = self._attributes[i]
0101             val = values[i]
0102             object.__setattr__(self, attr, val)
0103
0104     # return column names for INSERT
0105     def columnNames(cls, useSeq=False, defaultVales=None, skipDefaultAttr=False):
0106         if defaultVales is None:
0107             defaultVales = {}
0108         ret = ""
0109         for attr in cls._attributes:
0110             if skipDefaultAttr and (attr in cls._seqAttrMap or attr in defaultVales):
0111                 continue
0112             if ret != "":
0113                 ret += ","
0114             if useSeq and attr in cls._seqAttrMap:
0115                 ret += f"{cls._seqAttrMap[attr]}"
0116                 continue
0117             if attr in defaultVales:
0118                 arg = defaultVales[attr]
0119                 if arg is None:
0120                     ret += "NULL"
0121                 elif isinstance(arg, str):
0122                     ret += f"'{arg}'"
0123                 else:
0124                     ret += f"{arg}"
0125                 continue
0126             ret += attr
0127         return ret
0128
0129     columnNames = classmethod(columnNames)
0130
0131     # return expression of bind variables for INSERT
0132     def bindValuesExpression(cls, useSeq=True):
0133         ret = "VALUES("
0134         for attr in cls._attributes:
0135             if useSeq and attr in cls._seqAttrMap:
0136                 ret += f"{cls._seqAttrMap[attr]},"
0137             else:
0138                 ret += f":{attr},"
0139         ret = ret[:-1]
0140         ret += ")"
0141         return ret
0142
0143     bindValuesExpression = classmethod(bindValuesExpression)
0144
0145     # return an expression of bind variables for UPDATE to update only changed attributes
0146     def bindUpdateChangesExpression(self):
0147         ret = ""
0148         for attr in self._attributes:
0149             if attr in self._changedAttrs:
0150                 ret += f"{attr}=:{attr},"
0151         ret = ret[:-1]
0152         ret += " "
0153         return ret
0154
0155     # convert to job's FileSpec
0156     def convertToJobFileSpec(self, datasetSpec, setType=None, useEventService=False):
0157         jobFileSpec = JobFileSpec()
0158         jobFileSpec.fileID = self.fileID
0159         jobFileSpec.datasetID = datasetSpec.datasetID
0160         jobFileSpec.jediTaskID = datasetSpec.jediTaskID
0161         jobFileSpec.lfn = self.lfn
0162         jobFileSpec.GUID = self.GUID
0163         if setType is None:
0164             jobFileSpec.type = self.type
0165         else:
0166             jobFileSpec.type = setType
0167         jobFileSpec.scope = self.scope
0168         jobFileSpec.fsize = self.fsize
0169         jobFileSpec.checksum = self.checksum
0170         jobFileSpec.attemptNr = self.attemptNr
0171         # dataset attribute
0172         if datasetSpec is not None:
0173             # dataset
0174             if datasetSpec.containerName not in [None, ""]:
0175                 jobFileSpec.dataset = datasetSpec.containerName
0176             else:
0177                 jobFileSpec.dataset = datasetSpec.datasetName
0178             if self.type in datasetSpec.getInputTypes() or setType in datasetSpec.getInputTypes():
0179                 # prodDBlock
0180                 jobFileSpec.prodDBlock = datasetSpec.datasetName
0181                 # storage token
0182                 if datasetSpec.storageToken not in ["", None]:
0183                     jobFileSpec.dispatchDBlockToken = datasetSpec.storageToken
0184             else:
0185                 # destinationDBlock
0186                 jobFileSpec.destinationDBlock = datasetSpec.datasetName
0187                 # storage token
0188                 if datasetSpec.storageToken not in ["", None]:
0189                     jobFileSpec.destinationDBlockToken = datasetSpec.storageToken.split("/")[0]
0190                 # destination
0191                 if datasetSpec.destination not in ["", None]:
0192                     jobFileSpec.destinationSE = datasetSpec.destination
0193                 # set prodDBlockToken for Event Service
0194                 if useEventService and datasetSpec.getObjectStore() is not None:
0195                     jobFileSpec.prodDBlockToken = f"objectstore^{datasetSpec.getObjectStore()}"
0196                 # allow no output
0197                 if datasetSpec.isAllowedNoOutput():
0198                     jobFileSpec.allowNoOutput()
0199         # return
0200         return jobFileSpec
0201
0202     # convert from job's FileSpec
0203     def convertFromJobFileSpec(self, jobFileSpec):
0204         self.fileID = jobFileSpec.fileID
0205         self.datasetID = jobFileSpec.datasetID
0206         self.jediTaskID = jobFileSpec.jediTaskID
0207         self.lfn = jobFileSpec.lfn
0208         self.GUID = jobFileSpec.GUID
0209         self.type = jobFileSpec.type
0210         self.scope = jobFileSpec.scope
0211         self.fsize = jobFileSpec.fsize
0212         self.checksum = jobFileSpec.checksum
0213         self.attemptNr = jobFileSpec.attemptNr
0214         # convert NULL to None
0215         for attr in self._attributes:
0216             val = getattr(self, attr)
0217             if val == "NULL":
0218                 object.__setattr__(self, attr, None)
0219         # return
0220         return
0221
0222     # get effective number of events
0223     def getEffectiveNumEvents(self):
0224         if self.endEvent is not None and self.startEvent is not None:
0225             evtCounts = self.endEvent - self.startEvent + 1
0226             if evtCounts > 0:
0227                 return evtCounts
0228             return 1
0229         if self.nEvents is not None and self.nEvents > 0:
0230             return self.nEvents
0231         return 1
0232
0233     # extract fields string
0234     def extractFieldsStr(self, fieldNumList):
0235         tmpFieldStr = ""
0236         try:
0237             tmpMidStrList = re.split("\.|_tid\d+", self.lfn)
0238             for tmpFieldNum in fieldNumList:
0239                 tmpFieldStr += "." + tmpMidStrList[tmpFieldNum - 1]
0240         except Exception:
0241             pass
0242         return tmpFieldStr