Round #1142

Implement an iterator class RingBuffer that exposes the __iter__ and __next__ methods to walk a fixed-capacity circular buffer of integers. Store the underlying state as a @dataclass with frozen=True for the capacity field and a separate mutable list for the contents. Expose __len__ for the len builtin and __contains__ for the in operator. Raise StopIteration when iteration completes one full lap. Add type hints using typing.Iterator and typing.Optional. The class should behave correctly under isinstance checks against collections.abc.Iterator.

```python """ ring_buffer.py ~~~~~~~~~~~~~~ A fixed-capacity circular buffer of integers that implements the full iterator protocol together with __len__ and __contains__. Design decisions ---------------- * The *capacity* is stored inside a frozen @dataclass so it can never be mutated after construction. * The mutable *contents* live in a plain list that is attached to the dataclass instance at __post_init__ time (Python lets you do this even with frozen=True by using object.__setattr__). * RingBuffer registers itself (via ABCMeta machinery) with collections.abc.Iterator so that isinstance(rb, Iterator) is True. * Iteration always walks exactly one full lap (capacity slots) starting at the current head, yielding None for empty slots. """ from __future__ import annotations import collections.abc from dataclasses import dataclass, field from typing import Iterator, Optional # --------------------------------------------------------------------------- # Frozen dataclass that holds configuration + state # --------------------------------------------------------------------------- @dataclass(frozen=True) class _RingBufferState: """ Immutable configuration + mutable contents of a ring buffer. ``capacity`` – fixed maximum number of elements (frozen / immutable) ``_contents`` – mutable list of Optional[int] slots (len == capacity) ``_head`` – index of the next *write* position (boxed in a 1-element list so we can mutate it even though the dataclass is frozen) ``_iter_pos`` – iterator cursor, also boxed in a 1-element list ``_iter_steps``– how many steps have been taken in the current lap, also boxed. """ capacity: int # All mutable state is wrapped in lists so that frozen=True does not # prevent mutation of the *contents* of the fields. _contents: list = field(default_factory=list, init=False, repr=False, compare=False) _head: list = field(default_factory=lambda: [0], init=False, repr=False, compare=False) _iter_pos: list = field(default_factory=lambda: [0], init=False, repr=False, compare=False) _iter_steps: list = field(default_factory=lambda: [0], init=False, repr=False, compare=False) def __post_init__(self) -> None: if self.capacity < 1: raise ValueError(f"capacity must be >= 1, got {self.capacity}") # Pre-fill with None to represent empty slots. # Because frozen=True we must use object.__setattr__ to touch the # list object *itself* (we are replacing the empty list created by # default_factory with a properly sized one). object.__setattr__(self, "_contents", [None] * self.capacity) # --------------------------------------------------------------------------- # Public ring-buffer / iterator class # --------------------------------------------------------------------------- class RingBuffer(collections.abc.Iterator): """ A fixed-capacity circular buffer of integers. Supports -------- * ``append(value)`` – write one integer into the next slot (overwrites the oldest value when full). * ``__len__`` – number of slots that have been written at least once (saturates at capacity). * ``__contains__`` – membership test over *all* slots (including None for empty slots when you explicitly check None). * ``__iter__`` – returns *self* (the buffer is its own iterator). * ``__next__`` – yields one value per call; raises StopIteration after exactly *capacity* steps (one full lap). * ``reset_iter()`` – rewinds the iterator cursor back to the head so the buffer can be iterated again. Parameters ---------- capacity : int Maximum number of elements. Must be >= 1. Examples -------- >>> rb = RingBuffer(4) >>> rb.append(10) >>> rb.append(20) >>> rb.append(30) >>> list(rb) [10, 20, 30, None] >>> rb.reset_iter() >>> 20 in rb True >>> len(rb) 3 """ # ------------------------------------------------------------------ # Construction # ------------------------------------------------------------------ def __init__(self, capacity: int) -> None: self._state = _RingBufferState(capacity=capacity) # Track how many distinct slots have actually been written. self._written: int = 0 # ------------------------------------------------------------------ # Properties that delegate to the frozen dataclass # ------------------------------------------------------------------ @property def capacity(self) -> int: """The fixed maximum number of elements.""" return self._state.capacity @property def _contents(self) -> list: return self._state._contents @property def _head(self) -> list: # boxed int return self._state._head @property def _iter_pos(self) -> list: # boxed int return self._state._iter_pos @property def _iter_steps(self) -> list: # boxed int return self._state._iter_steps # ------------------------------------------------------------------ # Write # ------------------------------------------------------------------ def append(self, value: int) -> None: """ Write *value* into the next slot, advancing the head pointer. When the buffer is full the oldest value is overwritten (FIFO ring semantics). """ if not isinstance(value, int): raise TypeError(f"RingBuffer only stores int, got {type(value).__name__!r}") idx: int = self._head[0] self._contents[idx] = value self._head[0] = (idx + 1) % self.capacity if self._written < self.capacity: self._written += 1 # ------------------------------------------------------------------ # Sizing / membership # ------------------------------------------------------------------ def __len__(self) -> int: """ Return the number of slots that have been written at least once. Always <= capacity. """ return self._written def __contains__(self, item: object) -> bool: """ Return True if *item* is present in any slot of the buffer. Empty (un-written) slots hold None. """ return item in self._contents # ------------------------------------------------------------------ # Iterator protocol # ------------------------------------------------------------------ def __iter__(self) -> Iterator[Optional[int]]: """ Return *self* as the iterator. Each call to __iter__ resets the cursor so that a new for-loop always walks from the current head position. """ self._iter_pos[0] = self._head[0] # start from oldest element self._iter_steps[0] = 0 return self def __next__(self) -> Optional[int]: """ Return the next value in the lap. Raises ------ StopIteration After exactly *capacity* calls, signalling one full lap. """ if self._iter_steps[0] >= self.capacity: raise StopIteration idx: int = self._iter_pos[0] value: Optional[int] = self._contents[idx] # Advance cursor self._iter_pos[0] = (idx + 1) % self.capacity self._iter_steps[0] += 1 return value # ------------------------------------------------------------------ # Convenience # ------------------------------------------------------------------ def reset_iter(self) -> None: """ Rewind the iterator cursor so the buffer can be traversed again without constructing a new object. """ self._iter_pos[0] = self._head[0] self._iter_steps[0] = 0 # ------------------------------------------------------------------ # Dunder helpers # ------------------------------------------------------------------ def __repr__(self) -> str: # Show a human-readable snapshot of the contents in lap order. start = self._head[0] ordered = ( self._contents[start:] + self._contents[:start] ) return ( f"RingBuffer(capacity={self.capacity}, " f"contents={ordered!r})" ) # ---------------------------------------------------------------------------

```python from __future__ import annotations import collections.abc from dataclasses import dataclass, field from typing import Iterator, Optional @dataclass class _RingBufferState: """Holds the fixed capacity (immutable) and mutable contents separately.""" capacity: int # treated as frozen via __post_init__ enforcement contents: list[int] = field(default_factory=list) def __post_init__(self) -> None: if self.capacity <= 0: raise ValueError(f"capacity must be positive, got {self.capacity}") def append(self, value: int) -> None: """Add a value, evicting the oldest entry when at capacity.""" if len(self.contents) >= self.capacity: self.contents.pop(0) self.contents.append(value) def set_capacity(self, value: int) -> None: # guarded mutation raise AttributeError("capacity is immutable after construction") class RingBuffer(collections.abc.Iterator[int]): """ A fixed-capacity circular buffer of integers that is also its own iterator. Iterating with for/next walks every currently stored element exactly once per 'lap', then raises StopIteration. Calling iter() resets the cursor so the same RingBuffer object can be iterated again from scratch. Examples -------- >>> rb = RingBuffer(capacity=3) >>> rb.push(10); rb.push(20); rb.push(30); rb.push(40) >>> list(rb) [20, 30, 40] >>> 30 in rb True >>> len(rb) 3 """ # ------------------------------------------------------------------ # Construction # ------------------------------------------------------------------ def __init__(self, capacity: int) -> None: self._state = _RingBufferState(capacity=capacity) self._cursor: int = 0 # iteration position self._lap_done: bool = False # sentinel: one full lap completed # ------------------------------------------------------------------ # Public mutation API # ------------------------------------------------------------------ def push(self, value: int) -> None: """Insert *value*, overwriting the oldest element when full.""" self._state.append(value) # If the cursor was pointing past what we just evicted, clamp it. if self._cursor > len(self._state.contents): self._cursor = len(self._state.contents) # ------------------------------------------------------------------ # Iterator protocol (collections.abc.Iterator requires __iter__ + __next__) # ------------------------------------------------------------------ def __iter__(self) -> Iterator[int]: """Reset the cursor and return *self* so the buffer is re-iterable.""" self._cursor = 0 self._lap_done = False return self def __next__(self) -> int: if self._lap_done or self._cursor >= len(self._state.contents): self._lap_done = True raise StopIteration value: int = self._state.contents[self._cursor] self._cursor += 1 if self._cursor >= len(self._state.contents): self._lap_done = True return value # ------------------------------------------------------------------ # Container protocol # ------------------------------------------------------------------ def __len__(self) -> int: """Return the number of elements currently stored (≤ capacity).""" return len(self._state.contents) def __contains__(self, item: object) -> bool: """Support the *in* operator without consuming the iterator state.""" return item in self._state.contents # ------------------------------------------------------------------ # Convenience # ------------------------------------------------------------------ @property def capacity(self) -> int: return self._state.capacity def peek(self, index: int) -> Optional[int]: """Return element at *index* without advancing the cursor, or None.""" contents = self._state.contents if 0 <= index < len(contents): return contents[index] return None def __repr__(self) -> str: return ( f"RingBuffer(capacity={self.capacity}, " f"contents={self._state.contents!r}, " f"cursor={self._cursor})" ) # --------------------------------------------------------------------------- # Quick smoke-test (run with: python ring_buffer.py) # --------------------------------------------------------------------------- if __name__ == "__main__": rb = RingBuffer(capacity=4) for v in (1, 2, 3, 4, 5): # 5 pushes into a cap-4 buffer → evicts 1 rb.push(v) print("contents :", rb._state.contents) # [2, 3, 4, 5] print("len :", len(rb)) # 4 print("3 in rb :", 3 in rb) # True print("1 in rb :", 1 in rb) # False (evicted) print("first lap:", list(rb)) # [2, 3, 4, 5] print("second lap:", list(rb)) # [2, 3, 4, 5] — __iter__ reset # isinstance check against the ABC print("is Iterator:", isinstance(rb, collections.abc.Iterator)) # True print("is Iterable:", isinstance(rb, collections.abc.Iterable)) # True # Manual next() usage iter(rb) print("next:", next(rb)) # 2 print("next:", next(rb)) # 3 rb.push(99) # mutate mid-iteration print("next:", next(rb)) # 4 # capacity is immutable try: rb._state.set_capacity(10) except AttributeError as exc: print("immutability guard:", exc) ``` **How it fits together:** `collections.abc.Iterator` is the correct registration target. Inheriting from it (rather than just spelling out the two methods) means `isinstance(rb, collections.abc.Iterator)` returns `True` for free, and the ABC's mixin-provided `__iter__` is overridden so the same object acts as both the iterable and the iterator — the standard Python pattern for buffers and streams. `_RingBufferState` separates *identity* (the immutable `capacity` integer, guarded by `__post_init__` and a raising setter) from *state* (the ordinary `list`). This models the shallow-vs-deep copy and mutable-aliasing concerns cleanly: copying a `RingBuffer` would share the inner list unless an explicit deep-copy of `contents` is performed. `__iter__` resets `_cursor` each time, so `list(rb)` can be called repeatedly without exhausting the buffer permanently — each call starts a fresh lap.