progress.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/core/progress.py
 
"""
Purpose
-------
Provide reusable progress display for long-running OCT commands.
 
Responsibilities
----------------
- ProgressTracker: scrolling task checklist with per-task progress bar.
- Dashboard: fixed-size terminal display with in-place line updates and a
  background timer thread that refreshes the footer every second.
- Auto-detect TTY vs pipe; Dashboard prints final state on non-TTY.
- Thread-safe for concurrent callers (e.g. linter ThreadPoolExecutor).
 
Diagnostics
-----------
Domain: OCT-CORE
Levels:
    L3 — task lifecycle events
    L4 — tick and render calls
 
Contracts
---------
- ProgressTracker outputs to the provided stream (default: stderr).
- Dashboard outputs to the provided stream (default: stdout).
- quiet=True suppresses all output (for --json modes).
- Dashboard.finish() prints final line state on non-TTY streams.
"""
 
from __future__ import annotations
 
import sys
import threading
import time
from contextlib import contextmanager
from typing import TextIO
 
 
_BAR_WIDTH = 20
 
 
class ProgressTracker:
    """Task checklist with optional per-task progress bar."""
 
    def __init__(
        self,
        tasks: list[str],
        *,
        title: str = "",
        stream: TextIO | None = None,
        quiet: bool = False,
    ) -> None:
        self._tasks = list(tasks)
        self._title = title
        self._stream = stream or sys.stderr
        self._quiet = quiet
        self._is_tty = hasattr(self._stream, "isatty") and self._stream.isatty()
 
        self._lock = threading.Lock()
        self._current: str | None = None
        self._progress: int = 0
        self._total: int = 0
        self._task_start: float = 0.0
        self._done: dict[str, tuple[str, str]] = {}
 
        if not self._quiet and self._title:
            self._write_line(self._title)
 
    def start_task(self, name: str, total: int = 0) -> None:
        if self._quiet or name not in self._tasks:
            return
        with self._lock:
            self._current = name
            self._progress = 0
            self._total = total
            self._task_start = time.monotonic()
        if self._is_tty:
            self._render_status()
        else:
            self._write_line(f"  [start] {name}")
 
    def tick(self, n: int = 1) -> None:
        if self._quiet:
            return
        with self._lock:
            self._progress += n
        if self._is_tty:
            self._render_status()
 
    def complete_task(self, name: str, *, summary: str = "") -> None:
        if self._quiet or name not in self._tasks:
            return
        with self._lock:
            self._current = None
        marker = "[x]"
        line = f"  {marker} {name}"
        if summary:
            line += f"  {summary}"
        self._done[name] = ("done", summary)
        if self._is_tty:
            self._clear_status()
        self._write_line(line)
 
    def fail_task(self, name: str, *, summary: str = "") -> None:
        if self._quiet or name not in self._tasks:
            return
        with self._lock:
            self._current = None
        marker = "[!]"
        line = f"  {marker} {name}"
        if summary:
            line += f"  {summary}"
        self._done[name] = ("fail", summary)
        if self._is_tty:
            self._clear_status()
        self._write_line(line)
 
    def skip_task(self, name: str, *, summary: str = "") -> None:
        if self._quiet or name not in self._tasks:
            return
        with self._lock:
            self._current = None
        marker = "[-]"
        line = f"  {marker} {name}"
        if summary:
            line += f"  {summary}"
        self._done[name] = ("skip", summary)
        if self._is_tty:
            self._clear_status()
        self._write_line(line)
 
    @contextmanager
    def task(self, name: str, total: int = 0):
        self.start_task(name, total=total)
        try:
            yield self
            self.complete_task(name)
        except Exception:
            self.fail_task(name)
            raise
 
    def finish(self) -> None:
        if self._quiet:
            return
        if self._is_tty:
            self._clear_status()
        for t in self._tasks:
            if t not in self._done:
                self._write_line(f"  [ ] {t}")
 
    # -- internal rendering --------------------------------------------------
 
    def _write_line(self, text: str) -> None:
        try:
            self._stream.write(text + "\n")
            self._stream.flush()
        except (OSError, ValueError):
            pass
 
    def _render_status(self) -> None:
        with self._lock:
            name = self._current
            progress = self._progress
            total = self._total
            start = self._task_start
        if name is None:
            return
        if total > 0:
            pct = min(progress / total, 1.0)
            filled = int(pct * _BAR_WIDTH)
            bar = "=" * filled + ">" * (1 if filled < _BAR_WIDTH else 0)
            bar = bar.ljust(_BAR_WIDTH)
            text = f"  Running: {name}  [{bar}] {int(pct * 100)}%  {progress}/{total}"
        else:
            elapsed = time.monotonic() - start
            text = f"  Running: {name}  ... elapsed {int(elapsed)}s"
        try:
            self._stream.write(f"\r{text}\033[K")
            self._stream.flush()
        except (OSError, ValueError):
            pass
 
    def _clear_status(self) -> None:
        try:
            self._stream.write("\r\033[K")
            self._stream.flush()
        except (OSError, ValueError):
            pass
 
 
class Dashboard:
    """Fixed-size terminal display with in-place line updates and live timer.
 
    Prints a full template on render(), then overwrites individual lines via
    ANSI cursor movement as data becomes available.  A background daemon thread
    refreshes the footer line every second so elapsed-time counters stay live.
 
    On non-TTY streams the template is NOT printed during render(); instead,
    update_line() silently accumulates changes and finish() prints the final
    state in one shot — clean output for CI and pipes.
    """
 
    def __init__(
        self,
        *,
        stream: TextIO | None = None,
        quiet: bool = False,
    ) -> None:
        self._stream = stream or sys.stdout
        self._quiet = quiet
        self._is_tty = hasattr(self._stream, "isatty") and self._stream.isatty()
        self._lock = threading.Lock()
        self._lines: list[str] = []
        self._line_count: int = 0
        self._footer_line: int = -1
        self._task_name: str | None = None
        self._task_start: float = 0.0
        self._overall_start: float = 0.0
        self._tasks_done: int = 0
        self._tasks_total: int = 0
        self._stop_event = threading.Event()
        self._timer_thread: threading.Thread | None = None
 
    def render(self, lines: list[str], *, footer_line: int = -1,
               tasks_total: int = 0) -> None:
        """Print all lines.  *footer_line* is the index auto-updated by the
        background timer (default: third-from-last)."""
        self._lines = list(lines)
        self._line_count = len(lines)
        self._footer_line = footer_line if footer_line >= 0 else max(self._line_count - 3, 0)
        self._tasks_total = tasks_total
        self._overall_start = time.monotonic()
        if self._quiet:
            return
        if self._is_tty:
            for line in self._lines:
                self._stream.write(line + "\n")
            self._stream.flush()
            self._stop_event.clear()
            self._timer_thread = threading.Thread(
                target=self._timer_loop, daemon=True,
            )
            self._timer_thread.start()
 
    def update_line(self, index: int, text: str) -> None:
        """Replace the line at *index* with *text*."""
        if index < 0 or index >= self._line_count:
            return
        with self._lock:
            self._lines[index] = text
            if self._is_tty and not self._quiet:
                up = self._line_count - index
                self._stream.write(
                    f"\033[{up}A\r{text}\033[K\033[{up}B\r",
                )
                self._stream.flush()
 
    def set_task(self, name: str) -> None:
        """Set current task name and reset the per-task elapsed timer."""
        with self._lock:
            self._task_name = name
            self._task_start = time.monotonic()
        self._update_footer()
 
    def complete_task(self) -> None:
        """Increment done counter and clear the current task name."""
        with self._lock:
            self._tasks_done += 1
            self._task_name = None
 
    def finish(self) -> None:
        """Stop the timer thread and render final state."""
        self._stop_event.set()
        if self._timer_thread:
            self._timer_thread.join(timeout=2)
            self._timer_thread = None
        if self._quiet:
            return
        elapsed = time.monotonic() - self._overall_start
        final = f"  All {self._tasks_total} checks completed in {elapsed:.1f}s"
        self.update_line(self._footer_line, final)
        if not self._is_tty:
            for line in self._lines:
                self._stream.write(line + "\n")
            self._stream.flush()
 
    # -- background timer ----------------------------------------------------
 
    def _timer_loop(self) -> None:
        while not self._stop_event.is_set():
            self._stop_event.wait(1.0)
            if not self._stop_event.is_set():
                self._update_footer()
 
    def _update_footer(self) -> None:
        with self._lock:
            name = self._task_name
            start = self._task_start
            done = self._tasks_done
            total = self._tasks_total
        if name is None:
            return
        elapsed = time.monotonic() - start
        text = f"  [{done}/{total}] {name} ... elapsed {int(elapsed)}s"
        self.update_line(self._footer_line, text)