hourly: multi-variable requests, levels compatibility, concurrent downloads

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]

docs/src/api.md

@@ -14,6 +14,7 @@ CDSCredentials
 submit_cds_request
 poll_request_status
 download_cds_file
+CopernicusClimateDataStore.request_with_retries
 ```
 
 ## Utility Functions

src/CopernicusClimateDataStore.jl

@@ -8,25 +8,44 @@ 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...)
 
 Download ERA5 hourly data using the CDS API. This function provides compatibility
 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
 
 """

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,16 +152,21 @@ 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)
     headers = ["PRIVATE-TOKEN" => credentials.key]
 
     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

test/runtests.jl

@@ -29,6 +29,54 @@ 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.
+        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