INNER CODE UNIT · Python

get_lines_from

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

    def get_lines_from(self, start_idx: int) -> list[tuple[int, str]]:
        with self._lock:
            return [item for item in self._lines if item[0] >= start_idx]

    def wait_for_lines_or_done(
        self, next_idx: int, timeout: float = 0.5
    ) -> tuple[list[tuple[int, str]], bool]:
        """Wait until lines >= next_idx are available or process is marked done."""
        with self._new_line_cond:
            available = [item for item in self._lines if item[0] >= next_idx]
            if available or self._done:
                return available, self._done
            self._new_line_cond.wait(timeout=timeout)
            available = [item for item in self._lines if item[0] >= next_idx]
            return available, self._done


def _drain_stdout_worker(proc: subprocess.Popen, log_buffer: TrainLogBuffer) -> None:

View source record →