copapy.ringbuffer#

class copapy.ringbuffer[T](x, n, init=0)#

Bases: Generic[T]

Ring buffer holding the values of x at the last n runs of the task computing it, the current value included. At each run the current value is written in place to one slot of the buffer (no copy of the buffer), so the cost per run is independent of n. Before n runs the remaining slots hold the initial value.

>>> rb = cp.ringbuffer(x, 1000)
>>> mean = cp.filters.mean(rb.data)  # Order of the slots is irrelevant
>>> delayed = rb[250]                # x 250 runs ago
>>> fir = (rb.window() * taps).sum() # Oldest value first

The buffer is computed by the task computing x. To use it in another task, data and index must be outputs of the task computing it (like for delay). Values read from the buffer are constants for the gradient, only rb[0] (x itself) is differentiable.

n#

Number of values in the buffer

shape#

Shape of one value: () for scalars

Type:

tuple[int, ...]

index#

Slot of the current value in data (0 to n - 1)

Type:

value[int]

data#

The buffer as array of shape (n, *shape) in the order of the slots. It is the memory of the buffer, reading it costs no copy.

Type:

array[Any]

Ring buffer of the last n values of x.

Overloads:

  • x (value[float] | float), n (int), init (Any) = 0

  • x (value[int] | int), n (int), init (Any) = 0

  • x (T), n (int), init (Any) = 0

Parameters:

  • x (Any) – Value, vector, tensor, quaternion or array of the current run. Python numbers are converted to constant values.

  • n (int) – Number of values in the buffer (at least 1)

  • init (Any) – Value of the slots before they are written: a number for all elements or the shape of x

property oldest: T#

Oldest value in the buffer (n - 1 runs ago), e.g. for a moving sum updated by the value entering and the value leaving the window.

window()#

Values of the buffer in time order, the oldest value first and the current value last: a copy of the buffer (array of shape (n, *shape)).

Return type:

array[Any]