Skip to content

Downloader Service

Overview

The sat_download.services.downloader module provides a high-level facade for searching and downloading satellite images.


API Reference

SatelliteImageDownloader

SatelliteImageDownloader(api: SatelliteAPI, verbose=0)

A class to handle satellite image searching and downloading operations.

This class provides a simplified interface to search for satellite images using various filters and download them.

Parameters:

Name Type Description Default
api SatelliteAPI

The satellite API client to use for API operations

required
See Also

sat_download.api.base.SatelliteAPI : Base class for satellite API clients sat_download.api.odata.ODataAPI : Implementation for Copernicus Data Space API sat_download.api.usgs.USGSAPI : Implementation for USGS Earth Explorer API

Source code in sat_download\services\downloader.py
def __init__(self, api : SatelliteAPI, verbose = 0) -> None:
    self.api = api
    self.verbose = verbose

Functions

search

search(filters: SearchFilters) -> SearchResults

Perform a standard search operation using specified filters.

Parameters:

Name Type Description Default
filters SearchFilters

The search filters to apply to the search

required

Returns:

Type Description
SearchResults

The results from the search operation

Notes

Exceptions are caught and printed to console.

Source code in sat_download\services\downloader.py
def search(self, filters : SearchFilters) -> SearchResults:
    """
    Perform a standard search operation using specified filters.

    Parameters
    ----------
    filters : SearchFilters
        The search filters to apply to the search

    Returns
    -------
    SearchResults
        The results from the search operation

    Notes
    -----
    Exceptions are caught and printed to console.
    """
    try:
        return self.api.search(filters)        
    except Exception as exc:
        print(exc)
bulk_search(filters: SearchFilters) -> SearchResults

Perform a bulk search operation using specified filters.

Parameters:

Name Type Description Default
filters SearchFilters

The search filters to apply to the bulk search

required

Returns:

Type Description
SearchResults

The results from the bulk search operation

Notes

Exceptions are caught and printed to console.

Source code in sat_download\services\downloader.py
def bulk_search(self, filters: SearchFilters) -> SearchResults:
    """
    Perform a bulk search operation using specified filters.

    Parameters
    ----------
    filters : SearchFilters
        The search filters to apply to the bulk search

    Returns
    -------
    SearchResults
        The results from the bulk search operation

    Notes
    -----
    Exceptions are caught and printed to console.
    """
    try:
        return self.api.bulk_search(filters)        
    except Exception as exc:
        print(exc)

bulk_download

bulk_download(
    images: SearchResults, outdir: str
) -> List[str | None]

Download multiple satellite images in bulk.

Parameters:

Name Type Description Default
images SearchResults

The search results containing image IDs and metadata for the images to download.

required
outdir str

The output directory where the images will be saved.

required

Returns:

Type Description
List[str | None]

A list of file paths for successfully downloaded images. If a download fails, the corresponding entry in the list will be None.

Notes
  • This method ensures that the output directory exists before starting the downloads.
  • Each download is attempted individually, and exceptions are logged without halting the process.
  • Logs provide detailed information about successful downloads, warnings for failed downloads, and errors encountered during the process.
Source code in sat_download\services\downloader.py
def bulk_download(self, images: SearchResults, outdir: str) -> List[str | None]:
    """
    Download multiple satellite images in bulk.

    Parameters
    ----------
    images : SearchResults
        The search results containing image IDs and metadata for the images to download.
    outdir : str
        The output directory where the images will be saved.

    Returns
    -------
    List[str | None]
        A list of file paths for successfully downloaded images. If a download fails, 
        the corresponding entry in the list will be None.

    Notes
    -----
    - This method ensures that the output directory exists before starting the downloads.
    - Each download is attempted individually, and exceptions are logged without halting the process.
    - Logs provide detailed information about successful downloads, warnings for failed downloads, 
      and errors encountered during the process.
    """
    try:
        os.makedirs(outdir, exist_ok=True)
        return [ self.api.download(download_id, os.path.join(outdir, image.filename), self.verbose) for download_id, image in images.items() ]
    except Exception as exc:
        print(exc)

Usage Examples

Basic Workflow

from sat_download.api.odata import ODataAPI
from sat_download.services.downloader import SatelliteImageDownloader
from sat_download.data_types.search import SearchFilters
from sat_download.enums import COLLECTIONS

# 1. Create API client
api = ODataAPI(username="user", password="pass")

# 2. Create downloader service
downloader = SatelliteImageDownloader(api, verbose=1)

# 3. Define search filters
filters = SearchFilters(
    collection=COLLECTIONS.SENTINEL_2.value,
    processing_level="L2A",
    start_date="2024-01-01",
    end_date="2024-01-31",
    tile_id="30TWM"
)

# 4. Search for images
results = downloader.search(filters)
print(f"Found {len(results)} images")

# 5. Download images
downloaded = downloader.bulk_download(results, "./images")
print(f"Downloaded {len([f for f in downloaded if f])} files")

Bulk Search for Large Date Ranges

# For large date ranges, use bulk_search
filters = SearchFilters(
    collection=COLLECTIONS.SENTINEL_2.value,
    start_date="2023-01-01",
    end_date="2023-12-31",  # 1 year
    tile_id="30TWM"
)

# bulk_search handles API limits automatically
results = downloader.bulk_search(filters)
print(f"Found {len(results)} images over the year")

Multi-Provider Workflow

from sat_download.api.odata import ODataAPI
from sat_download.api.usgs import USGSAPI
from sat_download.services.downloader import SatelliteImageDownloader

# Copernicus downloader
copernicus_api = ODataAPI(username="user", password="pass")
copernicus_dl = SatelliteImageDownloader(copernicus_api, verbose=1)

# USGS downloader
usgs_api = USGSAPI(username="user", password="token")
usgs_dl = SatelliteImageDownloader(usgs_api, verbose=1)

# Search both providers
sentinel_filters = SearchFilters(
    collection=COLLECTIONS.SENTINEL_2.value,
    start_date="2024-01-01",
    end_date="2024-01-31"
)

landsat_filters = SearchFilters(
    collection=COLLECTIONS.LANDSAT_8.value,
    start_date="2024-01-01",
    end_date="2024-01-31"
)

sentinel_results = copernicus_dl.search(sentinel_filters)
landsat_results = usgs_dl.search(landsat_filters)

# Download to separate directories
copernicus_dl.bulk_download(sentinel_results, "./sentinel")
usgs_dl.bulk_download(landsat_results, "./landsat")

Workflow Diagram

flowchart TD
    User([User]) --> Create[Create API Client]
    Create --> Downloader[Create SatelliteImageDownloader]
    Downloader --> Filters[Define SearchFilters]

    Filters --> SearchType{Search Type?}
    SearchType -->|Simple| Search[search]
    SearchType -->|Bulk| BulkSearch[bulk_search]

    Search --> Results[SearchResults]
    BulkSearch --> Results

    Results --> Download{Download?}
    Download -->|Yes| BulkDownload[bulk_download]
    Download -->|No| End([Done])

    BulkDownload --> Files[Downloaded Files]
    Files --> End

    style User fill:#e1f5fe
    style End fill:#c8e6c9

Error Handling

The SatelliteImageDownloader handles errors gracefully:

# Errors are caught internally and printed
results = downloader.search(filters)  # Returns None on error

# Check for successful results
if results:
    downloaded = downloader.bulk_download(results, "./images")

    # Check which downloads succeeded
    for filepath in downloaded:
        if filepath:
            print(f"Success: {filepath}")
        else:
            print("Download failed")
else:
    print("Search failed or no results")

Facade Pattern Benefits

The SatelliteImageDownloader implements the Facade Pattern:

Benefit Description
Simplicity Single interface for all operations
Error Handling Centralized exception management
Directory Management Automatic directory creation
Decoupling Client isolated from API complexity

Without Facade (Complex)

# Manual approach requires more code
api = ODataAPI(username, password)
filters = SearchFilters(...)

try:
    results = api.search(filters)
except Exception as e:
    print(e)

os.makedirs("./images", exist_ok=True)
for image_id, image in results.items():
    try:
        filepath = os.path.join("./images", image.filename)
        api.download(image_id, filepath, verbose=1)
    except Exception as e:
        print(e)

With Facade (Simple)

# Facade simplifies the workflow
api = ODataAPI(username, password)
downloader = SatelliteImageDownloader(api, verbose=1)

filters = SearchFilters(...)
results = downloader.search(filters)
downloader.bulk_download(results, "./images")

Configuration

Verbosity Levels

Level Description
0 Silent mode (no output)
1 Progress bars for downloads
# Silent downloads
downloader = SatelliteImageDownloader(api, verbose=0)

# With progress bars
downloader = SatelliteImageDownloader(api, verbose=1)

References