lit_sb3.core.running_mean_std
[docs]
module
lit_sb3.core.running_mean_std
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57 | import numpy as np
class RunningMeanStd: # TODO replace with https://lightning.ai/docs/torchmetrics/stable/aggregation/running_mean.html
def __init__(self, epsilon: float = 1e-4, shape: tuple[int, ...] = ()) -> None:
"""Calculates the running mean and std of a data stream.
https://en.wikipedia.org/wiki/Algorithms_for_calculating_variance#Parallel_algorithm.
Args:
epsilon: helps with arithmetic issues
shape: the shape of the data stream's output
"""
self.mean = np.zeros(shape, np.float64)
self.var = np.ones(shape, np.float64)
self.count = epsilon
def copy(self) -> "RunningMeanStd":
"""Returns.
Return a copy of the current object.
"""
new_object = RunningMeanStd(shape=self.mean.shape)
new_object.mean = self.mean.copy()
new_object.var = self.var.copy()
new_object.count = float(self.count)
return new_object
def combine(self, other: "RunningMeanStd") -> None:
"""Combine stats from another ``RunningMeanStd`` object.
Args:
other: The other object to combine with.
"""
self.update_from_moments(other.mean, other.var, other.count)
def update(self, arr: np.ndarray) -> None:
batch_mean = np.mean(arr, axis=0)
batch_var = np.var(arr, axis=0)
batch_count = arr.shape[0]
self.update_from_moments(batch_mean, batch_var, batch_count)
def update_from_moments(self, batch_mean: np.ndarray, batch_var: np.ndarray, batch_count: float) -> None:
delta = batch_mean - self.mean
tot_count = self.count + batch_count
new_mean = self.mean + delta * batch_count / tot_count
m_a = self.var * self.count
m_b = batch_var * batch_count
m_2 = m_a + m_b + np.square(delta) * self.count * batch_count / (self.count + batch_count)
new_var = m_2 / (self.count + batch_count)
new_count = batch_count + self.count
self.mean = new_mean
self.var = new_var
self.count = new_count
|