heros.datasource.observables¶

Classes¶

`BoundaryChecker`	Check whether the observed value is in it's boundaries.
`Converter`
`Observable`
`ObservableProcessor`	An ObservableProcessor takes a configuration dict for multiple observables when it is instantiated.

Functions¶

lambdify(expression[, globals_dict])

Turn a string into a lambda function with the given globals.

Module Contents¶

heros.datasource.observables.lambdify(expression: str, globals_dict: dict | None = None)¶: Turn a string into a lambda function with the given globals. By default it enables a set of numpy functions in the expression. Further functions can be added by providing it in globals_dict.

class heros.datasource.observables.BoundaryChecker(boundaries: list)¶

Check whether the observed value is in it’s boundaries. Therefore provide an integer feedback with rising importance where boundary = 0 says that everything is fine. 1, 2, 3 correspond to higher warning / error or fault situations.

boundaries¶

_check_bounds(value)¶

__call__(return_value: heros.datasource.types.DatasourceReturnValue)¶

class heros.datasource.observables.Converter(conversion_code: str, unit: str | None = None)¶

_conversion_code¶

_convert_lambda¶

_unit = None¶

__call__(return_value: heros.datasource.types.DatasourceReturnValue)¶

class heros.datasource.observables.Observable(id, definition: dict)¶

_id¶

definition¶

_boundary_checker¶

_converter¶

property id¶

process(return_value: heros.datasource.types.DatasourceReturnValue)¶

class heros.datasource.observables.ObservableProcessor(obs_def: dict)¶

An ObservableProcessor takes a configuration dict for multiple observables when it is instantiated. An object of this class can be called with a :class:DataSourceReturnSet as an argument.

observables¶

observables_ids¶

__call__(datasource_return_set)¶