INNER CODE UNIT · Python

TrainLogBuffer

MakazhanAlpamys/Soup · src/soup_cli/ui/app.py:200

class TrainLogBuffer:
    """Thread-safe bounded ring buffer for subprocess output lines."""

    def __init__(self, maxlen: int = 10000):
        self._maxlen = maxlen
        self._lines: list[tuple[int, str]] = []
        self._lock = threading.Lock()
        self._new_line_cond = threading.Condition(self._lock)
        self._done = False
        self._total_emitted = 0

    def append(self, text: str) -> None:
        with self._new_line_cond:
            idx = self._total_emitted
            self._total_emitted += 1
            self._lines.append((idx, text))
            if len(self._lines) > self._maxlen:
                self._lines.pop(0)

View source record →