progress_handler

Module that holds the ProgressHandler class and Song Tracker class.

ClientSongDownload(song, progress, message, path=None) dataclass

Represents the download progress of a single song for a client.

ProgressHandler(simple_tui=False, update_callback=None, web_ui=False)

Class for handling the progress of a download, including the progress bar.

Arguments

• simple_tui: Whether or not to use the simple TUI.
• update_callback: A callback to call when the progress bar is updated.

Source code in spotdl/download/progress_handler.py

def __init__(
    self,
    simple_tui: bool = False,
    update_callback: Optional[Callable[[Any, str], None]] = None,
    web_ui: bool = False,
):
    """
    Initialize the progress handler.

    ### Arguments
    - simple_tui: Whether or not to use the simple TUI.
    - update_callback: A callback to call when the progress bar is updated.
    """

    self.songs: List[Song] = []
    self.song_count: int = 0
    self.overall_progress = 0
    self.overall_total = 100
    self.overall_completed_tasks = 0
    self.update_callback = update_callback
    self.previous_overall = self.overall_completed_tasks

    self.simple_tui = simple_tui
    self.web_ui = web_ui
    self.quiet = logger.getEffectiveLevel() < 10
    self.overall_task_id: Optional[TaskID] = None

    self.progress_tracker = ProgressTracker()

    if not self.simple_tui:
        console = get_console()

        self.rich_progress_bar = Progress(
            SizedTextColumn(
                "[white]{task.description}",
                overflow="ellipsis",
                width=int(console.width / 3),
            ),
            SizedTextColumn(
                "{task.fields[message]}", width=18, style="nonimportant"
            ),
            BarColumn(bar_width=None, finished_style="green"),
            "[progress.percentage]{task.percentage:>3.0f}%",
            TimeRemainingColumn(),
            # Normally when you exit the progress context manager (or call stop())
            # the last refreshed display remains in the terminal with the cursor on
            # the following line. You can also make the progress display disappear on
            # exit by setting transient=True on the Progress constructor
            transient=True,
        )

        # Basically a wrapper for rich's: with ... as ...
        self.rich_progress_bar.__enter__()

add_song(song)

Adds a song to the list of songs.

Arguments

• song: The song to add.

Source code in spotdl/download/progress_handler.py

def add_song(self, song: Song) -> None:
    """
    Adds a song to the list of songs.

    ### Arguments
    - song: The song to add.
    """

    self.songs.append(song)
    self.set_song_count(len(self.songs))

close()

Close the Tui Progress Handler.

Source code in spotdl/download/progress_handler.py

def close(self) -> None:
    """
    Close the Tui Progress Handler.
    """

    if not self.simple_tui:
        self.rich_progress_bar.stop()

    logging.shutdown()

get_new_tracker(song)

Get a new progress tracker.

Arguments

• song: The song to track.

Returns

• A new progress tracker.

Source code in spotdl/download/progress_handler.py

def get_new_tracker(self, song: Song) -> "SongTracker":
    """
    Get a new progress tracker.

    ### Arguments
    - song: The song to track.

    ### Returns
    - A new progress tracker.
    """

    return SongTracker(self, song)

set_song_count(count)

Set the number of songs to download.

Arguments

• count: The number of songs to download.

Source code in spotdl/download/progress_handler.py

def set_song_count(self, count: int) -> None:
    """
    Set the number of songs to download.

    ### Arguments
    - count: The number of songs to download.
    """

    self.song_count = count
    self.overall_total = 100 * count

    if not self.simple_tui:
        if self.song_count > 4:
            self.overall_task_id = self.rich_progress_bar.add_task(
                description="Total",
                message=(
                    f"{self.overall_completed_tasks}/{int(self.overall_total / 100)} "
                    "complete"
                ),
                total=self.overall_total,
                visible=(not self.quiet),
            )

set_songs(songs)

Sets the list of songs to be downloaded.

Arguments

• songs: The list of songs to download.

Source code in spotdl/download/progress_handler.py

def set_songs(self, songs: List[Song]) -> None:
    """
    Sets the list of songs to be downloaded.

    ### Arguments
    - songs: The list of songs to download.
    """

    self.songs = songs
    self.set_song_count(len(songs))

update_overall()

Update the overall progress bar.

Source code in spotdl/download/progress_handler.py

def update_overall(self) -> None:
    """
    Update the overall progress bar.
    """

    if not self.simple_tui:
        # If the overall progress bar exists
        if self.overall_task_id is not None:
            self.rich_progress_bar.update(
                self.overall_task_id,
                message=f"{self.overall_completed_tasks}/"
                f"{int(self.overall_total / 100)} "
                "complete",
                completed=self.overall_progress,
            )
    else:
        if self.previous_overall != self.overall_completed_tasks:
            logger.info(
                "%s/%s complete", self.overall_completed_tasks, self.song_count
            )
            self.previous_overall = self.overall_completed_tasks

ProgressHandlerError

Bases: Exception

Base class for all exceptions raised by ProgressHandler subclasses.

ProgressTracker

Tracks the progress of each song download. Similar to Rich Progress but without any TUI elements.

add(song)

Add a song to the progress tracker.

Source code in spotdl/download/progress_handler.py

def add(self, song: Song):
    """
    Add a song to the progress tracker.
    """
    # check if exists
    if song.url in self.songs:
        return
    self.songs[song.url] = ClientSongDownload(
        song=song, progress=0, message="Processing"
    )

remove(song)

Remove a song from the progress tracker.

Source code in spotdl/download/progress_handler.py

def remove(self, song: Song):
    """
    Remove a song from the progress tracker.
    """
    if song.url in self.songs:
        del self.songs[song.url]

set_path(song, path)

Set the download path for a song.

Source code in spotdl/download/progress_handler.py

def set_path(self, song: Song, path: str):
    """
    Set the download path for a song.
    """
    if song.url in self.songs:
        self.songs[song.url].path = path

update(song, progress, message)

Update the progress of a song in the progress tracker.

Source code in spotdl/download/progress_handler.py

def update(self, song: Song, progress: int, message: str):
    """
    Update the progress of a song in the progress tracker.
    """
    if song.url in self.songs:
        self.songs[song.url].progress = progress
        self.songs[song.url].message = message
    else:
        self.songs[song.url] = ClientSongDownload(
            song=song, progress=progress, message=message
        )

SizedTextColumn(text_format, style='none', justify='left', markup=True, highlighter=None, overflow=None, width=20)

Bases: ProgressColumn

Custom sized text column based on the Rich library.

Arguments

• text_format: The format string to use for the text.
• style: The style to use for the text.
• justify: The justification to use for the text.
• markup: Whether or not the text should be rendered as markup.
• highlighter: A Highlighter to use for highlighting the text.
• overflow: The overflow method to use for truncating the text.
• width: The maximum width of the text.

Source code in spotdl/download/progress_handler.py

def __init__(
    self,
    text_format: str,
    style: StyleType = "none",
    justify: JustifyMethod = "left",
    markup: bool = True,
    highlighter: Optional[Highlighter] = None,
    overflow: Optional[OverflowMethod] = None,
    width: int = 20,
) -> None:
    """
    A column containing text.

    ### Arguments
    - text_format: The format string to use for the text.
    - style: The style to use for the text.
    - justify: The justification to use for the text.
    - markup: Whether or not the text should be rendered as markup.
    - highlighter: A Highlighter to use for highlighting the text.
    - overflow: The overflow method to use for truncating the text.
    - width: The maximum width of the text.
    """

    self.text_format = text_format
    self.justify: JustifyMethod = justify
    self.style = style
    self.markup = markup
    self.highlighter = highlighter
    self.overflow: Optional[OverflowMethod] = overflow
    self.width = width
    super().__init__()

render(task)

Render the Column.

Arguments

• task: The Task to render.

Returns

• A Text object.

Source code in spotdl/download/progress_handler.py

def render(self, task: Task) -> Text:
    """
    Render the Column.

    ### Arguments
    - task: The Task to render.

    ### Returns
    - A Text object.
    """

    _text = self.text_format.format(task=task)
    if self.markup:
        text = Text.from_markup(_text, style=self.style, justify=self.justify)
    else:
        text = Text(_text, style=self.style, justify=self.justify)
    if self.highlighter:
        self.highlighter.highlight(text)

    text.truncate(max_width=self.width, overflow=self.overflow, pad=True)
    return text

SongTracker(parent, song)

Class to track the progress of a song.

Arguments

• parent: The parent Tui Progress Handler.

Source code in spotdl/download/progress_handler.py

def __init__(self, parent, song: Song) -> None:
    """
    Initialize the Tui Song Tracker.

    ### Arguments
    - parent: The parent Tui Progress Handler.
    """

    self.parent: "ProgressHandler" = parent
    self.song = song

    # Clean up the song name
    # from weird unicode characters
    self.song_name = "".join(
        char
        for char in self.song.display_name
        if char not in [chr(i) for i in BAD_CHARS]
    )

    self.progress: int = 0
    self.old_progress: int = 0
    self.status = ""
    self.path: Optional[str] = None

    self.parent.progress_tracker.add(self.song)
    if not self.parent.simple_tui:
        self.task_id = self.parent.rich_progress_bar.add_task(
            description=escape(self.song_name),
            message="Processing",
            total=100,
            completed=self.progress,
            start=False,
            visible=(not self.parent.quiet),
        )

ffmpeg_progress_hook(progress)

Updates the progress.

Arguments

• progress: The progress to update to.

Source code in spotdl/download/progress_handler.py

def ffmpeg_progress_hook(self, progress: int) -> None:
    """
    Updates the progress.

    ### Arguments
    - progress: The progress to update to.
    """

    if self.parent.simple_tui and not self.parent.web_ui:
        self.progress = 50
    else:
        self.progress = 50 + int(progress * 0.45)

    self.update("Converting")

notify_complete(status='Done')

Notifies the progress handler that the song has been downloaded and converted.

Arguments

• status: The status to display.

Source code in spotdl/download/progress_handler.py

def notify_complete(self, status="Done") -> None:
    """
    Notifies the progress handler that the song has been downloaded and converted.

    ### Arguments
    - status: The status to display.
    """

    self.progress = 100
    self.update(status)

notify_conversion_complete(status='Embedding metadata')

Notifies the progress handler that the song has been converted.

Arguments

• status: The status to display.

Source code in spotdl/download/progress_handler.py

def notify_conversion_complete(self, status="Embedding metadata") -> None:
    """
    Notifies the progress handler that the song has been converted.

    ### Arguments
    - status: The status to display.
    """

    self.progress = 95
    self.update(status)

notify_download_complete(status='Converting')

Notifies the progress handler that the song has been downloaded.

Arguments

• status: The status to display.

Source code in spotdl/download/progress_handler.py

def notify_download_complete(self, status="Converting") -> None:
    """
    Notifies the progress handler that the song has been downloaded.

    ### Arguments
    - status: The status to display.
    """

    self.progress = 50
    self.update(status)

notify_download_skip(status='Skipped')

Notifies the progress handler that the song has been skipped.

Arguments

• status: The status to display.

Source code in spotdl/download/progress_handler.py

def notify_download_skip(self, status="Skipped") -> None:
    """
    Notifies the progress handler that the song has been skipped.

    ### Arguments
    - status: The status to display.
    """

    self.progress = 100
    self.update(status)

notify_error(message, traceback, finish=False)

Logs an error message.

Arguments

• message: The message to log.
• traceback: The traceback of the error.
• finish: Whether to finish the task.

Source code in spotdl/download/progress_handler.py

def notify_error(
    self, message: str, traceback: Exception, finish: bool = False
) -> None:
    """
    Logs an error message.

    ### Arguments
    - message: The message to log.
    - traceback: The traceback of the error.
    - finish: Whether to finish the task.
    """

    self.update("Error")
    if finish:
        self.progress = 100

    if logger.getEffectiveLevel() == logging.DEBUG:
        logger.exception(message)
    else:
        logger.error("%s: %s", traceback.__class__.__name__, traceback)

set_path(path)

Sets the path of the song.

Arguments

• path: The path to set.

Source code in spotdl/download/progress_handler.py

def set_path(self, path: str) -> None:
    """
    Sets the path of the song.

    ### Arguments
    - path: The path to set.
    """

    self.path = path
    self.parent.progress_tracker.set_path(self.song, path)

update(message='')

Called at every event.

Arguments

• message: The message to display.

Source code in spotdl/download/progress_handler.py

def update(self, message=""):
    """
    Called at every event.

    ### Arguments
    - message: The message to display.
    """

    old_message = self.status
    self.status = message

    # The change in progress since last update
    delta = self.progress - self.old_progress

    self.parent.progress_tracker.update(self.song, self.progress, message)
    if self.progress == 100 or message == "Error":
        if not self.parent.web_ui:
            self.parent.progress_tracker.remove(self.song)

    if not self.parent.simple_tui:
        # Update the progress bar
        # `start_task` called every time to ensure progress is remove from indeterminate state
        self.parent.rich_progress_bar.start_task(self.task_id)
        self.parent.rich_progress_bar.update(
            self.task_id,
            description=escape(self.song_name),
            message=message,
            completed=self.progress,
        )

        # Refresh the progress bar to show the changes before it gets removed in case of 100%
        self.parent.rich_progress_bar.refresh()

        # If task is complete
        if self.progress == 100 or message == "Error":
            self.parent.overall_completed_tasks += 1
            self.parent.rich_progress_bar.remove_task(self.task_id)
    else:
        # If task is complete
        if self.progress == 100 or message == "Error":
            self.parent.overall_completed_tasks += 1

        # When running web ui print progress
        # only one time when downloading/converting/embedding
        if self.parent.web_ui and old_message != self.status:
            logger.info("%s: %s", self.song_name, message)
        elif not self.parent.web_ui and delta:
            logger.info("%s: %s", self.song_name, message)

    # Update the overall progress bar
    if self.parent.song_count == self.parent.overall_completed_tasks:
        self.parent.overall_progress = self.parent.song_count * 100
    else:
        self.parent.overall_progress += delta

    self.parent.update_overall()
    self.old_progress = self.progress

    if self.parent.update_callback:
        self.parent.update_callback(self, message)

yt_dlp_progress_hook(data)

Updates the progress.

Arguments

• progress: The progress to update to.

Source code in spotdl/download/progress_handler.py

def yt_dlp_progress_hook(self, data: Dict[str, Any]) -> None:
    """
    Updates the progress.

    ### Arguments
    - progress: The progress to update to.
    """

    if data["status"] == "downloading":
        file_bytes = data.get("total_bytes")
        if file_bytes is None:
            file_bytes = data.get("total_bytes_estimate")

        downloaded_bytes = data.get("downloaded_bytes")
        if self.parent.simple_tui and not self.parent.web_ui:
            self.progress = 50
        elif file_bytes and downloaded_bytes:
            self.progress = downloaded_bytes / file_bytes * 50

        self.update("Downloading")