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 cerataincertain 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()
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 richtqdm
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 cerataincertain 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()
