conddb2hdf5 Namespace Reference

Classes
class	DBDataProduct
class	DBGlobalTag
class	DBPayload
class	DBPayloadIterator
class	DBTag
class	IOVSyncValue

Functions
def	_checkMerge (previousIOV, newIOV, debugCopy, nExistingDataProducts)
def	_connect (db, init, read_only, args, as_admin=False)
def	_exists (session, primary_key, value)
def	_high (n)
def	_inserted_before (_IOV, timestamp)
def	_low (n)
def	_parse_timestamp (timestamp)
def	canonicalProductName (product)
def	connect (args, init=False, read_only=True, as_admin=False)
def	external_process_get_payloads_objtype_data (queue, args, payloads)
def	get_payloads_objtype_data (session, payloads)
def	globalTagInfo (session, name)
def	main ()
def	mergeIOVs (previousIOV, newIOV)
def	parseSince (time_type, since)
def	previousSyncValue (syncValue)
def	recordToType (record)
def	sinceToIOV (sinceList, time_type)
def	tagInfo (session, name, snapshot)
def	timeTypeName (time_type)
def	writeTag (tagsGroup, time_type, IOV_payloads, payloadToRefs, originalTagNames, recName)
def	writeTagImpl (tagsGroup, name, recName, time_type, IOV_payloads, payloadToRefs, originalTagNames)

Function Documentation

_checkMerge()

def conddb2hdf5._checkMerge (   previousIOV,   newIOV,   debugCopy,   nExistingDataProducts  )

private

Definition at line 382 of file conddb2hdf5.py.

References join().

Referenced by mergeIOVs().

def _checkMerge(previousIOV, newIOV, debugCopy, nExistingDataProducts):
    #sanity check
    #check proper number of entries
    previousSince = -1
    for i,e in enumerate(previousIOV):
        if len(e[1]) != nExistingDataProducts+1:
            raise RuntimeError("entry %i has wrong number of elements %i instead of %i"%(i,len(e[1]),nExistingDataProducts+1))
        if previousSince >= e[0]:
            #print(previousIOV,newIOV)
            raise RuntimeError("IOV not in order for index %i"%i)
        previousSince = e[0]

    previousIndex = 0
    debugIndex =0
    while debugIndex < len(debugCopy) and previousIndex < len(previousIOV):
        previousSince = previousIOV[previousIndex][0]
        debugSince = debugCopy[debugIndex][0]
        #print("debugSince: %i, prevSince: %i"%(debugSince,previousSince))
        #print(debugCopy)
        #print(previousIOV)
        if debugSince != previousSince:
            previousIndex +=1
            continue
        if debugCopy[debugIndex][1] != previousIOV[previousIndex][1][:nExistingDataProducts]:
            raise RuntimeError("packaged were not properly copied for index %i original:%s new:%s"%(debugIndex,",".join(debugCopy[debugIndex][1]),",".join(previousIOV[previousIndex][1][:nExistingDataProducts])))
        debugIndex +=1
        previousIndex +=1
    if debugIndex != len(debugCopy):
        raise RuntimeError("failed to copy forward index %i"%debugIndex)
    newIndex = 0
    previousIndex = 0
    while newIndex < len(newIOV) and previousIndex < len(previousIOV):
        previousSince = previousIOV[previousIndex][0]
        newSince = newIOV[newIndex][0]
        if newSince != previousSince:
            previousIndex +=1
            continue
        if previousIOV[previousIndex][1][-1] != newIOV[newIndex][1]:
            raise RuntimeError("failed to append package at index %i"%newIndex)
        previousIndex +=1
        newIndex +=1
    if newIndex != len(newIOV):
        raise RuntimeError("failed to merge IOV entry %i"%newIndex)
    

_connect()

def conddb2hdf5._connect (   db,   init,   read_only,   args,   as_admin = False  )

private

Definition at line 72 of file conddb2hdf5.py.

Referenced by connect().

def _connect(db, init, read_only, args, as_admin=False):

    logging.debug('Preparing connection to %s ...', db)

    url = conddb.make_url( db, read_only)
    pretty_url = url
    if url.drivername == 'oracle+frontier':
        ws = url.host.rsplit('%2F')
        if ws is not None:
            pretty_url = 'frontier://%s/%s' %(ws[-1],url.database)
    connTo = '%s [%s]' %(db,pretty_url)
    logging.info('Connecting to %s', connTo)
    logging.debug('DB url: %s',url)
    verbose= 0 
    if args.verbose is not None:
       verbose = args.verbose - 1
    connection = conddb.connect(url, args.authPath, verbose, as_admin)


    if not read_only:
        if connection.is_read_only:
            raise Exception('Impossible to edit a read-only database.')

        if connection.is_official:
            if args.force:
                if not args.yes:
                    logging.warning('You are going to edit an official database. If you are not one of the Offline DB experts but have access to the password for other reasons, please stop now.')
            else:
                raise Exception('Editing official databases is forbidden. Use the official DropBox to upload conditions. If you need a special intervention on the database, see the contact help: %s' % conddb.contact_help)
        # for sqlite we trigger the implicit schema creation
        if url.drivername == 'sqlite':
            if init:
                connection.init()
    if not connection._is_valid:
        raise Exception('No valid schema found in the database.')

    return connection

_exists()

def conddb2hdf5._exists (   session,   primary_key,   value  )

private

Definition at line 61 of file conddb2hdf5.py.

References submitPVResolutionJobs.count, and ALCARECOTkAlBeamHalo_cff.filter.

Referenced by globalTagInfo(), and tagInfo().

def _exists(session, primary_key, value):
    ret = None
    try: 
        ret = session.query(primary_key).\
            filter(primary_key == value).\
            count() != 0
    except sqlalchemy.exc.OperationalError:
        pass

    return ret

_high()

def conddb2hdf5._high (   n )

private

Definition at line 130 of file conddb2hdf5.py.

References createfilelist.int.

Referenced by parseSince().

def _high(n):
    return int(n) >> 32

_inserted_before()

def conddb2hdf5._inserted_before (   _IOV,   timestamp  )

private

To be used inside filter().

Definition at line 30 of file conddb2hdf5.py.

References _parse_timestamp().

Referenced by tagInfo().

def _inserted_before(_IOV,timestamp):
    '''To be used inside filter().
    '''

    if timestamp is None:
        # XXX: Returning None does not get optimized (skipped) by SQLAlchemy,
        #      and returning True does not work in Oracle (generates "and 1"
        #      which breaks Oracle but not SQLite). For the moment just use
        #      this dummy condition.
        return sqlalchemy.literal(True) == sqlalchemy.literal(True)

    return _IOV.insertion_time <= _parse_timestamp(timestamp)

_low()

def conddb2hdf5._low (   n )

private

Definition at line 133 of file conddb2hdf5.py.

References createfilelist.int.

Referenced by parseSince().

def _low(n):
    return int(n) & 0xffffffff

#end from conddb

#based on conddb._dump_payload

_parse_timestamp()

def conddb2hdf5._parse_timestamp (   timestamp )

private

Definition at line 43 of file conddb2hdf5.py.

Referenced by _inserted_before(), and conddblib._inserted_before().

def _parse_timestamp(timestamp):
    try:
        return datetime.datetime.strptime(timestamp, '%Y-%m-%d %H:%M:%S.%f')
    except ValueError:
        pass

    try:
        return datetime.datetime.strptime(timestamp, '%Y-%m-%d %H:%M:%S')
    except ValueError:
        pass

    try:
        return datetime.datetime.strptime(timestamp, '%Y-%m-%d')
    except ValueError:
        pass

    raise Exception("Could not parse timestamp '%s'" % timestamp)

canonicalProductName()

def conddb2hdf5.canonicalProductName

(

product

)

Definition at line 541 of file conddb2hdf5.py.

Referenced by conddb2hdf5.DBPayloadIterator.__next__().

def canonicalProductName(product):
    return __typedefs.get(product,product)

connect()

def conddb2hdf5.connect	(		args,
			init = False,
			read_only = True,
			as_admin = False
	)

Definition at line 111 of file conddb2hdf5.py.

References _connect(), and DeadROC_duringRun.dir.

Referenced by external_process_get_payloads_objtype_data(), and main().

def connect(args, init=False, read_only=True, as_admin=False):
    args.force = args.force if 'force' in dir(args) else False

    if 'destdb' in args:
        if args.destdb is None:
            args.destdb = args.db
        if args.db == args.destdb:
            conn1 = _connect(args.destdb, init, read_only, args)
            return conn1, conn1
        conn1 = _connect( args.db, init, True, args)
        conn2url = conddb.make_url(args.destdb, False)
        if conn2url.drivername == 'sqlite' and not os.path.exists(args.destdb): 
            init = True
        conn2 = _connect(args.destdb, init, False, args)
        return conn1, conn2

    return _connect( args.db, init, read_only, args, as_admin)

external_process_get_payloads_objtype_data()

def conddb2hdf5.external_process_get_payloads_objtype_data	(		queue,
			args,
			payloads
	)

Definition at line 146 of file conddb2hdf5.py.

References connect(), and get_payloads_objtype_data().

def external_process_get_payloads_objtype_data(queue, args, payloads):
    connection = connect(args)
    session = connection.session()
    queue.put(get_payloads_objtype_data(session, payloads))
#local

get_payloads_objtype_data()

def conddb2hdf5.get_payloads_objtype_data	(		session,
			payloads
	)

Definition at line 139 of file conddb2hdf5.py.

References python.cmstools.all(), and ALCARECOTkAlBeamHalo_cff.filter.

Referenced by external_process_get_payloads_objtype_data().

def get_payloads_objtype_data(session, payloads):

    Payload = session.get_dbtype(conddb.Payload)
    table = session.query(Payload.hash, Payload.object_type, Payload.data).\
        filter(Payload.hash.in_(payloads)).order_by(Payload.hash).all()
    return table

globalTagInfo()

def conddb2hdf5.globalTagInfo	(		session,
			name
	)

Definition at line 330 of file conddb2hdf5.py.

References _exists(), python.cmstools.all(), and ALCARECOTkAlBeamHalo_cff.filter.

def globalTagInfo(session,name):
    GlobalTag = session.get_dbtype(conddb.GlobalTag)
    GlobalTagMap = session.get_dbtype(conddb.GlobalTagMap)
    try:
        is_global_tag = _exists(session, GlobalTag.name, name)
        if is_global_tag:
            return session.query(GlobalTagMap.record, GlobalTagMap.label, GlobalTagMap.tag_name).\
                filter(GlobalTagMap.global_tag_name == name).\
                order_by(GlobalTagMap.record, GlobalTagMap.label).\
                all()
    except sqlalchemy.exc.OperationalError:
        sys.stderr.write("No table for GlobalTags found in DB.\n\n")
    return None

main()

def conddb2hdf5.main

(

)

Definition at line 544 of file conddb2hdf5.py.

References connect(), beamvalidation.exit(), print(), and hdf5Writer.writeH5File().

def main():
    parser = argparse.ArgumentParser(description='Read from CMS Condition DB and write to HDF5 file')
    parser.add_argument('--db', '-d', default='pro', help='Database to run the command on. Run the help subcommand for more information: conddb help')
    parser.add_argument('name', nargs='+', help="Name of the global tag.")
    parser.add_argument('--verbose', '-v', action='count', help='Verbosity level. -v prints debugging information of this tool, like tracebacks in case of errors. -vv prints, in addition, all SQL statements issued. -vvv prints, in addition, all results returned by queries.')
    parser.add_argument('--authPath','-a', default=None, help='Path of the authentication .netrc file. Default: the content of the COND_AUTH_PATH environment variable, when specified.')
    parser.add_argument('--snapshot', '-T', default=None, help="Snapshot time. If provided, the output will represent the state of the IOVs inserted into database up to the given time. The format of the string must be one of the following: '2013-01-20', '2013-01-20 10:11:12' or '2013-01-20 10:11:12.123123'.")
    parser.add_argument('--exclude', '-e', nargs='*', help = 'list of records to exclude from the file (can not be used with --include)')
    parser.add_argument('--include', '-i', nargs='*', help = 'lost of the only records that should be included in the file (can not be used with --exclude')
    parser.add_argument('--output', '-o', default='test.h5cond', help='name of hdf5 output file to write')
    parser.add_argument('--compressor', '-c', default='zlib', choices =['zlib','lzma','none'], help="compress data using 'zlib', 'lzma' or 'none'")    
    args = parser.parse_args()

    if args.exclude and args.include:
        print("Can not use --exclude and --include at the same time")
        exit(-1)

    connection = connect(args)
    session = connection.session()

    excludeRecords = set()
    if args.exclude:
        excludeRecords = set(args.exclude)
    includeRecords = set()
    if args.include:
        includeRecords = set(args.include)

    writeH5File(args.output, args.name, excludeRecords, includeRecords, lambda x: DBGlobalTag(args, session,  x), args.compressor )
    

mergeIOVs()

def conddb2hdf5.mergeIOVs	(		previousIOV,
			newIOV
	)

Definition at line 427 of file conddb2hdf5.py.

References _checkMerge(), and mps_setup.append.

Referenced by conddb2hdf5.DBTag.iovsNPayloadNames().

def mergeIOVs(previousIOV, newIOV):
    debugCopy = copy.deepcopy(previousIOV)
    previousSize = len(previousIOV)
    newSize = len(newIOV)
    previousIndex = 0
    newIndex =0
    nExistingDataProducts = len(previousIOV[0][1])
    while newIndex < newSize and previousIndex < previousSize:

        previousSince = previousIOV[previousIndex][0]
        newSince = newIOV[newIndex][0]
        if previousSince == newSince:
            previousIOV[previousIndex][1].append(newIOV[newIndex][1])
            newIndex +=1
            previousIndex +=1
            continue
        elif newSince < previousSince:
            if previousIndex == 0:
                payloads = [None]*nExistingDataProducts
                payloads.append(newIOV[newIndex][1])
                previousIOV.insert(0,[newSince,payloads])
            else:
                payloads = previousIOV[previousIndex-1][1][:nExistingDataProducts]
                payloads.append(newIOV[newIndex][1])
                previousIOV.insert(previousIndex,[newSince,payloads])
            newIndex +=1
            previousIndex +=1
            previousSize +=1
        elif newSince > previousSince:
            if newIndex == 0:
                previousIOV[previousIndex][1].append(None)
            else:
                if len(previousIOV[previousIndex][1]) == nExistingDataProducts:
                    previousIOV[previousIndex][1].append(newIOV[newIndex-1][1])
            previousIndex +=1
    if newIndex != newSize:
        #print("NEED TO EXTEND")
        #need to append to end
        previousPayloads = previousIOV[-1][1]
        while newIndex != newSize:
            newPayloads = previousPayloads[:]
            newPayloads[nExistingDataProducts] = newIOV[newIndex][1]
            previousIOV.append([newIOV[newIndex][0], newPayloads])
            newIndex +=1
    if previousIndex != previousSize:
        #need to add new item to all remaining entries
        while previousIndex < previousSize:
            previousIOV[previousIndex][1].append(newIOV[-1][1])
            previousIndex +=1
    _checkMerge(previousIOV, newIOV, debugCopy, nExistingDataProducts)
    return previousIOV

parseSince()

def conddb2hdf5.parseSince	(		time_type,
			since
	)

Definition at line 306 of file conddb2hdf5.py.

References _high(), and _low().

Referenced by sinceToIOV().

def parseSince(time_type, since):
    if time_type == conddb.TimeType.Time.value:
        return (_high(since), _low(since))
    if time_type == conddb.TimeType.Run.value:
        return (_high(since), 0)
    if time_type == conddb.TimeType.Lumi.value:
        return (_high(since), _low(since))

previousSyncValue()

def conddb2hdf5.previousSyncValue

(

syncValue

)

Definition at line 314 of file conddb2hdf5.py.

Referenced by sinceToIOV().

def previousSyncValue(syncValue):
    if syncValue[1] == 0:
        return (syncValue[0]-1, 0xffffffff)
    return (syncValue[0], syncValue[1]-1)
    

recordToType()

def conddb2hdf5.recordToType

(

record

)

Definition at line 512 of file conddb2hdf5.py.

Referenced by conddb2hdf5.DBTag.__type().

def recordToType(record):
    import subprocess
    return subprocess.run(["condRecordToDataProduct",record], capture_output = True, check=True, text=True).stdout

sinceToIOV()

def conddb2hdf5.sinceToIOV	(		sinceList,
			time_type
	)

Definition at line 319 of file conddb2hdf5.py.

References parseSince(), and previousSyncValue().

Referenced by conddb2hdf5.DBTag.iovsNPayloadNames().

def sinceToIOV(sinceList, time_type):
    firstValues = []
    lastValues = []
    for since in sinceList:
        syncValue = parseSince(time_type, since)
        firstValues.append(syncValue)
        if len(firstValues) != 1:
            lastValues.append(previousSyncValue(syncValue))
    lastValues.append((0xFFFFFFFF,0xFFFFFFFF))
    return [firstValues,lastValues]
    

tagInfo()

def conddb2hdf5.tagInfo	(		session,
			name,
			snapshot
	)

Definition at line 344 of file conddb2hdf5.py.

References _exists(), _inserted_before(), python.cmstools.all(), ALCARECOTkAlBeamHalo_cff.filter, and createfilelist.int.

Referenced by conddb2hdf5.DBTag.iovsNPayloadNames().

def tagInfo(session, name, snapshot):
    Tag = session.get_dbtype(conddb.Tag)
    IOV = session.get_dbtype(conddb.IOV)
    is_tag = _exists(session, Tag.name, name)
    if is_tag:
        time_type = session.query(Tag.time_type).\
            filter(Tag.name == name).\
            scalar()
            
        rawTagInfo = session.query(IOV.since, IOV.insertion_time, IOV.payload_hash).\
                     filter(
                         IOV.tag_name == name,
                         _inserted_before(IOV,snapshot),
                     ).\
                     order_by(IOV.since.desc(), IOV.insertion_time.desc()).\
                    from_self().\
                     order_by(IOV.since, IOV.insertion_time).\
                     all()
        filteredTagInfo = []
        lastSince = -1
        for since,insertion,payload in rawTagInfo:
            if lastSince == since:
                continue
            lastSince = since
            if time_type == conddb.TimeType.Run.value:
                #need to make Run and RunLumi directly comparable since some records
                # use a mix of the two for their IOVs
                since = int(since) << 32
            filteredTagInfo.append((since,payload))

        if time_type == conddb.TimeType.Run.value:
            time_type = conddb.TimeType.Lumi.value

        return time_type, filteredTagInfo
#                     [sinceLabel, 'Insertion Time', 'Payload', 'Object Type'],
#                     filters = [_since_filter(time_type), None, None, None],
#        )
   

timeTypeName()

def conddb2hdf5.timeTypeName

(

time_type

)

Definition at line 297 of file conddb2hdf5.py.

References str.

Referenced by conddb2hdf5.DBTag.time_type().

def timeTypeName(time_type):
    if time_type == conddb.TimeType.Time.value:
        return 'time'
    if time_type == conddb.TimeType.Run.value or time_type == conddb.TimeType.Lumi.value:
        return 'run_lumi'
    raise RuntimeError("unknown since time %s:"% str(time_type))
                       
        

writeTag()

def conddb2hdf5.writeTag	(		tagsGroup,
			time_type,
			IOV_payloads,
			payloadToRefs,
			originalTagNames,
			recName
	)

Definition at line 505 of file conddb2hdf5.py.

References writeTagImpl().

def writeTag(tagsGroup, time_type, IOV_payloads, payloadToRefs, originalTagNames, recName):
    name = originalTagNames[0]
    if len(originalTagNames) != 1:
        name = name+"@joined"
    return writeTagImpl(tagsGroup, name, recName, time_type, IOV_payloads, payloadToRefs, originalTagNames)
    

writeTagImpl()

def conddb2hdf5.writeTagImpl	(		tagsGroup,
			name,
			recName,
			time_type,
			IOV_payloads,
			payloadToRefs,
			originalTagNames
	)

Definition at line 479 of file conddb2hdf5.py.

Referenced by writeTag().

def writeTagImpl(tagsGroup, name, recName, time_type, IOV_payloads, payloadToRefs, originalTagNames):
    tagGroup = tagsGroup.create_group(name)
    tagGroup.attrs["time_type"] = time_type.encode("ascii") #timeTypeName(time_type).encode("ascii")
    tagGroup.attrs["db_tags"] = [x.encode("ascii") for x in originalTagNames]
    tagGroup.attrs["record"] = recName.encode("ascii")
    firstValues = [x[0] for x in IOV_payloads]
    lastValues = [x[1] for x in IOV_payloads]
    syncValueType = np.dtype([("high", np.uint32),("low", np.uint32)])
    first_np = np.empty(shape=(len(IOV_payloads),), dtype=syncValueType)
    first_np['high'] = [ x.high for x in firstValues]
    first_np['low'] = [ x.low for x in firstValues]
    last_np = np.empty(shape=(len(lastValues),), dtype=syncValueType)
    last_np['high'] = [ x.high for x in lastValues]
    last_np['low'] = [ x.low for x in lastValues]
    #tagGroup.create_dataset("first",data=np.array(firstValues), dtype=syncValueType)
    #tagGroup.create_dataset("last", data=np.array(lastValues),dtype=syncValueType)
    payloads = [ [ payloadToRefs[y] for y in x[2]] for x in IOV_payloads]
    compressor = None
    if len(first_np) > 100:
        compressor = 'gzip'
    tagGroup.create_dataset("first",data=first_np, compression = compressor)
    tagGroup.create_dataset("last",data=last_np, compression = compressor)
    tagGroup.create_dataset("payload", data=payloads, dtype=h5py.ref_dtype, compression = compressor)
    return tagGroup.ref