# SPDX-FileCopyrightText: 2024-2025 ModelCloud.ai # SPDX-FileCopyrightText: 2024-2025 qubitium@modelcloud.ai # SPDX-License-Identifier: Apache-2.0 # Contact: qubitium@modelcloud.ai, x.com/qubitium import datetime import os import re import sys import threading import time from contextlib import contextmanager, nullcontext from dataclasses import dataclass, replace from enum import Enum from functools import lru_cache, wraps from typing import Iterable, Optional, Union, TYPE_CHECKING, Sequence from warnings import warn from . import LogBar from .logbar import ( _running_in_notebook_environment, attach_progress_bar, detach_progress_bar, _record_progress_activity, mark_progress_bar_dirty, render_lock, render_progress_stack, _log_headless_state_once, ) if TYPE_CHECKING: # pragma: no cover - type hints without runtime import cycle from .logbar import LogBar as LogBarType from .drawing import ( ANSI_BOLD_RESET, ANSI_RESET, BLOCK_PARTIAL_CHARS, SUBCELL_RESOLUTION, TITLE_BASE_COLOR, TITLE_HIGHLIGHT_COLOR, CellBarRenderer, cached_display_atoms, strip_ansi, truncate_ansi, visible_length, ) from .terminal import RenderBackendState, render_backend_state, terminal_size from .util import auto_iterable logger = LogBar.shared() @contextmanager def _fallback_nullcontext(): """Small local stand-in for environments without `nullcontext()` support.""" yield def _safe_nullcontext(_nullcontext=nullcontext, _fallback=_fallback_nullcontext): """Return a working nullcontext object across supported runtimes.""" if callable(_nullcontext): return _nullcontext() return _fallback() def _render_locked(method): """Serialize one progress-bar API call through the shared render lock.""" @wraps(method) def wrapper(self, *args, **kwargs): # In headless/CI mode this progress bar does not produce visual output, # so the render lock is unnecessary for progress-bar API calls. if getattr(self, "_headless", False): return method(self, *args, **kwargs) context, _ = self._render_lock_context() with context: return method(self, *args, **kwargs) return wrapper @lru_cache(maxsize=1) def _env_animation_enabled() -> bool: """Read the global animation toggle once per process.""" value = os.environ.get("LOGBAR_ANIMATION", "1") return str(value).strip().lower() not in {"0", "false", "off", "no"} def _normalize_output_interval(value: Optional[Union[str, int]]) -> int: """Normalize logical progress render intervals to a safe integer >= 1.""" if value is None: return 1 try: normalized = int(str(value).strip()) except (TypeError, ValueError): return 1 return max(1, normalized) @lru_cache(maxsize=1) def _env_progress_output_interval() -> int: """Global default for how many logical progress updates occur per render.""" return _normalize_output_interval(os.environ.get("LOGBAR_PROGRESS_OUTPUT_INTERVAL", "1")) def _fg_256(code: int) -> str: """Build a 256-color foreground SGR escape.""" return f"\033[38;5;{code}m" def _fg_rgb(red: int, green: int, blue: int) -> str: """Build a truecolor foreground SGR escape.""" return f"\033[38;2;{red};{green};{blue}m" def _clamp_rgb(value: int) -> int: """Clamp one RGB channel to the ANSI truecolor range.""" return max(0, min(255, value)) def _resolve_color(color: Optional[str]) -> str: """Normalize named, numeric, hex, or raw ANSI colors into SGR escapes.""" if not color: return "" if color.startswith("\033["): return color named = _STYLE_COLOR_NAMES.get(color.lower()) if isinstance(color, str) else None if named: return named if isinstance(color, str) and color.isdigit(): return _fg_256(int(color)) if isinstance(color, str): hex_match = re.fullmatch(r"#?([0-9a-fA-F]{3}|[0-9a-fA-F]{6})", color) if hex_match: hex_value = hex_match.group(1) if len(hex_value) == 3: r, g, b = (int(ch * 2, 16) for ch in hex_value) else: r = int(hex_value[0:2], 16) g = int(hex_value[2:4], 16) b = int(hex_value[4:6], 16) return _fg_rgb(_clamp_rgb(r), _clamp_rgb(g), _clamp_rgb(b)) return str(color) @dataclass(frozen=True) class ProgressStyle: """Immutable description of how a progress bar should be drawn.""" name: str fill_char: str = "█" empty_char: str = "-" fill_ramp: Sequence[str] = BLOCK_PARTIAL_CHARS fill_colors: Sequence[str] = () empty_color: str = "" gradient: bool = False head_char: Optional[str] = None head_color: Optional[str] = None def with_fill_char(self, char: str) -> "ProgressStyle": """Return a copy with a different fully filled glyph.""" return replace(self, fill_char=char) def with_empty_char(self, char: str) -> "ProgressStyle": """Return a copy with a different empty-cell glyph.""" return replace(self, empty_char=char) def with_fill_ramp(self, chars: Optional[Sequence[str]]) -> "ProgressStyle": """Return a copy with a new partial-fill ramp.""" return replace(self, fill_ramp=tuple(chars) if chars is not None else ()) def with_colors( self, fill: Optional[Sequence[str]] = None, empty: Optional[str] = None, gradient: Optional[bool] = None, head_color: Optional[str] = None, ) -> "ProgressStyle": """Return a copy with updated palette, empty, or head colors.""" style = self if fill is not None: style = replace(style, fill_colors=tuple(fill)) if empty is not None: style = replace(style, empty_color=empty) if gradient is not None: style = replace(style, gradient=gradient) if head_color is not None: style = replace(style, head_color=head_color) return style def with_head_char(self, char: Optional[str]) -> "ProgressStyle": """Return a copy with a new head glyph.""" return replace(self, head_char=char) def render(self, filled: int, empty: int) -> tuple[str, str]: """Render the bar using whole filled and empty cells.""" return self.render_units( total_cells=filled + empty, filled_units=filled * SUBCELL_RESOLUTION, ) def render_units( self, total_cells: int, filled_units: int, *, units_per_cell: int = SUBCELL_RESOLUTION, ) -> tuple[str, str]: """Render the bar using sub-cell units for smoother progress.""" renderer = self._bar_renderer(units_per_cell=units_per_cell) result = renderer.render_units( total_cells=total_cells, filled_units=filled_units, ) plain = result.plain width = max(0, int(total_cells)) resolution = max(1, int(units_per_cell)) total_units = width * resolution filled_units = max(0, min(int(filled_units), total_units)) full_cells, partial_units = divmod(filled_units, resolution) occupied_cells = min(width, full_cells + (1 if partial_units else 0)) rendered = self._render_colored_bar(plain, occupied_cells) return plain, rendered def _render_colored_bar(self, plain: str, occupied_cells: int) -> str: """Apply fill, head, and empty colors to an already-rasterized bar.""" if not self.fill_colors and not self.empty_color and not self.head_color: return plain width = len(plain) if width == 0: return "" segments: list[str] = [] head_idx = occupied_cells - 1 if self.head_color and occupied_cells > 0 else -1 fill_end = head_idx if head_idx >= 0 else occupied_cells palette = self.fill_colors if palette and fill_end > 0: fill_plain = plain[:fill_end] palette_len = len(palette) if self.gradient and palette_len > 1 and occupied_cells > 1: scale_n = occupied_cells - 1 scale_d = palette_len - 1 start = 0 p = 0 while start < fill_end: end = ((p + 1) * scale_n + scale_d - 1) // scale_d if end > fill_end: end = fill_end if end <= start: end = start + 1 color = palette[p] if color: segments.append(color) segments.append(fill_plain[start:end]) if color: segments.append(ANSI_RESET) start = end p += 1 elif palette_len == 1: color = palette[0] if color: segments.append(color) segments.append(fill_plain) segments.append(ANSI_RESET) else: segments.append(fill_plain) else: for idx in range(fill_end): color = palette[idx % palette_len] if color: segments.append(color) segments.append(fill_plain[idx]) segments.append(ANSI_RESET) else: segments.append(fill_plain[idx]) elif fill_end > 0: segments.append(plain[:fill_end]) if head_idx >= 0: segments.append(self.head_color) segments.append(plain[head_idx]) segments.append(ANSI_RESET) if occupied_cells < width: empty_color = self.empty_color if empty_color: segments.append(empty_color) segments.append(plain[occupied_cells:]) segments.append(ANSI_RESET) else: segments.append(plain[occupied_cells:]) return "".join(segments) @lru_cache(maxsize=4) def _bar_renderer(self, units_per_cell: int = SUBCELL_RESOLUTION) -> CellBarRenderer: """Build the low-level bar renderer for this style.""" return CellBarRenderer( fill_char=self.fill_char, empty_char=self.empty_char, head_char=self.head_char, partial_chars=self.fill_ramp, units_per_cell=units_per_cell, ) def _select_color(self, idx: int, total: int) -> str: """Choose the fill color for one occupied bar cell.""" if not self.fill_colors: return "" palette = self.fill_colors if self.gradient and len(palette) > 1 and total > 1: pos = idx / (total - 1) palette_index = min(int(pos * (len(palette) - 1)), len(palette) - 1) return palette[palette_index] return palette[idx % len(palette)] def _register_default_styles() -> dict[str, ProgressStyle]: """Create the built-in named progress styles shipped with LogBar.""" styles = {} def add(style: ProgressStyle): """Register one built-in style by name in the local style map.""" styles[style.name] = style add( ProgressStyle( name="emerald_glow", fill_char="█", empty_char="░", fill_colors=( _STYLE_COLOR_NAMES["emerald"], _STYLE_COLOR_NAMES["spring"], ), empty_color=_STYLE_COLOR_NAMES["slate"], gradient=True, head_char="█", head_color=_STYLE_COLOR_NAMES["mint"], ) ) add( ProgressStyle( name="sunset", fill_char="▉", empty_char="·", fill_colors=( _STYLE_COLOR_NAMES["amber"], _STYLE_COLOR_NAMES["peach"], _STYLE_COLOR_NAMES["rose"], ), empty_color=_STYLE_COLOR_NAMES["evening"], gradient=True, head_char="▉", head_color=_STYLE_COLOR_NAMES["carnation"], ) ) add( ProgressStyle( name="ocean", fill_char="▓", empty_char="░", fill_colors=( _STYLE_COLOR_NAMES["deep_sea"], _STYLE_COLOR_NAMES["aqua"], _STYLE_COLOR_NAMES["foam"], ), empty_color=_STYLE_COLOR_NAMES["slate"], gradient=True, head_char="▓", head_color=_STYLE_COLOR_NAMES["foam"], ) ) add( ProgressStyle( name="mono", fill_char="█", empty_char="-", ) ) add( ProgressStyle( name="matrix", fill_char="▮", empty_char="·", fill_colors=( _STYLE_COLOR_NAMES["emerald"], _STYLE_COLOR_NAMES["matrix"], ), empty_color=_STYLE_COLOR_NAMES["charcoal"], gradient=False, head_char="▮", head_color=_STYLE_COLOR_NAMES["lime"], ) ) return styles _STYLE_COLOR_NAMES = { "emerald": _fg_rgb(47, 107, 85), "spring": _fg_rgb(72, 160, 120), "mint": _fg_rgb(152, 214, 173), "slate": _fg_256(240), "amber": _fg_256(214), "peach": _fg_256(217), "rose": _fg_256(211), "carnation": _fg_256(205), "evening": _fg_256(236), "deep_sea": _fg_256(24), "aqua": _fg_256(45), "foam": _fg_256(122), "matrix": _fg_256(34), "charcoal": _fg_256(237), "lime": _fg_256(118), } _STYLE_LOCK = threading.RLock() _PROGRESS_STYLES = _register_default_styles() _DEFAULT_STYLE_NAME = "emerald_glow" def progress_style_names() -> list[str]: """Return the registered style names in stable sorted order.""" with _STYLE_LOCK: return sorted(_PROGRESS_STYLES) def get_progress_style(style: Union[str, ProgressStyle]) -> ProgressStyle: """Resolve either a style name or an already-built style object.""" if isinstance(style, ProgressStyle): return style with _STYLE_LOCK: if style in _PROGRESS_STYLES: return _PROGRESS_STYLES[style] raise KeyError(f"Unknown progress style '{style}'. Available styles: {', '.join(progress_style_names())}") def register_progress_style(style: ProgressStyle) -> None: """Register or replace a named progress style.""" with _STYLE_LOCK: _PROGRESS_STYLES[style.name] = style # TODO FIXME: what does this do exactly? class ProgressBarWarning(Warning): """Warning subclass used for compatibility with older cleanup behavior.""" def __init__(self, msg, fp_write=None, *a, **k): """Optionally emit the warning immediately to a file-like writer.""" if fp_write is not None: fp_write("\n" + self.__class__.__name__ + ": " + str(msg).rstrip() + '\n') else: super().__init__(msg, *a, **k) class RenderMode(str, Enum): """Control whether progress bars redraw automatically during iteration.""" AUTO = "AUTO" # pb will auto draw() at the START of each itereation MANUAL = "MANUAL" # pb will not call draw() in each iteration and user must call draw() class ProgressBar: """Deterministic progress bar that can render inline or in a shared stack.""" @classmethod def available_styles(cls) -> list[str]: """List the names of all registered progress styles.""" return progress_style_names() @classmethod def register_style(cls, style: ProgressStyle) -> None: """Register a style through the class-level convenience API.""" register_progress_style(style) @classmethod def set_default_style(cls, style: Union[str, ProgressStyle]) -> ProgressStyle: """Set the process-wide default style for new progress bars.""" with _STYLE_LOCK: if isinstance(style, ProgressStyle): register_progress_style(style) resolved = style else: resolved = get_progress_style(style) global _DEFAULT_STYLE_NAME _DEFAULT_STYLE_NAME = resolved.name return resolved @classmethod def default_style(cls) -> ProgressStyle: """Return the current process-wide default style.""" with _STYLE_LOCK: return get_progress_style(_DEFAULT_STYLE_NAME) def __init__( self, iterable: Union[Iterable, int, dict, set], owner: Optional["LogBarType"] = None, *, output_interval: Optional[int] = None, ): """Create a progress bar around an iterable or a total step count.""" self._iterating = False # state: in init or active iteration self._render_mode = RenderMode.AUTO self._title = "" self._title_plain = "" self._title_width = 0 self._title_padding = "" self._subtitle = "" self._subtitle_plain = "" self._subtitle_width = 0 self._subtitle_padding = "" self._style = self.default_style() self._style_name = self._style.name self.closed = False # active state # max info length over the life ot the pb self.max_title_len = 0 self.max_subtitle_len = 0 # auto convert simple types into iterable auto_iter = auto_iterable(iterable) self.iterable = auto_iter if auto_iter else iterable self.bar_length = 0 self.current_iter_step = 0 self.time = time.time() self._title_animation_start = self.time self._title_animation_period = 0.1 self._output_interval = _normalize_output_interval( _env_progress_output_interval() if output_interval is None else output_interval ) self._last_output_step: Optional[int] = None # Auto-detect headless/AI-agent/notebook/CI environments and suppress # high-frequency redraws so logs do not get polluted with terminal # control sequences in remote or non-interactive shells. _backend_state = render_backend_state( stream=sys.stdout, size_provider=terminal_size, notebook=_running_in_notebook_environment(), ) self._headless = _backend_state.headless if self._headless: _log_headless_state_once(_backend_state) self.ui_show_left_steps = True # show [1 of 100] on left side self.ui_show_left_steps_offset = 0 self._owner_logger = owner self._attached_logger: Optional["LogBarType"] = None self._attached = False self._last_rendered_line = "" self._logbar_dirty_tracked = True self._next_title_refresh_at = 0.0 def _update_title_cache(self, title: str) -> None: """Cache plain-text and width metadata for the current title.""" self._title = title self._title_plain = strip_ansi(title) self._title_width = visible_length(title) self._title_padding = " " * max(0, self.max_title_len - self._title_width) def _update_subtitle_cache(self, subtitle: str) -> None: """Cache plain-text and width metadata for the current subtitle.""" self._subtitle = subtitle self._subtitle_plain = strip_ansi(subtitle) self._subtitle_width = visible_length(subtitle) self._subtitle_padding = " " * max(0, self.max_subtitle_len - self._subtitle_width) def _mark_dirty(self): """Mark the bar for stack redraw if it is currently attached.""" if self._attached: mark_progress_bar_dirty(self) return self @_render_locked def _tick_background_refresh(self, now: float) -> bool: """Advance time-based title animation state for shared refresh loops.""" if not self._attached or not self._title or not self._should_animate_title(): self._next_title_refresh_at = 0.0 return False period = max(self._title_animation_period, 1e-6) if self._next_title_refresh_at <= 0: self._next_title_refresh_at = now + period return False if now < self._next_title_refresh_at: return False self._next_title_refresh_at = now + period return True @_render_locked def set(self, show_left_steps: Optional[bool] = None, left_steps_offset: Optional[int] = None, ): """Update left-side step labels without changing progress state.""" if show_left_steps is not None: self.ui_show_left_steps = show_left_steps if left_steps_offset is not None: self.ui_show_left_steps_offset = left_steps_offset return self._mark_dirty() @_render_locked def style(self, style: Union[str, ProgressStyle]): """Swap the active named or custom drawing style.""" resolved = get_progress_style(style) self._style = resolved self._style_name = resolved.name return self._mark_dirty() @_render_locked def output_interval(self, interval: int): """Render after at least `interval` logical progress updates.""" self._output_interval = _normalize_output_interval(interval) return self._mark_dirty() @_render_locked def fill(self, fill: Union[str, ProgressStyle] = "█", empty: Optional[str] = None): """Override the fill and optional empty glyphs for this bar.""" if isinstance(fill, ProgressStyle): return self.style(fill) if isinstance(fill, str): with _STYLE_LOCK: if fill in _PROGRESS_STYLES: return self.style(fill) if not isinstance(fill, str) or not fill: raise ValueError("fill must be a non-empty string or a named style") previous_style = self._style style = previous_style.with_fill_char(fill) if empty is not None: style = style.with_empty_char(empty) if previous_style.head_char is None: style = style.with_head_char(None) elif previous_style.head_char == previous_style.fill_char: style = style.with_head_char(fill) self._style = style self._style_name = style.name return self._mark_dirty() @_render_locked def colors( self, fill: Optional[Union[str, Sequence[str]]] = None, empty: Optional[str] = None, gradient: Optional[bool] = None, head: Optional[str] = None, ): """Override fill, empty, gradient, or head colors for this bar.""" style = self._style if fill is not None: if isinstance(fill, (list, tuple)): palette = tuple(_resolve_color(c) for c in fill if c) else: resolved = _resolve_color(fill) palette = (resolved,) if resolved else () default_gradient = gradient if gradient is not None else (len(palette) > 1) style = style.with_colors(fill=palette, gradient=default_gradient) elif gradient is not None: style = style.with_colors(gradient=gradient) if empty is not None: style = style.with_colors(empty=_resolve_color(empty)) if head is not None: style = style.with_head_char(style.head_char or style.fill_char) style = style.with_colors(head_color=_resolve_color(head)) self._style = style self._style_name = style.name return self._mark_dirty() @_render_locked def head(self, char: Optional[str] = None, color: Optional[str] = None): """Configure the moving head glyph and optional head color.""" style = self._style if char is not None: if char: style = style.with_head_char(char) else: style = style.with_head_char(None) if color is not None: style = style.with_colors(head_color=_resolve_color(color)) self._style = style self._style_name = style.name return self._mark_dirty() @_render_locked def title(self, title: str): """Set the title shown before the bar and update width tracking.""" if title == self._title: return self if self._iterating and self._render_mode != RenderMode.MANUAL: logger.warn("ProgressBar: Title should not be updated after iteration has started unless in `manual` render mode.") if self._headless: self._title = title self._title_plain = title return self title_len = visible_length(title) if title_len > self.max_title_len: self.max_title_len = title_len previous_title = self._title self._update_title_cache(title) # Only reset the animation clock when transitioning from no title to # an initial title. For dynamic titles (updated every frame) we want # to preserve the elapsed time so the highlight keeps sweeping across # the text instead of snapping back to the first character. if not previous_title and title: self._title_animation_start = time.time() self._next_title_refresh_at = 0.0 return self._mark_dirty() @_render_locked def subtitle(self, subtitle: str): """Set the subtitle shown between the title and the bar.""" if subtitle == self._subtitle: return self if self._iterating and self._render_mode != RenderMode.MANUAL: logger.warn("ProgressBar: Sub-title should not be updated after iteration has started unless in `manual` render mode.") if self._headless: self._subtitle = subtitle self._subtitle_plain = subtitle return self subtitle_len = visible_length(subtitle) if subtitle_len > self.max_subtitle_len: self.max_subtitle_len = subtitle_len self._update_subtitle_cache(subtitle) return self._mark_dirty() # set render mode @_render_locked def mode(self, mode: RenderMode): """Switch between automatic and manual redraw modes.""" self._render_mode = mode return self._mark_dirty() @_render_locked def auto(self): """Enable redraws on each logical iteration step.""" self._render_mode = RenderMode.AUTO return self._mark_dirty() @_render_locked def manual(self ): """Disable automatic redraws so callers drive rendering explicitly.""" self._render_mode = RenderMode.MANUAL return self._mark_dirty() def _render_lock_context(self): """Return a context manager for the shared render lock when available.""" lock_provider = render_lock if callable(render_lock) else None if lock_provider is None: return _safe_nullcontext(), False try: lock_obj = lock_provider() except Exception: return _safe_nullcontext(), False if lock_obj is None: return _safe_nullcontext(), False if hasattr(lock_obj, "__enter__") and hasattr(lock_obj, "__exit__"): return lock_obj, True acquire = getattr(lock_obj, "acquire", None) release = getattr(lock_obj, "release", None) if callable(acquire) and callable(release): @contextmanager def _managed_lock(): """Normalize bare acquire/release locks into a context manager.""" acquire() try: yield lock_obj finally: release() return _managed_lock(), True return _safe_nullcontext(), False def _fallback_detach_registry(self) -> None: """Best-effort detach path for interpreter shutdown edge cases.""" try: from . import logbar as logbar_module except Exception: return bars = getattr(logbar_module, "_attached_progress_bars", None) if not isinstance(bars, list): return state_lock = getattr(logbar_module, "_STATE_LOCK", None) lock_context = _safe_nullcontext() if state_lock is not None: if hasattr(state_lock, "__enter__") and hasattr(state_lock, "__exit__"): lock_context = state_lock else: acquire = getattr(state_lock, "acquire", None) release = getattr(state_lock, "release", None) if callable(acquire) and callable(release): @contextmanager def _state_lock_ctx(): """Wrap the shared state lock in a context manager interface.""" acquire() try: yield finally: release() lock_context = _state_lock_ctx() with lock_context: if self in bars: try: bars.remove(self) except ValueError: pass @_render_locked def attach(self, logger: Optional["LogBarType"] = None): """Attach the bar to a logger so it renders in the shared stack.""" if self.closed: return self if self._attached: return self if self._headless: return self target_logger = logger or self._owner_logger or LogBar.shared() self._attached_logger = target_logger self._owner_logger = target_logger attach_progress_bar(self) self._attached = True self._next_title_refresh_at = 0.0 render_fn = render_progress_stack if callable(render_progress_stack) else None context, lock_held = self._render_lock_context() with context: if render_fn is not None: try: render_fn(lock_held=lock_held) except Exception: if not sys.is_finalizing(): raise return self @_render_locked def detach(self): """Detach the bar from stack management and stop future redraws.""" if not self._attached: return self detach_fn = detach_progress_bar if callable(detach_progress_bar) else None render_fn = render_progress_stack if callable(render_progress_stack) else None context, lock_held = self._render_lock_context() exc_to_raise: Optional[BaseException] = None detached_successfully = False try: with context: if detach_fn is not None: try: detach_fn(self) except Exception as exc: if sys.is_finalizing(): self._fallback_detach_registry() else: exc_to_raise = exc else: self._fallback_detach_registry() if render_fn is not None: try: render_fn(lock_held=lock_held) except Exception as exc: if not sys.is_finalizing() and exc_to_raise is None: exc_to_raise = exc if exc_to_raise is not None: raise exc_to_raise detached_successfully = True finally: if detached_successfully: self._attached = False self._attached_logger = None self._last_rendered_line = "" self._next_title_refresh_at = 0.0 return self def _render_position(self) -> int: """Progress coordinate used for render throttling.""" return self.step() def _should_render(self, force: bool = False, allow_repeat: bool = False) -> bool: """Return whether this logical state change should emit a new frame.""" if force: return True if not self._last_rendered_line: return True last_output_step = self._last_output_step if last_output_step is None: return True current_step = self._render_position() if allow_repeat and current_step <= last_output_step: return True return (current_step - last_output_step) >= self._output_interval def _resolve_rendered_line( self, columns: int, force: bool = False, allow_repeat: bool = False, backend_state: Optional[RenderBackendState] = None, style_enabled: Optional[bool] = None, ) -> Optional[str]: """Return the next rendered line, honoring draw throttling rules.""" if not self._should_render(force=force, allow_repeat=allow_repeat): return None rendered_line = self._render_snapshot( columns, backend_state=backend_state, style_enabled=style_enabled, ) self._last_output_step = self._render_position() return rendered_line def _render_backend_state(self, backend_state: Optional[RenderBackendState] = None) -> RenderBackendState: """Resolve backend capabilities for the current draw call.""" if backend_state is not None: return backend_state return render_backend_state( stream=sys.stdout, size_provider=terminal_size, notebook=_running_in_notebook_environment(), ) @_render_locked def draw(self, force: bool = False): """Render the current progress state either inline or in the stack.""" if self._headless: return # Even skipped draws count as activity. This prevents the background # refresher from redrawing the same bar immediately and undoing the # throttle when progress is advancing quickly. _record_progress_activity() backend_state = render_backend_state( stream=sys.stdout, size_provider=terminal_size, notebook=_running_in_notebook_environment(), ) columns = backend_state.columns render_fn = render_progress_stack if callable(render_progress_stack) else None style_enabled = backend_state.supports_styling if self._attached and render_fn is not None else backend_state.supports_ansi rendered_line = self._resolve_rendered_line( columns, force=force, allow_repeat=True, backend_state=backend_state, style_enabled=style_enabled, ) if rendered_line is None: return context, lock_held = self._render_lock_context() if not self._attached or render_fn is None: with context: try: # Clear the whole line before drawing the new inline frame # so characters from a previous longer frame do not linger. clear_line = "\033[2K" if backend_state.supports_ansi else "" print(f'\r{clear_line}{rendered_line}', end='', flush=True) except Exception: if not sys.is_finalizing(): raise return with context: try: render_fn( lock_held=lock_held, precomputed={self: rendered_line}, columns_hint=columns, backend_state=backend_state, ) except Exception: if not sys.is_finalizing(): raise def calc_time(self, iteration, total_steps=None): """Return elapsed and estimated remaining time for the progress line.""" if total_steps is None: total_steps = len(self) used_time = int(time.time() - self.time) formatted_time = str(datetime.timedelta(seconds=used_time)) completed = iteration if self._render_mode == RenderMode.AUTO and iteration > 0: completed = iteration - 1 remaining = str(datetime.timedelta(seconds=int((used_time / max(completed, 1)) * total_steps))) return f"{formatted_time} / {remaining}" def _render_snapshot( self, columns: Optional[int] = None, *, backend_state: Optional[RenderBackendState] = None, style_enabled: Optional[bool] = None, ) -> str: """Build a full rendered line snapshot for the current progress state.""" backend_state = self._render_backend_state(backend_state) if columns is None: columns = backend_state.columns supports_styling = backend_state.supports_styling if style_enabled is None else bool(style_enabled) total_steps = len(self) effective_total = total_steps if total_steps else 1 current_step = self.step() percent_num = current_step / float(effective_total) percent = ("{0:.1f}").format(100 * percent_num) log_text = f"{self.calc_time(current_step, total_steps)} [{current_step}/{total_steps}] {percent}%" pre_bar_size = 0 if self._title: pre_bar_size += self.max_title_len + 1 if self._subtitle: pre_bar_size += self.max_subtitle_len + 1 if self.ui_show_left_steps: left_current = current_step - self.ui_show_left_steps_offset left_total = total_steps - self.ui_show_left_steps_offset self.ui_show_left_steps_text = f"[{left_current} of {left_total}] " self.ui_show_left_steps_text_max_len = len(self.ui_show_left_steps_text) pre_bar_size += self.ui_show_left_steps_text_max_len # Titles and subtitles are cached separately so repeated redraws only # pay for width math when the text actually changes. padding = self._title_padding + self._subtitle_padding available_columns = columns if columns is not None and columns > 0 else 0 log_text_width = len(log_text) bar_length = max(0, available_columns - pre_bar_size - log_text_width - 2) if available_columns else 0 self.bar_length = bar_length total_units = bar_length * SUBCELL_RESOLUTION if current_step >= effective_total and total_units > 0: filled_units = total_units elif total_units > 0: # Keep very early progress visible by promoting tiny non-zero # values to the first sub-cell instead of leaving the bar empty. filled_units = int((current_step * total_units) / effective_total) if 0 < current_step < effective_total and filled_units == 0: filled_units = 1 else: filled_units = 0 bar_plain, bar_rendered = self._style.render_units( total_cells=bar_length, filled_units=filled_units, ) if not supports_styling: bar_rendered = bar_plain plain_width = pre_bar_size + bar_length + log_text_width + 2 rendered_line = self._render_line( bar_plain=bar_plain, bar_rendered=bar_rendered, log_text=log_text, pre_bar_padding=padding, columns=columns, supports_styling=supports_styling, animate_title=self._should_animate_title( backend_state=backend_state, style_enabled=supports_styling, ), plain_width=plain_width, ) self._last_rendered_line = rendered_line return rendered_line def _render_line( self, bar_plain: str, log_text: str, pre_bar_padding: str = "", columns: Optional[int] = None, bar_rendered: Optional[str] = None, supports_styling: bool = True, animate_title: Optional[bool] = None, plain_width: Optional[int] = None, ) -> str: """Assemble the final visible line from title, bar, and status fields.""" segments_plain = [] segments_rendered = [] def append_segment(text: str, rendered: Optional[str] = None, plain: Optional[str] = None): """Accumulate plain and styled output in lockstep for width accounting.""" segments_plain.append(plain if plain is not None else text) segments_rendered.append(rendered if rendered is not None else text) if animate_title is None: animate_title = supports_styling and self._should_animate_title(style_enabled=supports_styling) if self._title: if animate_title: animated_title = self._animated_text(self._title) append_segment(self._title, animated_title, self._title_plain) else: append_segment(self._title, plain=self._title_plain) append_segment(" ") if self._subtitle: subtitle_text = f"{self._subtitle} " append_segment(subtitle_text, plain=f"{self._subtitle_plain} ") if pre_bar_padding: append_segment(pre_bar_padding) if self.ui_show_left_steps: left_steps_text = self.ui_show_left_steps_text if not self._title and animate_title: append_segment(left_steps_text, self._animated_text(left_steps_text)) else: append_segment(left_steps_text) # Keep a plain-text mirror of the line so width measurement and # truncation stay correct even when the rendered form contains ANSI. plain_bar_segment = f"{bar_plain}| {log_text}" rendered_bar_segment = ( f"{bar_rendered}| {log_text}" if supports_styling and bar_rendered is not None else None ) append_segment(plain_bar_segment, rendered_bar_segment) plain_out = ''.join(segments_plain) rendered_out = ''.join(segments_rendered) if supports_styling else plain_out if columns is not None: if plain_width is None: plain_width = visible_length(plain_out) if plain_width > columns: rendered_out = ( self._truncate_ansi(rendered_out, columns) if supports_styling else strip_ansi(self._truncate_ansi(plain_out, columns)) ) elif plain_width < columns: pad = " " * (columns - plain_width) plain_out += pad rendered_out += pad return rendered_out def _animated_text(self, text: str) -> str: """Apply the sweeping highlight effect to one title-like string.""" if not text: return "" period = max(self._title_animation_period, 1e-6) elapsed = time.time() - self._title_animation_start text_len = visible_length(text) if text_len == 0: return text # pause for a few beats once the highlight reaches the end to calm flicker pause_steps = 5 cycle_length = text_len + pause_steps cycle_step = int(elapsed / period) % cycle_length highlight_idx = cycle_step if cycle_step < text_len else None parts = [TITLE_BASE_COLOR] visible_idx = 0 for is_ansi, token, width in cached_display_atoms(text): if is_ansi: parts.append(token) continue highlight_active = ( highlight_idx is not None and width > 0 and visible_idx <= highlight_idx < (visible_idx + width) ) if highlight_active: parts.append(TITLE_HIGHLIGHT_COLOR) parts.append(token) parts.append(ANSI_BOLD_RESET) parts.append(TITLE_BASE_COLOR) else: parts.append(token) visible_idx += width parts.append(ANSI_RESET) return ''.join(parts) def _truncate_ansi(self, text: str, limit: int) -> str: """Delegate ANSI-aware truncation through the shared drawing helpers.""" return truncate_ansi(text, limit) def _should_animate_title( self, backend_state: Optional[RenderBackendState] = None, style_enabled: Optional[bool] = None, ) -> bool: """Decide whether title animation is supported on the active backend.""" if not _env_animation_enabled(): return False try: state = self._render_backend_state(backend_state) except Exception: return False if state.headless: return False if style_enabled is None: style_enabled = state.supports_styling return bool(style_enabled) and (state.supports_cursor or state.notebook) def __bool__(self): """Mirror the truthiness of the wrapped iterable when defined.""" if self.iterable is None: raise TypeError('bool() undefined when iterable == total == None') return bool(self.iterable) def __len__(self): """Return the total number of logical steps when it can be inferred.""" return ( self.iterable.shape[0] if hasattr(self.iterable, "shape") else len(self.iterable) if hasattr(self.iterable, "__len__") else self.iterable.__length_hint__() if hasattr(self.iterable, "__length_hint__") else getattr(self, "total", None)) # TODO FIXME: I have no cluse why the try/catch is catching nothing here @_render_locked def __reversed__(self): """Reverse the underlying iterable and reuse the normal iterator path.""" try: original = self.iterable except AttributeError: raise TypeError("'progress' object is not reversible") else: self.iterable = reversed(self.iterable) return self.__iter__() finally: self.iterable = original @_render_locked def __contains__(self, item): """Proxy containment checks to the wrapped iterable when possible.""" contains = getattr(self.iterable, '__contains__', None) return contains(item) if contains is not None else item in self.__iter__() def __enter__(self): """Support `with ProgressBar(...) as pb` usage.""" return self # TODO FIXME: I don't understand the exception here. What are we catching? yield error? def __exit__(self, exc_type, exc_value, traceback): """Ensure the bar closes cleanly when leaving a context manager.""" try: self.close() except AttributeError: # maybe eager thread cleanup upon external error if (exc_type, exc_value, traceback) == (None, None, None): raise # TODO FIXME: what does this do exactly? warn("AttributeError ignored", ProgressBarWarning, stacklevel=2) def __del__(self): """Best-effort cleanup when the bar is garbage-collected.""" self.close() # TODO FIXME: what does this do exactly? where is this `pos` attr magically coming from? I don't see it anywhere @property def _comparable(self): """Expose a stable ordering key compatible with older integrations.""" return abs(getattr(self, "pos", 1 << 31)) def __hash__(self): """Keep instances hashable so they can live in dirty tracking sets.""" return id(self) def step(self) -> int: """Return the current logical progress position.""" return self.current_iter_step @_render_locked def next(self): """Advance progress by one logical step.""" self.current_iter_step += 1 return self def __iter__(self): """Iterate the wrapped iterable while updating and rendering progress.""" iterable = self.iterable for obj in iterable: # update running state if not self._iterating: self.iterating = True self.next() if self._render_mode == RenderMode.AUTO: self.draw() yield obj self.close() return @_render_locked def close(self): """Emit a final frame if needed, then detach the bar from rendering.""" if self.closed: return # Force the last observed progress state to render even when the final # step does not land on the configured output interval, e.g. 15 with an # interval of 10. Without this, the closing detach would erase the bar # before a 100% snapshot is ever emitted. if self.step() > 0 and (self._last_output_step is None or self.step() != self._last_output_step): self.draw(force=True) self.closed = True self.detach() class RollingProgressBar(ProgressBar): """Indeterminate progress indicator with a rolling highlight.""" def __init__( self, owner: Optional["LogBarType"] = None, *, output_interval: Optional[int] = None, interval: float = 0.5, tail_length: int = 4, ): """Create an indeterminate spinner-style progress bar.""" # Honor the global output interval default so callers can throttle # spinner redraws with LOGBAR_PROGRESS_OUTPUT_INTERVAL, while still # defaulting to one frame per tick for smooth local animation. super().__init__( iterable=1, owner=owner, output_interval=( _env_progress_output_interval() if output_interval is None else output_interval ), ) self._interval = max(0.05, float(interval)) self._tail_length = max(1, int(tail_length)) self._phase = 0 self._pending_spinner_steps = 0 self.ui_show_left_steps = False self._next_spinner_refresh_at = 0.0 @_render_locked def attach(self, logger: Optional["LogBarType"] = None): """Attach and seed the next spinner refresh deadline.""" super().attach(logger) self._next_spinner_refresh_at = time.monotonic() + self._interval return self @_render_locked def detach(self): """Detach and clear any pending spinner refresh deadline.""" self._next_spinner_refresh_at = 0.0 return super().detach() @_render_locked def close(self): """Stop the spinner without forcing a terminal 100% frame.""" if self.closed: return self.closed = True self.detach() @_render_locked def pulse(self) -> "RollingProgressBar": """Advance the animation a single frame and redraw immediately.""" self._advance_phase() self._next_spinner_refresh_at = time.monotonic() + self._interval self.draw() return self def _ensure_auto_updates(self) -> None: """Compatibility no-op kept for older spinner APIs.""" return None def _stop_auto_updates(self) -> None: """Compatibility no-op kept for older spinner APIs.""" return None @_render_locked def _tick_background_refresh(self, now: float) -> bool: """Advance spinner animation when its wall-clock interval elapses.""" changed = super()._tick_background_refresh(now) if not self._attached or self.closed: self._next_spinner_refresh_at = 0.0 return changed if self._next_spinner_refresh_at <= 0: self._next_spinner_refresh_at = now + self._interval return changed if now < self._next_spinner_refresh_at: return changed steps = max(1, int((now - self._next_spinner_refresh_at) // self._interval) + 1) # Reset the wall-clock deadline immediately so missed ticks don't pile up. self._next_spinner_refresh_at = now + self._interval # Throttle the spinner's phase advances with the same output interval used # for determinate bars (default from LOGBAR_PROGRESS_OUTPUT_INTERVAL). self._pending_spinner_steps += steps if self._pending_spinner_steps >= self._output_interval: advance = self._pending_spinner_steps self._pending_spinner_steps = 0 self._advance_phase(steps=advance) return True return changed def _advance_phase(self, steps: int = 1) -> None: """Move the spinner head forward by one or more frames.""" self._phase = (self._phase + max(1, int(steps))) % 1_000_000 def _render_position(self) -> int: """Use animation phase instead of step count for redraw throttling.""" return self._phase def _render_snapshot( self, columns: Optional[int] = None, *, backend_state: Optional[RenderBackendState] = None, style_enabled: Optional[bool] = None, ) -> str: """Build one spinner snapshot using elapsed time instead of percent.""" backend_state = self._render_backend_state(backend_state) if columns is None: columns = backend_state.columns supports_styling = backend_state.supports_styling if style_enabled is None else bool(style_enabled) elapsed_seconds = max(0, int(time.time() - self.time)) elapsed = str(datetime.timedelta(seconds=elapsed_seconds)) log_text = f"elapsed {elapsed}".strip() pre_bar_size = 0 if self._title: pre_bar_size += self.max_title_len + 1 if self._subtitle: pre_bar_size += self.max_subtitle_len + 1 padding = self._title_padding + self._subtitle_padding available_columns = columns if columns is not None and columns > 0 else 0 log_text_width = len(log_text) bar_length = max(0, available_columns - pre_bar_size - log_text_width - 2) if available_columns else 0 self.bar_length = bar_length bar_plain, bar_rendered = self._render_animation( bar_length, supports_styling=supports_styling, ) self._last_rendered_line = self._render_line( bar_plain=bar_plain, bar_rendered=bar_rendered, log_text=log_text, pre_bar_padding=padding, columns=columns, supports_styling=supports_styling, animate_title=self._should_animate_title( backend_state=backend_state, style_enabled=supports_styling, ), ) return self._last_rendered_line def _render_animation(self, bar_length: int, *, supports_styling: bool = True) -> tuple[str, str]: """Render the rolling head and trailing tail for the spinner bar.""" if bar_length <= 0: return "", "" fill_char = self._style.fill_char or "█" empty_char = self._style.empty_char or " " head_char = self._style.head_char or fill_char tail = min(self._tail_length, bar_length) if tail <= 0: tail = 1 width = bar_length phase_index = self._phase % width if width else 0 active: dict[int, int] = {} for offset in range(tail): idx = (phase_index - offset) % width if idx not in active: active[idx] = offset plain_chars = [empty_char] * width for idx, offset in active.items(): plain_chars[idx] = head_char if offset == 0 else fill_char bar_plain = ''.join(plain_chars) if not supports_styling or not (self._style.fill_colors or self._style.head_color or self._style.empty_color): return bar_plain, bar_plain segments: list[str] = [] current_color: Optional[str] = None for idx, char in enumerate(plain_chars): color = "" offset = active.get(idx) if offset is not None: if offset == 0 and self._style.head_color: color = self._style.head_color else: palette = self._style.fill_colors if palette: if self._style.gradient and len(palette) > 1: palette_index = min(offset, len(palette) - 1) else: palette_index = idx % len(palette) color = palette[palette_index] elif self._style.empty_color: color = self._style.empty_color if color != current_color: if current_color: segments.append(ANSI_RESET) if color: segments.append(color) current_color = color segments.append(char) if current_color: segments.append(ANSI_RESET) return bar_plain, ''.join(segments)