Fix ERA5 CDS API endpoint for new v1 API

- Submit: /api/retrieve/v1/processes/{dataset}/execution
- Request body wrapped in {"inputs": ...}
- Poll: /api/retrieve/v1/jobs/{jobId}
- Status field: "status" not "state", values: accepted/running/successful/failed
- Download: GET {job_url}/results, follow asset href link
- Add ERA5 credentials to k8s secret manifest
This commit is contained in:
Graham McIntire 2026-04-07 12:32:20 -05:00
parent fc65e2f731
commit fbf24622fa

View file

@ -106,15 +106,15 @@ defmodule Microwaveprop.Weather.Era5Client do
end
defp submit_request(dataset, request, api_key) do
url = "#{@cds_url}/retrieve/#{dataset}"
url = "#{@cds_url}/retrieve/v1/processes/#{dataset}/execution"
case Req.post(url,
json: request,
json: %{"inputs" => request},
headers: [{"PRIVATE-TOKEN", api_key}],
receive_timeout: 30_000
) do
{:ok, %{status: status, body: body}} when status in [200, 201, 202] ->
job_id = body["request_id"] || body["jobId"]
job_id = body["jobID"] || body["request_id"] || body["jobId"]
if job_id do
Logger.info("ERA5: submitted job #{job_id} for #{dataset}")
@ -138,24 +138,20 @@ defmodule Microwaveprop.Weather.Era5Client do
end
defp poll_and_download(job_id, api_key, attempt) do
url = "#{@cds_url}/tasks/#{job_id}"
url = "#{@cds_url}/retrieve/v1/jobs/#{job_id}"
case Req.get(url, headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 10_000) do
{:ok, %{status: 200, body: %{"state" => "completed"} = body}} ->
download_url = body["location"] || get_in(body, ["result", "location"])
{:ok, %{status: 200, body: %{"status" => "successful"}}} ->
results_url = "#{url}/results"
download_result(results_url, api_key)
if download_url do
download_result(download_url, api_key)
else
{:error, "ERA5: completed but no download URL in #{inspect(body)}"}
end
{:ok, %{status: 200, body: %{"state" => state}}} when state in ["queued", "running"] ->
{:ok, %{status: 200, body: %{"status" => status}}}
when status in ["accepted", "running"] ->
Process.sleep(@poll_interval_ms)
poll_and_download(job_id, api_key, attempt + 1)
{:ok, %{status: 200, body: %{"state" => "failed"} = body}} ->
{:error, "ERA5 job failed: #{inspect(body["error"])}"}
{:ok, %{status: 200, body: %{"status" => "failed"} = body}} ->
{:error, "ERA5 job failed: #{inspect(body["message"])}"}
{:ok, %{status: status, body: body}} ->
{:error, "ERA5 poll HTTP #{status}: #{inspect(body)}"}
@ -166,18 +162,37 @@ defmodule Microwaveprop.Weather.Era5Client do
end
defp download_result(url, api_key) do
case Req.get(url, headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 120_000, into: :self) do
{:ok, %{status: 200, body: body}} ->
{:ok, body}
case Req.get(url, headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 120_000) do
{:ok, %{status: 200, body: %{"asset" => %{"value" => %{"href" => href}}}}} ->
# New CDS returns a JSON with download link
download_file(href, api_key)
{:ok, %{status: status}} ->
{:error, "ERA5 download HTTP #{status}"}
{:ok, %{status: 200, headers: headers, body: body}} ->
# Direct binary download
content_type = %Req.Response{headers: headers} |> Req.Response.get_header("content-type") |> List.first("")
if String.contains?(content_type, "grib") or is_binary(body) do
{:ok, body}
else
{:error, "ERA5: unexpected response format"}
end
{:ok, %{status: status, body: body}} ->
{:error, "ERA5 download HTTP #{status}: #{inspect(body)}"}
{:error, reason} ->
{:error, "ERA5 download failed: #{inspect(reason)}"}
end
end
defp download_file(href, api_key) do
case Req.get(href, headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 120_000) do
{:ok, %{status: 200, body: body}} -> {:ok, body}
{:ok, %{status: status}} -> {:error, "ERA5 file download HTTP #{status}"}
{:error, reason} -> {:error, "ERA5 file download failed: #{inspect(reason)}"}
end
end
defp build_profile(lat, lon, valid_time, single_grib, pressure_grib) do
# Decode GRIB2 data using existing infrastructure
with {:ok, single_points} <- extract_point(single_grib, lat, lon),