python.AthHLT Namespace Reference

Classes
class	CondDB

Functions
def	get_sor_params (run_number)
def	get_trigconf_keys (run_number, lb_number)
def	getCACfg (jopath)
def	reload_from_json (filename, suppress_args=[])

Variables
	log = logging.getLogger('athenaHLT')
def	d = get_sor_params(327265)

Function Documentation

get_sor_params()

def python.AthHLT.get_sor_params

(

run_number

)

Definition at line 28 of file AthHLT.py.

def get_sor_params(run_number):
   from CoolConvUtilities import AtlCoolLib
 
   log.info('Reading SOR record for run %s from COOL', run_number)
 
   cdb = CondDB(run_number)
   dbcon = AtlCoolLib.readOpen('COOLONL_TDAQ/%s' % cdb.db_instance())
   folder = dbcon.getFolder(cdb.sor_folder())
 
   # need to keep sor variable while using payload (cannot do the following in
   # one single line nor overwrite sor). Otherwise: 1) GC comes into play;
   # 2) the object is deleted; 3) since it's a shared_ptr, the internal
   # cool::IObject also gets deleted; 4) payload is not valid any longer
   try:
      sor = folder.findObject(run_number << 32, 0)
   except Exception:
      return None        # This can happen for unknown run numbers
 
   payload = sor.payload()
   d = {k: payload[k] for k in payload}
   return d
 
 
@cache

get_trigconf_keys()

def python.AthHLT.get_trigconf_keys	(		run_number,
			lb_number
	)

Read Trigger keys from COOL

Definition at line 52 of file AthHLT.py.

def get_trigconf_keys(run_number, lb_number):
   """Read Trigger keys from COOL"""
 
   from TrigConfStorage.TriggerCoolUtil import TriggerCoolUtil
 
   cdb = CondDB(run_number)
   db = TriggerCoolUtil.GetConnection(cdb.db_instance())
   run_range = [[run_number,run_number]]
   d = {}
   d['SMK'] = TriggerCoolUtil.getHLTConfigKeys(db, run_range)[run_number]['SMK']
 
   def findKey(keys):
      for (key, firstLB, lastLB) in keys:
         if lb_number>=firstLB and lb_number<=lastLB:
            return key
      return None
 
   # Find L1/HLT prescale key
   d['LVL1PSK'] = findKey(TriggerCoolUtil.getL1ConfigKeys(db, run_range)[run_number]['LVL1PSK'])
   d['HLTPSK'] = findKey(TriggerCoolUtil.getHLTPrescaleKeys(db, run_range)[run_number]['HLTPSK2'])
 
   return d
 
 

getCACfg()

def python.AthHLT.getCACfg

(

jopath

)

Return the CA Cfg function based on joboptions path.
The format is MODULE[.FNC]. If no FNC is given, 'main' will be tried.

Definition at line 76 of file AthHLT.py.

def getCACfg(jopath):
   """Return the CA Cfg function based on joboptions path.
   The format is MODULE[.FNC]. If no FNC is given, 'main' will be tried."""
 
   import importlib
 
   sys.path.append('.')   # temporarily add local directory to search path
 
   # try to import module as given:
   try:
      module = importlib.import_module(jopath)
   except ModuleNotFoundError:
      if '.' not in jopath:
         raise
      # or interpret as module.fnc:
      mod_name, fnc_name = jopath.rsplit('.', maxsplit=1)
      module = importlib.import_module(mod_name)
   else:
      # if the first import worked we are using the 'main(flags)' function:
      fnc_name = 'main'
 
   sys.path.pop()
 
   log.info("Loading %s.%s", module.__name__, fnc_name)
   return getattr(module, fnc_name)
 
 

reload_from_json()

def python.AthHLT.reload_from_json	(		filename,
			suppress_args = []
	)

Re-launch athenaHLT from the given json file. Optionally suppress
the list of command line args (e.g. flags).

Definition at line 103 of file AthHLT.py.

def reload_from_json(filename, suppress_args=[]):
   """Re-launch athenaHLT from the given json file. Optionally suppress
   the list of command line args (e.g. flags)."""
 
   # Remove all command line args that are not compatible with running from JSON:
   argv = []
   for arg_index, arg in enumerate(sys.argv):
      if arg == '--dump-config-reload':
         continue
      if arg in ['--precommand', '-c', '--postcommand', '-C']:
         continue
      if arg_index > 0 and sys.argv[arg_index-1] in ['--precommand', '-c', '--postcommand', '-C']:
         continue
      if arg.startswith('--precommand') or arg.startswith('--postcommand'):
         continue
      if arg in suppress_args:
         continue
      argv.append(arg)
 
   argv[-1] = filename
   log.info('Restarting %s from %s ...', argv[0], argv[-1])
   sys.stdout.flush()
   sys.stderr.flush()
   os.execvp(argv[0], argv)
 
 
#
# Testing (used as ctest)
#

Variable Documentation

d

def python.AthHLT.d = get_sor_params(327265)

Definition at line 139 of file AthHLT.py.

log

python.AthHLT.log = logging.getLogger('athenaHLT')

Definition at line 6 of file AthHLT.py.