EventDisplaysOnlineHelpers Namespace Reference

Functions
def	GetBFields ()
def	WaitForPartition (partitionName=None)
def	EventCanBeSeenByPublic (projectTags)
def	GetUniqueJobID ()
def	GetRunNumber (partitionName)

Variables
def	runNumber = GetRunNumber()
def	bFields = GetBFields()
def	isPublicEvent = EventCanBeSeenByPublic(['data24_13p6TeV'])

Function Documentation

EventCanBeSeenByPublic()

def EventDisplaysOnlineHelpers.EventCanBeSeenByPublic

(

projectTags

)

Definition at line 60 of file EventDisplaysOnlineHelpers.py.

def EventCanBeSeenByPublic(projectTags):
# Is the data allowed to be seen by the general public on atlas live and in the CCC
  mlog = logging.getLogger( 'EventDisplays' )
  try:
    partition = IPCPartition('ATLAS')
    RunParams = ISObject(partition, 'RunParams.RunParams', 'RunParams')
    RunParams.checkout()
 
    ready4physics = ISInfoAny()
    ISInfoDictionary(partition).getValue('RunParams.Ready4Physics', ready4physics)
    print("physicsReady: %s " % ready4physics.get())
 
    physicsReady = ISObject(partition, 'RunParams.Ready4Physics','Ready4PhysicsInfo')
    physicsReady.checkout()
    print("Ready for physics: %r" % (physicsReady.ready4physics))
    print("RunParams.T0_project_tag", RunParams.T0_project_tag)
 
    sendToPublicStream = False
    if physicsReady.ready4physics and RunParams.T0_project_tag in projectTags:
      sendToPublicStream = True
 
    return sendToPublicStream
 
  except Exception:
    mlog.warning('Failed to get bool for EventCanBeSeenByPublic, this is expected for offline test and GM test paritions')
    return False
 

GetBFields()

def EventDisplaysOnlineHelpers.GetBFields

(

)

Definition at line 7 of file EventDisplaysOnlineHelpers.py.

def GetBFields():
  mlog = logging.getLogger( 'EventDisplays' )
 
  #BFields are read from initial partition
  partition='initial'
  mlog.debug("Trying to read magnetic field configuration from partition %s", partition)
 
  #now try and read the information from IS
  try :
    #Get hold of the initial partition
    ipcPart = IPCPartition(partition)
    if not ipcPart.isValid():
      raise UserWarning("Partition %s invalid - cannot access magnetic field setting"%partition)
    #Get the current and valid status
    toroidCurrent = ISObject(ipcPart,'DCS_GENERAL.MagnetToroidsCurrent.value','DdcFloatInfo')
    solenoidCurrent = ISObject(ipcPart,'DCS_GENERAL.MagnetSolenoidCurrent.value','DdcFloatInfo')
    toroidInvalid = ISObject(ipcPart,'DCS_GENERAL.MagnetToroidsCurrent.invalid','DdcIntInfo')
    solenoidInvalid = ISObject(ipcPart,'DCS_GENERAL.MagnetSolenoidCurrent.invalid','DdcIntInfo')
    toroidCurrent.checkout()
    solenoidCurrent.checkout()
    toroidInvalid.checkout()
    solenoidInvalid.checkout()
    #And calculate the flags
    solenoidOn=((solenoidCurrent.value > 1000.) and (solenoidInvalid.value == 0))
    toroidOn=((toroidCurrent.value > 1000.) and (toroidInvalid.value == 0))
  except UserWarning as err:
    mlog.error(err)
    #Should always be able to access initial parititon
    mlog.fatal("Failed to read magnetic field configuration from IS, aborting")
    sys.exit(1)
 
  #print the result
  mlog.info("Magnetic field in solenoid is %s", (solenoidOn and "ON") or "OFF")
  mlog.info("Magnetic field in toroid is %s", (toroidOn and "ON") or "OFF")
 
  #finally return our values
  return (solenoidOn,toroidOn)
 
 

GetRunNumber()

def EventDisplaysOnlineHelpers.GetRunNumber

(

partitionName

)

Definition at line 97 of file EventDisplaysOnlineHelpers.py.

def GetRunNumber(partitionName):
  part = IPCPartition(partitionName)
  RunParams = ISObject(part, 'RunParams.RunParams', 'RunParams')
  RunParams.checkout()
  run_number = RunParams.getAttributeValue('run_number')
  return run_number
 

GetUniqueJobID()

def EventDisplaysOnlineHelpers.GetUniqueJobID

(

)

Definition at line 87 of file EventDisplaysOnlineHelpers.py.

def GetUniqueJobID():
  # Setup unique output files (so that multiple Athena jobs on the same machine don't interfere)
  import os
  jobId = os.environ.get('TDAQ_APPLICATION_NAME', '').split(':')
  if not len(jobId) == 5:
    from random import randint
    jobId = ['Athena-EventProcessor', 'Athena-EventDisplays-Segment', 'EventDisplays-Rack', 'tmp', '%d' % randint(0, 999)]
 
  return jobId
 

WaitForPartition()

def EventDisplaysOnlineHelpers.WaitForPartition

(

partitionName = None

)

Definition at line 46 of file EventDisplaysOnlineHelpers.py.

def WaitForPartition(partitionName=None):
  mlog = logging.getLogger( 'EventDisplays' )
  partitionUp = False
  while not partitionUp:
    try:
      partition = IPCPartition(partitionName)
      runParams = ISObject(partition, "RunParams.RunParams", "RunParams")
      runParams.checkout()
      partitionUp = True
 
    except Exception:
      mlog.info("%s partition is not up, sleeping for 30 seconds", partitionName)
      time.sleep(30)
 

Variable Documentation

bFields

def EventDisplaysOnlineHelpers.bFields = GetBFields()

Definition at line 107 of file EventDisplaysOnlineHelpers.py.

isPublicEvent

def EventDisplaysOnlineHelpers.isPublicEvent = EventCanBeSeenByPublic(['data24_13p6TeV'])

Definition at line 109 of file EventDisplaysOnlineHelpers.py.

runNumber

def EventDisplaysOnlineHelpers.runNumber = GetRunNumber()

Definition at line 105 of file EventDisplaysOnlineHelpers.py.