| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253 |
- import logging
- import re
- import dataclasses
- from .common import MatchSeries
- from structures.measurement import Measurement24v, Measurement480v
- from structures.plant import S7State, CompactLogixState
- from structures.correlated import CorrelatedMeasurements
- from structures.common import BaseMeasurement
- logger = logging.getLogger(__name__)
- ALLOWED_NAMES = \
- [x.name for x in dataclasses.fields(Measurement24v)] + \
- [x.name for x in dataclasses.fields(Measurement480v)] + \
- [x.name for x in dataclasses.fields(CompactLogixState)] + \
- [x.name for x in dataclasses.fields(S7State)] + \
- [x.name for x in dataclasses.fields(CorrelatedMeasurements)] + \
- ['sum', 'min', 'max', 'avg', 'count', 'last']
- ALLOWED_NAMES = set([name for name in ALLOWED_NAMES if not name.startswith('_')])
- @dataclasses.dataclass(frozen=True)
- class Selection(BaseMeasurement):
- value: str
- class ComplexSelector():
- def __init__(self, parent, selector) -> None:
- self._selector = selector
- self._compiled = compile(selector, "<string>", "eval")
- # Validate allowed names
- for name in self._compiled.co_names:
- if name not in ALLOWED_NAMES:
- raise NameError(f"The use of '{name}' is not allowed in '{selector}'")
- def execute(self, values):
- for measurement in values:
- try:
- value = eval(self._compiled, {"__builtins__": {
- 'sum': sum,
- 'min': min,
- 'max': max,
- 'avg': lambda x: sum(x) / len(x),
- 'count': len,
- 'last': lambda x: x[-1],
- }}, measurement.__dict__)
- yield Selection(
- timestamp = measurement.timestamp,
- series = re.match(r"[\w_\.\-\(\)]+", self._selector).group(0),
- source = "selection",
- value=value
- )
- except Exception as e:
- logger.error(f"Error while evaluating selector '{self._selector}': {e}")
|
