python.atlas_oh.OHInputModule Class Reference

Inheritance diagram for python.atlas_oh.OHInputModule:

Collaboration diagram for python.atlas_oh.OHInputModule:

Public Member Functions
def	__init__ (self)
None	configure (self, Mapping[str, Any] options)
None	setSelectors (self, Collection[Pattern] selectors)
Generator[HistObject, None, None]	iterate (self, dryrun)
Iterable[HistObject]	__iter__ (self)
Iterable[HistObject]	warmup (self)

Public Attributes
	source
	classwarnings
	selectors
	provider
	prefix
	interval

Detailed Description

Definition at line 32 of file atlas_oh.py.

Constructor & Destructor Documentation

__init__()

def python.atlas_oh.OHInputModule.__init__

(

self

)

Definition at line 33 of file atlas_oh.py.

    def __init__(self):
        self.source = None
        self.classwarnings = set()
        self.selectors = None
 

Member Function Documentation

__iter__()

Iterable[HistObject] python.atlas_oh.OHInputModule.__iter__

(

self

)

Definition at line 110 of file atlas_oh.py.

    def __iter__(self) -> Iterable[HistObject]:
        return self.iterate(dryrun=False)
 

configure()

None python.atlas_oh.OHInputModule.configure	(		self,
		Mapping[str, Any]	options
	)

Configure this module. Potential elements of "options":
source: a string in the format partition/server
prefix: string to prepend to histogram names.

Definition at line 38 of file atlas_oh.py.

    def configure(self, options: Mapping[str, Any]) -> None:
        """
        Configure this module. Potential elements of "options":
        source: a string in the format partition/server
        prefix: string to prepend to histogram names.
        """
        from ispy import IPCPartition
        if 'source' not in options:
            raise ValueError("Must specify 'source' as an "
                             "option to OHInputModule")
        self.source = options['source']
        self.partition, self.server, self.provider = self.source.split(';')
        self.prefix = options.get('prefix', '')
        if not IPCPartition(self.partition).isValid():
            raise ValueError(f'Input partition {self.partition} does not exist')
        log = logging.getLogger(self.__class__.__name__)
        log.info(f'Using source {self.source}')
        self.interval = options.get('interval', 5)
        log.info(f'Check interval is {self.interval} seconds')
 

iterate()

Generator[HistObject, None, None] python.atlas_oh.OHInputModule.iterate	(		self,
			dryrun
	)

Connect to OH; iterate over matching histograms 

Definition at line 62 of file atlas_oh.py.

    def iterate(self, dryrun) -> Generator[HistObject, None, None]:
        """ Connect to OH; iterate over matching histograms """
        import ROOT
        ROOT.TH1.AddDirectory(ROOT.kFALSE)
        from oh import OHSubscriber
        from ispy import IPCPartition, InfoReader
        import time
        from collections import deque
        log = logging.getLogger(self.__class__.__name__)
        opartition = IPCPartition(self.partition)
 
        # Set up two reading methods: a subscriber to new histograms and
        # a first pass over all histograms.
        queue = deque()
        rec = Receiver(queue, f'{self.server}.{self.provider}.{self.prefix}')
        subs = OHSubscriber(opartition, self.server, rec, True)
        subs.subscribe_re2(self.server, self.provider, '.*', True)
 
        reader = InfoReader(opartition, self.server, rf'{self.provider}\..*')
        reader.update()
        log.debug('First pass update')
        for k in reader.objects:
            toread = k.replace(f'{self.server}.{self.provider}.','')
            obj = readobj(opartition, self.server, self.provider, toread, dryrun)
            log.debug('OH input read '
                      f'{k} as {toread.replace(self.prefix, "")}')
            yield HistObject(toread.replace(self.prefix, '', 1), obj)
 
        if not dryrun:
            log.info('First pass complete, entering update loop')
            del reader
            try:
                while True:
                    try:
                        time.sleep(self.interval)
                        if len(queue) > 0:
                            log.info('Update detected')
                        while True:
                            yield queue.popleft()
                    except IndexError:
                        pass
                    log.debug('Completed update')
            finally:
                subs.unsubscribe_re2(self.server, self.provider, '.*')
 
        return
        yield
 

setSelectors()

None python.atlas_oh.OHInputModule.setSelectors	(		self,
		Collection[Pattern]	selectors
	)

Do more later 

Definition at line 58 of file atlas_oh.py.

    def setSelectors(self, selectors: Collection[Pattern]) -> None:
        """ Do more later """
        self.selectors = selectors
 

warmup()

Iterable[HistObject] python.atlas_oh.OHInputModule.warmup

(

self

)

Definition at line 113 of file atlas_oh.py.

    def warmup(self) -> Iterable[HistObject]:
        return self.iterate(dryrun=True)
 
 

Member Data Documentation

classwarnings

python.atlas_oh.OHInputModule.classwarnings

Definition at line 35 of file atlas_oh.py.

interval

python.atlas_oh.OHInputModule.interval

Definition at line 55 of file atlas_oh.py.

prefix

python.atlas_oh.OHInputModule.prefix

Definition at line 50 of file atlas_oh.py.

provider

python.atlas_oh.OHInputModule.provider

Definition at line 49 of file atlas_oh.py.

selectors

python.atlas_oh.OHInputModule.selectors

Definition at line 36 of file atlas_oh.py.

source

python.atlas_oh.OHInputModule.source

Definition at line 34 of file atlas_oh.py.

---

The documentation for this class was generated from the following file:

• atlas_oh.py