Download a list of specific Sentinel-1 granuale from copernicus dataspace

download_s1_safe.py

#!/usr/bin/env python
# /// script
# dependencies = ["requests", "tqdm"]
# ///
"""Download Sentinel-1 SAFE files from Copernicus Dataspace."""

import sys
from netrc import netrc
from pathlib import Path

import requests
from tqdm.auto import tqdm


def download_s1_safe(
    safe_names: str | list[str], output_dir: str | Path = Path()
) -> list[Path]:
    """Download Sentinel-1 SAFE files from Copernicus Dataspace.

    Parameters
    ----------
    safe_names : str or list of str
        Single SAFE filename or list of SAFE filenames to download
    output_dir : str or Path
        Directory where SAFE files will be saved

    Returns
    -------
    list of Path
        Paths to downloaded SAFE files

    Notes
    -----
    Requires a .netrc file with credentials for 'dataspace.copernicus.eu'

    """
    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    if isinstance(safe_names, str):
        safe_names = [safe_names]

    token = _get_dataspace_token()
    downloaded_files = []
    headers = {"Authorization": f"Bearer {token}"}

    for safe_name in safe_names:
        output_path = output_dir / safe_name
        if output_path.suffix != ".zip":
            output_path = output_path.with_suffix(output_path.suffix + ".zip")
        print(f"Downloading {safe_name} to {output_path}")

        # Query for product ID
        query_response = requests.get(
            "https://catalogue.dataspace.copernicus.eu/odata/v1/Products",
            params={"$filter": f"Name eq '{safe_name}'"},
            headers=headers,
        )
        query_response.raise_for_status()
        product_id = query_response.json()["value"][0]["Id"]
        print(f"Found product ID {product_id}")

        download_url = f"https://download.dataspace.copernicus.eu/odata/v1/Products({product_id})/$value"
        response = requests.get(download_url, headers=headers, stream=True)
        response.raise_for_status()
        total_size = int(response.headers.get("content-length", 0))
        with (
            open(output_path, "wb") as f,
            tqdm(
                total=total_size,
                unit="iB",
                unit_scale=True,
                unit_divisor=1024,
                desc=safe_name,
            ) as pbar,
        ):
            for data in response.iter_content(chunk_size=10 * 4096):
                size = f.write(data)
                pbar.update(size)

        downloaded_files.append(output_path)

    return downloaded_files


def _get_dataspace_token() -> str:
    # Get credentials from .netrc, request token
    host = "dataspace.copernicus.eu"
    username, _, password = netrc().authenticators(host)
    auth_response = requests.post(
        "https://identity.dataspace.copernicus.eu/auth/realms/CDSE/protocol/openid-connect/token",
        data={
            "client_id": "cdse-public",
            "username": username,
            "password": password,
            "grant_type": "password",
        },
    )
    auth_response.raise_for_status()
    token = auth_response.json()["access_token"]
    return token


if __name__ == "__main__":
    filenames = sys.argv[1:]
    download_s1_safe(filenames)

Usage:

pipx run https://gist.githubusercontent.com/scottstanie/08fee9a35f73cd96129039d72d384077/raw/423b843c7c513cdc1e2e32dac1b21f420f0bfbfe/download_s1_safe.py S1C_IW_SLC__1SDV_20250201T231315_20250201T231342_000845_000ED0_ACE6.SAFE S1C_IW_SLC__1SDV_20250120T231315_20250120T231342_000670_0006C0_2225.SAFE