PyAthenaComps.py

Go to the documentation of this file.

# Copyright (C) 2002-2026 CERN for the benefit of the ATLAS collaboration
 
# @file: PyAthenaComps.py
# @purpose: a set of Python classes for PyAthena
# @author: Sebastien Binet <binet@cern.ch>
 
__doc__     = """Module containing a set of Python base classes for PyAthena"""
__author__  = "Sebastien Binet <binet@cern.ch>"
 

__all__ = [ 'StatusCode',
            'Alg',
            'Svc',
            'AlgTool',
            'Aud',
            'algs',
            'services',
            ]
 

from AthenaCommon.Configurable  import *  # noqa: F401, F403
from AthenaPython.Configurables import (CfgPyAlgorithm,
                                        CfgPyService,
                                        CfgPyAlgTool,
                                        CfgPyAud)
 

import weakref
class WeakRefHolder(object):
    """A convenient class to access py-components by name
    """
    pass
def __wk_setattr__(self, k, v):
    self.__setattr__(k, weakref.ref(v))
 
services = WeakRefHolder()
services.__doc__ = """The list of PyAthena::Svc which have been instantiated"""
services.__setattr__ = __wk_setattr__
 
algs = WeakRefHolder()
algs.__doc__ = """The list of PyAthena::Alg which have been instantiated"""
algs.__setattr__ = __wk_setattr__
 
del __wk_setattr__
 

def _get_prop_value(pycomp, propname):
    if propname in pycomp.properties() and  pycomp.properties()[propname] != pycomp.propertyNoValue:
        return pycomp.properties()[propname]
    else:
        return pycomp.getDefaultProperty(propname)
 
 

class StatusCode:
    Recoverable = 2
    Success     = 1
    Failure     = 0
 

class Alg( CfgPyAlgorithm ):
    """
    Base class from which all concrete algorithm classes should
    be derived.
 
    In order for a concrete algorithm class to do anything
    useful the methods initialize(), execute() and finalize()
    should be overridden.
    """
    def __init__(self, name = None, **kw):
        if name is None: name = kw.get('name', self.__class__.__name__)
        kw.setdefault('OutputLevel', 3) #INFO
        
        super(Alg, self).__init__(name, **kw)
        self._pyath_evtstore = None # handle to the evt store
        self._pyath_detstore = None # handle to the det store
        self._pyath_condstore = None # handle to the cond store
        self._ctx = None
        self.__component_type__="Algorithm"
        self.name=name
        return
 
    @property
    def evtStore(self):
        if not self._pyath_evtstore:
            import AthenaPython.PyAthena as PyAthena
            self._pyath_evtstore = PyAthena.py_svc('StoreGateSvc/StoreGateSvc')
        return self._pyath_evtstore
    
    @property
    def detStore(self):
        if not self._pyath_detstore:
            import AthenaPython.PyAthena as PyAthena
            self._pyath_detstore = PyAthena.py_svc('StoreGateSvc/DetectorStore')
        return self._pyath_detstore
    
    @property
    def condStore(self):
        if not self._pyath_condstore:
            import AthenaPython.PyAthena as PyAthena
            self._pyath_condstore = PyAthena.py_svc('StoreGateSvc/ConditionStore')
        return self._pyath_condstore
    
    def sysInitialize(self):
        if hasattr(self, 'OutputLevel'):
            self.msg.setLevel(self.OutputLevel)
        else:
            self.msg.setLevel(_get_prop_value(self,'OutputLevel'))
        return self.initialize()
    
    def initialize(self):
        return StatusCode.Success
 
    def sysReinitialize(self):
        if hasattr(self, 'OutputLevel'):
            self.msg.setLevel(self.OutputLevel)
        else:
            self.msg.setLevel(_get_prop_value(self,'OutputLevel'))
        return self.reinitialize()
    
    def reinitialize(self):
        return StatusCode.Success
 
    def sysExecute(self, cppcontext):
        import cppyy
        self._ctx = cppyy.bind_object(cppcontext, "EventContext")
        try:
            ret = self.execute()
        finally:
            self._ctx = None
        return ret
    
    def execute(self):
        return StatusCode.Success
 
    def sysFinalize(self):
        return self.finalize()
    
    def finalize(self):
        return StatusCode.Success
 
    def sysStart(self):
        return self.start()
    
    def start(self):
        return StatusCode.Success
 
    def sysStop(self):
        return self.stop()
    
    def stop(self):
        return StatusCode.Success
 
    def filterPassed(self):
        """Did this algorithm pass or fail its filter criterion for the last event?"""
        return self._cppHandle.filterPassed(self._ctx)
 
    def setFilterPassed(self, state):
        """Set the filter passed flag to the specified state"""
        return self._cppHandle.setFilterPassed(state, self._ctx)
    
    def resetExecuted(self):
       self.setExecuted(False)
       return StatusCode.Success
    
    def setExecuted(self,state):
       return self._cppHandle.setExecuted(state)
    
    def isExecuted(self):
       return self._cppHandle.isExecuted()
 
    def getContext(self):
       return self._ctx
    
    pass # PyAthena.Alg
 

class Svc( CfgPyService ):
    """Base class for all services
    """
    def __init__(self, name = None, **kw):
        kw.setdefault('OutputLevel', 3) #INFO
        if name is None: name = kw.get('name', self.__class__.__name__)
        
        super(Svc, self).__init__(name, **kw)
        self.__component_type__ = "Service"
        return
 
    def sysInitialize(self):
        if hasattr(self, 'OutputLevel'):
            self.msg.setLevel(self.OutputLevel)
        else:
            self.msg.setLevel(_get_prop_value(self,'OutputLevel'))
        return self.initialize()
    
    def initialize(self):
        return StatusCode.Success
 
    def sysReinitialize(self):
        if hasattr(self, 'OutputLevel'):
            self.msg.setLevel(self.OutputLevel)
        else:
            self.msg.setLevel(_get_prop_value(self,'OutputLevel'))
        return self.reinitialize()
    
    def reinitialize(self):
        return StatusCode.Success
 
    def sysFinalize(self):
        return self.finalize()
    
    def finalize(self):
        return StatusCode.Success
 
    def sysStart(self):
        return self.start()
    
    def start(self):
        return StatusCode.Success
 
    def sysStop(self):
        return self.stop()
    
    def stop(self):
        return StatusCode.Success
    
    pass # PyAthena.Svc
 

class AlgTool( CfgPyAlgTool ):
    """
    Base class from which all concrete algtool classes should be derived.
    """
    
    def __init__(self, name=None, parent=None, **kw):
        kw.setdefault('OutputLevel', 3) #INFO
        if name is None: name = kw.get('name', self.__class__.__name__)
        if parent is not None:
            if isinstance(parent, str): name = "%s.%s" % (parent,name)
            else:                       name = "%s.%s" % (parent.name(),name)
        
        super(AlgTool, self).__init__(name, **kw)
        self._pyath_evtstore = None # handle to the evt store
        self._pyath_detstore = None # handle to the det store
        self._pyath_condstore = None # handle to the cond store
        self.__component_type__ = "AlgTool"
        return
 
    @property
    def evtStore(self):
        if not self._pyath_evtstore:
            import AthenaPython.PyAthena as PyAthena
            self._pyath_evtstore = PyAthena.py_svc('StoreGateSvc/StoreGateSvc')
        return self._pyath_evtstore
    
    @property
    def detStore(self):
        if not self._pyath_detstore:
            import AthenaPython.PyAthena as PyAthena
            self._pyath_detstore = PyAthena.py_svc('StoreGateSvc/DetectorStore')
        return self._pyath_detstore
    
    @property
    def condStore(self):
        if not self._pyath_condstore:
            import AthenaPython.PyAthena as PyAthena
            self._pyath_condstore = PyAthena.py_svc('StoreGateSvc/ConditionStore')
        return self._pyath_condstore
    
    def sysInitialize(self):
        if hasattr(self, 'OutputLevel'):
            self.msg.setLevel(self.OutputLevel)
        else:
            self.msg.setLevel(_get_prop_value(self,'OutputLevel'))
        return self.initialize()
    
    def initialize(self):
        raise NotImplementedError(
            "You have to implement PyAthena.AlgTool.initialize() !"
            )
 
    def sysReinitialize(self):
        if hasattr(self, 'OutputLevel'):
            self.msg.setLevel(self.OutputLevel)
        else:
            self.msg.setLevel(_get_prop_value(self,'OutputLevel'))
        return self.reinitialize()
    
    def reinitialize(self):
        self.msg.info( "==> re-initialize..." )
        return StatusCode.Success
 
    def sysFinalize(self):
        return self.finalize()
    
    def finalize(self):
        raise NotImplementedError(
            "You have to implement PyAthena.AlgTool.finalize() !"
            )
 
    pass # PyAthena.AlgTool
 

class Aud( CfgPyAud ):
    """
    Base class from which all concrete auditor classes should be derived.
    """
    def __init__(self, name=None, **kw):
        kw.setdefault('OutputLevel', 3) #INFO
        if name is None: name = kw.get('name', self.__class__.__name__)
        
        super(Aud, self).__init__(name, **kw)
        return
 
    def sysInitialize(self):
        if hasattr(self, 'OutputLevel'):
            self.msg.setLevel(self.OutputLevel)
        else:
            self.msg.setLevel(_get_prop_value(self,'OutputLevel'))
        return self.initialize()
    
    def initialize(self):
        raise NotImplementedError(
            "You have to implement PyAthena.Aud.initialize() !"
            )
 
    def sysFinalize(self):
        return self.finalize()
    
    def finalize(self):
        raise NotImplementedError(
            "You have to implement PyAthena.Aud.finalize() !"
            )
 
    
 
    def before(self, evt_name, comp_name):
        return
 
    def after(self, evt_name, comp_name, sc):
        return
    
    pass # PyAthena.Aud