python.AthHLT Namespace Reference

Classes
class	CondDB

Functions
dict[str, Any]|None	get_sor_params (run_number)
dict[str, Any]|None	get_eor_params_crest (run_number, str crest_server)
	get_trigconf_keys (run_number, lb_number)
	get_trigconf_keys_crest (run_number, lb_number, crest_server)
	getCACfg (jopath)
	fillFromUnparsedArgs (flags)
	reload_from_json (filename, suppress_args=[], jobOptions=None)

Variables
	log = logging.getLogger('athenaHLT')
list	unparsedArguments = []
	Command line arguments not parsed by athenaEF, i.e.
dict[str, Any]|None	d = get_sor_params(327265)

Function Documentation

fillFromUnparsedArgs()

fillFromUnparsedArgs

(

flags

)

Re-apply the flags given on the athenaEF command line.

Definition at line 107 of file AthHLT.py.

def fillFromUnparsedArgs(flags):
   """Re-apply the flags given on the athenaEF command line."""
 
   for flag_arg in unparsedArguments:
      flags.fillFromString(flag_arg)
 
 

get_eor_params_crest()

dict[str, Any] | None get_eor_params_crest	(		run_number,
		str	crest_server )

Definition at line 53 of file AthHLT.py.

def get_eor_params_crest(run_number, crest_server:str) -> dict[str, Any] | None:
   from TrigConfStorage.TriggerCrestUtil import TriggerCrestUtil
   log.info('Reading EOR record for run %s from Crest', run_number)
   return TriggerCrestUtil.getEORParams(run_number, server=crest_server)
 
@cache

get_sor_params()

dict[str, Any] | None get_sor_params

(

run_number

)

Definition at line 30 of file AthHLT.py.

def get_sor_params(run_number) -> dict[str, Any] | None:
   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())   # type: ignore
 
   # 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()

get_trigconf_keys	(		run_number,
			lb_number )

Read Trigger keys from COOL

Definition at line 59 of file AthHLT.py.

def get_trigconf_keys(run_number, lb_number):
   """Read Trigger keys from COOL"""
 
   from TrigConfStorage.TriggerCoolUtil import TriggerCoolUtil
   confKeys: dict[str, Any] = TriggerCoolUtil.getTrigConfKeys(run_number, lb_number)
   confKeys['DBAlias'] = confKeys.pop('DB', None)
   return confKeys
 
 
@cache

get_trigconf_keys_crest()

get_trigconf_keys_crest	(		run_number,
			lb_number,
			crest_server )

Read Trigger keys from CREST

Definition at line 69 of file AthHLT.py.

def get_trigconf_keys_crest(run_number, lb_number, crest_server):
   """Read Trigger keys from CREST"""
   from TrigConfStorage.TriggerCrestUtil import TriggerCrestUtil   
   log.info("Using CREST server %s", crest_server)
   api: CrestApi = TriggerCrestUtil.getCrestApi(server=crest_server)
   confKeys: dict[str, Any] = TriggerCrestUtil.getTrigConfKeys(run_number, lb_number, api=api)
   confKeys['DBAlias'] = confKeys.pop('DB', None)
   return confKeys
 

getCACfg()

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 78 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()

reload_from_json	(		filename,
			suppress_args = [],
			jobOptions = None )

Re-launch athenaHLT from the given json file. Optionally suppress
the list of command line args (e.g. flags).
jobOptions: the positional argument to be replaced by `filename`.

Definition at line 114 of file AthHLT.py.

def reload_from_json(filename, suppress_args=[], jobOptions=None):
   """Re-launch athenaHLT from the given json file. Optionally suppress
   the list of command line args (e.g. flags).
   jobOptions: the positional argument to be replaced by `filename`."""
 
   # 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)
 
   if jobOptions is not None and jobOptions in argv:
      argv[argv.index(jobOptions)] = filename
   else:
      argv[-1] = filename
   log.info('Restarting %s from %s ...', argv[0], filename)
   sys.stdout.flush()
   sys.stderr.flush()
   os.execvp(argv[0], argv)
 
 
#
# Testing (used as ctest)
#

Variable Documentation

d

python.AthHLT.d = get_sor_params(327265)

Definition at line 154 of file AthHLT.py.

log

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

Definition at line 8 of file AthHLT.py.

unparsedArguments

list python.AthHLT.unparsedArguments = []

Command line arguments not parsed by athenaEF, i.e.

Athena flags.

Definition at line 106 of file AthHLT.py.