boss.factory

Classes

BOSSObject

_MROCompatiblityBase

Class to interrupt __init__ calls from heroized classes to the HERO mixin.

Factory

HEROFactory

DatasourceHEROFactory

PolledDatasourceHEROFactory

Functions

wrap_hero_class(→ type)

Wrap the source class to establish HEROS compatibility.

Module Contents

class boss.factory.BOSSObject[source]
_stop(boss: heros.RemoteHERO)[source]
boss.factory.wrap_hero_class(source_class: type, name: str, arg_dict: dict, wrap_target: str | None = None) type[source]

Wrap the source class to establish HEROS compatibility.

BOSS supports different external libraries which require extra treatment to expose them correctly to the HERO network. This function returns a HEROS compatible representation of the source class.

Parameters:
  • source_class – The class to be wrapped

  • name – Name of the HERO to be created from the source_class

  • arg_dict – Arguments to be passed to source_class.__init__

  • wrap_target – By default (None or "auto"), the function tries to automatically find from which library the source class is and wrap it accordingly. By specifying wrap_target the wrapper function can be controlled or wrapping can be turned off by passing no_wrap.

class boss.factory._MROCompatiblityBase(*args, **kwargs)[source]

Class to interrupt __init__ calls from heroized classes to the HERO mixin.

When a heroized class calls super().__init__() on object, the method resolution order (MRO) may cause the HERO mixin’s __init__ to be invoked with incorrect arguments. This class acts as an intermediary in the inheritance chain, preventing the HERO mixin from receiving unintended arguments by intercepting the super() call.

class boss.factory.Factory[source]
_mixin_class: type
classmethod _build(classname: str, name: str, arg_dict: dict | None = None, extra_decorators: list[tuple[str, str]] | None = None, wrap_target: str | None = None, realm: str = 'heros', session_manager=None, tags: list | None = None)[source]
classmethod _get_init_replacement(source_class: type, name: str, realm: str, session_manager, tags: list | None) Callable[source]
Abstractmethod:

static _decorate_methods(source_class: Any, extra_decorators: list[tuple[str, str]]) Any[source]

Wrap methods on the class before instantiation.

class boss.factory.HEROFactory[source]

Bases: Factory

_mixin_class
classmethod build(classname: str, name: str, arg_dict: dict | None = None, extra_decorators: list[tuple[str, str]] | None = None, wrap_target: str | None = None, realm='heros', session_manager=None, tags: list | None = None)[source]
classmethod _get_init_replacement(source_class: type, name: str, realm: str, session_manager, tags: list | None) Callable[source]
class boss.factory.DatasourceHEROFactory[source]

Bases: HEROFactory

_mixin_class
_observables: dict
classmethod build(classname: str, name: str, arg_dict: dict | None = None, extra_decorators: list[tuple[str, str]] | None = None, wrap_target: str | None = None, observables: dict | None = None, realm='heros', session_manager=None, tags: list | None = None)[source]
classmethod _get_init_replacement(source_class: type, name: str, realm: str, session_manager, tags: list | None) Callable[source]
class boss.factory.PolledDatasourceHEROFactory[source]

Bases: Factory

_mixin_class
_observables: dict
_interval: float
classmethod build(classname: str, name: str, arg_dict: dict | None = None, extra_decorators: list[tuple[str, str]] | None = None, wrap_target: str | None = None, loop: asyncio.AbstractEventLoop = asyncio.new_event_loop(), interval: float = 5, observables: dict | None = None, realm='heros', session_manager=None, tags: list | None = None)[source]
classmethod _get_init_replacement(source_class: type, name: str, realm: str, session_manager, tags: list | None) Callable[source]