Source code for eventkit.ops.array

from collections import deque

from .op import Op
from ..util import NO_VALUE

import numpy as np


[docs]class Array(Op): __slots__ = ('_count', '_q') def __init__(self, count, source=None): Op.__init__(self, source) self._count = count self._q = deque()
[docs] def on_source(self, *args): self._q.append( args[0] if len(args) == 1 else args if args else NO_VALUE) if self._count and len(self._q) > self._count: self._q.popleft() self.emit(np.asarray(self._q))
[docs] def min(self) -> "ArrayMin": """ Minimum value. """ return ArrayMin(self)
[docs] def max(self) -> "ArrayMax": """ Maximum value. """ return ArrayMax(self)
[docs] def sum(self) -> "ArraySum": """ Summation. """ return ArraySum(self)
[docs] def prod(self) -> "ArrayProd": """ Product. """ return ArrayProd(self)
[docs] def mean(self) -> "ArrayMean": """ Mean value. """ return ArrayMean(self)
[docs] def std(self) -> "ArrayStd": """ Sample standard deviation. """ return ArrayStd(self)
[docs] def any(self) -> "ArrayAny": """ Test if any array value is true. """ return ArrayAny(self)
[docs] def all(self) -> "ArrayAll": """ Test if all array values are true. """ return ArrayAll(self)
[docs]class ArrayMin(Op): __slots__ = ()
[docs] def on_source(self, arg): self.emit(arg.min())
[docs]class ArrayMax(Op): __slots__ = ()
[docs] def on_source(self, arg): self.emit(arg.max())
[docs]class ArraySum(Op): __slots__ = ()
[docs] def on_source(self, arg): self.emit(arg.sum())
[docs]class ArrayProd(Op): __slots__ = ()
[docs] def on_source(self, arg): self.emit(arg.prod())
[docs]class ArrayMean(Op): __slots__ = ()
[docs] def on_source(self, arg): self.emit(arg.mean())
[docs]class ArrayStd(Op): __slots__ = ()
[docs] def on_source(self, arg): self.emit(arg.std(ddof=1) if len(arg) > 1 else np.nan)
[docs]class ArrayAny(Op): __slots__ = ()
[docs] def on_source(self, arg): self.emit(arg.any())
[docs]class ArrayAll(Op): __slots__ = ()
[docs] def on_source(self, arg): self.emit(arg.all())