ctbmodules.update_ctb_settings Namespace Reference

Functions
None	update_ctb_settings (Path db_file, Path jsonfile, str|None session_name)
list[str]	set_subsystems (conffwk.Configuration db, dict systems)
list[str]	set_beam (conffwk.Configuration db, dict beam)
list[str]	set_beam_llts (conffwk.Configuration db, dict triggers)
list[str]	set_crt (conffwk.Configuration db, dict crt)
list[str]	set_crt_llts (conffwk.Configuration db, dict triggers)
None	set_pds (conffwk.Configuration db, dict pds)
None	set_pds_llts (conffwk.Configuration db, dict triggers)
list[str]	set_hlts (conffwk.Configuration db, dict hlts)
None	set_hlt (conffwk.Configuration db, dict hlt)
list[str]	set_misc (conffwk.Configuration db, dict misc)
list[str]	set_random_trigger (conffwk.Configuration db, str trigger_id, dict trigger)
None	set_sockets (conffwk.Configuration db, dict sockets)

Function Documentation

set_beam()

list[str] ctbmodules.update_ctb_settings.set_beam	(	conffwk.Configuration	db,
		dict	beam )

Definition at line 49 of file update_ctb_settings.py.

def set_beam(db:conffwk.Configuration, beam:dict) -> list[str] :
 
    board = db.get_dals("CTBoardConf")[0]  
    oks = board.beam
    oks.channel_mask = beam["channel_mask"]
    oks.reshape_lengths = beam["reshape_lengths"]
    oks.delays = beam["delays"]
    db.update_dal(oks)
 
    return set_beam_llts(db, beam["triggers"])
 
 

set_beam_llts()

list[str] ctbmodules.update_ctb_settings.set_beam_llts	(	conffwk.Configuration	db,
		dict	triggers )

Definition at line 61 of file update_ctb_settings.py.

def set_beam_llts(db:conffwk.Configuration, triggers:dict) -> list[str] :
 
    enable = []
    for t in triggers :
        oks = db.get_dal("CTBLLT", t["id"])
        oks.mask = t["mask"]
        oks.description = t["description"]
        db.update_dal(oks)
        if t["enable"] :
            enable.append(t["id"])
    return enable
 
 

set_crt()

list[str] ctbmodules.update_ctb_settings.set_crt	(	conffwk.Configuration	db,
		dict	crt )

Definition at line 74 of file update_ctb_settings.py.

def set_crt(db:conffwk.Configuration, crt:dict) -> list[str] :
 
    oks = db.get_dals("CTBCRTSubsystem")[0]
    oks.pixelate = crt["pixelate"]
    oks.channel_mask = crt["channel_mask"]
    oks.reshape_lengths = crt["reshape_lengths"]
    oks.delays = crt["delays"]
    db.update_dal(oks)
 
    return set_crt_llts(db, crt["triggers"])
    

set_crt_llts()

list[str] ctbmodules.update_ctb_settings.set_crt_llts	(	conffwk.Configuration	db,
		dict	triggers )

Definition at line 85 of file update_ctb_settings.py.

def set_crt_llts(db:conffwk.Configuration, triggers:dict) -> list[str] :
 
    enable = []
    for t in triggers :
        oks = db.get_dal("CTBCountLLT", t["id"])
        oks.count = t["count"]
        oks.type = t["type"]
        oks.mask = t["mask"]
        oks.description = t["description"]
        db.update_dal(oks)
        if t["enable"] :
            enable.append(t["id"])
    return enable
        
 

set_hlt()

None ctbmodules.update_ctb_settings.set_hlt	(	conffwk.Configuration	db,
		dict	hlt )

Definition at line 132 of file update_ctb_settings.py.

def set_hlt(db:conffwk.Configuration, hlt:dict) -> None :
    oks = db.get_dal("CTBHLT", hlt["id"])
    oks.minc = hlt["minc"]
    oks.mexc = hlt["mexc"]
    oks.prescale = hlt["prescale"]
    oks.description = hlt["description"]
    db.update_dal(oks)
    

set_hlts()

list[str] ctbmodules.update_ctb_settings.set_hlts	(	conffwk.Configuration	db,
		dict	hlts )

Definition at line 122 of file update_ctb_settings.py.

def set_hlts(db:conffwk.Configuration, hlts:dict) -> list[str] :
 
    enable = []
    for d in hlts :
        set_hlt(db, d)
        if d["enable"] :
            enable.append(d["id"])
            
    return enable
 

set_misc()

list[str] ctbmodules.update_ctb_settings.set_misc	(	conffwk.Configuration	db,
		dict	misc )

Definition at line 140 of file update_ctb_settings.py.

def set_misc(db:conffwk.Configuration, misc:dict) -> list[str] :
 
    oks_misc = db.get_dals("CTBMisc")[0]
    oks_misc.ch_status = misc["ch_status"]
    db.update_dal(oks_misc)
 
    pulser = misc["pulser"]
    oks_pulser = oks_misc.pulser
    oks_pulser.enable = pulser["enable"]
    oks_pulser.frequency = pulser["frequency"]
    db.update_dal(oks_pulser)
 
    timing = misc["timing"]
    oks_timing = oks_misc.timing
    oks_timing.address = timing["address"]
    oks_timing.triggers = timing["triggers"]
    oks_timing.lockout = timing["lockout"]
 
    db.update_dal(oks_timing)
 
    to_be_enabled = []
    to_be_enabled += set_random_trigger(db, "HLT_0", misc["randomtrigger_1"])
    to_be_enabled += set_random_trigger(db, "LLT_0", misc["randomtrigger_2"])
 
    return to_be_enabled
 

set_pds()

None ctbmodules.update_ctb_settings.set_pds	(	conffwk.Configuration	db,
		dict	pds )

Definition at line 100 of file update_ctb_settings.py.

def set_pds(db:conffwk.Configuration, pds:dict) -> None :
    oks = db.get_dals("CTBPDSSubsystem")[0]
    oks.dac_thresholds = pds["dac_thresholds"]
    oks.channel_mask = pds["channel_mask"]
    oks.reshape_lengths = pds["reshape_lengths"]
    oks.delays = pds["delays"]
    db.update_dal(oks)
 
    set_pds_llts(db, pds["triggers"])
   
 

set_pds_llts()

None ctbmodules.update_ctb_settings.set_pds_llts	(	conffwk.Configuration	db,
		dict	triggers )

Definition at line 111 of file update_ctb_settings.py.

def set_pds_llts(db:conffwk.Configuration, triggers:dict) -> None :
    for t in triggers :
        oks = db.get_dal("CTBPDSLLT", t["id"])
        oks.id = t["id"]
        oks.description = t["description"]
        oks.enable = t["enable"]
        oks.mask= t["mask"]
        oks.type = t["mask"]
        oks.count = t["count"]
        db.update_dal(oks)
    

set_random_trigger()

list[str] ctbmodules.update_ctb_settings.set_random_trigger	(	conffwk.Configuration	db,
		str	trigger_id,
		dict	trigger )

Definition at line 166 of file update_ctb_settings.py.

def set_random_trigger(db:conffwk.Configuration, trigger_id:str, trigger:dict) -> list[str] :
    oks_trigger=db.get_dal("CTBRandomTrigger", trigger_id)
    oks_trigger.fixed_freq=trigger["fixed_freq"]
    oks_trigger.beam_mode=trigger["beam_mode"]
    oks_trigger.period=trigger["period"]
    oks_trigger.description=trigger["description"]
    db.update_dal(oks_trigger)
        
    if trigger["enable"] :
        return [trigger_id]
    return []
    
    

set_sockets()

None ctbmodules.update_ctb_settings.set_sockets	(	conffwk.Configuration	db,
		dict	sockets )

Definition at line 179 of file update_ctb_settings.py.

def set_sockets(db:conffwk.Configuration, sockets:dict) -> None :
 
    oks_sockets = db.get_dals("CTBSockets")[0]
    
    oks_receiver = oks_sockets.receiver
    receiver = sockets["receiver"]
    oks_receiver.rollover = receiver["rollover"]
    oks_receiver.port = receiver["port"]
    db.update_dal(oks_receiver)
    
    oks_monitor = oks_sockets.monitor
    monitor = sockets["monitor"]
    oks_monitor.enable = monitor["enable"]
    oks_monitor.port = monitor["port"]
    db.update_dal(oks_monitor)
    
    oks_statistics = oks_sockets.statistics
    stat = sockets["statistics"]
    oks_statistics.enable= stat["enable"]
    oks_statistics.port = stat["port"]
    oks_statistics.updt_period = stat["updt_period"]
    db.update_dal(oks_statistics)
 
        
    
    

set_subsystems()

list[str] ctbmodules.update_ctb_settings.set_subsystems	(	conffwk.Configuration	db,
		dict	systems )

Definition at line 40 of file update_ctb_settings.py.

def set_subsystems(db:conffwk.Configuration, systems:dict) -> list[str] :
 
    set_pds(db, systems["pds"])
 
    enable = []
    enable += set_crt(db, systems["crt"])
    enable += set_beam(db, systems["beam"])
    return enable
 

update_ctb_settings()

None ctbmodules.update_ctb_settings.update_ctb_settings	(	Path	db_file,
		Path	jsonfile,
		str|None	session_name )

Definition at line 7 of file update_ctb_settings.py.

def update_ctb_settings(db_file:Path, jsonfile:Path, session_name:str|None) -> None :
    print (f"Setting {db_file} to {jsonfile}")
 
    json_data = json.load(open(jsonfile))
    ctb = json_data["ctb"]
 
    db = conffwk.Configuration('oksconflibs:' + db_file)
    set_sockets(db, ctb["sockets"])
 
    
    to_be_enabled = set_misc( db, ctb["misc"])
    to_be_enabled += set_hlts( db, ctb["HLT"]["trigger"])  
    to_be_enabled += set_subsystems( db, ctb["subsystems"])
 
    db.commit()
    
    triggers = db.get_dals("CTBTrigger")
    all_triggers = [t.id for t in triggers]
    to_be_disabled = []
    for t in all_triggers :
        if t not in to_be_enabled :
            to_be_disabled.append(t)
 
    print("Enabling:",to_be_enabled)
    print("Disabling:",to_be_disabled)
 
    if not session_name :
        session_name = db.get_dals("Session")[0].id
        
    enable.enable(db_file, False, to_be_enabled, session_name)
    enable.enable(db_file, True, to_be_disabled, session_name)