From 363781c5051538e23be7c613045cab3f4b74b9f9 Mon Sep 17 00:00:00 2001 From: Javier Tausia Date: Mon, 26 Oct 2026 19:46:41 +0100 Subject: [PATCH 1/4] [JTH] downloaders: IMOSDownloader for the public IMOS (AODN) S3 bucket New provider bluemath_tk.downloaders.imos: lists a bucket prefix with S3 ListObjectsV2 (paged, sub-folders in parallel, modified_since), mirrors the files locally, parallel and resumable, checked against the size and the ETag (MD5). A file is downloaded again when its size changes or the bucket copy is newer (local files get the bucket time). Product SRS with dataset Wave-Wind-Altimetry-DM00 (18 altimeter missions, 1985-now). file_checksum moves to _transfer (shared with earthdata). Co-Authored-By: Claude Opus 5.5 --- bluemath_tk/downloaders/_transfer.py | 14 + .../earthdata/earthdata_downloader.py | 23 +- .../downloaders/imos/SRS/SRS_config.json | 12 + bluemath_tk/downloaders/imos/__init__.py | 11 + .../downloaders/imos/imos_downloader.py | 454 ++++++++++++++++++ tests/downloaders/test_imos.py | 192 ++++++++ tests/downloaders/test_structure.py | 1 + 7 files changed, 691 insertions(+), 16 deletions(-) create mode 100644 bluemath_tk/downloaders/imos/SRS/SRS_config.json create mode 100644 bluemath_tk/downloaders/imos/__init__.py create mode 100644 bluemath_tk/downloaders/imos/imos_downloader.py create mode 100644 tests/downloaders/test_imos.py diff --git a/bluemath_tk/downloaders/_transfer.py b/bluemath_tk/downloaders/_transfer.py index 4b36613..42979e7 100644 --- a/bluemath_tk/downloaders/_transfer.py +++ b/bluemath_tk/downloaders/_transfer.py @@ -6,6 +6,7 @@ file validation. """ +import hashlib import logging import os import threading @@ -202,6 +203,19 @@ def http_download( ) +def file_checksum(path: str, algorithm: str) -> str: + """ + Hex digest of a file; algorithm as providers write it ("MD5", "SHA-256", ...). + """ + + 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() + + def validate_file(path: str) -> None: """ Check that a downloaded file is readable, based on its magic bytes. diff --git a/bluemath_tk/downloaders/earthdata/earthdata_downloader.py b/bluemath_tk/downloaders/earthdata/earthdata_downloader.py index b28cad2..6ac94ad 100644 --- a/bluemath_tk/downloaders/earthdata/earthdata_downloader.py +++ b/bluemath_tk/downloaders/earthdata/earthdata_downloader.py @@ -14,10 +14,8 @@ 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 @@ -28,7 +26,13 @@ from .._base_downloaders import BaseDownloader, load_config from .._download_result import DownloadResult -from .._transfer import http_download, http_get, make_session, retry_call +from .._transfer import ( + file_checksum, + http_download, + http_get, + make_session, + retry_call, +) _HERE = os.path.dirname(__file__) URS_HOST = "urs.earthdata.nasa.gov" @@ -133,19 +137,6 @@ def parse_granule(item: dict) -> List[dict]: 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. diff --git a/bluemath_tk/downloaders/imos/SRS/SRS_config.json b/bluemath_tk/downloaders/imos/SRS/SRS_config.json new file mode 100644 index 0000000..8317a9d --- /dev/null +++ b/bluemath_tk/downloaders/imos/SRS/SRS_config.json @@ -0,0 +1,12 @@ +{ + "description": "IMOS Satellite Remote Sensing (SRS) facility, public AODN bucket on AWS S3", + "bucket_url": "https://imos-data.s3-ap-southeast-2.amazonaws.com/", + "datasets": { + "Wave-Wind-Altimetry-DM00": { + "description": "Multi-mission altimeter wave height and wind speed (Ribal and Young 2019, updated), 1 Hz along-track, one file per mission and 1x1 degree cell, 1985-present, calibrated and raw values with quality flags", + "prefix": "IMOS/SRS/Surface-Waves/Wave-Wind-Altimetry-DM00/", + "local_dir": "SRS/Surface-Waves/Wave-Wind-Altimetry-DM00", + "filename_pattern": "^IMOS_SRS-Surface-Waves_MW_.*-DM00\\.nc$" + } + } +} diff --git a/bluemath_tk/downloaders/imos/__init__.py b/bluemath_tk/downloaders/imos/__init__.py new file mode 100644 index 0000000..c2541e0 --- /dev/null +++ b/bluemath_tk/downloaders/imos/__init__.py @@ -0,0 +1,11 @@ +""" +Project: BlueMath_tk +Sub-Module: downloaders.imos +Author: GeoOcean Research Group, Universidad de Cantabria +Repository: https://github.com/GeoOcean/BlueMath_tk.git +Status: Under development (Working) +""" + +from .imos_downloader import IMOSDownloader + +__all__ = ["IMOSDownloader"] diff --git a/bluemath_tk/downloaders/imos/imos_downloader.py b/bluemath_tk/downloaders/imos/imos_downloader.py new file mode 100644 index 0000000..27a94b2 --- /dev/null +++ b/bluemath_tk/downloaders/imos/imos_downloader.py @@ -0,0 +1,454 @@ +""" +Downloader for IMOS data in the public AODN bucket on AWS S3 (no account needed). + +Products (configs in ``imos//_config.json``): + +- SRS: Satellite Remote Sensing facility, e.g. the multi-mission altimeter + wave/wind archive (``Wave-Wind-Altimetry-DM00``). + +Each dataset is a prefix of the bucket. Files are listed with the S3 +ListObjectsV2 API and mirrored locally with the same folders below the prefix. +IMOS rewrites whole files when it releases new data, so a file is downloaded +again when its size changes or the bucket copy is newer than the local one. +""" + +import os +import re +import threading +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime +from typing import List, Optional, Sequence, Tuple, Union +from xml.etree import ElementTree + +import pandas as pd +import requests + +from .._base_downloaders import BaseDownloader, load_config +from .._download_result import DownloadResult +from .._transfer import ( + file_checksum, + http_download, + http_get, + make_session, + retry_call, +) + +_HERE = os.path.dirname(__file__) +S3_NS = {"s3": "http://s3.amazonaws.com/doc/2006-03-01/"} +FILE_COLUMNS = ["key", "path", "url", "size", "etag", "last_modified"] + +TimeLike = Union[str, datetime, pd.Timestamp] + + +def parse_listing(xml: bytes) -> Tuple[List[dict], List[str], Optional[str]]: + """ + One page of an S3 ListObjectsV2 answer. + + Parameters + ---------- + xml : bytes + Body of the answer. + + Returns + ------- + Tuple[List[dict], List[str], Optional[str]] + Objects (``key``, ``size``, ``etag`` without quotes, ``last_modified`` + as a UTC timestamp), common prefixes (sub-folders, when a delimiter is + used) and the continuation token (None on the last page). + """ + + root = ElementTree.fromstring(xml) + objects = [ + { + "key": c.findtext("s3:Key", namespaces=S3_NS), + "size": int(c.findtext("s3:Size", namespaces=S3_NS)), + "etag": c.findtext("s3:ETag", default="", namespaces=S3_NS).strip('"'), + "last_modified": pd.Timestamp( + c.findtext("s3:LastModified", namespaces=S3_NS) + ), + } + for c in root.findall("s3:Contents", S3_NS) + ] + prefixes = [ + p.findtext("s3:Prefix", namespaces=S3_NS) + for p in root.findall("s3:CommonPrefixes", S3_NS) + ] + truncated = root.findtext("s3:IsTruncated", default="false", namespaces=S3_NS) + token = root.findtext("s3:NextContinuationToken", namespaces=S3_NS) + + return objects, prefixes, token if truncated == "true" else None + + +class IMOSDownloader(BaseDownloader): + """ + Downloader for IMOS datasets in the public AODN S3 bucket. + + Files are saved to base_path_to_download//. Existing files are skipped when their size + matches and they are not older than the bucket copy; downloaded files get + the bucket modification time, are checked against the S3 ETag (MD5 of + single-part uploads) and several are downloaded in parallel + (``max_workers``; files are small, so many workers help). + + Examples + -------- + >>> downloader = IMOSDownloader( + ... product="SRS", + ... base_path_to_download="/lustre/geocean/DATA/IMOS", + ... max_workers=16, + ... ) + >>> downloader.list_subfolders("Wave-Wind-Altimetry-DM00") # missions + >>> result = downloader.download_data( + ... "Wave-Wind-Altimetry-DM00", subfolders=["JASON-3"], dry_run=True + ... ) + """ + + products_configs = { + "SRS": load_config(_HERE, "SRS", "SRS_config.json"), + } + + def __init__( + self, + product: str, + base_path_to_download: str, + max_workers: int = 16, + debug: bool = True, + **kwargs, + ) -> None: + """ + Initialize the IMOSDownloader. + + Parameters + ---------- + product : str + The product to download data from (e.g. "SRS"). + base_path_to_download : str + Base path where downloaded files will be stored. + max_workers : int, optional + Number of files downloaded (and folders listed) in parallel. + Default is 16. + 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"IMOSDownloader-{product}", level="DEBUG" if debug else "INFO" + ) + self.logger.info(f"---- IMOS DOWNLOADER INITIALIZED ({product}) ----") + + @property + def session(self) -> requests.Session: + """ + HTTP session of the current thread. + + 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 list_subfolders(self, dataset: str) -> List[str]: + """ + Sub-folders of a dataset (e.g. the missions of the altimeter archive). + + Parameters + ---------- + dataset : str + The dataset name. + + Returns + ------- + List[str] + Folder names directly below the dataset prefix. + """ + + prefix = self._get_dataset_config(dataset)["prefix"] + _, prefixes = self._list(prefix, delimiter="/") + + return [p[len(prefix) :].rstrip("/") for p in prefixes] + + def list_files( + self, + dataset: str, + subfolders: Optional[Sequence[str]] = None, + modified_since: Optional[TimeLike] = None, + ) -> pd.DataFrame: + """ + Files of a dataset, from the bucket listing. + + Parameters + ---------- + dataset : str + The dataset name. Use list_datasets() to see available datasets. + subfolders : Sequence[str], optional + Only these sub-folders (e.g. missions). Default is all. They are + listed in parallel. + modified_since : str or datetime, optional + Only files modified in the bucket since then (UTC). + + Returns + ------- + pd.DataFrame + One row per file (columns FILE_COLUMNS; ``path`` is relative to + the dataset prefix), sorted by path. Files that do not match the + dataset ``filename_pattern`` are left out. + """ + + dataset_config = self._get_dataset_config(dataset) + prefix = dataset_config["prefix"] + if subfolders is None: + subfolders = self.list_subfolders(dataset) + prefixes = [f"{prefix}{s.strip('/')}/" for s in subfolders] + + with ThreadPoolExecutor(max_workers=self.max_workers) as executor: + listings = list(executor.map(lambda p: self._list(p)[0], prefixes)) + + files = pd.DataFrame( + [o for objects in listings for o in objects], + columns=["key", "size", "etag", "last_modified"], + ) + files["path"] = files["key"].str.slice(len(prefix)) + files["url"] = self.product_config["bucket_url"] + files["key"] + pattern = dataset_config.get("filename_pattern") + if pattern: + keep = files["key"].str.rsplit("/", n=1).str[-1].str.match(pattern) + if (~keep).any(): + self.logger.info( + f"{(~keep).sum()} files do not match {pattern} and are left out" + ) + files = files[keep] + if modified_since is not None: + since = pd.Timestamp(modified_since) + since = since.tz_localize("UTC") if since.tz is None else since + files = files[files["last_modified"] >= since] + files = files[FILE_COLUMNS].sort_values("path").reset_index(drop=True) + self.logger.info( + f"{len(files)} files in {prefix} ({files['size'].sum() / 1e9:,.1f} GB)" + ) + + return files + + def get_local_path(self, dataset: str, path: str) -> str: + """ + Local path of a file. + + Parameters + ---------- + dataset : str + The dataset name. + path : str + Path of the file below the dataset prefix (``path`` of list_files). + + Returns + ------- + str + The local path. + """ + + local_dir = self._get_dataset_config(dataset)["local_dir"] + + return os.path.normpath( + os.path.join(self.base_path_to_download, local_dir, path) + ) + + def download_data( + self, + dataset: str, + subfolders: Optional[Sequence[str]] = None, + modified_since: Optional[TimeLike] = None, + force: bool = False, + verify_existing: bool = True, + dry_run: bool = True, + ) -> DownloadResult: + """ + Download (or update) the files of a dataset. + + The same call keeps a local copy up to date: complete files that did + not change in the bucket are skipped. + + Parameters + ---------- + dataset : str + The dataset to download. Use list_datasets() to see available datasets. + subfolders : Sequence[str], optional + Only these sub-folders (e.g. missions). Default is all. + modified_since : str or datetime, optional + Only files modified in the bucket since then. + force : bool, optional + Download again even if the file is up to date. Default is False. + verify_existing : bool, optional + Compare existing files with the bucket (size and modification + time) 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 size of each file is in metadata["remote_sizes"]. + """ + + files = self.list_files(dataset, subfolders, modified_since) + + return self.download_files( + dataset, + files, + force=force, + verify_existing=verify_existing, + dry_run=dry_run, + ) + + def download_files( + self, + dataset: str, + files: pd.DataFrame, + force: bool = False, + verify_existing: bool = True, + dry_run: bool = True, + ) -> DownloadResult: + """ + Download the files of a table returned by list_files. + + Parameters + ---------- + dataset : str + The dataset the files belong to. + files : pd.DataFrame + Files to download (output of list_files, possibly filtered). + force, verify_existing, dry_run : bool, optional + As in download_data. + + Returns + ------- + DownloadResult + Result with information about downloaded, skipped, and error files. + """ + + self.logger.info(f"Checking {len(files)} 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.path), + force, + verify_existing, + dry_run, + ) + for row in files.itertuples(index=False) + ] + for future in futures: + result.merge(future.result()) + + return self.finalize_download_result(result) + + 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 _list( + self, prefix: str, delimiter: Optional[str] = None + ) -> Tuple[List[dict], List[str]]: + """All objects (and common prefixes) below a prefix, every page.""" + + params = {"list-type": "2", "prefix": prefix, "max-keys": "1000"} + if delimiter: + params["delimiter"] = delimiter + objects, prefixes, token = [], [], None + while True: + page = dict(params, **({"continuation-token": token} if token else {})) + response = retry_call( + lambda: http_get( + self.session, self.product_config["bucket_url"], params=page + ), + retries=self.retries, + backoff=self.retry_backoff, + logger=self.logger, + ) + page_objects, page_prefixes, token = parse_listing(response.content) + objects.extend(page_objects) + prefixes.extend(page_prefixes) + if token is None: + return objects, prefixes + + def _download_file( + self, + row, + output_file: str, + force: bool, + verify_existing: bool, + dry_run: bool, + ) -> DownloadResult: + """ + Download a single file unless the local copy is complete and up to date. + """ + + result = DownloadResult() + result.metadata["remote_sizes"] = {output_file: int(row.size)} + modified = row.last_modified.timestamp() + + if os.path.exists(output_file) and not force: + local_size = os.path.getsize(output_file) + local_time = os.path.getmtime(output_file) + if not verify_existing or ( + local_size == row.size and local_time >= modified + ): + result.add_skipped(output_file, "Already downloaded") + return result + reason = ( + "Updated in the bucket" + if local_size == row.size + else (f"Local size {local_size} differs from bucket size {row.size}") + ) + if dry_run: + result.add_skipped( + output_file, f"{reason}, would download again (dry run)" + ) + return result + self.logger.debug(f"{reason}, downloading again: {output_file}") + force = True + + def fetch(part_file: str) -> None: + http_download( + self.session, + row.url, + part_file, + expected_size=int(row.size), + resume=True, + ) + # The ETag is the MD5 of single-part uploads ("-" otherwise) + if re.fullmatch(r"[0-9a-f]{32}", row.etag or ""): + checksum = file_checksum(part_file, "MD5") + if checksum != row.etag: + os.remove(part_file) + raise IOError( + f"MD5 of {row.key} is {checksum}, ETag says {row.etag}" + ) + os.utime(part_file, (modified, modified)) + + self.fetch_file( + output_file, fetch, result, force=force, dry_run=dry_run, resumable=True + ) + + return result diff --git a/tests/downloaders/test_imos.py b/tests/downloaders/test_imos.py new file mode 100644 index 0000000..c77f9c9 --- /dev/null +++ b/tests/downloaders/test_imos.py @@ -0,0 +1,192 @@ +import hashlib +import os +import unittest +from collections import namedtuple +from unittest import mock + +import pandas as pd +from test_transfer import LocalServerTestCase + +from bluemath_tk.downloaders.imos import imos_downloader +from bluemath_tk.downloaders.imos.imos_downloader import ( + FILE_COLUMNS, + IMOSDownloader, + parse_listing, +) + +DATASET = "Wave-Wind-Altimetry-DM00" +PREFIX = "IMOS/SRS/Surface-Waves/Wave-Wind-Altimetry-DM00/" +File = namedtuple("File", FILE_COLUMNS) + + +def cell_name(mission, cell="044N-356E"): + return f"IMOS_SRS-Surface-Waves_MW_{mission}_FV02_{cell}-DM00.nc" + + +def listing(keys, prefixes=(), token=None): + """S3 ListObjectsV2 answer with the given keys (size 100, 2026-07-14).""" + contents = "".join( + f"{k}2026-07-14T10:06:44.000Z" + f""{'a' * 32}"100" + for k in keys + ) + common = "".join( + f"{p}" for p in prefixes + ) + tail = ( + f"true{token}" + if token + else "false" + ) + return ( + '' + '' + f"{contents}{common}{tail}" + ).encode() + + +class FakeResponse: + def __init__(self, content): + self.content = content + + +class TestParseListing(unittest.TestCase): + def test_objects_prefixes_and_token(self): + key = PREFIX + "SWOT/040N_340E/" + cell_name("SWOT") + objects, prefixes, token = parse_listing( + listing([key], prefixes=[PREFIX + "SWOT/"], token="abc") + ) + self.assertEqual(token, "abc") + self.assertEqual(prefixes, [PREFIX + "SWOT/"]) + self.assertEqual(objects[0]["key"], key) + self.assertEqual(objects[0]["size"], 100) + self.assertEqual(objects[0]["etag"], "a" * 32) + self.assertEqual( + objects[0]["last_modified"], pd.Timestamp("2026-07-14T10:06:44Z") + ) + + def test_last_page(self): + self.assertIsNone(parse_listing(listing([]))[2]) + + +class TestListFiles(unittest.TestCase): + def setUp(self): + self.downloader = IMOSDownloader( + product="SRS", base_path_to_download="/data", retries=1, retry_backoff=0 + ) + + def test_unknown_dataset(self): + with self.assertRaises(ValueError): + self.downloader.list_files("nope") + + def test_subfolders(self): + answer = listing([], prefixes=[PREFIX + "JASON-3/", PREFIX + "SWOT/"]) + with mock.patch.object( + imos_downloader, "http_get", lambda *a, **k: FakeResponse(answer) + ): + self.assertEqual( + self.downloader.list_subfolders(DATASET), ["JASON-3", "SWOT"] + ) + + def test_paging_pattern_and_modified_since(self): + swot = PREFIX + "SWOT/040N_340E/" + pages = { + None: listing([swot + cell_name("SWOT", "041N-341E")], token="t1"), + "t1": listing([swot + cell_name("SWOT", "040N-340E"), swot + "README.txt"]), + } + calls = [] + + def fake_get(session, url, params=None): + calls.append(dict(params)) + return FakeResponse(pages[params.get("continuation-token")]) + + with mock.patch.object(imos_downloader, "http_get", fake_get): + files = self.downloader.list_files(DATASET, subfolders=["SWOT"]) + newer = self.downloader.list_files( + DATASET, subfolders=["SWOT"], modified_since="2026-08-01" + ) + + self.assertEqual(list(files.columns), FILE_COLUMNS) + self.assertEqual( + list(files["path"]), + [ + "SWOT/040N_340E/" + cell_name("SWOT", "040N-340E"), + "SWOT/040N_340E/" + cell_name("SWOT", "041N-341E"), + ], + ) + self.assertTrue(files["url"].iloc[0].startswith("https://imos-data.s3")) + self.assertEqual(calls[0]["prefix"], PREFIX + "SWOT/") + self.assertNotIn("continuation-token", calls[0]) + self.assertEqual(calls[1]["continuation-token"], "t1") + self.assertTrue(newer.empty) + + def test_local_path_mirrors_the_bucket(self): + self.assertEqual( + self.downloader.get_local_path(DATASET, "SWOT/040N_340E/f.nc"), + os.path.normpath( + "/data/SRS/Surface-Waves/Wave-Wind-Altimetry-DM00/SWOT/040N_340E/f.nc" + ), + ) + + +class TestIMOSDownloadFile(LocalServerTestCase): + def setUp(self): + super().setUp() + self.downloader = IMOSDownloader( + product="SRS", + base_path_to_download=self.tmp_dir.name, + retries=1, + retry_backoff=0, + ) + self.output_file = os.path.join(self.tmp_dir.name, "sub", "data.bin") + self.modified = pd.Timestamp("2026-07-14T10:06:44Z") + + def file(self, etag=None, modified=None): + return File( + key="data.bin", + path="data.bin", + url=f"{self.base_url}/data.bin", + size=len(self.content), + etag=hashlib.md5(self.content).hexdigest() if etag is None else etag, + last_modified=self.modified if modified is None else modified, + ) + + def download(self, row, dry_run=False): + return self.downloader._download_file( + row, self.output_file, force=False, verify_existing=True, dry_run=dry_run + ) + + def test_download_sets_bucket_time_and_skips(self): + result = self.download(self.file()) + self.assertEqual(result.downloaded_files, [self.output_file]) + self.assertEqual(os.path.getmtime(self.output_file), self.modified.timestamp()) + result = self.download(self.file()) + self.assertEqual(result.skipped_files, [self.output_file]) + + def test_newer_in_bucket_is_downloaded_again(self): + self.download(self.file()) + newer = self.file(modified=self.modified + pd.Timedelta(days=30)) + result = self.download(newer, dry_run=True) + self.assertIn( + "Updated in the bucket", result.metadata["skip_reasons"][self.output_file] + ) + result = self.download(newer) + self.assertEqual(result.downloaded_files, [self.output_file]) + self.assertEqual( + os.path.getmtime(self.output_file), + (self.modified + pd.Timedelta(days=30)).timestamp(), + ) + + def test_wrong_md5_is_an_error(self): + result = self.download(self.file(etag="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_multipart_etag_is_not_a_checksum(self): + result = self.download(self.file(etag="0" * 32 + "-3")) + self.assertEqual(result.downloaded_files, [self.output_file]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/downloaders/test_structure.py b/tests/downloaders/test_structure.py index a7a41e6..412ff11 100644 --- a/tests/downloaders/test_structure.py +++ b/tests/downloaders/test_structure.py @@ -12,6 +12,7 @@ "copernicus": "CopernicusDownloader", "csiro": "CSIRODownloader", "earthdata": "EarthdataDownloader", + "imos": "IMOSDownloader", "ecmwf": "ECMWFDownloader", "noaa": "NOAADownloader", } From 9a45a99097772668595352acb4ba0bd4f0bbedd2 Mon Sep 17 00:00:00 2001 From: Javier Tausia Date: Mon, 26 Oct 2026 19:46:52 +0100 Subject: [PATCH 2/4] [JTH] docs: IMOS downloader section Co-Authored-By: Claude Opus 5.5 --- docs/source/downloaders.rst | 39 +++++++++++++++++++++++++++++++++++++ 1 file changed, 39 insertions(+) diff --git a/docs/source/downloaders.rst b/docs/source/downloaders.rst index 13ceeac..9c92e52 100644 --- a/docs/source/downloaders.rst +++ b/docs/source/downloaders.rst @@ -32,6 +32,12 @@ Providers and products - 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. + * - ``imos`` + - ``IMOSDownloader`` + - SRS + - IMOS data in the public AODN bucket on AWS S3 (no account), e.g. the + multi-mission altimeter wave/wind archive. Mirrors a bucket prefix, + re-downloads files that changed, checked against the S3 ETag (MD5). * - ``noaa`` - ``NOAADownloader`` - NDBC @@ -251,6 +257,39 @@ 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``). +IMOS: altimeter wave and wind archive +------------------------------------- + +The IMOS multi-mission altimeter archive (Ribal and Young 2019, updated by +IMOS) has one file per mission and 1x1 degree cell, 1985 to present, in the +public AODN bucket. No account is needed. + +.. code-block:: python + + from bluemath_tk.downloaders.imos import IMOSDownloader + + downloader = IMOSDownloader( + product="SRS", + base_path_to_download="/lustre/geocean/DATA/IMOS", + max_workers=32, # files are small (~300 kB): many workers + ) + downloader.list_subfolders("Wave-Wind-Altimetry-DM00") # the missions + + result = downloader.download_data( + "Wave-Wind-Altimetry-DM00", + subfolders=["JASON-3", "SWOT"], # default: all missions + dry_run=True, + ) + print(result.summary()) + +Files mirror the bucket below the dataset prefix: +``/SRS/Surface-Waves/Wave-Wind-Altimetry-DM00//<20x20 folder>/``. +IMOS rewrites whole files at each release, so a file is downloaded again when +its size changes or the bucket copy is newer than the local one (downloaded +files get the bucket modification time). ``modified_since`` only lists files +changed since a date; ``list_files`` returns the table of files (path, URL, +size, ETag, modification time) and ``download_files`` downloads such a table. + NOAA and ECMWF -------------- From 68e02b32f358bbcf762b03b232f058fef73fea1b Mon Sep 17 00:00:00 2001 From: Javier Tausia Date: Mon, 26 Oct 2026 19:58:26 +0100 Subject: [PATCH 3/4] [JTH] downloaders: DownloadResult stays fast with hundreds of thousands of files add_* and merge checked "file not in list", which is quadratic: a dry run of the IMOS archive (736,000 files) did not finish in 10 min. A set next to each list makes the check O(1) (rebuilt if a list is changed directly). Co-Authored-By: Claude Opus 5.5 --- bluemath_tk/downloaders/_download_result.py | 31 ++++++++++++++------- tests/downloaders/test_transfer.py | 17 +++++++++++ 2 files changed, 38 insertions(+), 10 deletions(-) diff --git a/bluemath_tk/downloaders/_download_result.py b/bluemath_tk/downloaders/_download_result.py index 0809486..34da7e1 100644 --- a/bluemath_tk/downloaders/_download_result.py +++ b/bluemath_tk/downloaders/_download_result.py @@ -112,13 +112,11 @@ def add_error( if context: error_dict["context"] = context self.errors.append(error_dict) - if file_path not in self.error_files: - self.error_files.append(file_path) + self._append_unique("error_files", file_path) def add_downloaded(self, file_path: str): """Add a successfully downloaded file.""" - if file_path not in self.downloaded_files: - self.downloaded_files.append(file_path) + self._append_unique("downloaded_files", file_path) def add_skipped(self, file_path: str, reason: str = ""): """ @@ -131,11 +129,26 @@ def add_skipped(self, file_path: str, reason: str = ""): reason : str, optional Reason why the file was skipped. """ - if file_path not in self.skipped_files: - self.skipped_files.append(file_path) + self._append_unique("skipped_files", file_path) if reason: self.metadata.setdefault("skip_reasons", {})[file_path] = reason + def _append_unique(self, name: str, file_path: str) -> None: + """ + Append to one of the file lists unless already there. + + A set kept next to each list makes the check O(1), so results of + hundreds of thousands of files stay fast. It is rebuilt if the list was + changed directly. + """ + files = getattr(self, name) + seen = self.__dict__.setdefault("_seen", {}).get(name) + if seen is None or len(seen) != len(files): + seen = self._seen[name] = set(files) + if file_path not in seen: + seen.add(file_path) + files.append(file_path) + def merge(self, other: "DownloadResult") -> "DownloadResult": """ Add the files, errors and metadata of another result to this one. @@ -155,11 +168,9 @@ def merge(self, other: "DownloadResult") -> "DownloadResult": for file_path in other.downloaded_files: self.add_downloaded(file_path) for file_path in other.skipped_files: - if file_path not in self.skipped_files: - self.skipped_files.append(file_path) + self._append_unique("skipped_files", file_path) for file_path in other.error_files: - if file_path not in self.error_files: - self.error_files.append(file_path) + self._append_unique("error_files", file_path) self.errors.extend(other.errors) for key, value in other.metadata.items(): if isinstance(value, dict): diff --git a/tests/downloaders/test_transfer.py b/tests/downloaders/test_transfer.py index 6bab0c3..78d83a4 100644 --- a/tests/downloaders/test_transfer.py +++ b/tests/downloaders/test_transfer.py @@ -1,6 +1,7 @@ import os import tempfile import threading +import time import unittest from concurrent.futures import ThreadPoolExecutor from functools import partial @@ -318,6 +319,22 @@ def test_product_without_download_method(self): class TestDownloadResult(unittest.TestCase): + def test_many_files_and_direct_appends(self): + # A dry run of ~700,000 files merges one result per file: must not be + # quadratic. Lists changed directly still count as seen. + result = DownloadResult() + start = time.perf_counter() + for i in range(200_000): + other = DownloadResult() + other.add_skipped(f"f{i}", "Would download (dry run)") + result.merge(other) + self.assertLess(time.perf_counter() - start, 30) + self.assertEqual(len(result.skipped_files), 200_000) + result.skipped_files.append("direct") + result.add_skipped("direct") + result.add_skipped("f0") + self.assertEqual(len(result.skipped_files), 200_001) + def test_merge_and_save(self): result = DownloadResult(downloaded_files=["a"]) other = DownloadResult(downloaded_files=["a", "b"]) From 7e06bcfbe49039d11c07f37e145a9d9340221852 Mon Sep 17 00:00:00 2001 From: Javier Tausia Date: Mon, 26 Oct 2026 20:32:21 +0100 Subject: [PATCH 4/4] [JTH] downloaders: IMOS reads each local folder once instead of one stat per file On Lustre, 736,000 stat calls of missing files took over 30 min; with one os.scandir per folder the dry run of the whole IMOS archive takes 1 min. Dry runs no longer stat missing files a second time in fetch_file. Co-Authored-By: Claude Opus 5.5 --- .../downloaders/imos/imos_downloader.py | 54 +++++++++++++++++-- tests/downloaders/test_imos.py | 16 +++++- 2 files changed, 64 insertions(+), 6 deletions(-) diff --git a/bluemath_tk/downloaders/imos/imos_downloader.py b/bluemath_tk/downloaders/imos/imos_downloader.py index 27a94b2..9dd195e 100644 --- a/bluemath_tk/downloaders/imos/imos_downloader.py +++ b/bluemath_tk/downloaders/imos/imos_downloader.py @@ -343,17 +343,20 @@ def download_files( self.logger.info(f"Checking {len(files)} files of {self.product}/{dataset}") result = self.create_download_result() + paths = [self.get_local_path(dataset, path) for path in files["path"]] + local = self._scan_local(paths) with ThreadPoolExecutor(max_workers=self.max_workers) as executor: futures = [ executor.submit( self._download_file, row, - self.get_local_path(dataset, row.path), + path, + local.get(path), force, verify_existing, dry_run, ) - for row in files.itertuples(index=False) + for row, path in zip(files.itertuples(index=False), paths) ] for future in futures: result.merge(future.result()) @@ -392,25 +395,66 @@ def _list( if token is None: return objects, prefixes + def _scan_local(self, paths: List[str]) -> dict: + """ + Size and modification time of the local files among ``paths``. + + Each folder is read once (os.scandir) instead of one stat per file: on + Lustre, ~736,000 stat calls of missing files took over 30 minutes. + + Returns + ------- + dict + ``{path: (size, mtime)}`` of the files that exist. + """ + + def scan(folder: str) -> dict: + try: + with os.scandir(folder) as entries: + return { + os.path.join(folder, e.name): ( + e.stat().st_size, + e.stat().st_mtime, + ) + for e in entries + if e.is_file() + } + except FileNotFoundError: + return {} + + folders = sorted({os.path.dirname(p) for p in paths}) + local = {} + with ThreadPoolExecutor(max_workers=self.max_workers) as executor: + for found in executor.map(scan, folders): + local.update(found) + + return local + def _download_file( self, row, output_file: str, + local: Optional[Tuple[int, float]], force: bool, verify_existing: bool, dry_run: bool, ) -> DownloadResult: """ Download a single file unless the local copy is complete and up to date. + + ``local`` is the (size, mtime) of the local file, None if it does not exist. """ result = DownloadResult() result.metadata["remote_sizes"] = {output_file: int(row.size)} modified = row.last_modified.timestamp() - if os.path.exists(output_file) and not force: - local_size = os.path.getsize(output_file) - local_time = os.path.getmtime(output_file) + if local is None and dry_run: + result.add_skipped(output_file, "Would download (dry run)") + return result + + if local is not None and not force: + local_size, local_time = local if not verify_existing or ( local_size == row.size and local_time >= modified ): diff --git a/tests/downloaders/test_imos.py b/tests/downloaders/test_imos.py index c77f9c9..a586cd6 100644 --- a/tests/downloaders/test_imos.py +++ b/tests/downloaders/test_imos.py @@ -152,10 +152,24 @@ def file(self, etag=None, modified=None): ) def download(self, row, dry_run=False): + local = self.downloader._scan_local([self.output_file]).get(self.output_file) return self.downloader._download_file( - row, self.output_file, force=False, verify_existing=True, dry_run=dry_run + row, + self.output_file, + local, + force=False, + verify_existing=True, + dry_run=dry_run, ) + def test_dry_run_of_missing_file(self): + result = self.download(self.file(), dry_run=True) + self.assertEqual( + result.metadata["skip_reasons"][self.output_file], + "Would download (dry run)", + ) + self.assertFalse(os.path.exists(os.path.dirname(self.output_file))) + def test_download_sets_bucket_time_and_skips(self): result = self.download(self.file()) self.assertEqual(result.downloaded_files, [self.output_file])