I will address your immediate concern of separating creating and updating progress bars from the main task of searching and yielding audio files.
My approach would be to use a variation of the Observer pattern. Typically you have a class that does some processing and one or more "observers" that wish to be notified of state changes or events that occurs during that processing. The processing class would typically implement methods add_observer and remove_observer maintaining a list of observers to be notified. There would be a well-defined interface that observers must implement and as processing proceeds and the state of processing changes or some event occurs, each observer that is registered would be notified by invoking a well-defined method defined by the observer interface.
Since your main processing is done by a function, get_audio_files, rather than a class method, it will be more convenient to have this function take an optional argument that is an instance of interface DirEventObserverInterface:
from abc import ABC, abstractmethod
class DirEventObserverInterface(ABC):
"""Interface for an "observer" that can be passed to get_audio_files
that will be called at various stages in processing."""
@abstractmethod
def start_iteration(self, level: int, num_entries: int) -> None:
"""Called when a new level of subsdirectory is to be iterated."""
@abstractmethod
def update(self, level: int, description: str) -> None:
"""Called when a description for a subdirectory level is to be updated."""
@abstractmethod
def end_iteration(self, level: int) -> None:
"""Called when an iteration of a subdirectory has completed."""
And the signature for get_audio_files becomes:
def get_audio_files(source: Path, listener: DirEventObserverInterface | None=None) -> Generator[Path]:
...
In the following code I have made a couple of additional changes:
- You computed the
artists list to be all the directory entries for the passed path to get_audio_files and then in processing these entries you skip those entries that are not directories. I have changed this so that artists only contains directory entries. This provides a more accurate count of entries that need processing.
- When checking whether a directory entry is one of the types of audio files you are interested in I have added the additional check to make sure the entry in question is an actual file just to handle the case where you might have a directory whose name ends with, for ecample, ".flac".
from collections.abc import Generator
from pathlib import Path
from time import sleep
from abc import ABC, abstractmethod
from rich.console import Console
from rich.progress import BarColumn, MofNCompleteColumn, Progress, TextColumn
"""
Walks a given directory for Artist/Album/Track files
yielding any file with an audio format extension.
"""
class DirEventObserverInterface(ABC):
"""Interface for an "observer" that can be passed to get_audio_files
that will be called at various stages in processing."""
@abstractmethod
def start_iteration(self, level: int, num_entries: int) -> None:
"""Called when a new level of subsdirectory is to be iterated."""
@abstractmethod
def update(self, level: int, description: str) -> None:
"""Called when a description for a subdirectory level is to be updated."""
@abstractmethod
def end_iteration(self, level: int) -> None:
"""Called when an iteration of a subdirectory has completed."""
class DirEventObserver(DirEventObserverInterface):
"""An implementation of DirEventObserverInterface that uses the rich
package for displaying progress bars."""
def __init__(self):
console = Console(force_terminal=True)
self._progress = Progress(
TextColumn("[progress.description]{task.description:>40.40s}"),
BarColumn(complete_style="bar.finished"),
MofNCompleteColumn(),
console=console,
)
self._bars = {}
def start_iteration(self, level: int, num_entries: int, description: str) -> None:
"""Called when a new progress bar is to be created for a subdirectory
at a certain level."""
if not self._bars:
self._progress.start()
self._bars[level] = self._progress.add_task(description, total=num_entries)
def update(self, level: int, description: str) -> None:
"""Called when a progress bar's description is to be updated."""
self._progress.update(
self._bars[level],
advance=1,
description=description
)
def end_iteration(self, level: int) -> None:
"""Called when a progress bar is to be removed."""
self._progress.remove_task(self._bars[level])
del self._bars[level]
if not self._bars:
self._progress.stop()
def get_audio_files(source: Path, listener: DirEventObserverInterface | None=None) -> Generator[Path]:
"""
Return audio files found in given directory.
Given a path to a folder with an Artist/Album/Track audiofile
layout, each path to audio files inside is yielded.
"""
VALID_TRACK_EXTENSIONS = (".flac", ".mp3", ".m4a", ".aac")
artists = sorted(
(x for x in source.iterdir() if x.is_dir() and not x.name.startswith(".")),
key=lambda x: x.stem.casefold(),
)
if listener:
listener.start_iteration(level=1, num_entries=len(artists), description="")
# Per Artist (top-level)
for artist_path in artists:
if listener:
listener.update(level=1, description=artist_path.name)
# Per Album
for album_path in artist_path.iterdir():
if not album_path.is_dir():
continue
tracks = sorted(
x
for x in album_path.iterdir()
if x.is_file() and x.suffix.casefold() in VALID_TRACK_EXTENSIONS
)
if listener:
listener.start_iteration(level=2, num_entries=len(tracks), description=album_path.name)
# Per Track
for track in tracks:
if listener:
listener.update(level=2, description=f"{album_path.name}/{track.stem}")
yield track
if listener:
listener.end_iteration(level=2)
if listener:
listener.end_iteration(level=1)
def main() -> None:
source = Path("source")
for audio_file in get_audio_files(source, DirEventObserver()):
# analyze/transcode/update files
sleep(.05)
if __name__ == "__main__":
main()
If you wanted now to use the tqdm package for displaying progress, you only need to change the DirEventObserver class. But to simplify the code further, get_audio_files always requires a DirEventObserverInterface instance but by default one that does nothing is used:
from collections.abc import Generator
from pathlib import Path
from time import sleep
from abc import ABC, abstractmethod
from tqdm import tqdm
"""
Walks a given directory for Artist/Album/Track files
yielding any file with an audio format extension.
"""
class DirEventObserverInterface(ABC):
"""Interface for an "observer" that can be passed to get_audio_files
that will be called at various stages in processing."""
@abstractmethod
def start_iteration(self, level: int, num_entries: int) -> None:
"""Called when a new level of subsdirectory is to be iterated."""
@abstractmethod
def update(self, level: int, description: str) -> None:
"""Called when a description for a subdirectory level is to be updated."""
@abstractmethod
def end_iteration(self, level: int) -> None:
"""Called when an iteration of a subdirectory has completed."""
class DoNothingDirEventObserver(DirEventObserverInterface):
"""Interface for an "observer" that ignores all events."""
def start_iteration(self, level: int, num_entries: int) -> None:
pass
def update(self, level: int, description: str) -> None:
pass
def end_iteration(self, level: int) -> None:
pass
class DirEventObserver(DirEventObserverInterface):
"""An implementation of DirEventObserverInterface that uses the tqdm
package for displaying progress bars."""
def __init__(self):
self._bars = {}
def start_iteration(self, level: int, num_entries: int, description: str) -> None:
"""Called when a new progress bar is to be created for a subdirectory
at a certain level."""
bar = tqdm(total=num_entries, position=level, desc=description, leave=False)
self._bars[level] = bar
def update(self, level: int, description: str) -> None:
"""Called when a progress bar's description is to be updated."""
bar = self._bars[level]
bar.update()
bar.set_description(description)
bar.refresh()
def end_iteration(self, level: int) -> None:
"""Called when a progress bar is to be removed."""
pass
def get_audio_files(source: Path, listener: DirEventObserverInterface=DoNothingDirEventObserver()) -> Generator[Path]:
"""
Return audio files found in given directory.
Given a path to a folder with an Artist/Album/Track audiofile
layout, each path to audio files inside is yielded.
"""
VALID_TRACK_EXTENSIONS = (".flac", ".mp3", ".m4a", ".aac")
artists = sorted(
(x for x in source.iterdir() if x.is_dir() and not x.name.startswith(".")),
key=lambda x: x.stem.casefold(),
)
listener.start_iteration(level=1, num_entries=len(artists), description="")
# Per Artist (top-level)
for artist_path in artists:
listener.update(level=1, description=artist_path.name)
# Per Album
for album_path in artist_path.iterdir():
if not album_path.is_dir():
continue
tracks = sorted(
x
for x in album_path.iterdir()
if x.is_file() and x.suffix.casefold() in VALID_TRACK_EXTENSIONS
)
listener.start_iteration(level=2, num_entries=len(tracks), description=album_path.name)
# Per Track
for track in tracks:
listener.update(level=2, description=f"{album_path.name}/{track.stem}")
yield track
listener.end_iteration(level=2)
listener.end_iteration(level=1)
def main() -> None:
source = Path("source")
for audio_file in get_audio_files(source, DirEventObserver()):
# analyze/transcode/update files
sleep(.05)
if __name__ == "__main__":
main()
