Merge pull request #332 from fourMs/copilot/optimize-ffmpeg-commands

Add `show_progress()` to globally suppress MGT progress bars

Author: alexarje

musicalgestures/__init__.py

@@ -14,6 +14,7 @@
     ffmpeg_cmd,
     get_length,
     generate_outfilename,
+    show_progress,
 )
 from musicalgestures._mglist import MgList
 

musicalgestures/_utils.py

@@ -1,5 +1,31 @@
 from typing import Union, Tuple
 
+#: Module-level flag controlling whether progress bars are shown.
+#: Use :func:`show_progress` to change this setting.
+_SHOW_PROGRESS: bool = True
+
+
+def show_progress(enabled: bool) -> None:
+    """Enable or disable the MGT progress bars globally.
+
+    Disabling the progress bars is useful when running batch processing jobs or
+    when the output is captured by a logging framework where the repeated
+    ``\\r`` updates would clutter the log.
+
+    Args:
+        enabled (bool): Pass ``True`` to show progress bars (default behaviour)
+            or ``False`` to suppress them.
+
+    Examples:
+        >>> import musicalgestures as mg
+        >>> mg.show_progress(False)  # suppress all progress bars
+        >>> # … batch processing …
+        >>> mg.show_progress(True)   # re-enable for interactive use
+    """
+    global _SHOW_PROGRESS
+    _SHOW_PROGRESS = bool(enabled)
+
+
 class MgProgressbar():
     """
     Calls in a loop to create terminal progress bar.
@@ -132,6 +158,10 @@ def progress(self, iteration):
         """
         if self.finished:
             return
+        if not _SHOW_PROGRESS:
+            if iteration >= self.total:
+                self.finished = True
+            return
         import sys
         import shutil
 

tests/test_utils.py

@@ -1,5 +1,6 @@
 import musicalgestures
 from musicalgestures._utils import *
+import musicalgestures._utils as _utils_module
 import numpy as np
 import os
 import itertools
@@ -169,6 +170,71 @@ def test_repr(self):
         assert print(pb) == None
 
 
+class Test_show_progress:
+    def setup_method(self):
+        # Always restore the default state after each test
+        _utils_module._SHOW_PROGRESS = True
+
+    def teardown_method(self):
+        _utils_module._SHOW_PROGRESS = True
+
+    def test_show_progress_default_is_true(self):
+        assert _utils_module._SHOW_PROGRESS is True
+
+    def test_show_progress_disable(self):
+        show_progress(False)
+        assert _utils_module._SHOW_PROGRESS is False
+
+    def test_show_progress_enable(self):
+        _utils_module._SHOW_PROGRESS = False
+        show_progress(True)
+        assert _utils_module._SHOW_PROGRESS is True
+
+    def test_show_progress_exposed_on_module(self):
+        # Ensure show_progress is accessible via the top-level package
+        assert hasattr(musicalgestures, 'show_progress')
+        assert callable(musicalgestures.show_progress)
+
+    def test_progress_bar_suppressed_does_not_print(self, capsys):
+        show_progress(False)
+        pb = MgProgressbar(total=100)
+        for step in range(103):
+            pb.progress(step)
+        captured = capsys.readouterr()
+        assert captured.out == ""
+
+    def test_progress_bar_suppressed_marks_finished(self):
+        show_progress(False)
+        pb = MgProgressbar(total=100)
+        assert not pb.finished
+        pb.progress(200)
+        assert pb.finished
+
+    def test_progress_bar_suppressed_finished_returns_early(self):
+        show_progress(False)
+        pb = MgProgressbar(total=100)
+        pb.finished = True
+        # Should return immediately without touching _SHOW_PROGRESS check
+        assert pb.progress(50) == None
+
+    def test_progress_bar_enabled_after_disable(self, capsys):
+        show_progress(False)
+        pb = MgProgressbar(total=100)
+        pb.could_not_get_terminal_window = True
+        for step in range(50):
+            pb.progress(step)
+        show_progress(True)
+        pb2 = MgProgressbar(total=100)
+        pb2.could_not_get_terminal_window = True
+        import time
+        for step in range(103):
+            time.sleep(0.01)
+            pb2.progress(step)
+        captured = capsys.readouterr()
+        # pb2 ran with progress bars enabled so output should be non-empty
+        assert len(captured.out) > 0
+
+
 class Test_roundup:
     def test_positive(self):
         assert roundup(10, 4) == 12