Pb draw interval override (#38)

* add env LOGBAR_PROGRESS_OUTPUT_INTERVAL override

Signed-off-by: Qubitium <qubitium@modelcloud.ai>

* fix test

Signed-off-by: Qubitium <qubitium@modelcloud.ai>

* fix test

Signed-off-by: Qubitium <qubitium@modelcloud.ai>

* fix test

Signed-off-by: Qubitium <qubitium@modelcloud.ai>

---------

Signed-off-by: Qubitium <qubitium@modelcloud.ai>

Author: Qubitium

README.md

@@ -24,6 +24,8 @@
 - Built-in styling for progress bar fills, colors, gradients, and head glyphs.
 - Animated progress titles with a subtle sweeping highlight.
   Set `LOGBAR_ANIMATION=0` to disable the highlight animation.
+- Progress output throttling for reducing redraw churn in batch-heavy jobs.
+  Set `LOGBAR_PROGRESS_OUTPUT_INTERVAL=10` to render every 10 logical updates instead of every update.
 - Column-aware table printer with spans, width hints, and `fit` sizing.
 - Zero dependencies; works anywhere Python runs.
 
@@ -99,6 +101,15 @@ for _ in log.pb(500).title("Downloading"):
     time.sleep(0.05)
 ```
 
+When a workload updates progress very frequently, throttle redraw churn globally or per bar:
+
+```py
+for _ in log.pb(500, output_interval=10).title("Quantizing"):
+    time.sleep(0.01)
+```
+
+`output_interval=10` means LogBar will emit a fresh snapshot after roughly every 10 logical progress steps, while still forcing the last pending step to render before the bar closes. Set `LOGBAR_PROGRESS_OUTPUT_INTERVAL=10` to apply the same default process-wide.
+
 Manual mode gives full control when you need to interleave logging and redraws:
 
 ```py

logbar/logbar.py

@@ -319,7 +319,17 @@ def _render_progress_stack_locked(precomputed: Optional[dict] = None, columns_hi
 
         if rendered is None:
             try:
-                rendered = pb._render_snapshot(columns)
+                # Progress bars may throttle redraws. When a bar is not due for
+                # a fresh snapshot, keep its last rendered line in the stack
+                # instead of recomputing it opportunistically on another bar's
+                # redraw.
+                resolve_rendered = getattr(pb, "_resolve_rendered_line", None)
+                if callable(resolve_rendered):
+                    rendered = resolve_rendered(columns)
+                    if rendered is None:
+                        rendered = pb._last_rendered_line or ""
+                else:
+                    rendered = pb._render_snapshot(columns)
             except Exception:  # pragma: no cover - avoid breaking logging on render issues
                 rendered = pb._last_rendered_line or ""
         else:
@@ -565,10 +575,10 @@ def shared(cls, override_logger: Optional[bool] = False):
         return shared_logger
 
 
-    def pb(self, iterable: Iterable):
+    def pb(self, iterable: Iterable, *, output_interval: Optional[int] = None):
         from logbar.progress import ProgressBar
 
-        return ProgressBar(iterable, owner=self).attach(self)
+        return ProgressBar(iterable, owner=self, output_interval=output_interval).attach(self)
 
     def spinner(self, title: str = "", *, interval: float = 0.5, tail_length: int = 4):
         from logbar.progress import RollingProgressBar

logbar/progress.py

@@ -82,6 +82,27 @@ def _env_animation_enabled() -> bool:
     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:
     return f"\033[38;5;{code}m"
 
@@ -374,7 +395,13 @@ def set_default_style(cls, style: Union[str, ProgressStyle]) -> ProgressStyle:
     def default_style(cls) -> ProgressStyle:
         return get_progress_style(_DEFAULT_STYLE_NAME)
 
-    def __init__(self, iterable: Union[Iterable, int, dict, set], owner: Optional["LogBarType"] = None):
+    def __init__(
+        self,
+        iterable: Union[Iterable, int, dict, set],
+        owner: Optional["LogBarType"] = None,
+        *,
+        output_interval: Optional[int] = None,
+    ):
         self._iterating = False # state: in init or active iteration
 
         self._render_mode = RenderMode.AUTO
@@ -398,6 +425,10 @@ def __init__(self, iterable: Union[Iterable, int, dict, set], owner: Optional["L
         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
 
         self.ui_show_left_steps = True # show [1 of 100] on left side
         self.ui_show_left_steps_offset = 0
@@ -424,6 +455,12 @@ def style(self, style: Union[str, ProgressStyle]):
         self._style_name = resolved.name
         return self
 
+    def output_interval(self, interval: int):
+        """Render after at least `interval` logical progress updates."""
+
+        self._output_interval = _normalize_output_interval(interval)
+        return self
+
     def fill(self, fill: Union[str, ProgressStyle] = "█", empty: Optional[str] = None):
         if isinstance(fill, ProgressStyle) or (isinstance(fill, str) and fill in _PROGRESS_STYLES):
             return self.style(fill)
@@ -672,10 +709,50 @@ def detach(self):
 
         return self
 
-    def draw(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:
+        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,
+    ) -> Optional[str]:
+        if not self._should_render(force=force, allow_repeat=allow_repeat):
+            return None
+
+        rendered_line = self._render_snapshot(columns)
+        self._last_output_step = self._render_position()
+        return rendered_line
+
+    def draw(self, force: bool = False):
+        # 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()
         columns, _ = terminal_size()
-        rendered_line = self._render_snapshot(columns)
+        rendered_line = self._resolve_rendered_line(columns, force=force, allow_repeat=True)
+        if rendered_line is None:
+            return
 
         render_fn = render_progress_stack if callable(render_progress_stack) else None
         context, lock_held = self._render_lock_context()
@@ -976,6 +1053,13 @@ def close(self):
         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()
 
@@ -984,7 +1068,10 @@ class RollingProgressBar(ProgressBar):
     """Indeterminate progress indicator with a rolling highlight."""
 
     def __init__(self, owner: Optional["LogBarType"] = None, interval: float = 0.5, tail_length: int = 4):
-        super().__init__(iterable=1, owner=owner)
+        # Spinner animation already has its own wall-clock interval. Keep
+        # output throttling at 1 so the phase animation stays smooth unless a
+        # caller explicitly changes it later.
+        super().__init__(iterable=1, owner=owner, output_interval=1)
         self._interval = max(0.05, float(interval))
         self._tail_length = max(1, int(tail_length))
         self._phase = 0
@@ -1050,6 +1137,9 @@ def _animate_loop(self) -> None:
     def _advance_phase(self) -> None:
         self._phase = (self._phase + 1) % 1_000_000
 
+    def _render_position(self) -> int:
+        return self._phase
+
     def _render_snapshot(self, columns: Optional[int] = None) -> str:
         if columns is None:
             columns, _ = terminal_size()

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "LogBar"
-version = "0.2.3"
+version = "0.2.4"
 description = "A unified Logger and ProgressBar util with zero dependencies."
 readme = "README.md"
 requires-python = ">=3"

tests/conftest.py

@@ -0,0 +1,8 @@
+import sys
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parent.parent
+
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))

tests/pytest.ini

@@ -1,2 +1,3 @@
 [pytest]
-addopts = -s
+addopts = -s
+pythonpath = ..

tests/test_progress.py

@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: Apache-2.0
 # Contact: qubitium@modelcloud.ai, x.com/qubitium
 
+import os
 import subprocess
 import random
 import re
@@ -15,6 +16,7 @@
 from unittest.mock import patch
 
 from logbar import LogBar
+from logbar import progress as progress_module
 from logbar.progress import ProgressBar, TITLE_HIGHLIGHT_COLOR, ANSI_BOLD_RESET
 from logbar.logbar import _active_progress_bars
 
@@ -68,6 +70,33 @@ def generate_expanding_str_a_to_z():
 REVERSED_SAMPLES = reversed(SAMPLES)
 
 class TestProgress(unittest.TestCase):
+    def setUp(self):
+        self._saved_env = {
+            "LOGBAR_ANIMATION": os.environ.get("LOGBAR_ANIMATION"),
+            "LOGBAR_PROGRESS_OUTPUT_INTERVAL": os.environ.get("LOGBAR_PROGRESS_OUTPUT_INTERVAL"),
+        }
+
+        for key in self._saved_env:
+            os.environ.pop(key, None)
+
+        self._clear_progress_env_caches()
+
+    def tearDown(self):
+        for key, value in self._saved_env.items():
+            if value is None:
+                os.environ.pop(key, None)
+            else:
+                os.environ[key] = value
+
+        self._clear_progress_env_caches()
+
+    @staticmethod
+    def _clear_progress_env_caches():
+        for helper_name in ("_env_animation_enabled", "_env_progress_output_interval"):
+            helper = getattr(progress_module, helper_name, None)
+            cache_clear = getattr(helper, "cache_clear", None)
+            if callable(cache_clear):
+                cache_clear()
 
     def test_title_fixed_subtitle_dynamic(self):
         pb = log.pb(SAMPLES).title("TITLE:").manual()
@@ -170,6 +199,52 @@ def test_title_animation_respects_logbar_animation_env(self):
         )
         self.assertEqual(disabled.stdout.strip(), "0")
 
+    def test_progress_output_interval_respects_env(self):
+        cache_clear = progress_module._env_progress_output_interval.cache_clear
+
+        with patch.dict(os.environ, {"LOGBAR_PROGRESS_OUTPUT_INTERVAL": "10"}):
+            cache_clear()
+            pb = ProgressBar(range(5))
+            self.assertEqual(pb._output_interval, 10)
+            with redirect_stdout(StringIO()):
+                pb.close()
+
+        cache_clear()
+
+    def test_progress_output_interval_defaults_to_one(self):
+        cache_clear = progress_module._env_progress_output_interval.cache_clear
+
+        with patch.dict(os.environ, {}, clear=True):
+            cache_clear()
+            pb = ProgressBar(range(5))
+            self.assertEqual(pb._output_interval, 1)
+            with redirect_stdout(StringIO()):
+                pb.close()
+
+        cache_clear()
+
+    def test_progress_output_interval_skips_intermediate_draws_but_flushes_final_step(self):
+        columns = 96
+
+        with patch('logbar.progress.terminal_size', return_value=(columns, 24)), \
+             patch('logbar.logbar.terminal_size', return_value=(columns, 24)):
+            buffer = StringIO()
+            with redirect_stdout(buffer):
+                pb = log.pb(15, output_interval=10).manual()
+                for step in range(1, 16):
+                    pb.current_iter_step = step
+                    pb.draw()
+                pb.close()
+
+        lines = extract_rendered_lines(buffer.getvalue())
+        progress_lines = [line for line in lines if "/15]" in line]
+
+        self.assertTrue(any("[0/15]" in line for line in progress_lines))
+        self.assertTrue(any("[10/15]" in line for line in progress_lines))
+        self.assertTrue(any("[15/15]" in line and "100.0%" in line for line in progress_lines))
+        self.assertFalse(any("[9/15]" in line for line in progress_lines))
+        self.assertFalse(any("[14/15]" in line for line in progress_lines))
+
     def test_draw_respects_terminal_width(self):
         pb = log.pb(100).title("TITLE").subtitle("SUBTITLE").manual()
         pb.current_iter_step = 50