Buffer¶
The rite.collections.buffer subpackage provides circular and sliding window buffer implementations for efficient, fixed-size data handling.
Buffer Collections Module¶
Provides various buffer data structures for specialized data storage patterns.
Classes:¶
- CircularBuffer: Ring buffer with automatic overwriting
- RingBuffer: Alias for CircularBuffer
- BoundedBuffer: Size-limited buffer with overflow protection
- SlidingWindow: Moving window over a data stream
Classes¶
BoundedBuffer(maxsize: int, overflow_strategy: str = 'drop_oldest')
¶
BoundedBuffer Class¶
A size-limited buffer that prevents overflow with configurable behavior.
Parameters¶
maxsize : int Maximum number of elements the buffer can hold. overflow_strategy : str Strategy when full: 'block', 'drop_oldest', 'drop_newest', 'raise'
Initialize the bounded buffer.
maxsize: Maximum capacity of the buffer.
overflow_strategy: Behavior when buffer is full.
Attributes¶
capacity: int
property
¶
Return maximum number of elements the buffer can hold.
Functions¶
append(item: Any) -> bool
¶
Add an item to the buffer.
item: The item to add.
bool: True if item was added, False if dropped.
BufferError: If overflow_strategy is 'raise' and buffer is full.
clear() -> None
¶
Clear all items from buffer.
extend(items: list[Any]) -> None
¶
Extend buffer with multiple items.
items: List of items to add.
get(index: int) -> Any | None
¶
Return the item at a given index or None if out of bounds.
get_all() -> list[Any]
¶
Get all items in buffer.
list[Any]: All items in insertion order.
is_empty() -> bool
¶
Check if buffer is empty.
bool: True if buffer contains no items.
is_full() -> bool
¶
Check if buffer is full.
bool: True if buffer has reached maxsize.
peek() -> Any | None
¶
Return the oldest item without removing it.
CircularBuffer(size: int)
¶
CircularBuffer Class¶
A fixed-size buffer that overwrites the oldest data when full.
Attributes¶
size : int The maximum number of elements the buffer can hold. buffer : list[Any | None] The internal list storing buffer elements. index : int The current index for the next write operation. full : bool Indicates if the buffer has reached its maximum capacity.
Methods¶
append(value: Any) -> None: Adds a value to the buffer, overwriting the oldest element if full. get_all() -> list[Any | None]: Retrieves all elements in the buffer in the correct order. is_empty() -> bool: Checks if the buffer is empty. is_full() -> bool: Checks if the buffer is full.
Initializes a CircularBuffer instance.
Parameters:¶
size : int The maximum number of elements the buffer can hold.
Functions¶
append(value: Any) -> None
¶
Adds a value to the buffer. Overwrites the oldest element if the buffer is full.
Parameters:¶
value : Any The value to add to the buffer.
clear() -> None
¶
Clear the buffer.
get(index: int) -> Any | None
¶
Return item at index or None if out of bounds.
get_all() -> list[Any | None]
¶
Retrieves all elements in the buffer in the correct order.
Returns¶
list[Any | None]: A list of elements in the buffer, ordered from the oldest to the newest.
is_empty() -> bool
¶
SlidingWindow(size: int, aggregation_func: Callable[[list[Any]], Any] | None = None)
¶
SlidingWindow Class¶
A fixed-size window that slides over a data stream, useful for moving averages, rolling calculations, and stream processing.
Parameters¶
size : int The size of the sliding window. aggregation_func : Callable | None Optional function to apply to window contents.
Initialize the sliding window.
size: Window size (number of elements).
aggregation_func: Optional function to aggregate window data.
Functions¶
add(value: Any) -> Any | None
¶
Add a value to the window and return aggregated result.
value: The value to add to the window.
Any | None: Aggregated result if aggregation_func is set.
clear() -> None
¶
Clear all elements from window.
get_aggregate() -> Any | None
¶
Get aggregated value of current window.
Any | None: Aggregated result or None if no aggregation function.
get_window() -> list[Any]
¶
Get current window contents.
list[Any]: Current window values in order.
is_empty() -> bool
¶
Check if window is empty.
bool: True if window contains no elements.
is_full() -> bool
¶
Check if window is full.
bool: True if window has reached its size.
moving_average() -> float | None
¶
Calculate moving average of numeric window.
float | None: Average of window values or None if empty.
moving_max() -> Any | None
¶
Get maximum value in window.
Any | None: Maximum value or None if empty.
moving_min() -> Any | None
¶
Get minimum value in window.
Any | None: Minimum value or None if empty.
moving_sum() -> float | None
¶
Calculate moving sum of numeric window.
float | None: Sum of window values or None if empty.
Modules¶
bounded_buffer
¶
Bounded Buffer¶
A thread-safe bounded buffer that blocks or raises errors when capacity limits are reached.
Classes¶
BoundedBuffer(maxsize: int, overflow_strategy: str = 'drop_oldest')
¶
BoundedBuffer Class¶
A size-limited buffer that prevents overflow with configurable behavior.
Parameters¶
maxsize : int Maximum number of elements the buffer can hold. overflow_strategy : str Strategy when full: 'block', 'drop_oldest', 'drop_newest', 'raise'
Initialize the bounded buffer.
maxsize: Maximum capacity of the buffer.
overflow_strategy: Behavior when buffer is full.
Attributes¶
capacity: int
property
¶Return maximum number of elements the buffer can hold.
Functions¶
append(item: Any) -> bool
¶Add an item to the buffer.
item: The item to add.
bool: True if item was added, False if dropped.
BufferError: If overflow_strategy is 'raise' and buffer is full.
clear() -> None
¶Clear all items from buffer.
extend(items: list[Any]) -> None
¶Extend buffer with multiple items.
items: List of items to add.
get(index: int) -> Any | None
¶Return the item at a given index or None if out of bounds.
get_all() -> list[Any]
¶Get all items in buffer.
list[Any]: All items in insertion order.
is_empty() -> bool
¶Check if buffer is empty.
bool: True if buffer contains no items.
is_full() -> bool
¶Check if buffer is full.
bool: True if buffer has reached maxsize.
peek() -> Any | None
¶Return the oldest item without removing it.
circular_buffer
¶
Circular Buffer¶
A circular buffer (ring buffer) implementation that overwrites the oldest entries when the buffer reaches its maximum capacity.
Classes:¶
- CircularBuffer: A fixed-size buffer with circular indexing.
Features:¶
- Append new elements, overwriting the oldest if the buffer is full.
- Retrieve all elements in the correct order.
- Check if the buffer is empty or full.
Usage:¶
buffer = CircularBuffer(size=5)
buffer.append(1)
buffer.append(2)
print(buffer.get_all()) # Outputs: [1, 2]
buffer.append(3)
buffer.append(4)
buffer.append(5)
print(buffer.get_all()) # Outputs: [1, 2, 3, 4, 5]
buffer.append(6)
print(buffer.get_all()) # Outputs: [6, 2, 3, 4, 5]
Classes¶
CircularBuffer(size: int)
¶
CircularBuffer Class¶
A fixed-size buffer that overwrites the oldest data when full.
Attributes¶
size : int The maximum number of elements the buffer can hold. buffer : list[Any | None] The internal list storing buffer elements. index : int The current index for the next write operation. full : bool Indicates if the buffer has reached its maximum capacity.
Methods¶
append(value: Any) -> None: Adds a value to the buffer, overwriting the oldest element if full. get_all() -> list[Any | None]: Retrieves all elements in the buffer in the correct order. is_empty() -> bool: Checks if the buffer is empty. is_full() -> bool: Checks if the buffer is full.
Initializes a CircularBuffer instance.
Parameters:¶
size : int The maximum number of elements the buffer can hold.
Functions¶
append(value: Any) -> None
¶Adds a value to the buffer. Overwrites the oldest element if the buffer is full.
Parameters:¶
value : Any The value to add to the buffer.
clear() -> None
¶Clear the buffer.
get(index: int) -> Any | None
¶Return item at index or None if out of bounds.
get_all() -> list[Any | None]
¶Retrieves all elements in the buffer in the correct order.
Returns¶
list[Any | None]: A list of elements in the buffer, ordered from the oldest to the newest.
is_empty() -> bool
¶
sliding_window
¶
Sliding Window¶
A sliding window data structure for stream processing and moving calculations.
Classes¶
SlidingWindow(size: int, aggregation_func: Callable[[list[Any]], Any] | None = None)
¶
SlidingWindow Class¶
A fixed-size window that slides over a data stream, useful for moving averages, rolling calculations, and stream processing.
Parameters¶
size : int The size of the sliding window. aggregation_func : Callable | None Optional function to apply to window contents.
Initialize the sliding window.
size: Window size (number of elements).
aggregation_func: Optional function to aggregate window data.
Functions¶
add(value: Any) -> Any | None
¶Add a value to the window and return aggregated result.
value: The value to add to the window.
Any | None: Aggregated result if aggregation_func is set.
clear() -> None
¶Clear all elements from window.
get_aggregate() -> Any | None
¶Get aggregated value of current window.
Any | None: Aggregated result or None if no aggregation function.
get_window() -> list[Any]
¶Get current window contents.
list[Any]: Current window values in order.
is_empty() -> bool
¶Check if window is empty.
bool: True if window contains no elements.
is_full() -> bool
¶Check if window is full.
bool: True if window has reached its size.
moving_average() -> float | None
¶Calculate moving average of numeric window.
float | None: Average of window values or None if empty.
moving_max() -> Any | None
¶Get maximum value in window.
Any | None: Maximum value or None if empty.
moving_min() -> Any | None
¶Get minimum value in window.
Any | None: Minimum value or None if empty.
moving_sum() -> float | None
¶Calculate moving sum of numeric window.
float | None: Sum of window values or None if empty.
options: show_root_heading: true show_source: false heading_level: 2