Helpers

helpers

A module for managing file downloads with progress tracking in the console.

This module provides functionality for downloading files with visual progress indicators using the Rich library. It includes a signal handler for graceful interruption and a DownloadProgressBar class for concurrent file downloads with progress tracking.

Classes:

Name	Description
DownloadProgressBar: A class that provides visual progress tracking for file downloads.

Functions:

Name	Description
handle_sigint: Signal handler for SIGINT	console (Console): Rich Console instance for output rendering. done_event (Event): Threading Event used for signaling download interruption.

DownloadProgressBar

DownloadProgressBar()

A progress bar for downloading files.

This class provides a visual progress indicator for file downloads using the Rich library. It supports downloading multiple files concurrently with a progress bar showing download speed, completion percentage, and elapsed time.

Attributes:

Name	Type	Description
progress	Progress	A Rich Progress instance configured with custom columns for displaying download information.

Examples:

>>> downloader = DownloadProgressBar()
>>> urls = ['http://example.com/file1.zip', 'http://example.com/file2.zip']
>>> downloader.download(urls, '/path/to/destination')

Source code in eos_downloader/helpers/__init__.py

def __init__(self) -> None:
    self.progress = Progress(
        TextColumn(
            "💾  Downloading [bold blue]{task.fields[filename]}", justify="right"
        ),
        BarColumn(bar_width=None),
        "[progress.percentage]{task.percentage:>3.1f}%",
        "•",
        TransferSpeedColumn(),
        "•",
        DownloadColumn(),
        "•",
        TimeElapsedColumn(),
        "•",
        console=console,
    )

download

download(urls: Iterable[str], dest_dir: str) -> None

Download files from URLs concurrently to a destination directory.

This method downloads files from the provided URLs in parallel using a thread pool, displaying progress for each download in the console.

Parameters:

Name	Type	Description	Default
urls	Iterable[str]	An iterable of URLs to download files from.	required
dest_dir	str	The destination directory where files will be saved.	required

Returns:

Type	Description
None

Examples:

>>> downloader = DownloadProgressBar()
>>> urls = ["http://example.com/file1.txt", "http://example.com/file2.txt"]
>>> downloader.download(urls, "/path/to/destination")

Source code in eos_downloader/helpers/__init__.py

def download(self, urls: Iterable[str], dest_dir: str) -> None:
    """Download files from URLs concurrently to a destination directory.

    This method downloads files from the provided URLs in parallel using a thread pool,
    displaying progress for each download in the console.

    Parameters
    ----------
    urls : Iterable[str]
        An iterable of URLs to download files from.
    dest_dir : str
        The destination directory where files will be saved.

    Returns
    -------
    None

    Examples
    --------
    >>> downloader = DownloadProgressBar()
    >>> urls = ["http://example.com/file1.txt", "http://example.com/file2.txt"]
    >>> downloader.download(urls, "/path/to/destination")
    """
    with self.progress:
        with ThreadPoolExecutor(max_workers=4) as pool:
            futures = []
            for url in urls:
                filename = url.split("/")[-1].split("?")[0]
                dest_path = os.path.join(dest_dir, filename)
                task_id = self.progress.add_task(
                    "download", filename=filename, start=False
                )
                futures.append(pool.submit(self._copy_url, task_id, url, dest_path))

            for future in futures:
                future.result()  # Wait for all downloads to complete

handle_sigint

handle_sigint(signum: Any, frame: Any) -> None

Signal handler for SIGINT (Ctrl+C).

This function sets the done_event flag when SIGINT is received, allowing for graceful termination of the program.

Parameters:

Name	Type	Description	Default
signum	Any	Signal number.	required
frame	Any	Current stack frame object.	required

Returns:

Type	Description
None

Source code in eos_downloader/helpers/__init__.py

def handle_sigint(signum: Any, frame: Any) -> None:
    """
    Signal handler for SIGINT (Ctrl+C).

    This function sets the done_event flag when SIGINT is received,
    allowing for graceful termination of the program.

    Parameters
    ----------
    signum : Any
        Signal number.
    frame : Any
        Current stack frame object.

    Returns
    -------
    None
    """
    done_event.set()