IS
Size: a a a
IS
AA
IS
IS
IS
DV
DV
NA
DV
DV
DV
DV
DV
DK
DV
DV
DV
DV
class MeanNodeState(BaseModel):
values: List[float] = []
class MeanNodeConfig(BaseModel):
min_window: int = 3
max_window: int = 100
class MeanNode(BaseCDAGNode):
"""
Calculate estimated mean value
"""
name = "mean"
static_inputs = "x"
config_cls = MeanNodeConfig
state_cls = MeanNodeState
categories = [Category.STATISTICS]
def get_stats(self, values: np.array) -> float:
return np.mean(values)
def get_value(self) -> Optional[ValueType]:
x = self.get_input("x")
if x is None:
return None
self.state.values.insert(0, float(x))
# Trim
if len(self.state.values) >= self.config.max_window:
self.state.values = self.state.values[: self.config.max_window]
# Check window is filled
if len(self.state.values) < self.config.min_window:
return None
values = np.array(self.state.values)
return self.get_stats(values)
DV