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())