athenaEF Namespace Reference

Classes
class	RunParams
class	ConfigRunner
class	MyHelp

Functions
	get_trigconf_keys_from_oks (partition=None, webdaq_base=None, strict=False)
	get_run_params (args=None, from_is=False, partition=None, webdaq_base=None, strict=False, solenoid_current_override=None, toroids_current_override=None)
	load_from_json (json_file, run_params=None, num_threads=1, num_slots=1, ef_overrides=None)
	load_from_database (db_server, smk, l1psk=None, hltpsk=None, run_params=None, num_threads=1, num_slots=1, ef_overrides=None)
str	arg_sor_time (s)
	The following arg_* methods are used as custom types in argparse.
	arg_detector_mask (s)
	check_args (parser, args)
	update_run_params (args, flags)
	update_trigconf_keys (args, flags)
	start_oh_infrastructure (args)
	stop_oh_infrastructure (proc)
	main ()

Variables
	log = logging.getLogger('athenaEF')

Detailed Description

date"

# defaults
export USETCMALLOC=1
export USEIMF=1

# parse command line arguments
for a in ${@}
do
    case "$a" in
        --stdcmalloc)    USETCMALLOC=0;;
        --tcmalloc)      USETCMALLOC=1;;
        --stdcmath)      USEIMF=0;;
        --imf)           USEIMF=1;;
        --preloadlib*)   export ATHENA_ADD_PRELOAD=${a#*=};;
        --no-ers-signal-handlers)  export TDAQ_ERS_NO_SIGNAL_HANDLERS=1;;
    esac
done

# Do the actual preloading via LD_PRELOAD
source `which athena_preload.sh `

# Now resurrect ourselves as python script
python_path=`which python`
"exec" "$python_path" "-tt" "$0" "$@";

Function Documentation

arg_detector_mask()

athenaEF.arg_detector_mask

(

s

)

Convert detector mask to format expected by eformat

Definition at line 803 of file athenaEF.py.

def arg_detector_mask(s):
   """Convert detector mask to format expected by eformat"""
   if s=='all':
      return RunParams.DEFAULT_DETECTOR_MASK
   dmask = hex(int(s,16))                                    # Normalize input to hex-string
   dmask = dmask.lower().replace('0x', '').replace('l', '')  # remove markers
   return '0' * (32 - len(dmask)) + dmask                    # (pad with 0s)
 
 

arg_sor_time()

str athenaEF.arg_sor_time

(

s

)

The following arg_* methods are used as custom types in argparse.

Convert possible SOR time arguments to an OWLTime compatible string

Definition at line 795 of file athenaEF.py.

def arg_sor_time(s) -> str:
   """Convert possible SOR time arguments to an OWLTime compatible string"""
   fmt = '%d/%m/%y %H:%M:%S.%f'
   if s=='now':        return dt.now().strftime(fmt)
   elif s.isdigit():   return dt.fromtimestamp(float(s)/1e9).strftime(fmt)
   else:               return s
 
 

check_args()

athenaEF.check_args	(		parser,
			args )

Consistency check of command line arguments (same as athenaHLT.py)

Definition at line 812 of file athenaEF.py.

def check_args(parser, args):
   """Consistency check of command line arguments (same as athenaHLT.py)"""
 
   if not args.jobOptions and not args.use_database:
      parser.error("No job options file specified")
 
   if (not args.file and not args.dump_config_exit
       and (args.efdf_interface_library or 'TrigDFEmulator') == 'TrigDFEmulator'):
      parser.error("--file is required unless using --dump-config-exit or online efdf-interface-library")
 
   if args.use_crest and not args.use_database:
      parser.error("--use-crest requires --use-database")
 
   if args.oh_monitoring and args.online_environment:
      parser.error("--oh-monitoring (-M) and --online-environment are mutually exclusive.")
 

get_run_params()

athenaEF.get_run_params	(		args = None,
			from_is = False,
			partition = None,
			webdaq_base = None,
			strict = False,
			solenoid_current_override = None,
			toroids_current_override = None )

Get run parameters from the appropriate source.

This is the main entry point for obtaining run parameters. It provides
a single place to modify when adding new sources (like WEBDAQ).

Args:
   args: argparse Namespace with command-line arguments (optional)
   from_is: If True, try to read from WEBDAQ first
   partition: Partition name for IS access (defaults to TDAQ_PARTITION env var)
   webdaq_base: WEBDAQ base URL (defaults to TDAQ_WEBDAQ_BASE env var)
   strict: If True, raise an exception if IS read fails (for --online-environment)
   solenoid_current_override: Command-line override for solenoid current
   toroids_current_override: Command-line override for toroids current

Returns:
   RunParams instance

Raises:
   RuntimeError: If strict=True and IS read fails

Definition at line 489 of file athenaEF.py.

                   solenoid_current_override=None, toroids_current_override=None):
   """
   Get run parameters from the appropriate source.
   
   This is the main entry point for obtaining run parameters. It provides
   a single place to modify when adding new sources (like WEBDAQ).
   
   Args:
      args: argparse Namespace with command-line arguments (optional)
      from_is: If True, try to read from WEBDAQ first
      partition: Partition name for IS access (defaults to TDAQ_PARTITION env var)
      webdaq_base: WEBDAQ base URL (defaults to TDAQ_WEBDAQ_BASE env var)
      strict: If True, raise an exception if IS read fails (for --online-environment)
      solenoid_current_override: Command-line override for solenoid current
      toroids_current_override: Command-line override for toroids current
   
   Returns:
      RunParams instance
   
   Raises:
      RuntimeError: If strict=True and IS read fails
   """
   if from_is:
      return RunParams.from_is(partition=partition, webdaq_base=webdaq_base, strict=strict,
                               solenoid_current_override=solenoid_current_override,
                               toroids_current_override=toroids_current_override)
   elif args is not None:
      return RunParams.from_args(args)
   else:
      return RunParams()
 
 

get_trigconf_keys_from_oks()

athenaEF.get_trigconf_keys_from_oks	(		partition = None,
			webdaq_base = None,
			strict = False )

Read trigger configuration keys (SMK, L1PSK, HLTPSK) and DB info from OKS via WEBDAQ REST API.

This reads the keys from the partition's TriggerConfiguration object and its
related L1TriggerConfiguration and TriggerDBConnection objects.

OKS Structure:
- Partition -> TriggerConfiguration -> L1TriggerConfiguration (Lvl1PrescaleKey)
- Partition -> TriggerConfiguration -> TriggerDBConnection (SuperMasterKey)
- Partition -> TriggerConfiguration -> HLTImplementationDB (hltPrescaleKey)

Args:
   partition: The partition name (default: from TDAQ_PARTITION env var)
   webdaq_base: Base URL for webis_server (default: from TDAQ_WEBDAQ_BASE env var)
   strict: If True, raise an exception if OKS read fails (for --online-environment)

Returns:
   dict with keys: SMK, L1PSK, HLTPSK, db_alias (values may be None if not found)

Raises:
   RuntimeError: If strict=True and OKS read fails

Definition at line 326 of file athenaEF.py.

def get_trigconf_keys_from_oks(partition=None, webdaq_base=None, strict=False):
   """
   Read trigger configuration keys (SMK, L1PSK, HLTPSK) and DB info from OKS via WEBDAQ REST API.
   
   This reads the keys from the partition's TriggerConfiguration object and its
   related L1TriggerConfiguration and TriggerDBConnection objects.
   
   OKS Structure:
   - Partition -> TriggerConfiguration -> L1TriggerConfiguration (Lvl1PrescaleKey)
   - Partition -> TriggerConfiguration -> TriggerDBConnection (SuperMasterKey)
   - Partition -> TriggerConfiguration -> HLTImplementationDB (hltPrescaleKey)
   
   Args:
      partition: The partition name (default: from TDAQ_PARTITION env var)
      webdaq_base: Base URL for webis_server (default: from TDAQ_WEBDAQ_BASE env var)
      strict: If True, raise an exception if OKS read fails (for --online-environment)
   
   Returns:
      dict with keys: SMK, L1PSK, HLTPSK, db_alias (values may be None if not found)
   
   Raises:
      RuntimeError: If strict=True and OKS read fails
   """
   import requests
   
   # Determine the base URL
   if webdaq_base is None:
      webdaq_base = os.environ.get('TDAQ_WEBDAQ_BASE')
   
   if not webdaq_base:
      msg = "TDAQ_WEBDAQ_BASE not set, cannot read from OKS"
      if strict:
         raise RuntimeError(msg + " (required for --online-environment)")
      log.warning(msg)
      return {'SMK': None, 'L1PSK': None, 'HLTPSK': None, 'db_alias': None}
   
   # Determine partition
   if partition is None:
      partition = os.environ.get('TDAQ_PARTITION', 'ATLAS')
   
   log.info("Reading trigger configuration keys from OKS via WEBDAQ: %s (partition=%s)", 
            webdaq_base, partition)
   
   result = {'SMK': None, 'L1PSK': None, 'HLTPSK': None, 'db_alias': None}
   
   def extract_oks_data(response_json):
      """
      Extract data from OKS compact format: [name, type, attributes, relationships]
      Returns tuple (attributes_dict, relationships_dict)
      """
      if isinstance(response_json, list) and len(response_json) >= 4:
         return response_json[2], response_json[3]  # attributes, relationships
      elif isinstance(response_json, list) and len(response_json) >= 3:
         return response_json[2], {}  # attributes only
      return response_json, {}  # fallback
   
   def get_ref_id(ref):
      """Extract object ID from a relationship reference."""
      if isinstance(ref, list) and len(ref) >= 2:
         return ref[0]  # [id, class] format
      elif isinstance(ref, dict) and 'id' in ref:
         return ref['id']
      elif isinstance(ref, str):
         return ref
      return None
   
   # OKS API: GET /info/current/{partition}/oks/{class}/{name}?format=compact
   # Response format: [name, type, attributes, relationships]
   # - attributes: dict of simple values (strings, ints, etc.)
   # - relationships: dict of references to other objects
   try:
      url = f"{webdaq_base}/info/current/{partition}/oks/Partition/{partition}?format=compact"
      log.debug("Fetching Partition from OKS: %s", url)
      
      response = requests.get(url, timeout=10)
      if response.status_code == 200:
         part_attrs, part_rels = extract_oks_data(response.json())
         log.debug("Partition attributes: %s", part_attrs)
         log.debug("Partition relationships: %s", part_rels)
         
         # Get TriggerConfiguration reference from relationships
         trig_conf_id = None
         if 'TriggerConfiguration' in part_rels:
            trig_conf_id = get_ref_id(part_rels['TriggerConfiguration'])
         
         if trig_conf_id:
            log.debug("TriggerConfiguration ID: %s", trig_conf_id)
            
            # Get TriggerConfiguration object
            url = f"{webdaq_base}/info/current/{partition}/oks/TriggerConfiguration/{trig_conf_id}?format=compact"
            response = requests.get(url, timeout=10)
            if response.status_code == 200:
               trig_attrs, trig_rels = extract_oks_data(response.json())
               log.debug("TriggerConfiguration attributes: %s", trig_attrs)
               log.debug("TriggerConfiguration relationships: %s", trig_rels)
               
               # Get L1TriggerConfiguration for L1PSK (relationship 'l1')
               if 'l1' in trig_rels:
                  l1_id = get_ref_id(trig_rels['l1'])
                  if l1_id:
                     url = f"{webdaq_base}/info/current/{partition}/oks/L1TriggerConfiguration/{l1_id}?format=compact"
                     resp = requests.get(url, timeout=10)
                     if resp.status_code == 200:
                        l1_attrs, _ = extract_oks_data(resp.json())
                        log.debug("L1TriggerConfiguration attributes: %s", l1_attrs)
                        if 'Lvl1PrescaleKey' in l1_attrs:
                           result['L1PSK'] = int(l1_attrs['Lvl1PrescaleKey'])
                           log.info("Got L1PSK=%d from OKS", result['L1PSK'])
               
               # Get TriggerDBConnection for SMK and db_alias (relationship 'TriggerDBConnection')
               if 'TriggerDBConnection' in trig_rels:
                  db_id = get_ref_id(trig_rels['TriggerDBConnection'])
                  if db_id:
                     url = f"{webdaq_base}/info/current/{partition}/oks/TriggerDBConnection/{db_id}?format=compact"
                     resp = requests.get(url, timeout=10)
                     if resp.status_code == 200:
                        db_attrs, _ = extract_oks_data(resp.json())
                        log.debug("TriggerDBConnection attributes: %s", db_attrs)
                        if 'SuperMasterKey' in db_attrs:
                           result['SMK'] = int(db_attrs['SuperMasterKey'])
                           log.info("Got SMK=%d from OKS", result['SMK'])
                        if 'Alias' in db_attrs:
                           result['db_alias'] = db_attrs['Alias']
                           log.info("Got db_alias=%s from OKS", result['db_alias'])
               
               # Get HLTImplementationDB for HLTPSK (relationship 'hlt')
               if 'hlt' in trig_rels:
                  hlt_id = get_ref_id(trig_rels['hlt'])
                  if hlt_id:
                     url = f"{webdaq_base}/info/current/{partition}/oks/HLTImplementationDB/{hlt_id}?format=compact"
                     resp = requests.get(url, timeout=10)
                     if resp.status_code == 200:
                        hlt_attrs, _ = extract_oks_data(resp.json())
                        log.debug("HLTImplementationDB attributes: %s", hlt_attrs)
                        if 'hltPrescaleKey' in hlt_attrs:
                           result['HLTPSK'] = int(hlt_attrs['hltPrescaleKey'])
                           log.info("Got HLTPSK=%d from OKS", result['HLTPSK'])
      else:
         msg = f"Failed to fetch Partition from OKS: HTTP {response.status_code}"
         if strict:
            raise RuntimeError(msg + " (required for --online-environment)")
         log.warning(msg)
         
   except requests.exceptions.RequestException as e:
      msg = f"Error fetching trigger keys from OKS: {e}"
      if strict:
         raise RuntimeError(msg + " (required for --online-environment)")
      log.warning(msg)
   except (ValueError, KeyError, TypeError) as e:
      msg = f"Error parsing trigger keys from OKS: {e}"
      if strict:
         raise RuntimeError(msg + " (required for --online-environment)")
      log.warning(msg)
   
   # In strict mode, verify we got the required keys from OKS
   if strict:
      missing = [k for k in ['SMK', 'L1PSK', 'HLTPSK'] if result.get(k) is None]
      if missing:
         raise RuntimeError(f"Failed to get {', '.join(missing)} from OKS (required for --online-environment)")
   
   return result
 
 

load_from_database()

athenaEF.load_from_database	(		db_server,
			smk,
			l1psk = None,
			hltpsk = None,
			run_params = None,
			num_threads = 1,
			num_slots = 1,
			ef_overrides = None )

Load configuration from trigger database using the Super Master Key (SMK).

Returns a ConfigRunner that uses TrigConf::JobOptionsSvc with TYPE="DB"
to load configuration directly from the database, same as athenaHLT.

Definition at line 777 of file athenaEF.py.

                       num_threads=1, num_slots=1, ef_overrides=None):
   """
   Load configuration from trigger database using the Super Master Key (SMK).
   
   Returns a ConfigRunner that uses TrigConf::JobOptionsSvc with TYPE="DB"
   to load configuration directly from the database, same as athenaHLT.
   """
   log.info("Loading job options from database %s with SMK %d", db_server, smk)
   return ConfigRunner.from_database(db_server, smk, l1psk, hltpsk, run_params,
                                      num_threads=num_threads,
                                      num_slots=num_slots,
                                      ef_overrides=ef_overrides)
 
 

load_from_json()

athenaEF.load_from_json	(		json_file,
			run_params = None,
			num_threads = 1,
			num_slots = 1,
			ef_overrides = None )

Load configuration from a Gaudi joboptions JSON file.

Returns a ConfigRunner with a run() method that executes the configuration
using TrigConf::JobOptionsSvc with TYPE="FILE".

Definition at line 757 of file athenaEF.py.

def load_from_json(json_file, run_params=None, num_threads=1, num_slots=1, ef_overrides=None):
   """
   Load configuration from a Gaudi joboptions JSON file.
   
   Returns a ConfigRunner with a run() method that executes the configuration
   using TrigConf::JobOptionsSvc with TYPE="FILE".
   """
   with open(json_file, 'r') as f:
      jocat = json.load(f)
   
   if jocat.get('filetype') != 'joboptions':
      raise ValueError(f"Invalid JSON file type: {jocat.get('filetype')}, expected 'joboptions'")
   
   properties = jocat.get('properties', {})
   return ConfigRunner.from_json(json_file, run_params, properties,
                                  num_threads=num_threads,
                                  num_slots=num_slots,
                                  ef_overrides=ef_overrides)
 
 

main()

athenaEF.main

(

)

Definition at line 1083 of file athenaEF.py.

def main():
   parser = argparse.ArgumentParser(prog='athenaEF.py', formatter_class=
                                    lambda prog : argparse.ArgumentDefaultsHelpFormatter(prog, max_help_position=32, width=100),
                                    usage = '%(prog)s [OPTION]... -f FILE jobOptions',
                                    add_help=False)
   parser.expert_groups = []   # Keep list of expert option groups
 
   
   g = parser.add_argument_group('Options')
   g.add_argument('jobOptions', nargs='?', help='job options: CA module (package.module:function), pickle file (.pkl), or JSON file (.json)')
   g.add_argument('--threads', metavar='N', type=int, default=1, help='number of threads')
   g.add_argument('--concurrent-events', metavar='N', type=int, help='number of concurrent events if different from --threads')
   g.add_argument('--log-level', '-l', metavar='LVL', default='INFO', help='OutputLevel of athena')
   g.add_argument('--precommand', '-c', metavar='CMD', action='append', default=[],
                  help='Python commands executed before job options')
   g.add_argument('--postcommand', '-C', metavar='CMD', action='append', default=[],
                  help='Python commands executed after job options')
   g.add_argument('--interactive', '-i', action='store_true', help='interactive mode')
   g.add_argument('--help', '-h', nargs='?', choices=['all'], action=MyHelp, help='show help')
 
   g = parser.add_argument_group('Input/Output')
   g.add_argument('--file', '--filesInput', '-f', action='append', help='input RAW file')
   g.add_argument('--save-output', '-o', metavar='FILE', help='output file name')
   g.add_argument('--number-of-events', '--evtMax', '-n', metavar='N', type=int, default=None,
                  help='processes N events (default: from DB/config, -1 means all)')
   g.add_argument('--skip-events', '--skipEvents', '-k', metavar='N', type=int, default=None,
                  help='skip N first events')
   g.add_argument('--loop-files', action=argparse.BooleanOptionalAction, default=None,
                  help='loop over input files if no more events')
   g.add_argument('--efdf-interface-library', metavar='LIB', default=None,
                  help='name of the EFDF interface shared library to load (default: TrigDFEmulator)')
 
   
   g = parser.add_argument_group('Performance and debugging')
   g.add_argument('--perfmon', action='store_true', help='enable PerfMon')
   g.add_argument('--tcmalloc', action='store_true', default=True, help='use tcmalloc')
   g.add_argument('--stdcmalloc', action='store_true', help='use stdcmalloc')
   g.add_argument('--stdcmath', action='store_true', help='use stdcmath library')
   g.add_argument('--imf', action='store_true', default=True, help='use Intel math library')
   g.add_argument('--show-includes', '-s', action='store_true', help='show printout of included files')
 
   
   g = parser.add_argument_group('Conditions')
   g.add_argument('--run-number', '-R', metavar='RUN', type=int,
                  help='run number (if None, read from first event)')
   g.add_argument('--lb-number', '-L', metavar='LBN', type=int,
                  help='lumiblock number (if None, read from first event)')
   g.add_argument('--conditions-run', metavar='RUN', type=int, default=None,
                  help='reference run number for conditions lookup (use when IS run number has no COOL data)')
   g.add_argument('--sor-time', type=arg_sor_time,
                  help='The Start Of Run time. Three formats are accepted: '
                  '1) the string "now", for current time; '
                  '2) the number of nanoseconds since epoch (e.g. 1386355338658000000 or int(time.time() * 1e9)); '
                  '3) human-readable "20/11/18 17:40:42.3043". If not specified the sor-time is read from the conditions DB')
   g.add_argument('--detector-mask', metavar='MASK', type=arg_detector_mask,
                  help='detector mask (if None, read from the conditions DB), use string "all" to enable all detectors')
 
   
   g = parser.add_argument_group('Database')
   g.add_argument('--use-database', '-b', action='store_true',
                  help='configure from trigger database using SMK')
   g.add_argument('--db-server', metavar='DB', default='TRIGGERDB_RUN3', help='DB server name (alias)')
   g.add_argument('--smk', type=int, default=None, help='Super Master Key')
   g.add_argument('--l1psk', type=int, default=None, help='L1 prescale key')
   g.add_argument('--hltpsk', type=int, default=None, help='HLT prescale key')
   g.add_argument('--use-crest', action='store_true', default=False,
                  help='Use CREST for trigger configuration')
   g.add_argument('--crest-server', metavar='URL', default=None,
                  help='CREST server URL (defaults to flags.Trigger.crestServer)')
   g.add_argument('--dump-config', action='store_true', help='Dump joboptions JSON file')
   g.add_argument('--dump-config-exit', action='store_true', help='Dump joboptions JSON file and exit')
 
   
   g = parser.add_argument_group('Magnets')
   g.add_argument('--solenoid-current', type=float, default=None,
                  help='Solenoid current in Amperes (default: nominal current for offline running, required from IS online)')
   g.add_argument('--toroids-current', type=float, default=None,
                  help='Toroids current in Amperes (default: nominal current for offline running, required from IS online)')
 
   
   g = parser.add_argument_group('Online')
   g.add_argument('--online-environment', action='store_true',
                  help='Enable online environment: read run parameters from IS and trigger '
                       'configuration keys (SMK, L1PSK, HLTPSK) from OKS via WEBDAQ REST API')
   g.add_argument('--partition', metavar='NAME', default=None,
                  help='TDAQ partition name (defaults to TDAQ_PARTITION environment variable)')
   g.add_argument('--webdaq-base', metavar='URL', default=None,
                  help='WEBDAQ base URL (defaults to TDAQ_WEBDAQ_BASE environment variable)')
 
   
   g = parser.add_argument_group('Online Histogramming')
   g.add_argument('--oh-monitoring', '-M', action='store_true', default=False,
                  help='enable online histogram publishing via WebdaqHistSvc')
 
   
   g = parser.add_argument_group('Expert')
   parser.expert_groups.append(g)
   (args, unparsed_args) = parser.parse_known_args()
   check_args(parser, args)
 
   # set ROOT to batch mode (ATR-21890)
   from PyUtils.Helpers import ROOTSetup
   ROOTSetup(batch=True)
 
   # Enable ROOT thread safety
   import ROOT
   ROOT.ROOT.EnableThreadSafety()
 
   # set default Python OutputLevel and file inclusion
   import AthenaCommon.Logging
   AthenaCommon.Logging.log.setLevel(getattr(logging, args.log_level))
   AthenaCommon.Logging.log.setFormat("%(asctime)s  Py:%(name)-31s %(levelname)7s %(message)s")
   if args.show_includes:
      from AthenaCommon.Include import include
      include.setShowIncludes( True )
 
   # consistency checks for arguments
   if not args.concurrent_events:
      args.concurrent_events = args.threads
 
   # Update args and set athena flags
   from TrigPSC import PscConfig
   from TrigPSC.PscDefaultFlags import defaultOnlineFlags
   
   # Get flags with online defaults (same as athenaHLT)
   flags = defaultOnlineFlags()
 
   # set MessageSvc OutputLevel
   from AthenaCommon import Constants
   flags.Exec.OutputLevel = getattr(Constants, args.log_level)
 
   # Enable WebdaqHistSvc for online histogram publishing if requested
   if args.oh_monitoring:
      flags.Trigger.Online.useOnlineWebdaqHistSvc = True
      log.info("Enabled WebdaqHistSvc for online histogram publishing")
 
   # CREST configuration (same as athenaHLT)
   log.info("Using CREST for trigger configuration: %s", args.use_crest)
   if args.use_crest:
      flags.Trigger.useCrest = True
      if args.crest_server:
         flags.Trigger.crestServer = args.crest_server
      else:
         args.crest_server = flags.Trigger.crestServer
 
   update_run_params(args, flags)
 
   if args.use_database:
      # If HLTPSK was given on the command line OR from OKS (--online-environment),
      # we ignore what is stored in COOL and use the specified key directly from the DB.
      # This is needed because COOL may point to a different HLTPSK for the forced run number.
      PscConfig.forcePSK = (args.hltpsk is not None) or args.online_environment
      # Read trigger config keys from COOL/OKS if not specified
      update_trigconf_keys(args, flags)
 
   # Fill flags from command line (if not running from DB/JSON)
   if not args.use_database and args.jobOptions and not args.jobOptions.endswith('.json'):
      PscConfig.unparsedArguments = unparsed_args
      for flag_arg in unparsed_args:
         flags.fillFromString(flag_arg)
 
   PscConfig.interactive = args.interactive
   PscConfig.exitAfterDump = args.dump_config_exit
 
   # NOTE: Do NOT set flags.Input.Files here!
   # Like athenaHLT, we keep Input.Files=[] during configuration to ensure the
   # configuration is portable and doesn't depend on specific input file metadata.
   # Input files are passed to EFInterface for runtime use only.
 
   # Set conditions run number override (for test partitions with fake run numbers)
   if args.conditions_run is not None:
      log.info("Using conditions from reference run %d (overriding run %s for IOV lookup)",
               args.conditions_run, args.run_number)
      flags.Input.ConditionsRunNumber = args.conditions_run
 
   # Set number of events
   if args.number_of_events is not None and args.number_of_events > 0:
      flags.Exec.MaxEvents = args.number_of_events
 
   # Set skip events
   if args.skip_events is not None and args.skip_events > 0:
      flags.Exec.SkipEvents = args.skip_events
 
   # NOTE: Do NOT set flags.Concurrency.NumThreads or NumConcurrentEvents here.
   # Threading is set at runtime via iProperty after configure() - see ConfigRunner.run()
 
   # Enable PerfMon if requested
   flags.PerfMon.doFastMonMT = args.perfmon
 
   # Configure EF ByteStream services (mandatory to run without HLTMPPU)
   # This provides the data flow interface that would normally come from HLTMPPU
   flags.Trigger.Online.useEFByteStreamSvc = True
   # EFInterfaceSvc settings from the command line. 
   # Only options explicitly given are collected, anything else keeps the value from the DB/jobOptions configuration
   ef_files = args.file if args.file else []
   ef_overrides = {}
   if ef_files:
      ef_overrides['Files'] = ef_files
      # Metadata read from the input file - always more accurate than DB values
      ef_overrides.update({
         'T0ProjectTag' : args.T0_project_tag,
         'BeamType'     : args.beam_type,
         'BeamEnergy'   : args.beam_energy,
         'TriggerType'  : args.trigger_type,
         'Stream'       : args.stream,
         'Lumiblock'    : args.lumiblock,
         'DetMask'      : args.file_detector_mask,
      })
   if args.run_number is not None:          # from -R, IS, or the input file
      ef_overrides['RunNumber'] = args.run_number
   if args.save_output is not None:
      ef_overrides['OutputFileName'] = args.save_output
   if args.loop_files is not None:
      ef_overrides['LoopOverFiles'] = args.loop_files
   if args.number_of_events is not None:
      ef_overrides['NumEvents'] = args.number_of_events
   if args.skip_events is not None:
      ef_overrides['SkipEvents'] = args.skip_events
   if args.efdf_interface_library is not None:
      ef_overrides['EFDFInterfaceLibraryName'] = args.efdf_interface_library
 
   # Apply to the flags for the CA-module path (getEFInterfaceSvc reads these)
   _prop2flag = {'Files': 'Files', 'OutputFileName': 'OutputFileName',
                 'LoopOverFiles': 'LoopFiles', 'NumEvents': 'NumEvents',
                 'SkipEvents': 'SkipEvents', 'RunNumber': 'RunNumber',
                 'T0ProjectTag': 'T0ProjectTag', 'BeamType': 'BeamType',
                 'BeamEnergy': 'BeamEnergy', 'TriggerType': 'TriggerType',
                 'Stream': 'Stream', 'Lumiblock': 'Lumiblock', 'DetMask': 'DetMask',
                 'EFDFInterfaceLibraryName': 'LibraryName'}
   for prop, value in ef_overrides.items():
      setattr(flags.Trigger.Online.EFInterface, _prop2flag[prop], value)
 
   # Execute precommands
   if args.precommand:
      log.info("Executing precommand(s)")
      for cmd in args.precommand:
         log.info("  %s", cmd)
         exec(cmd, globals(), {'flags': flags})
 
   # Determine input type
   is_database = args.use_database
   is_pickle = False
   is_json = False
   
   if not is_database and args.jobOptions:
      jobOptions = args.jobOptions
      is_pickle = jobOptions.endswith('.pkl')
      is_json = jobOptions.endswith('.json')
 
   if is_database:
      # Load configuration from trigger database
      # Handle CREST vs standard DB access (same as athenaHLT)
      if args.use_crest:
         crestconn = TriggerCrestUtil.getCrestConnection(args.db_server)
         db_alias = f"{args.crest_server}/{crestconn}"
         log.info("Loading configuration via CREST from %s with SMK %d", db_alias, args.smk)
      else:
         db_alias = args.db_server
         log.info("Loading configuration from database %s with SMK %d", db_alias, args.smk)
      
      # Get run parameters for prepareForStart
      run_params = get_run_params(args).to_dict()
      acc = load_from_database(db_alias, args.smk, args.l1psk, args.hltpsk, run_params,
                               num_threads=args.threads, num_slots=args.concurrent_events,
                               ef_overrides=ef_overrides)
      log.info("Configuration loaded from database")
 
   elif is_pickle:
      # Load ComponentAccumulator from pickle file
      log.info("Loading configuration from pickle file: %s", jobOptions)
      with open(jobOptions, 'rb') as f:
         acc = pickle.load(f)
      log.info("Configuration loaded from pickle")
 
   elif is_json:
      # Load configuration from JSON file
      log.info("Loading configuration from JSON file: %s", jobOptions)
      # Get run parameters for prepareForStart
      run_params = get_run_params(args).to_dict()
      acc = load_from_json(jobOptions, run_params,
                           num_threads=args.threads, num_slots=args.concurrent_events,
                           ef_overrides=ef_overrides)
      log.info("Configuration loaded from JSON")
 
   else:
      # Load from CA module - follow the same pattern as athenaHLT/TrigPSCPythonCASetup:
      # 1. Build the full configuration with services
      # 2. Dump to JSON file
      # 3. Use AthHLT.reload_from_json to re-exec and reload from JSON
      # This preserves the ability to use the same JobOptionsSvc as athenaHLT
      log.info("Loading CA configuration from: %s", jobOptions)
      
      # Clone and lock flags for services configuration (as done in TrigPSCPythonCASetup)
      from AthenaConfiguration.ComponentAccumulator import ComponentAccumulator
      from AthenaConfiguration.MainServicesConfig import addMainSequences
      from TrigServices.TriggerUnixStandardSetup import commonServicesCfg
      from AthenaConfiguration.ComponentFactory import CompFactory
      
      locked_flags = flags.clone()
      locked_flags.lock()
      
      # Create base CA with framework services (like TrigPSCPythonCASetup)
      cfg = ComponentAccumulator(CompFactory.AthSequencer("AthMasterSeq", Sequential=True))
      cfg.setAppProperty('ExtSvcCreates', False)
      cfg.setAppProperty("MessageSvcType", "TrigMessageSvc")
      cfg.setAppProperty("JobOptionsSvcType", "TrigConf::JobOptionsSvc")
      
      # Add main sequences and common services (includes TrigServicesCfg)
      addMainSequences(locked_flags, cfg)
      cfg.merge(commonServicesCfg(locked_flags))
      
      # Now merge user CA config (with unlocked flags, as in TrigPSCPythonCASetup)
      cfg_func = AthHLT.getCACfg(jobOptions)
      cfg.merge(cfg_func(flags))
      
      # Execute postcommands before dumping (like TrigPSCPythonCASetup)
      if args.postcommand:
         log.info("Executing postcommand(s)")
         for cmd in args.postcommand:
            log.info("  %s", cmd)
            exec(cmd, globals(), {'flags': flags, 'cfg': cfg})
         args.postcommand = []  # Clear so we don't run them again later
      
      # Dump configuration to JSON (like TrigPSCPythonCASetup)
      fname = "HLTJobOptions"
      log.info("Dumping configuration to %s.pkl and %s.json", fname, fname)
      with open(f"{fname}.pkl", "wb") as f:
         cfg.store(f)
      
      from TrigConfIO.JsonUtils import create_joboptions_json
      create_joboptions_json(f"{fname}.pkl", f"{fname}.json")
      
      # Check for dump-and-exit
      if args.dump_config_exit:
         log.info("Configuration dumped to %s.json. Exiting...", fname)
         sys.exit(0)
 
      # Re-exec from the JSON (same as athenaHLT TrigPSCPythonCASetup -> AthHLT.reload_from_json -> os.execvp). 
      # This replaces the process image freeing up the configuration heap 
      log.info("Configuration dumped to %s.json. Re-exec...", fname)
      AthHLT.reload_from_json(f"{fname}.json", suppress_args=PscConfig.unparsedArguments + ['--dump-config'], jobOptions=args.jobOptions)
      
   # Execute postcommands
   if args.postcommand:
      log.info("Executing postcommand(s)")
      for cmd in args.postcommand:
         log.info("  %s", cmd)
         exec(cmd, globals(), {'flags': flags, 'acc': acc})
 
   # Dump configuration if requested
   if args.dump_config or args.dump_config_exit:
      fname = "HLTJobOptions"
      
      if is_database:
         # For DB mode, fetch properties via Python API
         from TrigConfIO.HLTTriggerConfigAccess import HLTJobOptionsAccess
         log.info("Fetching configuration from database for dump...")
         jo_access = HLTJobOptionsAccess(dbalias=acc.db_server, smkey=acc.smk)
         props = jo_access.algorithms()
         
         log.info("Dumping configuration to %s.json", fname)
         hlt_json = {'filetype': 'joboptions', 'properties': props}
         with open(f"{fname}.json", "w") as f:
            json.dump(hlt_json, f, indent=4, sort_keys=True, ensure_ascii=True)
            
      elif is_json:
         # For JSON mode, properties were already loaded
         props = acc.properties
         if props:
            log.info("Dumping configuration to %s.json", fname)
            hlt_json = {'filetype': 'joboptions', 'properties': props}
            with open(f"{fname}.json", "w") as f:
               json.dump(hlt_json, f, indent=4, sort_keys=True, ensure_ascii=True)
         else:
            log.warning("No properties available to dump")
            
      elif is_pickle:
         # For pickle-loaded ComponentAccumulator, gather properties
         app_props, msg_props, comp_props = acc.gatherProps()
         props = {"ApplicationMgr": app_props, "MessageSvc": msg_props}
         for comp, name, value in comp_props:
            props.setdefault(comp, {})[name] = value
         
         log.info("Dumping configuration to %s.json", fname)
         hlt_json = {'filetype': 'joboptions', 'properties': props}
         with open(f"{fname}.json", "w") as f:
            json.dump(hlt_json, f, indent=4, sort_keys=True, ensure_ascii=True)
      
      # Note: For CA module, dumping is already handled earlier 
      # before converting to ConfigRunner
      
      if args.dump_config_exit:
         log.info("Configuration dumped. Exiting...")
         sys.exit(0)
 
   # Run the application directly (like athena.py does)
   log.info("Starting Athena execution...")
   
   # Create worker directory structure that HLT services expect
   # (normally created by HLTMPPU/PSC). Worker ID 1 means single-worker, non-forked mode
   # and must match what we pass to hltUpdateAfterFork(worker_id=1) in ConfigRunner.run()
   worker_dir = os.path.join(os.getcwd(), "athenaHLT_workers", "athenaHLT-01")
   if not os.path.exists(worker_dir):
      log.info("Creating worker directory: %s", worker_dir)
      os.makedirs(worker_dir, exist_ok=True)
 
   # Start the private TDAQ infrastructure for -M
   oh_infra = start_oh_infrastructure(args) if args.oh_monitoring else None
   
   if args.interactive:
      log.info("Interactive mode - call acc.run() to execute")
      import code
      code.interact(local={'acc': acc, 'flags': flags})
   else:
      # Run the application
      from AthenaCommon import ExitCodes
      exitcode = 0
      try:
         # Pass maxEvents if explicitly set (including -1 for all events)
         sc = acc.run(args.number_of_events)
         if sc.isFailure():
            exitcode = ExitCodes.EXE_ALG_FAILURE
      except SystemExit as e:
         exitcode = ExitCodes.EXE_ALG_FAILURE if e.code == 1 else e.code
      except Exception:
         traceback.print_exc()
         exitcode = ExitCodes.UNKNOWN_EXCEPTION
      finally:
         stop_oh_infrastructure(oh_infra)
 
      log.info('Leaving with code %d: "%s"', exitcode, ExitCodes.what(exitcode))
      sys.exit(exitcode)
 
 

start_oh_infrastructure()

athenaEF.start_oh_infrastructure

(

args

)

Start a private TDAQ infrastructure (offline test of OH publication).

Definition at line 992 of file athenaEF.py.

def start_oh_infrastructure(args):
   """Start a private TDAQ infrastructure (offline test of OH publication)."""
   import shutil, socket, signal, subprocess, time
 
   infra_script = shutil.which('athenaEF_tdaq_infra.py')
   if infra_script is None:
      log.error("athenaEF_tdaq_infra.py not found on PATH (required for -M)")
      sys.exit(1)
 
   partition = args.partition or 'athenaEF'
   host = 'localhost'
   # Get a free port for webis
   s = socket.socket()
   s.bind((host, 0))
   port = s.getsockname()[1]
   s.close()
   oh_server = 'Histogramming'   # WebdaqHistSvc.OHServerName default
   run_number = args.run_number if args.run_number is not None else 0
 
   # Export variables for the -M case 
   os.environ['TDAQ_PARTITION']   = partition
   os.environ['TDAQ_WEBDAQ_BASE'] = f'http://{host}:{port}'
   os.environ['TDAQ_OH_SERVER']   = oh_server
 
   log.info("Starting private OH infrastructure: partition=%s, webdaq=%s, oh_server=%s",
            partition, os.environ['TDAQ_WEBDAQ_BASE'], oh_server)
 
   logfile = open('athenaEF_oh_infra.log', 'w')
 
   # PR_SET_PDEATHSIG so the infrastructure is torn down (SIGTERM -> oh_cp +
   # ipc_rm) if athenaEF dies unexpectedly, e.g. segfaults mid-run.
   from ctypes import cdll
   PR_SET_PDEATHSIG = 1
   def _pdeathsig():
      cdll['libc.so.6'].prctl(PR_SET_PDEATHSIG, signal.SIGTERM)
 
   proc = subprocess.Popen(
      [infra_script,
       '--partition',   partition,
       '--webdaq-port', str(port),
       '--oh-server',   oh_server,
       '--run-number',  str(run_number)],
      stdout=logfile, stderr=subprocess.STDOUT,
      preexec_fn=_pdeathsig, close_fds=True)
 
   # Wait for the readiness marker (or early failure / timeout)
   timeout = 120
   deadline = time.time() + timeout
   while time.time() < deadline:
      if proc.poll() is not None:
         log.error("OH infrastructure exited early (code %s); see %s", proc.returncode, logfile.name)
         sys.exit(1)
      with open(logfile.name) as f:
         if 'ATHENAEF_INFRA_READY' in f.read():
            log.info("OH infrastructure is ready")
            return proc
      time.sleep(1)
 
   log.error("OH infrastructure did not become ready within %d s; see %s", timeout, logfile.name)
   proc.terminate()
   sys.exit(1)
 
 

stop_oh_infrastructure()

athenaEF.stop_oh_infrastructure

(

proc

)

Terminate the private TDAQ infrastructure (SIGTERM triggers oh_cp + ipc_rm).

Definition at line 1055 of file athenaEF.py.

def stop_oh_infrastructure(proc):
   """Terminate the private TDAQ infrastructure (SIGTERM triggers oh_cp + ipc_rm)."""
   if proc is None or proc.poll() is not None:
      return
   import signal
   log.info("Stopping OH infrastructure")
   proc.send_signal(signal.SIGTERM)
   try:
      proc.wait(timeout=60)
   except Exception:
      proc.kill()
 
 

update_run_params()

athenaEF.update_run_params	(		args,
			flags )

Update run parameters from IS, file, or conditions DB

Definition at line 828 of file athenaEF.py.

def update_run_params(args, flags):
   """Update run parameters from IS, file, or conditions DB"""
 
   # If --online-environment is specified, try to read from Information Service first
   if getattr(args, 'online_environment', False):
      log.info("Reading run parameters from Information Service via WEBDAQ")
      # Pass command-line magnet values as overrides (if provided)
      # strict=True ensures we fail if IS read fails, rather than falling back to defaults
      # But if user provided magnet values on command line, those take precedence over IS
      solenoid_override = getattr(args, 'solenoid_current', None)
      toroids_override = getattr(args, 'toroids_current', None)
      
      run_params = get_run_params(from_is=True, 
                                  partition=getattr(args, 'partition', None),
                                  webdaq_base=getattr(args, 'webdaq_base', None),
                                  strict=True,
                                  solenoid_current_override=solenoid_override,
                                  toroids_current_override=toroids_override)
      # Update args with values from IS (if not already set on command line)
      if args.run_number is None and run_params.run_number is not None:
         args.run_number = run_params.run_number
         log.info("Using run_number=%d from IS", args.run_number)
      if args.lb_number is None and run_params.lb_number is not None:
         args.lb_number = run_params.lb_number
         log.info("Using lb_number=%d from IS", args.lb_number)
      if args.sor_time is None and run_params.sor_time is not None:
         args.sor_time = run_params.sor_time
         log.info("Using sor_time=%s from IS", args.sor_time)
      if args.detector_mask is None and run_params.detector_mask is not None:
         args.detector_mask = run_params.detector_mask
         log.info("Using detector_mask=%s from IS", args.detector_mask)
      # Update magnet currents from IS (run_params already has command-line overrides if provided)
      args.solenoid_current = run_params.solenoid_current
      args.toroids_current = run_params.toroids_current
      args.beam_type = run_params.beam_type
      args.beam_energy = run_params.beam_energy
 
   if (args.run_number is not None and args.lb_number is None) or (args.run_number is None and args.lb_number is not None):
      log.error("Both or neither of the options -R (--run-number) and -L (--lb-number) have to be specified")
 
   # Read metadata from input file (like HLTMPPy/runner.py getRunParamsFromFile)
   if args.file:
      from eformat import EventStorage
      dr = EventStorage.pickDataReader(args.file[0])
      if args.run_number is None:
         args.run_number = dr.runNumber()
         args.lb_number = dr.lumiblockNumber()
      args.T0_project_tag = dr.projectTag()
      args.beam_type = dr.beamType()
      args.beam_energy = dr.beamEnergy()
      args.trigger_type = dr.triggerType()
      args.stream = dr.stream()
      args.lumiblock = dr.lumiblockNumber()
      args.file_detector_mask = "{:032x}".format(dr.detectorMask())
   else:
      args.T0_project_tag = getattr(args, 'T0_project_tag', '')
      args.beam_type = getattr(args, 'beam_type', 0)
      args.beam_energy = getattr(args, 'beam_energy', 0)
      args.trigger_type = getattr(args, 'trigger_type', 0)
      args.stream = getattr(args, 'stream', '')
      args.lumiblock = getattr(args, 'lumiblock', 0)
      args.file_detector_mask = getattr(args, 'file_detector_mask', '00000000000000000000000000000000')
 
   sor_params = None
   if (args.sor_time is None or args.detector_mask is None) and args.run_number is not None:
      sor_params = AthHLT.get_sor_params(args.run_number)
      log.debug('SOR parameters: %s', sor_params)
      if sor_params is None:
         log.error("Run %d does not exist. If you want to use this run-number specify "
                   "remaining run parameters, e.g.: --sor-time=now --detector-mask=all", args.run_number)
         sys.exit(1)
 
   if args.sor_time is None and sor_params is not None:
      args.sor_time = arg_sor_time(str(sor_params['SORTime']))
 
   if args.detector_mask is None and sor_params is not None:
      dmask = sor_params['DetectorMask']
      if args.run_number < AthHLT.CondDB._run2:
         dmask = hex(dmask)
      args.detector_mask = arg_detector_mask(dmask)
   
   if args.dump_config_exit and not args.run_number:
      args.run_number = 0
 
   # Apply defaults for magnet currents if not set (offline mode only)
   # In online mode, magnets must come from IS or command line (handled above)
   if getattr(args, 'solenoid_current', None) is None:
      args.solenoid_current = RunParams.DEFAULT_SOLENOID_CURRENT
      log.debug("Using default solenoid_current=%.1f", args.solenoid_current)
   if getattr(args, 'toroids_current', None) is None:
      args.toroids_current = RunParams.DEFAULT_TOROIDS_CURRENT
      log.debug("Using default toroids_current=%.1f", args.toroids_current)
 
 

update_trigconf_keys()

athenaEF.update_trigconf_keys	(		args,
			flags )

Update trigger configuration keys from OKS, COOL, or CREST.

Priority order:
1. Command-line arguments (always take precedence)
2. OKS via WEBDAQ (if --online-environment is set)
3. CREST (if --use-crest is set)
4. COOL (default)

Definition at line 922 of file athenaEF.py.

def update_trigconf_keys(args, flags):
   """Update trigger configuration keys from OKS, COOL, or CREST.
   
   Priority order:
   1. Command-line arguments (always take precedence)
   2. OKS via WEBDAQ (if --online-environment is set)
   3. CREST (if --use-crest is set)
   4. COOL (default)
   """
 
   if args.smk is None or args.l1psk is None or args.hltpsk is None:
      trigconf = None
      
      # Try OKS first if --online-environment is set
      if getattr(args, 'online_environment', False):
         log.info("Reading trigger configuration keys from OKS (online environment)")
         # strict=True ensures we fail if OKS read fails, rather than falling back to COOL
         oks_keys = get_trigconf_keys_from_oks(
            partition=getattr(args, 'partition', None),
            webdaq_base=getattr(args, 'webdaq_base', None),
            strict=True
         )
         log.info("Retrieved trigger keys from OKS: %s", oks_keys)
         
         # With strict=True, we're guaranteed to have all keys or an exception was raised
         trigconf = {
            'SMK': oks_keys.get('SMK'),
            'LVL1PSK': oks_keys.get('L1PSK'),
            'HLTPSK': oks_keys.get('HLTPSK')
         }
         # Also update db_server if provided by OKS and not set on command line
         if oks_keys.get('db_alias') and args.db_server == 'TRIGGERDB_RUN3':
            args.db_server = oks_keys['db_alias']
            log.info("Using db_server=%s from OKS", args.db_server)
      
      # Fall back to CREST or COOL only if NOT in online-environment mode
      if trigconf is None:
         if args.use_crest:
            crest_server = args.crest_server or flags.Trigger.crestServer
            log.info("Reading trigger configuration keys from CREST for run %s", args.run_number)
            trigconf = AthHLT.get_trigconf_keys_crest(args.run_number, args.lb_number, crest_server)
            log.info("Retrieved trigger keys from CREST: %s", trigconf)
         else:
            log.info("Reading trigger configuration keys from COOL for run %s", args.run_number)
            trigconf = AthHLT.get_trigconf_keys(args.run_number, args.lb_number)
            log.info("Retrieved trigger keys from COOL: %s", trigconf)
 
      try:
         if args.smk is None:
            args.smk = trigconf['SMK']
            log.debug("Using SMK=%d from conditions DB/OKS", args.smk)
         else:
            log.debug("Using SMK=%d from command line (ignoring DB/OKS value %s)", args.smk, trigconf.get('SMK'))
         if args.l1psk is None:
            args.l1psk = trigconf['LVL1PSK']
            log.debug("Using L1PSK=%d from conditions DB/OKS", args.l1psk)
         else:
            log.debug("Using L1PSK=%d from command line (ignoring DB/OKS value %s)", args.l1psk, trigconf.get('LVL1PSK'))
         if args.hltpsk is None:
            args.hltpsk = trigconf['HLTPSK']
            log.debug("Using HLTPSK=%d from conditions DB/OKS", args.hltpsk)
         else:
            log.debug("Using HLTPSK=%d from command line (ignoring DB/OKS value %s)", args.hltpsk, trigconf.get('HLTPSK'))
      except KeyError:
         log.error("Cannot read trigger configuration keys from the conditions database for run %d", args.run_number)
         sys.exit(1)
   else:
      log.info("Using trigger configuration keys from command line: SMK=%d, L1PSK=%d, HLTPSK=%d",
               args.smk, args.l1psk, args.hltpsk)
 

Variable Documentation

log

athenaEF.log = logging.getLogger('athenaEF')

Definition at line 54 of file athenaEF.py.