From 3e23fe7b897b1357904a777e17917b860c1b2c03 Mon Sep 17 00:00:00 2001 From: "Gregory L. Wagner" Date: Sun, 12 Jul 2026 04:19:32 +0000 Subject: [PATCH 1/3] hourly: multi-variable requests, levels compatibility, concurrent downloads Three gaps in hourly() relative to the registered v0.1 (era5cli) interface that NumericalEarth's batched download extension relies on: - variables now accepts a Vector{String} as well as a String. Each variable is its own CDS request (matching era5cli semantics); requests are submitted concurrently via asyncmap, up to `threads` at a time, so a bundle of variables waits in the Copernicus queue together instead of sequentially. Multi-variable calls append the variable name to outputprefix so each request gets its own file; single-variable naming is unchanged. - The v0.1 `levels` keyword is accepted again as a spelling of pressure_levels. Previously it fell silently into additional_kw, so a v0.1-style pressure-level request downloaded the single-levels product without warning (e.g. u10 instead of u on levels). - Existing files are skipped per variable when overwrite=false, and the unused dates_str construction is removed. Adds offline tests pinning the file-naming and skip-existing contract. Co-Authored-By: Claude Fable 5 --- src/CopernicusClimateDataStore.jl | 64 ++++++++++++++++++++----------- test/runtests.jl | 29 ++++++++++++++ 2 files changed, 70 insertions(+), 23 deletions(-) diff --git a/src/CopernicusClimateDataStore.jl b/src/CopernicusClimateDataStore.jl index 1f143b9..1296eef 100644 --- a/src/CopernicusClimateDataStore.jl +++ b/src/CopernicusClimateDataStore.jl @@ -8,7 +8,7 @@ include("cds_client.jl") """ hourly(; variables, startyear, months, days, hours, area=nothing, - pressure_levels=nothing, format="netcdf", outputprefix="era5", + pressure_levels=nothing, levels=nothing, format="netcdf", outputprefix="era5", overwrite=false, threads=Threads.nthreads(), splitmonths=false, directory=".", additional_kw...) @@ -16,17 +16,36 @@ Download ERA5 hourly data using the CDS API. This function provides compatibilit with NumericalEarth's ERA5 download interface. # Arguments +- `variables`: Variable name(s) - String or Vector{String}. Each variable is its own + CDS request; requests are submitted concurrently (up to `threads` at a time) + so the whole bundle waits in the Copernicus queue together. - `pressure_levels`: Optional pressure levels in hPa (e.g., [1000, 850, 500]). If provided, downloads from pressure-levels dataset instead of single-levels. +- `levels`: The v0.1 (era5cli) spelling of `pressure_levels`: a vector of levels in hPa + selects the pressure-levels dataset; `:surface` and `nothing` mean single-levels. -Returns a vector of downloaded file paths. +Returns a vector of downloaded file paths, one per variable. + +# File naming +A single variable keeps the v0.2 names (`outputprefix.nc` for a single date/hour, +`outputprefix_year_month_day.nc` otherwise); with multiple variables the variable +name is appended to `outputprefix` so each request gets its own file. """ -function hourly(; variables::String, startyear::Int, months, days, hours, - area=nothing, pressure_levels=nothing, format::String="netcdf", +function hourly(; variables::Union{String, AbstractVector{String}}, startyear::Int, months, days, hours, + area=nothing, pressure_levels=nothing, levels=nothing, format::String="netcdf", outputprefix::String="era5", overwrite::Bool=false, threads::Int=Threads.nthreads(), splitmonths::Bool=false, directory::String=".", additional_kw...) + variables_arr = variables isa String ? [variables] : collect(variables) + + # `levels` must be handled explicitly: an unrecognized keyword lands silently in + # `additional_kw`, so a v0.1-style pressure-level request would download the + # single-levels product without warning. + if isnothing(pressure_levels) && levels isa AbstractVector + pressure_levels = levels + end + # Convert single values to arrays months_arr = months isa AbstractVector ? months : [months] days_arr = days isa AbstractVector ? days : [days] @@ -35,17 +54,10 @@ function hourly(; variables::String, startyear::Int, months, days, hours, # Format hours as two-digit strings hours_str = [string(h, pad=2) * ":00" for h in hours_arr] - # Format date strings - dates_str = String[] - for month in months_arr, day in days_arr - push!(dates_str, string(startyear, "-", string(month, pad=2), "-", string(day, pad=2))) - end - # Build request parameters request_params = Dict( "product_type" => "reanalysis", "format" => format, - "variable" => variables, "year" => string(startyear), "month" => [string(m, pad=2) for m in months_arr], "day" => [string(d, pad=2) for d in days_arr], @@ -71,27 +83,33 @@ function hourly(; variables::String, startyear::Int, months, days, hours, request_params["pressure_level"] = [string(Int(p)) for p in pl_arr] end - # Generate output filename mkpath(directory) # If requesting single date/hour, use outputprefix as-is (for NumericalEarth compatibility) # Otherwise append date for batched downloads - if length(months_arr) == 1 && length(days_arr) == 1 && length(hours_arr) == 1 - output_file = joinpath(directory, "$(outputprefix).nc") - else - output_file = joinpath(directory, "$(outputprefix)_$(startyear)_$(first(months_arr))_$(first(days_arr)).nc") - end + single_date = length(months_arr) == 1 && length(days_arr) == 1 && length(hours_arr) == 1 + date_tag = single_date ? "" : "_$(startyear)_$(first(months_arr))_$(first(days_arr))" - # Skip if file exists and not overwriting - if isfile(output_file) && !overwrite - return [output_file] + function output_file(variable) + variable_tag = length(variables_arr) == 1 ? "" : "_$(variable)" + return joinpath(directory, "$(outputprefix)$(variable_tag)$(date_tag).nc") end - # Submit download request + # Skip files that exist when not overwriting + pending = overwrite ? variables_arr : filter(variable -> !isfile(output_file(variable)), variables_arr) + dataset_id = pressure_levels === nothing ? "reanalysis-era5-single-levels" : "reanalysis-era5-pressure-levels" - retrieve(dataset_id, request_params, output_file) - return [output_file] + # One CDS request per variable, submitted concurrently + if !isempty(pending) + asyncmap(pending; ntasks=max(threads, 1)) do variable + params = copy(request_params) + params["variable"] = variable + retrieve(dataset_id, params, output_file(variable)) + end + end + + return map(output_file, variables_arr) end """ diff --git a/test/runtests.jl b/test/runtests.jl index 3d1c01a..62de1f0 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -29,6 +29,35 @@ using CopernicusClimateDataStore @test isdefined(CopernicusClimateDataStore, :download_cds_file) end + @testset "hourly file naming and skip-existing (offline)" begin + # A single variable keeps the historical names; existing files short-circuit + # the request entirely, so these run without credentials or network. + mktempdir() do dir + single = joinpath(dir, "era5.nc") + touch(single) + paths = hourly(variables="2m_temperature", startyear=2020, + months=1, days=1, hours=0, directory=dir) + @test paths == [single] + + dated = joinpath(dir, "era5_2020_1_1.nc") + touch(dated) + paths = hourly(variables="2m_temperature", startyear=2020, + months=1, days=1, hours=[0, 12], directory=dir) + @test paths == [dated] + end + + # Multiple variables get per-variable files, returned in input order + mktempdir() do dir + t2m = joinpath(dir, "era5_2m_temperature.nc") + u10 = joinpath(dir, "era5_10m_u_component_of_wind.nc") + touch(t2m) + touch(u10) + paths = hourly(variables=["2m_temperature", "10m_u_component_of_wind"], + startyear=2020, months=1, days=1, hours=0, directory=dir) + @test paths == [t2m, u10] + end + end + @testset "ERA5 Download Integration Test" begin # Only run if CDS credentials are available has_credentials = try From a4261ad00fa4296637fff4761c904192809ee9c1 Mon Sep 17 00:00:00 2001 From: "Gregory L. Wagner" Date: Tue, 14 Jul 2026 21:07:10 +0000 Subject: [PATCH 2/3] Fast-start poll backoff, transient-error retries; bump to 0.2.1 Two request-lifecycle fixes that matter for batched multi-request use: - poll_request_status polled at a fixed 10 s; small extractions complete in ~15-25 s, so up to 10 s per request was dead time. Polling now starts at 1 s and backs off exponentially to poll_interval, so short requests return promptly while long queue waits still poll gently. - The CDS gateway intermittently answers valid requests with transient errors (e.g. 502). Submit, status poll, results fetch, and file download now retry with exponential backoff (request_with_retries) instead of dying on the first blip - which previously took down every request of a concurrent bundle, possibly minutes into a queue wait. Co-Authored-By: Claude Fable 5 --- Project.toml | 2 +- src/cds_client.jl | 38 +++++++++++++++++++++++++++++++++----- test/runtests.jl | 19 +++++++++++++++++++ 3 files changed, 53 insertions(+), 6 deletions(-) diff --git a/Project.toml b/Project.toml index 1df4a9f..eac32f6 100644 --- a/Project.toml +++ b/Project.toml @@ -1,6 +1,6 @@ name = "CopernicusClimateDataStore" uuid = "bce3f73f-acea-4481-bd86-df89ecc2cb46" -version = "0.2.0" +version = "0.2.1" authors = ["NumericalEarth and lovely collaborators"] [deps] diff --git a/src/cds_client.jl b/src/cds_client.jl index 714c1f1..7669a2e 100644 --- a/src/cds_client.jl +++ b/src/cds_client.jl @@ -99,6 +99,28 @@ function parse_cdsapi_rc(path::String) return CDSCredentials(url, key) end +""" + request_with_retries(f; attempts=4, initial_delay=1.0, what="CDS request") + +Call `f()`, retrying with exponential backoff on failure. The CDS gateway +intermittently answers valid requests with transient errors (e.g. 502); without +retries one blip kills a request that may be minutes into its queue wait — and +takes every other request of a concurrent bundle down with it. +""" +function request_with_retries(f; attempts=4, initial_delay=1.0, what="CDS request") + delay = initial_delay + for attempt in 1:attempts + try + return f() + catch e + attempt < attempts || rethrow() + @warn "$what failed (attempt $attempt/$attempts); retrying in $(round(delay, digits=1)) s" exception=e + sleep(delay) + delay *= 2 + end + end +end + """ submit_cds_request(credentials, dataset, params) @@ -117,7 +139,7 @@ function submit_cds_request(credentials::CDSCredentials, dataset::String, params # v2 requires params wrapped in "inputs" body = JSON3.write(Dict("inputs" => params)) - response = HTTP.post(endpoint, headers, body) + response = request_with_retries(() -> HTTP.post(endpoint, headers, body); what="CDS submit") # Extract status endpoint from Location header location = String(Dict(response.headers)["location"]) @@ -130,6 +152,10 @@ end Poll CDS request status until completion or timeout. Returns download URL when ready. + +Polling starts fast (1 s) and backs off exponentially to `poll_interval`, so small +requests that complete in seconds aren't held up to a full fixed interval, while +long queue waits still poll gently. """ function poll_request_status(credentials::CDSCredentials, status_endpoint::String; max_wait=3600, poll_interval=5, verbose=true) @@ -137,9 +163,10 @@ function poll_request_status(credentials::CDSCredentials, status_endpoint::Strin start_time = time() last_status = "" + delay = min(1.0, poll_interval) while time() - start_time < max_wait - response = HTTP.get(status_endpoint, headers) + response = request_with_retries(() -> HTTP.get(status_endpoint, headers); what="CDS status poll") result = JSON3.read(String(response.body)) status = result.status @@ -159,7 +186,7 @@ function poll_request_status(credentials::CDSCredentials, status_endpoint::Strin if status == "successful" # Get results endpoint and extract actual download URL results_url = status_endpoint * "/results" - results_response = HTTP.get(results_url, headers) + results_response = request_with_retries(() -> HTTP.get(results_url, headers); what="CDS results fetch") results_json = JSON3.read(String(results_response.body)) # Extract download URL from asset.value.href @@ -169,7 +196,8 @@ function poll_request_status(credentials::CDSCredentials, status_endpoint::Strin error("CDS request failed. Check https://cds.climate.copernicus.eu/requests for details.") end - sleep(poll_interval) + sleep(delay) + delay = min(delay * 1.6, poll_interval) end error("Request timed out after $(max_wait)s") @@ -185,7 +213,7 @@ function download_cds_file(url::String, output_path::String, credentials::CDSCre # Add authentication header headers = ["PRIVATE-TOKEN" => credentials.key] - Downloads.download(url, output_path; headers) + request_with_retries(() -> Downloads.download(url, output_path; headers); what="CDS file download") return output_path end diff --git a/test/runtests.jl b/test/runtests.jl index 62de1f0..eb030b9 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -29,6 +29,25 @@ using CopernicusClimateDataStore @test isdefined(CopernicusClimateDataStore, :download_cds_file) end + @testset "request_with_retries" begin + # Succeeds once the transient failures stop + calls = Ref(0) + result = CopernicusClimateDataStore.request_with_retries(; initial_delay=0.01) do + calls[] += 1 + calls[] < 3 ? error("transient") : :ok + end + @test result == :ok + @test calls[] == 3 + + # Rethrows after exhausting attempts + calls[] = 0 + @test_throws ErrorException CopernicusClimateDataStore.request_with_retries(; attempts=2, initial_delay=0.01) do + calls[] += 1 + error("persistent") + end + @test calls[] == 2 + end + @testset "hourly file naming and skip-existing (offline)" begin # A single variable keeps the historical names; existing files short-circuit # the request entirely, so these run without credentials or network. From 678863e8d09a42ed8f8ed15beefdb2dec93309f3 Mon Sep 17 00:00:00 2001 From: "Gregory L. Wagner" Date: Tue, 14 Jul 2026 21:37:59 +0000 Subject: [PATCH 3/3] Add request_with_retries to the API reference Documenter's missing_docs check fails on docstrings absent from the manual; list the new helper alongside the other request-lifecycle helpers. Co-Authored-By: Claude Fable 5 --- docs/src/api.md | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/src/api.md b/docs/src/api.md index 4974e6a..4f003f4 100644 --- a/docs/src/api.md +++ b/docs/src/api.md @@ -14,6 +14,7 @@ CDSCredentials submit_cds_request poll_request_status download_cds_file +CopernicusClimateDataStore.request_with_retries ``` ## Utility Functions