Add progress bars to several commands and plugins.

Many long-running commands produce little or no feedback in the terminal to
indicate that they're progressing, and none of them provide estimates of how
long the operation will run. This change introduces the `enlighten` python
package, which displays progress bars akin to TQDM below the existing terminal
output.

To support consistent use and presentation of the progress bars, and to
allow for future modification, we introduce a method to beets.ui -
beets.ui.iprogress_bar - which can be used by Beets' core commands and all
Beets plugins. Example usage is provided in the methods' documentation.

The Enlighten library does not work as well in Windows PowerShell as it
does in a linux terminal (manually tested in Zsh), so the progress bars
are disabled in Windows environments. Resolving these issues and enabling
them in Windows is left as future work.

Author: peterjdolan

beets/ui/__init__.py

@@ -32,18 +32,29 @@
 from difflib import SequenceMatcher
 from functools import cache
 from itertools import chain
-from typing import Any, Callable, Literal
+from typing import (
+    Any,
+    Callable,
+    Generator,
+    Iterable,
+    Literal,
+    Sequence,
+    TypeVar,
+)
 
 import confuse
+import enlighten
 
 from beets import config, library, logging, plugins, util
 from beets.dbcore import db
 from beets.dbcore import query as db_query
 from beets.util import as_string
 from beets.util.functemplate import template
 
+is_windows = sys.platform == "win32"
+
 # On Windows platforms, use colorama to support "ANSI" terminal colors.
-if sys.platform == "win32":
+if is_windows:
     try:
         import colorama
     except ImportError:
@@ -1325,6 +1336,60 @@ def add_all_common_options(self):
         self.add_format_option()
 
 
+M = library.Album | library.Item | Any
+def iprogress_bar(sequence: Sequence[M], **kwargs) -> Generator[M, None, None]:
+    """Construct and manage an `enlighten.Counter` progress bar while iterating.
+
+    Example usage:
+    ```
+    for album in ui.iprogress_bar(
+        lib.albums(), desc="Updating albums", unit="albums"):
+        do_something_to(album)
+    ```
+
+    If the progress bar is iterating over an Album or an Item, then it will detect
+    whether or not the item has been modified, and will color-code the progress bar
+    with white and blue to indicate total progress and the portion of items that have
+    been modified.
+
+    Args:
+        sequence: An `Iterable` sequence to iterate over. If provided, and the
+            sequence can return its length, then the length will be used as the
+            total for the counter. The counter will be updated for each item
+            in the sequence.
+        kwargs: Additional keyword arguments to pass to the `enlighten.Counter`
+            constructor.
+
+    Yields:
+        The items from the sequence.
+    """
+    if sequence is None:
+        log.error("sequence must not be None")
+        return
+
+    # If sequence is not None, and can return its length, then use that as the total.
+    if "total" not in kwargs and hasattr(sequence, "__len__"):
+        kwargs["total"] = len(sequence)
+
+    # Disabled in windows environments. See above for details
+    with enlighten.Manager(enabled=not is_windows) as manager:
+        with manager.counter(**kwargs) as counter:
+            change_counter = counter.add_subcounter("blue")
+
+            for item in sequence:
+                revision = None
+                if hasattr(item, '_revision'):
+                    revision = item._revision
+
+                # Yield the item, allowing it to be modified, or not.
+                yield item
+
+                if revision and item._revision != revision:
+                    change_counter.update()
+                else:
+                    counter.update()
+
+
 # Subcommand parsing infrastructure.
 #
 # This is a fairly generic subcommand parser for optparse. It is

beetsplug/autobpm.py

@@ -21,7 +21,7 @@
 import numpy as np
 
 from beets.plugins import BeetsPlugin
-from beets.ui import Subcommand, should_write
+from beets.ui import Subcommand, iprogress_bar, should_write
 
 if TYPE_CHECKING:
     from beets.importer import ImportTask
@@ -56,7 +56,7 @@ def imported(self, _, task: ImportTask) -> None:
         self.calculate_bpm(task.imported_items())
 
     def calculate_bpm(self, items: list[Item], write: bool = False) -> None:
-        for item in items:
+        for item in iprogress_bar(items, desc="Calculating BPM", unit="items"):
             path = item.filepath
             if bpm := item.bpm:
                 self._log.info("BPM for {} already exists: {}", path, bpm)

beetsplug/badfiles.py

@@ -137,6 +137,7 @@ def check_item(self, item):
         error_lines = []
 
         if status > 0:
+            success = False
             error_lines.append(
                 f"{ui.colorize('text_error', dpath)}: checker exited with"
                 f" status {status}"
@@ -145,16 +146,18 @@ def check_item(self, item):
                 error_lines.append(f"  {line}")
 
         elif errors > 0:
+            success = False
             error_lines.append(
                 f"{ui.colorize('text_warning', dpath)}: checker found"
                 f" {status} errors or warnings"
             )
             for line in output:
                 error_lines.append(f"  {line}")
         elif self.verbose:
+            success = True
             error_lines.append(f"{ui.colorize('text_success', dpath)}: ok")
 
-        return error_lines
+        return success, error_lines
 
     def on_import_task_start(self, task, session):
         if not self.config["check_on_import"].get(False):
@@ -163,9 +166,8 @@ def on_import_task_start(self, task, session):
         checks_failed = []
 
         for item in task.items:
-            error_lines = self.check_item(item)
-            if error_lines:
-                checks_failed.append(error_lines)
+            _, error_lines = self.check_item(item)
+            checks_failed.append(error_lines)
 
         if checks_failed:
             task._badfiles_checks_failed = checks_failed
@@ -200,8 +202,19 @@ def command(self, lib, opts, args):
         self.verbose = opts.verbose
 
         def check_and_print(item):
-            for error_line in self.check_item(item):
-                ui.print_(error_line)
+            with ui.changes_and_errors_pbars(
+                total=len(item),
+                desc="Checking item",
+                unit="items",
+                color="white",
+            ) as (_, n_unchanged, n_errors):
+                success, error_lines = self.check_item(item)
+                if success:
+                    n_unchanged.update()
+                else:
+                    n_errors.update()
+                    for line in error_lines:
+                        ui.print_(line)
 
         par_map(check_and_print, items)
 

beetsplug/chroma.py

@@ -254,7 +254,11 @@ def submit_cmd_func(lib, opts, args):
         )
 
         def fingerprint_cmd_func(lib, opts, args):
-            for item in lib.items(args):
+            for item in ui.iprogress_bar(
+                lib.items(args),
+                desc="Fingerprinting items",
+                unit="items",
+            ):
                 fingerprint_item(self._log, item, write=ui.should_write())
 
         fingerprint_cmd.func = fingerprint_cmd_func

beetsplug/duplicates.py

@@ -17,6 +17,7 @@
 import os
 import shlex
 
+from beets import ui
 from beets.library import Album, Item
 from beets.plugins import BeetsPlugin
 from beets.ui import Subcommand, UserError, print_
@@ -284,9 +285,13 @@ def _group_by(self, objs, keys, strict):
         import collections
 
         counts = collections.defaultdict(list)
-        for obj in objs:
+        for obj in ui.iprogress_bar(
+            objs,
+            desc="Finding duplicates",
+            unit="items",
+        ):
             values = [getattr(obj, k, None) for k in keys]
-            values = [v for v in values if v not in (None, "")]
+            values = list(filter(lambda v: v not in (None, ""), values))
             if strict and len(values) < len(keys):
                 self._log.debug(
                     "some keys {} on item {.filepath} are null or empty: skipping",

beetsplug/embedart.py

@@ -120,7 +120,11 @@ def embed_func(lib, opts, args):
                 if not opts.yes and not _confirm(items, not opts.file):
                     return
 
-                for item in items:
+                for item in ui.iprogress_bar(
+                    items,
+                    desc="Embedding artwork",
+                    unit="items",
+                ):
                     art.embed_item(
                         self._log,
                         item,
@@ -155,7 +159,11 @@ def embed_func(lib, opts, args):
                 if not opts.yes and not _confirm(items, not opts.url):
                     os.remove(tempimg)
                     return
-                for item in items:
+                for item in ui.iprogress_bar(
+                    items,
+                    desc="Embedding artwork",
+                    unit="items",
+                ):
                     art.embed_item(
                         self._log,
                         item,
@@ -172,7 +180,11 @@ def embed_func(lib, opts, args):
                 # Confirm with user.
                 if not opts.yes and not _confirm(albums, not opts.file):
                     return
-                for album in albums:
+                for album in ui.iprogress_bar(
+                    albums,
+                    desc="Embedding artwork",
+                    unit="albums",
+                ):
                     art.embed_album(
                         self._log,
                         album,

beetsplug/fetchart.py

@@ -1564,7 +1564,12 @@ def batch_fetch_art(
         """Fetch album art for each of the albums. This implements the manual
         fetchart CLI command.
         """
-        for album in albums:
+
+        for album in ui.iprogress_bar(
+            albums,
+            desc="Fetching album art",
+            unit="albums",
+        ):
             if (
                 album.artpath
                 and not force

beetsplug/ftintitle.py

@@ -117,7 +117,11 @@ def func(lib, opts, args):
             keep_in_artist_field = self.config["keep_in_artist"].get(bool)
             write = ui.should_write()
 
-            for item in lib.items(args):
+            for item in ui.iprogress_bar(
+                lib.items(args),
+                desc="Analyzing songs",
+                unit="songs",
+            ):
                 if self.ft_in_title(item, drop_feat, keep_in_artist_field):
                     item.store()
                     if write:

beetsplug/lastgenre/__init__.py

@@ -525,7 +525,11 @@ def lastgenre_func(lib, opts, args):
 
             if opts.album:
                 # Fetch genres for whole albums
-                for album in lib.albums(args):
+                for album in ui.iprogress_bar(
+                    lib.albums(args),
+                    desc="Fetching genres",
+                    unit="albums",
+                ):
                     album.genre, src = self._get_genre(album)
                     self._log.info(
                         'genre for album "{0.album}" ({1}): {0.genre}',
@@ -554,7 +558,11 @@ def lastgenre_func(lib, opts, args):
             else:
                 # Just query singletons, i.e. items that are not part of
                 # an album
-                for item in lib.items(args):
+                for item in ui.iprogress_bar(
+                    lib.items(args),
+                    desc="Fetching genres",
+                    unit="tracks",
+                ):
                     item.genre, src = self._get_genre(item)
                     item.store()
                     self._log.info(

beetsplug/lyrics.py

@@ -1048,7 +1048,11 @@ def func(lib: Library, opts, args) -> None:
             # import_write config value.
             self.config.set(vars(opts))
             items = list(lib.items(args))
-            for item in items:
+            for item in ui.iprogress_bar(
+                items,
+                desc="Fetching lyrics",
+                unit="items",
+            ):
                 self.add_item_lyrics(item, ui.should_write())
                 if item.lyrics and opts.print:
                     ui.print_(item.lyrics)