From fcec5bade5e6872ffb3584fadc183fcde5807360 Mon Sep 17 00:00:00 2001 From: Javier Tausia Date: Mon, 26 Oct 2026 11:11:50 +0100 Subject: [PATCH 1/4] [JTH] downloaders: remove the AVISO provider It was out of date (L3 path gives 404, L2 points to the superseded PGC0, cycles hard-coded) and unused. SWOT L2 now comes from NASA Earthdata. Co-Authored-By: Claude Opus 5.5 --- .../downloaders/aviso/SWOT/SWOT_config.json | 39 --- bluemath_tk/downloaders/aviso/__init__.py | 11 - .../downloaders/aviso/aviso_downloader.py | 247 ------------------ docs/source/downloaders.rst | 22 +- tests/downloaders/test_structure.py | 1 - 5 files changed, 2 insertions(+), 318 deletions(-) delete mode 100644 bluemath_tk/downloaders/aviso/SWOT/SWOT_config.json delete mode 100644 bluemath_tk/downloaders/aviso/__init__.py delete mode 100644 bluemath_tk/downloaders/aviso/aviso_downloader.py diff --git a/bluemath_tk/downloaders/aviso/SWOT/SWOT_config.json b/bluemath_tk/downloaders/aviso/SWOT/SWOT_config.json deleted file mode 100644 index c1598f2..0000000 --- a/bluemath_tk/downloaders/aviso/SWOT/SWOT_config.json +++ /dev/null @@ -1,39 +0,0 @@ -{ - "datasets": { - "swot-l3-expert": { - "description": "SWOT L3 Expert Product", - "url": "https://tds-odatis.aviso.altimetry.fr/thredds/catalog/dataset-l3-swot-karin-nadir-validated/l3_lr_ssh/v1_0/Expert/catalog.html", - "ftp_base_path": "/swot_products/l3_karin_nadir/l3_lr_ssh/v1_0/Expert/", - "cycles": [ - "cycle_001", - "cycle_002", - "cycle_003", - "cycle_004", - "cycle_005", - "cycle_006", - "cycle_007", - "cycle_008", - "cycle_009" - ] - }, - "swot-l2-expert": { - "description": "SWOT L2 Expert Product", - "url": "https://tds-odatis.aviso.altimetry.fr/thredds/catalog/dataset-l2-swot-karin-lr-ssh-validated/PGC0/Expert/catalog.html", - "ftp_base_path": "/swot_products/l2_karin/l2_lr_ssh/PGC0/Expert/", - "cycles": [ - "cycle_001", - "cycle_002", - "cycle_003", - "cycle_004", - "cycle_005", - "cycle_006", - "cycle_007", - "cycle_008", - "cycle_009" - ] - } - }, - "ftp_server": "ftp-access.aviso.altimetry.fr", - "docs": "https://www.earthdata.nasa.gov/s3fs-public/2024-06/D-109532_SWOT_UserHandbook_20240502.pdf", - "api": "https://podaac.jpl.nasa.gov/SWOT?tab=datasets-information§ions=about" -} \ No newline at end of file diff --git a/bluemath_tk/downloaders/aviso/__init__.py b/bluemath_tk/downloaders/aviso/__init__.py deleted file mode 100644 index 006dcf4..0000000 --- a/bluemath_tk/downloaders/aviso/__init__.py +++ /dev/null @@ -1,11 +0,0 @@ -""" -Project: BlueMath_tk -Sub-Module: downloaders.aviso -Author: GeoOcean Research Group, Universidad de Cantabria -Repository: https://github.com/GeoOcean/BlueMath_tk.git -Status: Under development (Working) -""" - -from .aviso_downloader import AvisoDownloader - -__all__ = ["AvisoDownloader"] diff --git a/bluemath_tk/downloaders/aviso/aviso_downloader.py b/bluemath_tk/downloaders/aviso/aviso_downloader.py deleted file mode 100644 index 6c6d911..0000000 --- a/bluemath_tk/downloaders/aviso/aviso_downloader.py +++ /dev/null @@ -1,247 +0,0 @@ -""" -Downloader for AVISO+ altimetry products, served over FTP. - -Products (configs in ``aviso//_config.json``): - -- SWOT: SWOT L2/L3 KaRIn expert products, one directory per cycle. -""" - -import ftplib -import os -import posixpath -from typing import List, Optional - -from .._base_downloaders import BaseDownloader, load_config -from .._download_result import DownloadResult - -_HERE = os.path.dirname(__file__) - - -class AvisoDownloader(BaseDownloader): - """ - Downloader for AVISO+ altimetry data, served over FTP (requires an account). - - Downloads all .nc files of the requested cycles of a dataset. - The FTP connection is opened on first use, re-opened after failures, and - closed with ``close()`` or when used as a context manager. - - Examples - -------- - >>> with AvisoDownloader( - ... product="SWOT", - ... base_path_to_download="./swot_data", - ... username="your_username", - ... password="your_password" - ... ) as downloader: - ... result = downloader.download_data( - ... dataset="swot-l3-expert", - ... cycles=["cycle_001"], - ... force=False, - ... dry_run=False - ... ) - """ - - products_configs = { - "SWOT": load_config(_HERE, "SWOT", "SWOT_config.json"), - } - - def __init__( - self, - product: str, - base_path_to_download: str, - username: str, - password: str, - debug: bool = True, - **kwargs, - ) -> None: - """ - Initialize the AvisoDownloader. - - Parameters - ---------- - product : str - The product to download data from (e.g., "SWOT"). - base_path_to_download : str - Base path where downloaded files will be stored. - username : str - AVISO FTP username. - password : str - AVISO FTP password. - debug : bool, optional - If True, sets logger to DEBUG level. Default is True. - **kwargs - Passed to BaseDownloader (e.g., retries, retry_backoff). - - Raises - ------ - ValueError - If the product configuration is not found or FTP server is not specified. - """ - - super().__init__( - product=product, - base_path_to_download=base_path_to_download, - debug=debug, - **kwargs, - ) - - self.set_logger_name( - f"AvisoDownloader-{product}", level="DEBUG" if debug else "INFO" - ) - - self._ftp_server = self.product_config.get("ftp_server") - if self._ftp_server is None: - raise ValueError("FTP server not found in product configuration") - self._username = username - self._password = password - self._client: Optional[ftplib.FTP] = None - self._exclude_attributes.extend(["_client", "_password"]) - - self.logger.info(f"---- AVISO DOWNLOADER INITIALIZED ({product}) ----") - - def __enter__(self) -> "AvisoDownloader": - return self - - def __exit__(self, *exc) -> None: - self.close() - - @property - def client(self) -> ftplib.FTP: - """ - FTP client connection, opened and logged in on first use. - - Returns - ------- - ftplib.FTP - FTP client instance. - """ - if self._client is None: - self._client = ftplib.FTP(self._ftp_server, timeout=120) - self._client.login(self._username, self._password) - return self._client - - def close(self) -> None: - """ - Close the FTP connection, if open. - """ - if self._client is not None: - try: - self._client.quit() - except Exception: - self._client.close() - self._client = None - - def download_data_swot( - self, - dataset: str, - cycles: Optional[List[str]] = None, - force: bool = False, - dry_run: bool = True, - ) -> DownloadResult: - """ - Download SWOT data for a specific dataset. - - Downloads all .nc files from specified cycles. Files are saved to: - base_path_to_download/dataset/cycle/filename.nc - - Parameters - ---------- - dataset : str - The dataset to download (e.g., "swot-l3-expert"). - Use list_datasets() to see available datasets. - cycles : List[str], optional - List of cycle folder names to download (e.g., ["cycle_001", "cycle_002"]). - If None, uses cycles from dataset configuration. Default is None. - force : bool, optional - Force re-download even if file exists. Default is False. - dry_run : bool, optional - If True, only check what would be downloaded. Default is True. - - Returns - ------- - DownloadResult - Result with all downloaded files and download statistics. - - Raises - ------ - ValueError - If dataset is not found or no cycles are available. - """ - - if dataset not in self.list_datasets(): - raise ValueError( - f"Dataset '{dataset}' not found. Available: {self.list_datasets()}" - ) - - dataset_config = self.product_config["datasets"][dataset] - ftp_base_path = dataset_config["ftp_base_path"] - if cycles is None: - cycles = dataset_config.get("cycles", []) - if not cycles: - raise ValueError( - f"No cycles specified for dataset '{dataset}' and cycles parameter not provided" - ) - - self.logger.info(f"Downloading dataset: {dataset}, cycles: {cycles}") - result = self.create_download_result() - - for cycle in cycles: - remote_dir = posixpath.join(ftp_base_path, cycle) - try: - files = self._list_nc_files(remote_dir) - except Exception as e: - self.logger.error(f"Error listing {remote_dir}: {e}") - result.add_error(remote_dir, e) - continue - if not files: - self.logger.warning(f"No files found in cycle {cycle}") - continue - - for filename in files: - local_path = os.path.join( - self.base_path_to_download, dataset, cycle, filename - ) - remote_path = posixpath.join(remote_dir, filename) - - def fetch(part_file: str, remote_path: str = remote_path) -> None: - try: - with open(part_file, "wb") as f: - self.client.retrbinary(f"RETR {remote_path}", f.write) - except Exception: - # The connection may be broken: reconnect on next attempt - self.close() - raise - - self.fetch_file(local_path, fetch, result, force=force, dry_run=dry_run) - - return self.finalize_download_result(result) - - def _list_nc_files(self, remote_dir: str) -> List[str]: - """ - List the .nc files (names only) in a directory of the FTP server. - - Parameters - ---------- - remote_dir : str - Absolute FTP directory (e.g., "/swot_products/.../Expert/cycle_001"). - - Returns - ------- - List[str] - Sorted list of .nc filenames. - """ - - try: - names = self.client.nlst(remote_dir) - except ftplib.error_perm as e: - # Some servers answer "550 No files found" for empty directories - if str(e).startswith("550"): - return [] - raise - except (ftplib.error_temp, OSError, EOFError): - self.close() - names = self.client.nlst(remote_dir) - - return sorted( - {posixpath.basename(name) for name in names if name.endswith(".nc")} - ) diff --git a/docs/source/downloaders.rst b/docs/source/downloaders.rst index 612999e..d7d6902 100644 --- a/docs/source/downloaders.rst +++ b/docs/source/downloaders.rst @@ -30,10 +30,6 @@ Providers and products - ``NOAADownloader`` - NDBC - NDBC buoy bulk parameters, wave spectra and directional spectra. - * - ``aviso`` - - ``AvisoDownloader`` - - SWOT - - AVISO+ FTP server (requires an account). * - ``ecmwf`` - ``ECMWFDownloader`` - OpenData @@ -197,8 +193,8 @@ Files are saved to Yearly files with the previous naming (``__01_02_..._12.nc``) are recognised and not downloaded again. -NOAA, AVISO and ECMWF ---------------------- +NOAA and ECMWF +-------------- .. code-block:: python @@ -210,20 +206,6 @@ NOAA, AVISO and ECMWF data_type="wave_spectra", buoy_id="41001", years=[2022, 2023], dry_run=False ) -.. code-block:: python - - from bluemath_tk.downloaders.aviso import AvisoDownloader - - with AvisoDownloader( - product="SWOT", - base_path_to_download="./swot", - username="user", - password="password", - ) as downloader: - result = downloader.download_data( - dataset="swot-l3-expert", cycles=["cycle_001"], dry_run=False - ) - .. code-block:: python from bluemath_tk.downloaders.ecmwf import ECMWFDownloader diff --git a/tests/downloaders/test_structure.py b/tests/downloaders/test_structure.py index 6fc1280..259e5ce 100644 --- a/tests/downloaders/test_structure.py +++ b/tests/downloaders/test_structure.py @@ -9,7 +9,6 @@ # provider package -> downloader class PROVIDERS = { - "aviso": "AvisoDownloader", "copernicus": "CopernicusDownloader", "csiro": "CSIRODownloader", "ecmwf": "ECMWFDownloader", From 59ca531a7b96f221a276f527f8a583dbd9d3e0ff Mon Sep 17 00:00:00 2001 From: Javier Tausia Date: Mon, 26 Oct 2026 11:15:41 +0100 Subject: [PATCH 2/4] [JTH] downloaders: EarthdataDownloader for NASA CMR collections (SWOT) New provider bluemath_tk.downloaders.earthdata: CMR granule search (paged with CMR-Search-After; time range, bbox, updated_since), HTTPS downloads with the Earthdata login from ~/.netrc, parallel and resumable, checked against the CMR size and checksum. Product SWOT with dataset L2_LR_SSH_WINDWAVE_D (files saved by month of the pass start). Co-Authored-By: Claude Opus 5.5 --- .../earthdata/SWOT/SWOT_config.json | 13 + bluemath_tk/downloaders/earthdata/__init__.py | 11 + .../earthdata/earthdata_downloader.py | 539 ++++++++++++++++++ docs/source/downloaders.rst | 58 ++ tests/downloaders/test_earthdata.py | 259 +++++++++ tests/downloaders/test_structure.py | 1 + 6 files changed, 881 insertions(+) create mode 100644 bluemath_tk/downloaders/earthdata/SWOT/SWOT_config.json create mode 100644 bluemath_tk/downloaders/earthdata/__init__.py create mode 100644 bluemath_tk/downloaders/earthdata/earthdata_downloader.py create mode 100644 tests/downloaders/test_earthdata.py diff --git a/bluemath_tk/downloaders/earthdata/SWOT/SWOT_config.json b/bluemath_tk/downloaders/earthdata/SWOT/SWOT_config.json new file mode 100644 index 0000000..3c1afe4 --- /dev/null +++ b/bluemath_tk/downloaders/earthdata/SWOT/SWOT_config.json @@ -0,0 +1,13 @@ +{ + "description": "Surface Water and Ocean Topography (SWOT) mission, CNES/NASA, distributed by PO.DAAC", + "cmr_url": "https://cmr.earthdata.nasa.gov/search/granules.umm_json", + "datasets": { + "L2_LR_SSH_WINDWAVE_D": { + "description": "KaRIn L2 low-rate wind and wave product (swh_karin, wind_speed_karin, sig0_karin), 2 km swath grid, one file per half-orbit pass, version D", + "short_name": "SWOT_L2_LR_SSH_WINDWAVE_D", + "filename_pattern": "^SWOT_L2_LR_SSH_WindWave_.*\\.nc$", + "local_dir": "SWOT/L2_LR_SSH_WINDWAVE_D/{year}-{month:02d}", + "start": "2023-03" + } + } +} diff --git a/bluemath_tk/downloaders/earthdata/__init__.py b/bluemath_tk/downloaders/earthdata/__init__.py new file mode 100644 index 0000000..90a55c9 --- /dev/null +++ b/bluemath_tk/downloaders/earthdata/__init__.py @@ -0,0 +1,11 @@ +""" +Project: BlueMath_tk +Sub-Module: downloaders.earthdata +Author: GeoOcean Research Group, Universidad de Cantabria +Repository: https://github.com/GeoOcean/BlueMath_tk.git +Status: Under development (Working) +""" + +from .earthdata_downloader import EarthdataDownloader + +__all__ = ["EarthdataDownloader"] diff --git a/bluemath_tk/downloaders/earthdata/earthdata_downloader.py b/bluemath_tk/downloaders/earthdata/earthdata_downloader.py new file mode 100644 index 0000000..b28cad2 --- /dev/null +++ b/bluemath_tk/downloaders/earthdata/earthdata_downloader.py @@ -0,0 +1,539 @@ +""" +Downloader for NASA Earthdata collections (PO.DAAC and the other NASA DAACs). + +Products (configs in ``earthdata//_config.json``): + +- SWOT: Surface Water and Ocean Topography mission, KaRIn L2 products. + +Granules are found with the NASA CMR search API (public, no login) and +downloaded over HTTPS. Downloads redirect to the Earthdata login +(urs.earthdata.nasa.gov), which reads the credentials from ``~/.netrc``:: + + machine urs.earthdata.nasa.gov login password + +Each dataset of a product is one CMR collection, given by its short name. +""" + +import hashlib +import netrc +import os +import re +import threading +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime +from typing import List, Optional, Sequence, Union + +import pandas as pd +import requests + +from .._base_downloaders import BaseDownloader, load_config +from .._download_result import DownloadResult +from .._transfer import http_download, http_get, make_session, retry_call + +_HERE = os.path.dirname(__file__) +URS_HOST = "urs.earthdata.nasa.gov" +CMR_PAGE_SIZE = 2000 +GRANULE_COLUMNS = [ + "granule_ur", + "filename", + "url", + "size", + "checksum_algorithm", + "checksum", + "start_time", + "end_time", + "revision_date", +] + +TimeLike = Union[str, datetime, pd.Timestamp] + + +def cmr_time_range(start: Optional[TimeLike], end: Optional[TimeLike]) -> str: + """ + CMR ``temporal`` parameter for a time range. + + Strings are read at their own resolution, so ``end="2024-06"`` covers the + whole of June 2024 and ``end="2024-06-15"`` the whole day. + + Parameters + ---------- + start, end : str or datetime, optional + First and last time (both included). None leaves that side open. + + Returns + ------- + str + ``","`` in ISO 8601 UTC, with empty sides when open. + """ + + def to_period(value: TimeLike) -> pd.Period: + if isinstance(value, str): + return pd.Period(value) + return pd.Period(pd.Timestamp(value), freq="s") + + parts = [] + if start is not None: + parts.append(to_period(start).start_time.strftime("%Y-%m-%dT%H:%M:%SZ")) + else: + parts.append("") + if end is not None: + parts.append(to_period(end).end_time.strftime("%Y-%m-%dT%H:%M:%SZ")) + else: + parts.append("") + + return ",".join(parts) + + +def parse_granule(item: dict) -> List[dict]: + """ + Data files of one CMR granule (UMM-JSON item). + + Parameters + ---------- + item : dict + One element of ``items`` in a ``granules.umm_json`` response. + + Returns + ------- + List[dict] + One row per HTTPS data file ("GET DATA" link), with the columns in + GRANULE_COLUMNS. Size and checksum come from the granule's + ``DataGranule`` information and are None if it does not list the file. + """ + + umm = item["umm"] + files_info = { + f.get("Name"): f + for f in umm.get("DataGranule", {}).get("ArchiveAndDistributionInformation", []) + } + time_range = umm.get("TemporalExtent", {}).get("RangeDateTime", {}) + + rows = [] + for link in umm.get("RelatedUrls", []): + url = link.get("URL", "") + if link.get("Type") != "GET DATA" or not url.startswith("https://"): + continue + filename = url.rsplit("/", 1)[-1] + info = files_info.get(filename, {}) + checksum = info.get("Checksum", {}) + rows.append( + { + "granule_ur": umm.get("GranuleUR"), + "filename": filename, + "url": url, + "size": info.get("SizeInBytes"), + "checksum_algorithm": checksum.get("Algorithm"), + "checksum": checksum.get("Value"), + "start_time": time_range.get("BeginningDateTime"), + "end_time": time_range.get("EndingDateTime"), + "revision_date": item.get("meta", {}).get("revision-date"), + } + ) + + return rows + + +def file_checksum(path: str, algorithm: str) -> str: + """ + Hex digest of a file ("MD5", "SHA-256", ... as written in CMR). + """ + + digest = hashlib.new(algorithm.lower().replace("-", "")) + with open(path, "rb") as f: + for block in iter(lambda: f.read(8 * 1024 * 1024), b""): + digest.update(block) + + return digest.hexdigest() + + +class EarthdataDownloader(BaseDownloader): + """ + Downloader for NASA Earthdata collections, found through the CMR search API. + + Files are saved to base_path_to_download/, where + ``local_dir`` may use ``{year}`` and ``{month}`` of the granule start time. + Existing files are compared with the size listed in CMR, transfers are + resumable and checked against the CMR checksum, and several files are + downloaded in parallel (``max_workers``). + + Examples + -------- + >>> downloader = EarthdataDownloader( + ... product="SWOT", + ... base_path_to_download="/lustre/geocean/DATA/NASA", + ... ) + >>> granules = downloader.search_granules( + ... "L2_LR_SSH_WINDWAVE_D", start="2024-06", end="2024-06", + ... bbox=(2.5, 51.0, 7.5, 54.0), + ... ) + >>> result = downloader.download_data( + ... "L2_LR_SSH_WINDWAVE_D", start="2024-06", end="2024-06", + ... bbox=(2.5, 51.0, 7.5, 54.0), dry_run=False, + ... ) + """ + + products_configs = { + "SWOT": load_config(_HERE, "SWOT", "SWOT_config.json"), + } + + def __init__( + self, + product: str, + base_path_to_download: str, + max_workers: int = 4, + debug: bool = True, + **kwargs, + ) -> None: + """ + Initialize the EarthdataDownloader. + + Parameters + ---------- + product : str + The product to download data from (e.g. "SWOT"). + base_path_to_download : str + Base path where downloaded files will be stored. + max_workers : int, optional + Number of files downloaded in parallel. Default is 4. + debug : bool, optional + If True, sets logger to DEBUG level. Default is True. + **kwargs + Passed to BaseDownloader (e.g., retries, retry_backoff). + """ + + super().__init__( + product=product, + base_path_to_download=base_path_to_download, + debug=debug, + **kwargs, + ) + + self.max_workers = max_workers + self._local = threading.local() + self._exclude_attributes.append("_local") + + self.set_logger_name( + f"EarthdataDownloader-{product}", level="DEBUG" if debug else "INFO" + ) + self.logger.info(f"---- EARTHDATA DOWNLOADER INITIALIZED ({product}) ----") + + @property + def session(self) -> requests.Session: + """ + HTTP session of the current thread. + + requests reads ``~/.netrc`` for the Earthdata login host when a download + redirects there; the session then keeps the login cookies. + + Returns + ------- + requests.Session + Session with retries for connection errors and 5xx responses. + """ + if not hasattr(self._local, "session"): + self._local.session = make_session() + return self._local.session + + def search_granules( + self, + dataset: str, + start: Optional[TimeLike] = None, + end: Optional[TimeLike] = None, + bbox: Optional[Sequence[float]] = None, + updated_since: Optional[TimeLike] = None, + ) -> pd.DataFrame: + """ + Data files of a dataset, from the CMR granule search. + + Parameters + ---------- + dataset : str + The dataset name. Use list_datasets() to see available datasets. + start, end : str or datetime, optional + Time range (both included, see cmr_time_range). Granules that + overlap it are returned. Default is the whole dataset. + bbox : Sequence[float], optional + (west, south, east, north) in degrees, longitudes in -180/180. + updated_since : str or datetime, optional + Only granules added or changed in CMR since then. + + Returns + ------- + pd.DataFrame + One row per file (columns GRANULE_COLUMNS), sorted by start time. + Files that do not match the dataset ``filename_pattern`` are left out. + """ + + dataset_config = self._get_dataset_config(dataset) + if start is None: + start = dataset_config.get("start") + params = { + "short_name": dataset_config["short_name"], + "page_size": CMR_PAGE_SIZE, + "sort_key": "start_date", + } + if start is not None or end is not None: + params["temporal"] = cmr_time_range(start, end) + if bbox is not None: + if len(bbox) != 4: + raise ValueError("bbox must be (west, south, east, north)") + params["bounding_box"] = ",".join(str(float(v)) for v in bbox) + if updated_since is not None: + params["updated_since"] = pd.Timestamp(updated_since).strftime( + "%Y-%m-%dT%H:%M:%SZ" + ) + + rows = [] + search_after = None + while True: + headers = {"CMR-Search-After": search_after} if search_after else {} + + def get_page() -> requests.Response: + response = http_get( + self.session, + self.product_config["cmr_url"], + params=params, + headers=headers, + ) + response.json() # retry truncated responses too + return response + + response = retry_call( + get_page, + retries=self.retries, + backoff=self.retry_backoff, + logger=self.logger, + ) + items = response.json().get("items", []) + for item in items: + rows.extend(parse_granule(item)) + search_after = response.headers.get("CMR-Search-After") + if not items or not search_after: + break + + granules = pd.DataFrame(rows, columns=GRANULE_COLUMNS) + pattern = dataset_config.get("filename_pattern") + if pattern: + keep = granules["filename"].str.match(pattern) + if (~keep).any(): + self.logger.info( + f"{(~keep).sum()} files do not match {pattern} and are left out" + ) + granules = granules[keep] + granules = granules.drop_duplicates("filename").sort_values("start_time") + self.logger.info( + f"{len(granules)} files in {dataset_config['short_name']}" + f" ({granules['size'].sum() / 1e9:,.1f} GB)" + ) + + return granules.reset_index(drop=True) + + def get_local_path(self, dataset: str, filename: str, start_time: TimeLike) -> str: + """ + Local path of a file. + + Parameters + ---------- + dataset : str + The dataset name. + filename : str + The file name. + start_time : str or datetime + Start time of the granule, used to fill ``{year}`` and ``{month}`` + of the dataset ``local_dir``. + + Returns + ------- + str + The local path. + """ + + dataset_config = self._get_dataset_config(dataset) + start_time = pd.Timestamp(start_time) + local_dir = dataset_config["local_dir"].format( + year=start_time.year, month=start_time.month + ) + + return os.path.normpath( + os.path.join(self.base_path_to_download, local_dir, filename) + ) + + def download_data( + self, + dataset: str, + start: Optional[TimeLike] = None, + end: Optional[TimeLike] = None, + bbox: Optional[Sequence[float]] = None, + updated_since: Optional[TimeLike] = None, + force: bool = False, + verify_existing: bool = True, + dry_run: bool = True, + ) -> DownloadResult: + """ + Download the files of a dataset that match a search. + + The search arguments are those of search_granules. The same call can be + repeated to keep a dataset up to date: complete files are skipped. + + Parameters + ---------- + dataset : str + The dataset to download. Use list_datasets() to see available datasets. + start, end : str or datetime, optional + Time range (both included). Default is the whole dataset. + bbox : Sequence[float], optional + (west, south, east, north) in degrees. + updated_since : str or datetime, optional + Only granules added or changed in CMR since then. + force : bool, optional + Download again even if the file exists. Default is False. + verify_existing : bool, optional + Compare the size of existing files with the CMR size and download + them again if they differ. Default is True. + dry_run : bool, optional + If True, only check what would be downloaded. Default is True. + + Returns + ------- + DownloadResult + Result with information about downloaded, skipped, and error files. + The CMR size of each file is in metadata["remote_sizes"]. + """ + + granules = self.search_granules(dataset, start, end, bbox, updated_since) + + return self.download_granules( + dataset, + granules, + force=force, + verify_existing=verify_existing, + dry_run=dry_run, + ) + + def download_granules( + self, + dataset: str, + granules: pd.DataFrame, + force: bool = False, + verify_existing: bool = True, + dry_run: bool = True, + ) -> DownloadResult: + """ + Download the files of a table returned by search_granules. + + Parameters + ---------- + dataset : str + The dataset the files belong to. + granules : pd.DataFrame + Files to download (output of search_granules, possibly filtered). + force, verify_existing, dry_run : bool, optional + As in download_data. + + Returns + ------- + DownloadResult + Result with information about downloaded, skipped, and error files. + """ + + if not dry_run: + self.check_credentials() + + self.logger.info(f"Checking {len(granules)} files of {self.product}/{dataset}") + result = self.create_download_result() + with ThreadPoolExecutor(max_workers=self.max_workers) as executor: + futures = [ + executor.submit( + self._download_file, + row, + self.get_local_path(dataset, row.filename, row.start_time), + force, + verify_existing, + dry_run, + ) + for row in granules.itertuples(index=False) + ] + for future in futures: + result.merge(future.result()) + + return self.finalize_download_result(result) + + @staticmethod + def check_credentials() -> None: + """ + Check that ``~/.netrc`` has an entry for the Earthdata login. + + Raises + ------ + ValueError + If ``~/.netrc`` is missing or has no ``urs.earthdata.nasa.gov`` entry. + """ + + try: + entry = netrc.netrc().authenticators(URS_HOST) + except (FileNotFoundError, netrc.NetrcParseError) as e: + raise ValueError(f"Cannot read ~/.netrc: {e}") from e + if entry is None: + raise ValueError( + f"~/.netrc has no entry for {URS_HOST}: add " + f"'machine {URS_HOST} login password ' (chmod 600)" + ) + + def _get_dataset_config(self, dataset: str) -> dict: + if dataset not in self.list_datasets(): + raise ValueError( + f"Dataset '{dataset}' not found. Available: {self.list_datasets()}" + ) + return self.product_config["datasets"][dataset] + + def _download_file( + self, + granule, + output_file: str, + force: bool, + verify_existing: bool, + dry_run: bool, + ) -> DownloadResult: + """ + Download a single file, comparing an existing file with the CMR size. + """ + + result = DownloadResult() + size = None if pd.isna(granule.size) else int(granule.size) + if size is not None: + result.metadata["remote_sizes"] = {output_file: size} + + if os.path.exists(output_file) and not force: + local_size = os.path.getsize(output_file) + if not verify_existing or size is None or local_size == size: + result.add_skipped(output_file, "Already downloaded") + return result + message = f"Local size {local_size} differs from CMR size {size}" + if dry_run: + result.add_skipped( + output_file, f"{message}, would download again (dry run)" + ) + return result + self.logger.warning(f"{message}, downloading again: {output_file}") + force = True + + def fetch(part_file: str) -> None: + http_download( + self.session, granule.url, part_file, expected_size=size, resume=True + ) + if isinstance(granule.checksum, str) and isinstance( + granule.checksum_algorithm, str + ): + checksum = file_checksum(part_file, granule.checksum_algorithm) + if checksum != granule.checksum.lower(): + os.remove(part_file) + raise IOError( + f"{granule.checksum_algorithm} of {granule.filename} is " + f"{checksum}, CMR says {granule.checksum}" + ) + + self.fetch_file( + output_file, fetch, result, force=force, dry_run=dry_run, resumable=True + ) + + return result diff --git a/docs/source/downloaders.rst b/docs/source/downloaders.rst index d7d6902..13ceeac 100644 --- a/docs/source/downloaders.rst +++ b/docs/source/downloaders.rst @@ -26,6 +26,12 @@ Providers and products - ``CopernicusDownloader`` - ERA5, CERRA - Copernicus Climate Data Store (``cdsapi``). Parallel requests. + * - ``earthdata`` + - ``EarthdataDownloader`` + - SWOT + - NASA Earthdata collections (PO.DAAC, ...), found with the CMR search API + and downloaded over HTTPS with an Earthdata login (``~/.netrc``). + Parallel, resumable, checked against the CMR size and checksum. * - ``noaa`` - ``NOAADownloader`` - NDBC @@ -193,6 +199,58 @@ Files are saved to Yearly files with the previous naming (``__01_02_..._12.nc``) are recognised and not downloaded again. +NASA Earthdata: SWOT +-------------------- + +Each dataset is one NASA CMR collection. The search is public; downloads need +a free Earthdata account in ``~/.netrc`` (``chmod 600``):: + + machine urs.earthdata.nasa.gov login password + +.. code-block:: python + + from bluemath_tk.downloaders.earthdata import EarthdataDownloader + + downloader = EarthdataDownloader( + product="SWOT", + base_path_to_download="/lustre/geocean/DATA/NASA", + max_workers=4, # parallel downloads + ) + downloader.list_datasets() # ['L2_LR_SSH_WINDWAVE_D'] + + # Table of files (URL, size, checksum, start/end time), sorted by time + granules = downloader.search_granules( + "L2_LR_SSH_WINDWAVE_D", + start="2024-06", # strings are read at their resolution: + end="2024-06", # this is the whole of June 2024 + bbox=(2.5, 51.0, 7.5, 54.0), # west, south, east, north + ) + + result = downloader.download_data( + "L2_LR_SSH_WINDWAVE_D", + start="2024-06", + end="2024-06", + bbox=(2.5, 51.0, 7.5, 54.0), + dry_run=True, + ) + print(result.summary()) # includes the total size to download + +Files are saved to ``/SWOT/L2_LR_SSH_WINDWAVE_D//``, the month +of the granule start time. ``updated_since="2026-10-01"`` only lists granules +added or changed in CMR since then, which keeps a local copy up to date +cheaply. ``download_granules(dataset, granules)`` downloads a table returned by +``search_granules``, for example after filtering it. + +The downloader mirrors what CMR lists. For SWOT this means that a pass can be +there several times (product counter ``_01``, ``_02``, ... and processing +versions such as ``PGD0`` reprocessed / ``PID0`` forward processing); choosing +one version per pass is left to whoever reads the files. Files that do not +match the dataset ``filename_pattern`` (a few granules of other SWOT products +are listed in the WindWave collection) are left out. + +To add another NASA collection, add a dataset with its CMR ``short_name`` to a +product configuration (or a new ``/_config.json``). + NOAA and ECMWF -------------- diff --git a/tests/downloaders/test_earthdata.py b/tests/downloaders/test_earthdata.py new file mode 100644 index 0000000..ba6b846 --- /dev/null +++ b/tests/downloaders/test_earthdata.py @@ -0,0 +1,259 @@ +import hashlib +import os +import tempfile +import unittest +from collections import namedtuple +from datetime import datetime +from unittest import mock + +from test_transfer import LocalServerTestCase + +from bluemath_tk.downloaders.earthdata import earthdata_downloader +from bluemath_tk.downloaders.earthdata.earthdata_downloader import ( + GRANULE_COLUMNS, + EarthdataDownloader, + cmr_time_range, + parse_granule, +) + +DATASET = "L2_LR_SSH_WINDWAVE_D" +ARCHIVE = "https://archive.example/protected/SWOT_L2_LR_SSH_D" +Granule = namedtuple("Granule", GRANULE_COLUMNS) + + +def swot_name(product="WindWave", cycle_pass="016_085", times=None): + times = times or "20240601T045132_20240601T054300" + return f"SWOT_L2_LR_SSH_{product}_{cycle_pass}_{times}_PGD0_01.nc" + + +def cmr_item(name, start="2024-06-01T04:51:17.972Z", size=1000, md5="abc"): + """Minimal CMR UMM-JSON item, with the links of a real SWOT granule.""" + return { + "meta": {"concept-id": "G1-POCLOUD", "revision-date": "2026-03-11T04:39:38Z"}, + "umm": { + "GranuleUR": name.replace(".nc", "_swot"), + "TemporalExtent": { + "RangeDateTime": { + "BeginningDateTime": start, + "EndingDateTime": "2024-06-01T05:43:15.472Z", + } + }, + "DataGranule": { + "ArchiveAndDistributionInformation": [ + {"Name": name.replace(".nc", ".log"), "SizeInBytes": 5}, + { + "Name": name, + "SizeInBytes": size, + "Checksum": {"Value": md5, "Algorithm": "MD5"}, + }, + ] + }, + "RelatedUrls": [ + {"Type": "EXTENDED METADATA", "URL": f"{ARCHIVE}/{name}.iso.xml"}, + {"Type": "GET DATA", "URL": f"{ARCHIVE}/{name}"}, + {"Type": "GET DATA VIA DIRECT ACCESS", "URL": f"s3://bucket/{name}"}, + ], + }, + } + + +class FakeResponse: + def __init__(self, items, search_after=None): + self._items = items + self.headers = {"CMR-Search-After": search_after} if search_after else {} + + def json(self): + return {"items": self._items} + + +class TestCmrTimeRange(unittest.TestCase): + def test_resolution_of_strings(self): + self.assertEqual( + cmr_time_range("2024-06", "2024-06"), + "2024-06-01T00:00:00Z,2024-06-30T23:59:59Z", + ) + self.assertEqual( + cmr_time_range("2024-02-29", "2024-02-29"), + "2024-02-29T00:00:00Z,2024-02-29T23:59:59Z", + ) + + def test_open_sides_and_datetimes(self): + self.assertEqual(cmr_time_range("2024", None), "2024-01-01T00:00:00Z,") + self.assertEqual( + cmr_time_range(None, datetime(2024, 1, 2, 3, 4, 5)), + ",2024-01-02T03:04:05Z", + ) + + +class TestParseGranule(unittest.TestCase): + def test_only_https_data_files(self): + name = swot_name() + rows = parse_granule(cmr_item(name, size=12428786, md5="c1658")) + self.assertEqual(len(rows), 1) + row = rows[0] + self.assertEqual(sorted(row), sorted(GRANULE_COLUMNS)) + self.assertEqual(row["filename"], name) + self.assertEqual(row["url"], f"{ARCHIVE}/{name}") + self.assertEqual(row["size"], 12428786) + self.assertEqual((row["checksum_algorithm"], row["checksum"]), ("MD5", "c1658")) + self.assertEqual(row["start_time"], "2024-06-01T04:51:17.972Z") + + def test_file_not_listed_in_data_granule(self): + item = cmr_item("a.nc") + item["umm"]["DataGranule"]["ArchiveAndDistributionInformation"] = [] + row = parse_granule(item)[0] + self.assertIsNone(row["size"]) + self.assertIsNone(row["checksum"]) + + +class TestSearchGranules(unittest.TestCase): + def setUp(self): + self.downloader = EarthdataDownloader( + product="SWOT", base_path_to_download="/data", retries=1, retry_backoff=0 + ) + + def test_unknown_dataset(self): + with self.assertRaises(ValueError): + self.downloader.search_granules("nope") + + def test_paging_filters_and_pattern(self): + late = swot_name(cycle_pass="016_086", times="20240601T054300_20240601T063400") + early = swot_name() + stray = swot_name(product="Expert") + pages = [ + FakeResponse([cmr_item(late, start="2024-06-01T05:43:00Z")], "[1,2]"), + FakeResponse([cmr_item(early), cmr_item(stray)]), + ] + calls = [] + + def fake_get(session, url, params=None, headers=None): + calls.append((dict(params), dict(headers))) + return pages[len(calls) - 1] + + with mock.patch.object(earthdata_downloader, "http_get", fake_get): + granules = self.downloader.search_granules( + DATASET, + start="2024-06", + end="2024-06", + bbox=(2.5, 51, 7.5, 54), + updated_since="2026-10-01", + ) + + self.assertEqual(list(granules["filename"]), [early, late]) + params, headers = calls[0] + self.assertEqual(params["short_name"], "SWOT_L2_LR_SSH_WINDWAVE_D") + self.assertEqual( + params["temporal"], "2024-06-01T00:00:00Z,2024-06-30T23:59:59Z" + ) + self.assertEqual(params["bounding_box"], "2.5,51.0,7.5,54.0") + self.assertEqual(params["updated_since"], "2026-10-01T00:00:00Z") + self.assertEqual(headers, {}) + self.assertEqual(calls[1][1], {"CMR-Search-After": "[1,2]"}) + + def test_default_start_is_dataset_start(self): + calls = [] + + def fake_get(session, url, params=None, headers=None): + calls.append(dict(params)) + return FakeResponse([]) + + with mock.patch.object(earthdata_downloader, "http_get", fake_get): + granules = self.downloader.search_granules(DATASET) + self.assertEqual(calls[0]["temporal"], "2023-03-01T00:00:00Z,") + self.assertTrue(granules.empty) + + def test_local_path_uses_month_of_start(self): + self.assertEqual( + self.downloader.get_local_path(DATASET, "f.nc", "2024-05-31T23:58:00Z"), + os.path.normpath("/data/SWOT/L2_LR_SSH_WINDWAVE_D/2024-05/f.nc"), + ) + + +class TestCredentials(unittest.TestCase): + def check_with_netrc(self, content): + with tempfile.TemporaryDirectory() as home: + path = os.path.join(home, ".netrc") + with open(path, "w") as f: + f.write(content) + os.chmod(path, 0o600) + with mock.patch.dict(os.environ, {"HOME": home}): + EarthdataDownloader.check_credentials() + + def test_entry_found(self): + self.check_with_netrc( + "machine urs.earthdata.nasa.gov login user password secret\n" + ) + + def test_entry_missing(self): + with self.assertRaises(ValueError): + self.check_with_netrc("machine other.host login user password secret\n") + + +class TestEarthdataDownloadFile(LocalServerTestCase): + def setUp(self): + super().setUp() + self.downloader = EarthdataDownloader( + product="SWOT", + base_path_to_download=self.tmp_dir.name, + retries=1, + retry_backoff=0, + ) + self.output_file = os.path.join(self.tmp_dir.name, "data.bin") + self.md5 = hashlib.md5(self.content).hexdigest() + + def granule(self, size=None, checksum=None): + return Granule( + granule_ur="data", + filename="data.bin", + url=f"{self.base_url}/data.bin", + size=len(self.content) if size is None else size, + checksum_algorithm="MD5", + checksum=self.md5 if checksum is None else checksum, + start_time="2024-06-01T00:00:00Z", + end_time="2024-06-01T01:00:00Z", + revision_date="2026-01-01T00:00:00Z", + ) + + def download(self, granule, dry_run=False): + return self.downloader._download_file( + granule, + self.output_file, + force=False, + verify_existing=True, + dry_run=dry_run, + ) + + def test_download_and_skip(self): + result = self.download(self.granule()) + self.assertEqual(result.downloaded_files, [self.output_file]) + with open(self.output_file, "rb") as f: + self.assertEqual(f.read(), self.content) + self.assertEqual( + result.metadata["remote_sizes"], {self.output_file: len(self.content)} + ) + + result = self.download(self.granule()) + self.assertEqual(result.skipped_files, [self.output_file]) + + def test_wrong_checksum_is_an_error(self): + result = self.download(self.granule(checksum="0" * 32)) + self.assertEqual(result.error_files, [self.output_file]) + self.assertFalse(os.path.exists(self.output_file)) + self.assertFalse(os.path.exists(self.output_file + ".part")) + + def test_incomplete_local_file(self): + with open(self.output_file, "wb") as f: + f.write(self.content[:100]) + + result = self.download(self.granule(), dry_run=True) + self.assertIn( + "would download again", result.metadata["skip_reasons"][self.output_file] + ) + + result = self.download(self.granule()) + self.assertEqual(result.downloaded_files, [self.output_file]) + self.assertEqual(os.path.getsize(self.output_file), len(self.content)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/downloaders/test_structure.py b/tests/downloaders/test_structure.py index 259e5ce..a7a41e6 100644 --- a/tests/downloaders/test_structure.py +++ b/tests/downloaders/test_structure.py @@ -11,6 +11,7 @@ PROVIDERS = { "copernicus": "CopernicusDownloader", "csiro": "CSIRODownloader", + "earthdata": "EarthdataDownloader", "ecmwf": "ECMWFDownloader", "noaa": "NOAADownloader", } From 033730c0c85eeb858b99dd8c91c00584a39b63d3 Mon Sep 17 00:00:00 2001 From: Javier Tausia Date: Mon, 26 Oct 2026 11:21:32 +0100 Subject: [PATCH 3/4] [JTH] downloaders: open NetCDF files one at a time in validate_file netCDF-C/HDF5 are not thread safe. Parallel downloads (CSIRO, Earthdata) validate files from several threads at once, which gave HDF errors and segfaults (a SWOT download job crashed after 13 s). A lock serialises the netCDF4 open; the new test segfaults without it. Co-Authored-By: Claude Opus 5.5 --- bluemath_tk/downloaders/_transfer.py | 16 +++++++++++----- tests/downloaders/test_transfer.py | 8 ++++++++ 2 files changed, 19 insertions(+), 5 deletions(-) diff --git a/bluemath_tk/downloaders/_transfer.py b/bluemath_tk/downloaders/_transfer.py index 37c4bb9..4b36613 100644 --- a/bluemath_tk/downloaders/_transfer.py +++ b/bluemath_tk/downloaders/_transfer.py @@ -8,6 +8,7 @@ import logging import os +import threading import time from typing import Callable, Optional, Tuple, Type, TypeVar @@ -18,6 +19,10 @@ PART_SUFFIX = ".part" CHUNK_SIZE = 8 * 1024 * 1024 +# netCDF-C/HDF5 are not thread safe: opening files from several download threads +# at once gives HDF errors or segfaults, so validate_file opens one at a time +_NETCDF_LOCK = threading.Lock() + T = TypeVar("T") @@ -217,12 +222,13 @@ def validate_file(path: str) -> None: magic = f.read(8) if magic.startswith(b"CDF") or magic.startswith(b"\x89HDF"): - import netCDF4 + with _NETCDF_LOCK: + import netCDF4 - try: - netCDF4.Dataset(path).close() - except Exception as e: - raise IOError(f"Corrupt NetCDF file {path}: {e}") from e + try: + netCDF4.Dataset(path).close() + except Exception as e: + raise IOError(f"Corrupt NetCDF file {path}: {e}") from e elif magic.startswith(b"GRIB"): with open(path, "rb") as f: diff --git a/tests/downloaders/test_transfer.py b/tests/downloaders/test_transfer.py index 5209595..6bab0c3 100644 --- a/tests/downloaders/test_transfer.py +++ b/tests/downloaders/test_transfer.py @@ -2,6 +2,7 @@ import tempfile import threading import unittest +from concurrent.futures import ThreadPoolExecutor from functools import partial from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer @@ -161,6 +162,13 @@ def _write_netcdf(self) -> str: def test_valid_netcdf(self): validate_file(self._write_netcdf()) + def test_netcdf_from_several_threads(self): + # Download threads validate files at the same time: netCDF-C is not + # thread safe (HDF errors or segfaults without the lock) + path = self._write_netcdf() + with ThreadPoolExecutor(max_workers=8) as executor: + list(executor.map(lambda _: validate_file(path), range(200))) + def test_truncated_netcdf(self): path = self._write_netcdf() with open(path, "rb") as f: From 5e5cf5174175aed18fcbf38348a3b434d70f68b7 Mon Sep 17 00:00:00 2001 From: Javier Tausia Date: Mon, 26 Oct 2026 12:06:52 +0100 Subject: [PATCH 4/4] [JTH] downloaders: earthdata credentials test also works on Windows "~" is USERPROFILE on Windows, not HOME, so the test's temporary ~/.netrc was not found there. Co-Authored-By: Claude Opus 5.5 --- tests/downloaders/test_earthdata.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/downloaders/test_earthdata.py b/tests/downloaders/test_earthdata.py index ba6b846..b2a8d5c 100644 --- a/tests/downloaders/test_earthdata.py +++ b/tests/downloaders/test_earthdata.py @@ -176,7 +176,8 @@ def check_with_netrc(self, content): with open(path, "w") as f: f.write(content) os.chmod(path, 0o600) - with mock.patch.dict(os.environ, {"HOME": home}): + # "~" is HOME on POSIX and USERPROFILE on Windows + with mock.patch.dict(os.environ, {"HOME": home, "USERPROFILE": home}): EarthdataDownloader.check_credentials() def test_entry_found(self):