Merge pull request #235 from rsalmei/escape-codes-hook

ANSI Escape codes support in print hook

Author: rsalmei

CHANGELOG.md

@@ -1,6 +1,11 @@
 # Changelog
 
 
+## 3.1.1 - Apr 08, 2023
+- print hook support for ANSI Escape Codes (which avoid sending newlines when not needed)
+- typing support in `alive_it`, so the collection type is correctly identified.
+
+
 ## 3.1.0 - Mar 23, 2023
 - new resuming computations support with `skipped` items
 - new `max_cols` config setting, the number of columns to use if not possible to fetch it, like in jupyter and other platforms which doesn't support size

README.md

@@ -1,7 +1,7 @@
 from .core.configuration import config_handler
 from .core.progress import alive_bar, alive_it
 
-VERSION = (3, 1, 0)
+VERSION = (3, 1, 1)
 
 __author__ = 'Rogério Sampaio de Almeida'
 __email__ = '[email protected]'

alive_progress/__init__.py

@@ -30,7 +30,7 @@ def flush_buffers():
 
     def flush(stream):
         if buffers[stream]:
-            write(stream, '\n')
+            write(stream, '\n')  # when the current index is about to change, send a newline.
             stream.flush()
 
     def write(stream, part):
@@ -39,6 +39,9 @@ def write(stream, part):
 
         buffer = buffers[stream]
         if part != '\n':
+            if part.startswith('\x1b'):  # if the part starts with ESC, just send it.
+                stream.write(part)
+                return
             # this will generate a sequence of lines interspersed with None, which will later
             # be rendered as the indent filler to align additional lines under the same header.
             gen = chain.from_iterable(zip(repeat(None), part.splitlines(True)))

alive_progress/core/hook_manager.py

@@ -6,6 +6,7 @@
 import threading
 import time
 from contextlib import contextmanager
+from typing import Any, Callable, Collection, Iterable, Optional, TypeVar
 
 from .calibration import calibrated_fps, custom_fps
 from .configuration import config_handler
@@ -16,7 +17,7 @@
     gen_simple_exponential_smoothing
 
 
-def alive_bar(total=None, *, calibrate=None, **options):
+def alive_bar(total: Optional[int] = None, *, calibrate: Optional[int] = None, **options: Any):
     """An alive progress bar to keep track of lengthy operations.
     It has a spinner indicator, elapsed time, throughput and ETA.
     When the operation finishes, a receipt is displayed with statistics.
@@ -178,17 +179,17 @@ def set_title(title=None):
 
     if config.manual:
         def bar(percent):  # for manual progress modes, regardless of total.
-            hook_manager.flush_buffers()
+            hook_manager.flush_buffers()  # notify that the current index is about to change.
             run.percent = max(0., float(percent))
             bar_update_hook()
     elif not total:
         def bar(count=1):  # for unknown progress mode.
-            hook_manager.flush_buffers()
+            hook_manager.flush_buffers()  # notify that the current index is about to change.
             run.count += max(1, int(count))
             bar_update_hook()
     else:
         def bar(count=1, *, skipped=False):  # for definite progress mode.
-            hook_manager.flush_buffers()
+            hook_manager.flush_buffers()  # notify that the current index is about to change.
             count = max(1, int(count))
             run.count += count
             if not skipped:
@@ -289,7 +290,7 @@ def stats_end(f):
 
     if total or config.manual:  # we can track progress and therefore eta.
         def stats_run(f):
-            run.rate_text = rate_text(1)
+            run.rate_text = rate_text(1)  # although repeated below,
             run.eta_text = eta_text(gen_eta.send((current(), run.rate)))
             return f.format(rate=run.rate_text, unit=unit, eta=run.eta_text)
 
@@ -298,7 +299,7 @@ def stats_run(f):
         stats_default = '({eta}, {rate})'
     else:  # unknown progress.
         def stats_run(f):
-            run.rate_text = rate_text(1)
+            run.rate_text = rate_text(1)  # it won't be calculated if not needed.
             return f.format(rate=run.rate_text, eta='?')
 
         bar_repr = bar_repr.unknown
@@ -473,7 +474,12 @@ def _render_title(config, title=None):
     return combine_cells(fix_cells(title[:length - 1]), ('…',))
 
 
-def alive_it(it, total=None, *, finalize=None, calibrate=None, **options):
+T = TypeVar('T')
+
+
+def alive_it(it: Collection[T], total: Optional[int] = None, *,
+             finalize: Callable[[Any], None] = None,
+             calibrate: Optional[int] = None, **options: Any) -> Iterable[T]:
     """New iterator adapter in 2.0, which makes it simpler to monitor any processing.
 
     Simply wrap your iterable with `alive_it`, and process your items normally!
@@ -498,7 +504,7 @@ def alive_it(it, total=None, *, finalize=None, calibrate=None, **options):
     ... items = range(100000)
     ... bar = alive_it(items)
     ... for item in bar:
-    ...     bar.text(f'Wow, it works! Item: {item}')
+    ...     bar.text = f'Wow, it works! Item: {item}'
     ...     # process item.
 
     You can also send a `finalize` function to set the final receipt title and text, and any other
@@ -518,7 +524,7 @@ def alive_it(it, total=None, *, finalize=None, calibrate=None, **options):
 DB updated |████████████████████| 100k/100k [100%] in 2.6s (38.7k/s) 100000 entries changed
 
     Args:
-        it (iterable): the input iterable to be processed
+        it: the input iterable to be processed
         total: same as alive_bar
         finalize: a function to be called when the bar is going to finalize
         calibrate: same as alive_bar
@@ -546,7 +552,7 @@ def alive_it(it, total=None, *, finalize=None, calibrate=None, **options):
     return __AliveBarIteratorAdapter(it, finalize, __alive_bar(config, total, calibrate=calibrate))
 
 
-class __AliveBarIteratorAdapter:
+class __AliveBarIteratorAdapter(Iterable[T]):
     def __init__(self, it, finalize, inner_bar):
         self._it, self._finalize, self._inner_bar = it, finalize, inner_bar