Source code for common.stat_live

import numpy as np

[docs] class RunningStat(): # tracks realtime mean and standard deviation without storing any data def __init__(self, priors=None, max_trackable=-1): self.max_trackable = max_trackable if priors is not None: # initialize from history self.M = priors[0] self.S = priors[1] self.n = priors[2] self.M_last = self.M self.S_last = self.S else: self.reset()
[docs] def reset(self): self.M = 0. self.S = 0. self.M_last = 0. self.S_last = 0. self.n = 0
[docs] def push_data(self, new_data): # short term memory hack if self.max_trackable < 0 or self.n < self.max_trackable: self.n += 1 if self.n == 0: self.M_last = new_data self.M = self.M_last self.S_last = 0. else: self.M = self.M_last + (new_data - self.M_last) / self.n self.S = self.S_last + (new_data - self.M_last) * (new_data - self.M) self.M_last = self.M self.S_last = self.S
[docs] def mean(self): return self.M
[docs] def variance(self): if self.n >= 2: return self.S / (self.n - 1.) else: return 0
[docs] def std(self): return np.sqrt(self.variance())
[docs] def params_to_save(self): return [self.M, self.S, self.n]
[docs] class RunningStatFilter(): def __init__(self, raw_priors=None, filtered_priors=None, max_trackable=-1): self.raw_stat = RunningStat(raw_priors, -1) self.filtered_stat = RunningStat(filtered_priors, max_trackable)
[docs] def reset(self): self.raw_stat.reset() self.filtered_stat.reset()
[docs] def push_and_update(self, new_data): _std_last = self.raw_stat.std() self.raw_stat.push_data(new_data) _delta_std = self.raw_stat.std() - _std_last if _delta_std <= 0: self.filtered_stat.push_data(new_data) else: pass
# self.filtered_stat.push_data(self.filtered_stat.mean()) # class SequentialBayesian():