Utils Download

download

CLASS	DESCRIPTION
URLValidationError	Raised when URL validation fails.
DownloadError	Base exception for download-related errors.
DownloadResult	Container for download result metadata.

FUNCTION	DESCRIPTION
download_url	Download file from URL.
download_urls	Download multiple files from URLs.
list_datasets	List available datasets.
get_dataset_info	Get detailed information about a specific dataset.
download_dataset	Download a dataset from the registry.

Classes

URLValidationError

Bases: ValueError

Raised when URL validation fails.

DownloadError

Bases: Exception

Base exception for download-related errors.

DownloadResult(url: str, filepath: str, success: bool, file_size: int | None = None, duration: float | None = None, error: str | None = None)

Container for download result metadata.

Source code in spectre/utils/download.py

def __init__(
    self,
    url: str,
    filepath: str,
    success: bool,
    file_size: int | None = None,
    duration: float | None = None,
    error: str | None = None,
):
    self.url = url
    self.filepath = filepath
    self.success = success
    self.file_size = file_size
    self.duration = duration
    self.error = error

Functions

validate_url(url: str) -> None

Validate URL format and scheme.

PARAMETER	DESCRIPTION
url	URL to validate. TYPE: str

RAISES	DESCRIPTION
URLValidationError	If URL is invalid or unsupported scheme.

Source code in spectre/utils/download.py

def validate_url(url: str) -> None:
    """
    Validate URL format and scheme.

    Parameters
    ----------
    url : str
        URL to validate.

    Raises
    ------
    URLValidationError
        If URL is invalid or unsupported scheme.
    """
    if not isinstance(url, str) or not url.strip():
        raise URLValidationError("URL must be a non-empty string.")

    try:
        parsed = urlparse(url)
    except Exception as e:
        raise URLValidationError(f"Invalid URL format: {e}") from e

    if not parsed.scheme:
        raise URLValidationError("URL must include a scheme (http/https/ftp).")

    if parsed.scheme not in {"http", "https", "ftp"}:
        raise URLValidationError(f"Unsupported URL scheme: {parsed.scheme}")

    if not parsed.netloc:
        raise URLValidationError("URL must include a valid domain.")

download_url(url: str, filepath: str, overwrite: bool = False, show_progress: bool = True, validate: bool = True, max_retries: int = 3, retry_delay: float = 1.0, timeout: float = 30.0, progress_callback: Callable[[int, int], None] | None = None) -> DownloadResult

Download file from URL.

PARAMETER	DESCRIPTION
url	URL to download file from. TYPE: str
filepath	Filepath to save downloaded file to. TYPE: str
overwrite	If True, overwrite existing file, by default False. TYPE: bool DEFAULT: False
show_progress	Show download progress, by default True. TYPE: bool DEFAULT: True
validate	Validate URL before downloading, by default True. TYPE: bool DEFAULT: True
max_retries	Maximum number of retry attempts, by default 3. TYPE: int DEFAULT: 3
retry_delay	Delay between retries in seconds, by default 1.0. TYPE: float DEFAULT: 1.0
timeout	Download timeout in seconds, by default 30.0. TYPE: float DEFAULT: 30.0
progress_callback	Custom progress callback function(downloaded, total), by default None. TYPE: Callable[[int, int], None] | None DEFAULT: None

RETURNS	DESCRIPTION
DownloadResult	Download result with metadata.

RAISES	DESCRIPTION
URLValidationError	If URL validation fails.
DownloadError	If download fails after all retries.
ValueError	If filepath is invalid.

Source code in spectre/utils/download.py

def download_url(
    url: str,
    filepath: str,
    overwrite: bool = False,
    show_progress: bool = True,
    validate: bool = True,
    max_retries: int = 3,
    retry_delay: float = 1.0,
    timeout: float = 30.0,
    progress_callback: Callable[[int, int], None] | None = None,
) -> DownloadResult:
    """
    Download file from URL.

    Parameters
    ----------
    url : str
        URL to download file from.
    filepath : str
        Filepath to save downloaded file to.
    overwrite : bool, optional
        If True, overwrite existing file, by default False.
    show_progress : bool, optional
        Show download progress, by default True.
    validate : bool, optional
        Validate URL before downloading, by default True.
    max_retries : int, optional
        Maximum number of retry attempts, by default 3.
    retry_delay : float, optional
        Delay between retries in seconds, by default 1.0.
    timeout : float, optional
        Download timeout in seconds, by default 30.0.
    progress_callback : Callable[[int, int], None] | None, optional
        Custom progress callback function(downloaded, total), by default None.

    Returns
    -------
    DownloadResult
        Download result with metadata.

    Raises
    ------
    URLValidationError
        If URL validation fails.
    DownloadError
        If download fails after all retries.
    ValueError
        If filepath is invalid.
    """
    start_time = time.time()

    # Validate inputs
    if validate:
        validate_url(url)

    # Resolve and validate filepath
    path = Path(filepath).expanduser().resolve()
    if path.is_dir():
        raise ValueError("Specify full filepath, including filename.")

    # Check if file exists and overwrite setting
    if path.is_file():
        if not overwrite:
            return DownloadResult(
                url=url,
                filepath=str(path),
                success=True,
                file_size=path.stat().st_size,
                duration=0.0,
                error="File already exists (skipped)",
            )
        else:
            import warnings

            warnings.warn(f"Overwriting existing file: {path}", UserWarning)

    # Create parent directory
    path.parent.mkdir(parents=True, exist_ok=True)

    # Attempt download with retries
    last_error = None

    for attempt in range(max_retries):
        try:
            if progress_callback or show_progress:
                result = _download_with_progress(
                    url, str(path), timeout, progress_callback, show_progress
                )
            else:
                result = _download_simple(url, str(path), timeout)

            duration = time.time() - start_time
            return DownloadResult(
                url=url,
                filepath=str(path),
                success=True,
                file_size=result["size"],
                duration=duration,
            )

        except Exception as e:
            last_error = e
            if attempt < max_retries - 1:
                if show_progress:
                    print(f"Download attempt {attempt + 1} failed: {e}")
                    print(f"Retrying in {retry_delay} seconds...")
                time.sleep(retry_delay)
            continue

    # All retries failed
    duration = time.time() - start_time
    error_msg = _categorize_error(last_error)

    return DownloadResult(
        url=url,
        filepath=str(path),
        success=False,
        duration=duration,
        error=error_msg,
    )

download_urls(urls: list[str], filepaths: list[str] | str, overwrite: bool = False, show_progress: bool = True, **kwargs) -> list[DownloadResult]

Download multiple files from URLs.

PARAMETER	DESCRIPTION
urls	List of URLs to download. TYPE: list[str]
filepaths	List of filepaths or directory path. If string (directory), filenames will be inferred from URLs. TYPE: list[str] | str
overwrite	If True, overwrite existing files, by default False. TYPE: bool DEFAULT: False
show_progress	Show download progress, by default True. TYPE: bool DEFAULT: True
**kwargs	Additional arguments passed to download_url(). DEFAULT: {}

RETURNS	DESCRIPTION
list[DownloadResult]	List of download results.

RAISES	DESCRIPTION
ValueError	If urls and filepaths lengths do not match.

Examples:

>>> # Download to specific files
>>> results = download_urls(
...     ["http://example.com/file1.txt", "http://example.com/file2.txt"],
...     ["local1.txt", "local2.txt"],
... )
>>>
>>> # Download to directory (filenames from URLs)
>>> results = download_urls(
...     ["http://example.com/file1.txt", "http://example.com/file2.txt"],
...     "./",
... )

Source code in spectre/utils/download.py

def download_urls(
    urls: list[str],
    filepaths: list[str] | str,
    overwrite: bool = False,
    show_progress: bool = True,
    **kwargs,
) -> list[DownloadResult]:
    """
    Download multiple files from URLs.

    Parameters
    ----------
    urls : list[str]
        List of URLs to download.
    filepaths : list[str] | str
        List of filepaths or directory path. If string (directory),
        filenames will be inferred from URLs.
    overwrite : bool, optional
        If True, overwrite existing files, by default False.
    show_progress : bool, optional
        Show download progress, by default True.
    **kwargs
        Additional arguments passed to download_url().

    Returns
    -------
    list[DownloadResult]
        List of download results.

    Raises
    ------
    ValueError
        If urls and filepaths lengths do not match.

    Examples
    --------
    >>> # Download to specific files
    >>> results = download_urls(
    ...     ["http://example.com/file1.txt", "http://example.com/file2.txt"],
    ...     ["local1.txt", "local2.txt"],
    ... )
    >>>
    >>> # Download to directory (filenames from URLs)
    >>> results = download_urls(
    ...     ["http://example.com/file1.txt", "http://example.com/file2.txt"],
    ...     "./",
    ... )
    """
    if isinstance(filepaths, str):
        # Directory provided - infer filenames from URLs
        directory = Path(filepaths)
        filepaths = []
        for url in urls:
            filename = Path(urlparse(url).path).name
            if not filename:
                filename = f"download_{hash(url) % 10000}.dat"
            filepaths.append(str(directory / filename))

    if len(urls) != len(filepaths):
        raise ValueError(
            f"URLs count ({len(urls)}) must match filepaths count ({len(filepaths)})"
        )

    results = []

    # Sequential download for simplicity
    for i, (url, filepath) in enumerate(zip(urls, filepaths)):
        if show_progress:
            print(f"\nDownloading {i + 1}/{len(urls)}: {url}")

        result = download_url(
            url=url,
            filepath=filepath,
            overwrite=overwrite,
            show_progress=show_progress,
            **kwargs,
        )
        results.append(result)

        if show_progress:
            if result.success:
                size_str = f" ({result.file_size:,} bytes)" if result.file_size else ""
                print(f"✓ Completed: {filepath}{size_str}")
            else:
                print(f"✗ Failed: {result.error}")

    return results

list_datasets(tags: list[str] | None = None, show_details: bool = True) -> None

List available datasets.

PARAMETER	DESCRIPTION
tags	Filter datasets by tags, by default None. TYPE: list[str] | None DEFAULT: None
show_details	Show detailed information, by default True. TYPE: bool DEFAULT: True

Examples:

>>> list_datasets()
>>> list_datasets(tags=["molecular-dynamics"])
>>> list_datasets(show_details=False)  # Just names

Source code in spectre/utils/download.py

def list_datasets(tags: list[str] | None = None, show_details: bool = True) -> None:
    """
    List available datasets.

    Parameters
    ----------
    tags : list[str] | None, optional
        Filter datasets by tags, by default None.
    show_details : bool, optional
        Show detailed information, by default True.

    Examples
    --------
    >>> list_datasets()
    >>> list_datasets(tags=["molecular-dynamics"])
    >>> list_datasets(show_details=False)  # Just names
    """
    filtered_datasets = DATASETS

    if tags:
        filtered_datasets = {
            name: info
            for name, info in DATASETS.items()
            if any(tag in info.get("tags", []) for tag in tags)
        }

    if not filtered_datasets:
        print("No datasets found matching the criteria.")
        return

    print(f"\nAvailable Datasets ({len(filtered_datasets)} found):")
    print("-" * 50)

    for name, info in filtered_datasets.items():
        if show_details:
            print(f"\n {name}")
            print(f"   Description: {info['description']}")
            print(f"   Format: {info['format']}")
            if info["size_mb"]:
                print(f"   Size: {info['size_mb']} MB")
            print(f"   License: {info['license']}")
            if info["tags"]:
                print(f"   Tags: {', '.join(info['tags'])}")  # ty: ignore[no-matching-overload]
            print(f"   URL: {info['url']}")
        else:
            print(f"• {name} - {info['description']}")

    print()

get_dataset_info(dataset_name: str) -> dict

Get detailed information about a specific dataset.

PARAMETER	DESCRIPTION
dataset_name	Name of the dataset. TYPE: str

RETURNS	DESCRIPTION
dict	Dataset information.

RAISES	DESCRIPTION
KeyError	If dataset is not found.

Examples:

>>> info = get_dataset_info("adp-feat-dist-bf2")
>>> print(info["description"])

Source code in spectre/utils/download.py

def get_dataset_info(dataset_name: str) -> dict:
    """
    Get detailed information about a specific dataset.

    Parameters
    ----------
    dataset_name : str
        Name of the dataset.

    Returns
    -------
    dict
        Dataset information.

    Raises
    ------
    KeyError
        If dataset is not found.

    Examples
    --------
    >>> info = get_dataset_info("adp-feat-dist-bf2")
    >>> print(info["description"])
    """
    if dataset_name not in DATASETS:
        available = list(DATASETS.keys())
        raise KeyError(
            f"Dataset '{dataset_name}' not found. Available datasets: {available}"
        )

    return DATASETS[dataset_name].copy()

download_dataset(dataset_name: str, filepath: str | None = None, create_dataset: bool = True, overwrite: bool = False, show_progress: bool = True, **kwargs) -> tuple[DownloadResult, Any | None]

Download a dataset from the registry.

PARAMETER	DESCRIPTION
dataset_name	Name of the dataset to download. TYPE: str
filepath	Custom filepath, by default None (uses dataset name). TYPE: str | None DEFAULT: None
create_dataset	Create DatasetWeighted object, by default True. TYPE: bool DEFAULT: True
overwrite	Overwrite existing files, by default False. TYPE: bool DEFAULT: False
show_progress	Show download progress, by default True. TYPE: bool DEFAULT: True
**kwargs	Additional arguments passed to download_url() and create_dataset(). DEFAULT: {}

RETURNS	DESCRIPTION
tuple[DownloadResult, DatasetWeighted | None]	Download result and optionally created dataset.

RAISES	DESCRIPTION
KeyError	If dataset is not found.

Examples:

>>> # Download and create dataset
>>> result, dataset = download_dataset("adp-feat-dist-bf2")
>>> print(f"Downloaded: {result.success}")
>>> print(f"Dataset shape: {dataset.data.shape}")
>>>
>>> # Just download the file
>>> result, _ = download_dataset(
...     "adp-feat-dist-bf2",
...     filepath="my_data.txt",
...     create_dataset=False,
... )

Source code in spectre/utils/download.py

def download_dataset(
    dataset_name: str,
    filepath: str | None = None,
    create_dataset: bool = True,
    overwrite: bool = False,
    show_progress: bool = True,
    **kwargs,
) -> tuple[DownloadResult, Any | None]:
    """
    Download a dataset from the registry.

    Parameters
    ----------
    dataset_name : str
        Name of the dataset to download.
    filepath : str | None, optional
        Custom filepath, by default None (uses dataset name).
    create_dataset : bool, optional
        Create DatasetWeighted object, by default True.
    overwrite : bool, optional
        Overwrite existing files, by default False.
    show_progress : bool, optional
        Show download progress, by default True.
    **kwargs
        Additional arguments passed to download_url() and create_dataset().

    Returns
    -------
    tuple[DownloadResult, DatasetWeighted | None]
        Download result and optionally created dataset.

    Raises
    ------
    KeyError
        If dataset is not found.

    Examples
    --------
    >>> # Download and create dataset
    >>> result, dataset = download_dataset("adp-feat-dist-bf2")
    >>> print(f"Downloaded: {result.success}")
    >>> print(f"Dataset shape: {dataset.data.shape}")
    >>>
    >>> # Just download the file
    >>> result, _ = download_dataset(
    ...     "adp-feat-dist-bf2",
    ...     filepath="my_data.txt",
    ...     create_dataset=False,
    ... )
    """
    # Get dataset info
    if dataset_name not in DATASETS:
        available = list(DATASETS.keys())
        raise KeyError(
            f"Dataset '{dataset_name}' not found. Available datasets: {available}"
        )

    dataset_info = DATASETS[dataset_name]

    # Show dataset information
    if show_progress:
        print(f"\n📊 Downloading dataset: {dataset_name}")
        print(f"Description: {dataset_info['description']}")
        print(f"License: {dataset_info['license']}")
        print(f"Citation: {dataset_info['citation']}")
        print("-" * 50)

    # Determine filepath
    if filepath is None:
        # Infer extension from format
        ext_map = {"csv": ".csv", "text": ".txt", "plumed": ".dat", "numpy": ".npy"}
        ext = ext_map.get(dataset_info["format"], ".data")
        filepath = f"{dataset_name}{ext}"

    # Download the file
    download_result = download_url(
        url=dataset_info["url"],
        filepath=filepath,
        overwrite=overwrite,
        show_progress=show_progress,
        **kwargs,
    )

    dataset = None
    if create_dataset and download_result.success:
        try:
            # Import here to avoid circular imports
            from spectre.utils.io import create_dataset

            # Prepare arguments for dataset creation
            dataset_kwargs = {
                "target_cols": dataset_info.get("target_cols"),
                "weight_cols": dataset_info.get("weight_cols"),
                "format": dataset_info.get("format", "auto"),
            }

            # Remove None values and add any user kwargs
            dataset_kwargs = {k: v for k, v in dataset_kwargs.items() if v is not None}
            dataset_kwargs.update(kwargs)

            dataset = create_dataset(filepath, **dataset_kwargs)

            if show_progress:
                print("\n✅ Dataset created successfully!")
                print(f"   Data shape: {dataset.data.shape}")
                if dataset.target is not None:
                    target_shape = (
                        dataset.target.shape if dataset.target.ndim > 0 else "scalar"
                    )
                    print(f"   Target shape: {target_shape}")
                if dataset.weights is not None:
                    weight_shape = (
                        dataset.weights.shape if dataset.weights.ndim > 0 else "scalar"
                    )
                    print(f"   Weights shape: {weight_shape}")

        except Exception as e:
            if show_progress:
                print(f"\n⚠️  File downloaded but dataset creation failed: {e}")
                print(f"   You can still use the downloaded file: {filepath}")

    return download_result, dataset