From 8637253fda4bd2124d46d7008a7927e968fee224 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Thu, 9 Apr 2026 13:00:58 -0500 Subject: [PATCH] =?UTF-8?q?Batch=20ERA5=20fetches=20by=20month=20and=202?= =?UTF-8?q?=C2=B0=20tile?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per-point ERA5 fetches were tragically slow because every point-hour triggered its own asynchronous CDS job (submit → poll → assemble → download). For backfill this meant thousands of independent jobs queued against Copernicus. The new path groups requests by calendar month and a 2° × 2° lat/lon tile so one CDS cycle populates ~60k profiles at once, and Oban uniqueness on (year, month, tile_lat, tile_lon) collapses every duplicate enqueue. - Era5BatchClient builds the monthly CDS requests, extracts every (lat, lon, hour) from the GRIB2 blob with wgrib2, derives refractivity params, and bulk-inserts in 2k-row chunks with on_conflict: :nothing. fetch_month_into_db/1 short-circuits when the month-tile already has any cached profile. - Era5MonthBatchWorker runs the batch on the :era5 queue with a generous backoff (10m → 1d) and the uniqueness key above. - Era5FetchWorker is now a thin router: cache hit → :ok, cache miss → enqueue the month-batch for the point's tile-month and return. No more per-point CDS calls. - Wgrib2 grows extract_grid_messages/3 which preserves per-message datetimes by parsing `d=YYYYMMDDHH[MMSS]` from the inventory, so a single GRIB2 file carrying a whole month decodes correctly. - The era5_backfill mix task enqueues month-tile batches directly. --- .../weather/era5_batch_client.ex | 298 ++++++++++++++++++ lib/microwaveprop/weather/era5_client.ex | 24 +- lib/microwaveprop/weather/grib2/wgrib2.ex | 141 ++++++++- .../workers/era5_fetch_worker.ex | 50 +-- .../workers/era5_month_batch_worker.ex | 64 ++++ lib/mix/tasks/era5_backfill.ex | 65 ++-- .../weather/era5_batch_client_test.exs | 78 +++++ .../workers/era5_fetch_worker_test.exs | 41 +++ .../workers/era5_month_batch_worker_test.exs | 43 +++ 9 files changed, 744 insertions(+), 60 deletions(-) create mode 100644 lib/microwaveprop/weather/era5_batch_client.ex create mode 100644 lib/microwaveprop/workers/era5_month_batch_worker.ex create mode 100644 test/microwaveprop/weather/era5_batch_client_test.exs create mode 100644 test/microwaveprop/workers/era5_month_batch_worker_test.exs diff --git a/lib/microwaveprop/weather/era5_batch_client.ex b/lib/microwaveprop/weather/era5_batch_client.ex new file mode 100644 index 00000000..12febd7b --- /dev/null +++ b/lib/microwaveprop/weather/era5_batch_client.ex @@ -0,0 +1,298 @@ +defmodule Microwaveprop.Weather.Era5BatchClient do + @moduledoc """ + Fetches ERA5 reanalysis data from Copernicus Climate Data Store (CDS) one + month-tile at a time and bulk-inserts the resulting profiles into + `era5_profiles`. + + The per-point `Era5Client.fetch_profile/3` path is tragically slow because + every point-hour triggers its own asynchronous CDS job (submit → poll → + assemble → download). For backfill this balloons into thousands of + independent CDS jobs. This module instead groups requests by month and a + small lat/lon tile (2° × 2° by default) so one submit/poll/download cycle + produces many thousands of profile rows. + + Idempotency: `fetch_month_into_db/1` is a no-op when the requested + (year, month, tile) already has profile rows in `era5_profiles`. Combined + with Oban uniqueness on `Era5MonthBatchWorker`, redundant fetches are + collapsed. + """ + + import Ecto.Query + + alias Microwaveprop.Repo + alias Microwaveprop.Weather.Era5Client + alias Microwaveprop.Weather.Era5Profile + alias Microwaveprop.Weather.Grib2.Wgrib2 + alias Microwaveprop.Weather.SoundingParams + + require Logger + + # ERA5 hourly reanalysis runs on a 0.25° global grid. Tiles are aligned + # to an even-degree lattice (2° × 2°), producing 9 × 9 = 81 grid points + # per tile per hour. A month-tile expands to ~60k profiles. + @tile_degrees 2 + @era5_step 0.25 + + @doc """ + Returns the integer tile coordinate `{tile_lat, tile_lon}` for a point, + where `tile_lat` is the south edge and `tile_lon` is the west edge. + """ + def tile_for_point(lat, lon) do + {floor_to_tile(lat), floor_to_tile(lon)} + end + + defp floor_to_tile(value) do + value + |> Kernel./(@tile_degrees) + |> Float.floor() + |> trunc() + |> Kernel.*(@tile_degrees) + end + + @doc false + def tile_degrees, do: @tile_degrees + + @doc """ + Fetches one month of ERA5 data covering the given tile and stores every + resulting hourly profile in `era5_profiles`. Returns + `{:ok, inserted_count}` or `{:error, reason}`. + """ + def fetch_month_into_db(%{year: year, month: month, tile_lat: tile_lat, tile_lon: tile_lon}) do + if month_tile_cached?(year, month, tile_lat, tile_lon) do + Logger.info("ERA5Batch: #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon} already cached") + {:ok, 0} + else + do_fetch_month(year, month, tile_lat, tile_lon) + end + end + + defp do_fetch_month(year, month, tile_lat, tile_lon) do + area = tile_area(tile_lat, tile_lon) + days = days_in_month(year, month) + + Logger.info("ERA5Batch: fetching #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon} (#{days} days)") + + with {:ok, single_grib} <- fetch_single_level_month(year, month, days, area), + {:ok, pressure_grib} <- fetch_pressure_level_month(year, month, days, area), + {:ok, profiles} <- build_month_profiles(tile_lat, tile_lon, single_grib, pressure_grib) do + inserted = bulk_insert_profiles(profiles) + + Logger.info("ERA5Batch: stored #{inserted} profiles for #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon}") + + {:ok, inserted} + end + end + + defp month_tile_cached?(year, month, tile_lat, tile_lon) do + {first, last} = month_bounds(year, month) + + Era5Profile + |> where( + [p], + p.valid_time >= ^first and p.valid_time < ^last and + p.lat >= ^(tile_lat * 1.0) and p.lat <= ^(tile_lat * 1.0 + @tile_degrees) and + p.lon >= ^(tile_lon * 1.0) and p.lon <= ^(tile_lon * 1.0 + @tile_degrees) + ) + |> Repo.exists?() + end + + defp month_bounds(year, month) do + {:ok, first} = Date.new(year, month, 1) + {:ok, first_dt} = DateTime.new(first, ~T[00:00:00], "Etc/UTC") + last_dt = DateTime.add(first_dt, days_in_month(year, month) * 86_400, :second) + {first_dt, last_dt} + end + + defp days_in_month(year, month), do: Calendar.ISO.days_in_month(year, month) + + defp tile_area(tile_lat, tile_lon) do + # CDS area is [north, west, south, east] + Enum.map([tile_lat + @tile_degrees, tile_lon, tile_lat, tile_lon + @tile_degrees], &(&1 * 1.0)) + end + + defp pad(n), do: String.pad_leading(Integer.to_string(n), 2, "0") + + # -- CDS requests ---------------------------------------------------------- + + defp fetch_single_level_month(year, month, days, area) do + request = %{ + "product_type" => ["reanalysis"], + "variable" => Era5Client.single_level_vars(), + "year" => [Integer.to_string(year)], + "month" => [pad(month)], + "day" => Enum.map(1..days, &pad/1), + "time" => Enum.map(0..23, &"#{pad(&1)}:00"), + "area" => area, + "data_format" => "grib" + } + + Era5Client.submit_and_download("reanalysis-era5-single-levels", request) + end + + defp fetch_pressure_level_month(year, month, days, area) do + request = %{ + "product_type" => ["reanalysis"], + "variable" => Era5Client.pressure_level_vars(), + "pressure_level" => Era5Client.pressure_levels(), + "year" => [Integer.to_string(year)], + "month" => [pad(month)], + "day" => Enum.map(1..days, &pad/1), + "time" => Enum.map(0..23, &"#{pad(&1)}:00"), + "area" => area, + "data_format" => "grib" + } + + Era5Client.submit_and_download("reanalysis-era5-pressure-levels", request) + end + + # -- GRIB extraction ------------------------------------------------------- + + defp grid_spec_for_tile(tile_lat, tile_lon) do + count = round(@tile_degrees / @era5_step) + 1 + + %{ + lon_start: tile_lon * 1.0, + lon_count: count, + lon_step: @era5_step, + lat_start: tile_lat * 1.0, + lat_count: count, + lat_step: @era5_step + } + end + + defp build_month_profiles(tile_lat, tile_lon, single_grib, pressure_grib) do + grid_spec = grid_spec_for_tile(tile_lat, tile_lon) + + with {:ok, single_msgs} <- + Wgrib2.extract_grid_messages(single_grib, ":(TMP|DPT|PRES|UGRD|VGRD|PWAT|HPBL):", grid_spec), + {:ok, pressure_msgs} <- Wgrib2.extract_grid_messages(pressure_grib, ":(TMP|DPT|HGT):", grid_spec) do + # Index both message sets by (datetime, lat, lon) → var:level map. + points_single = index_messages(single_msgs) + points_pressure = index_messages(pressure_msgs) + + all_keys = + MapSet.union( + MapSet.new(Map.keys(points_single)), + MapSet.new(Map.keys(points_pressure)) + ) + + profiles = + all_keys + |> Enum.map(fn {dt, lat, lon} -> + single_fields = Map.get(points_single, {dt, lat, lon}, %{}) + pressure_fields = Map.get(points_pressure, {dt, lat, lon}, %{}) + build_profile_attrs(lat, lon, dt, single_fields, pressure_fields) + end) + |> Enum.reject(&is_nil/1) + + {:ok, profiles} + end + end + + defp index_messages(messages) do + Enum.reduce(messages, %{}, fn msg, acc -> + key_name = "#{msg.var}:#{msg.level}" + + Enum.reduce(msg.values, acc, fn {{lat, lon}, value}, inner -> + Map.update( + inner, + {msg.datetime, lat, lon}, + %{key_name => value}, + &Map.put(&1, key_name, value) + ) + end) + end) + end + + defp build_profile_attrs(_lat, _lon, nil, _single, _pressure), do: nil + + defp build_profile_attrs(lat, lon, datetime, single_fields, pressure_fields) do + temp_c = kelvin_to_celsius(single_fields["TMP:2 m above ground"]) + dewpoint_c = kelvin_to_celsius(single_fields["DPT:2 m above ground"]) + pressure_mb = pa_to_mb(single_fields["PRES:surface"]) + hpbl_m = single_fields["HPBL:surface"] + pwat_mm = single_fields["PWAT:entire atmosphere"] + + profile = + Era5Client.pressure_levels() + |> Enum.map(fn level_str -> + level = String.to_integer(level_str) + pres_key = "#{level} mb" + + %{ + "pres" => level * 1.0, + "tmpc" => kelvin_to_celsius(pressure_fields["TMP:#{pres_key}"]), + "dwpc" => kelvin_to_celsius(pressure_fields["DPT:#{pres_key}"]), + "hght" => geopotential_to_height(pressure_fields["HGT:#{pres_key}"]) + } + end) + |> Enum.reject(&is_nil(&1["tmpc"])) + + if profile == [] and is_nil(temp_c) do + nil + else + params = SoundingParams.derive(profile) + + base = %{ + valid_time: datetime, + lat: lat, + lon: lon, + profile: profile, + hpbl_m: hpbl_m, + pwat_mm: pwat_mm, + surface_temp_c: temp_c, + surface_dewpoint_c: dewpoint_c, + surface_pressure_mb: pressure_mb + } + + if params do + Map.merge(base, %{ + surface_refractivity: params.surface_refractivity, + min_refractivity_gradient: params.min_refractivity_gradient, + ducting_detected: params.ducting_detected || false, + duct_characteristics: params.duct_characteristics + }) + else + base + end + end + end + + defp kelvin_to_celsius(nil), do: nil + defp kelvin_to_celsius(k), do: k - 273.15 + + defp pa_to_mb(nil), do: nil + defp pa_to_mb(pa), do: pa / 100.0 + + defp geopotential_to_height(nil), do: nil + defp geopotential_to_height(gp), do: gp / 9.80665 + + # -- Bulk insert ----------------------------------------------------------- + + # Insert in chunks to keep each query under Postgres' parameter limit + # (65,535 params / ~15 fields ≈ 4,000 rows per statement). + @chunk_size 2_000 + + defp bulk_insert_profiles(profiles) do + now = DateTime.truncate(DateTime.utc_now(), :second) + + profiles + |> Enum.chunk_every(@chunk_size) + |> Enum.reduce(0, fn chunk, acc -> + rows = + Enum.map(chunk, fn attrs -> + attrs + |> Map.put(:inserted_at, now) + |> Map.put(:updated_at, now) + end) + + {count, _} = + Repo.insert_all(Era5Profile, rows, + on_conflict: :nothing, + conflict_target: [:lat, :lon, :valid_time] + ) + + acc + count + end) + end +end diff --git a/lib/microwaveprop/weather/era5_client.ex b/lib/microwaveprop/weather/era5_client.ex index 5185b9ed..94c4fcfe 100644 --- a/lib/microwaveprop/weather/era5_client.ex +++ b/lib/microwaveprop/weather/era5_client.ex @@ -31,6 +31,19 @@ defmodule Microwaveprop.Weather.Era5Client do @pressure_level_vars ~w(temperature dewpoint_temperature geopotential) + @doc false + def pressure_levels, do: @pressure_levels + @doc false + def single_level_vars, do: @single_level_vars + @doc false + def pressure_level_vars, do: @pressure_level_vars + @doc false + def cds_url, do: @cds_url + @doc false + def poll_interval_ms, do: @poll_interval_ms + @doc false + def max_poll_attempts, do: @max_poll_attempts + @doc """ Fetch ERA5 profile for a single point and time. Returns {:ok, profile_attrs} or {:error, reason}. @@ -70,7 +83,7 @@ defmodule Microwaveprop.Weather.Era5Client do "data_format" => "grib" } - submit_and_download("reanalysis-era5-single-levels", request) + do_submit_and_download("reanalysis-era5-single-levels", request) end @doc """ @@ -89,10 +102,15 @@ defmodule Microwaveprop.Weather.Era5Client do "data_format" => "grib" } - submit_and_download("reanalysis-era5-pressure-levels", request) + do_submit_and_download("reanalysis-era5-pressure-levels", request) end - defp submit_and_download(dataset, request) do + @doc false + def submit_and_download(dataset, request) do + do_submit_and_download(dataset, request) + end + + defp do_submit_and_download(dataset, request) do api_key = api_key() if is_nil(api_key) do diff --git a/lib/microwaveprop/weather/grib2/wgrib2.ex b/lib/microwaveprop/weather/grib2/wgrib2.ex index 63e457e9..efa2aadd 100644 --- a/lib/microwaveprop/weather/grib2/wgrib2.ex +++ b/lib/microwaveprop/weather/grib2/wgrib2.ex @@ -98,8 +98,8 @@ defmodule Microwaveprop.Weather.Grib2.Wgrib2 do |> String.split("\n") |> Enum.flat_map(fn line -> case String.split(line, ":", parts: 8) do - [_n, _offset, _date, var, level | _] -> - [%{var: var, level: level}] + [_n, _offset, date, var, level | _] -> + [%{var: var, level: level, datetime: parse_wgrib2_date(date)}] _ -> [] @@ -107,6 +107,143 @@ defmodule Microwaveprop.Weather.Grib2.Wgrib2 do end) end + # wgrib2 inventory date looks like "d=20250101000000" or "d=2025010100". + # Returns a `DateTime` in UTC, or `nil` if it can't be parsed. + defp parse_wgrib2_date("d=" <> digits), do: parse_wgrib2_date_digits(digits) + defp parse_wgrib2_date(_), do: nil + + defp parse_wgrib2_date_digits(<>) do + {mi, s} = + case rest do + <> -> {mm, ss} + <> -> {mm, "00"} + _ -> {"00", "00"} + end + + with {year, ""} <- Integer.parse(y), + {month, ""} <- Integer.parse(m), + {day, ""} <- Integer.parse(d), + {hour, ""} <- Integer.parse(h), + {minute, ""} <- Integer.parse(mi), + {second, ""} <- Integer.parse(s), + {:ok, naive} <- NaiveDateTime.new(year, month, day, hour, minute, second), + {:ok, dt} <- DateTime.from_naive(naive, "Etc/UTC") do + DateTime.truncate(dt, :second) + else + _ -> nil + end + end + + defp parse_wgrib2_date_digits(_), do: nil + + @doc """ + Like `extract_grid/3`, but returns a flat list of per-message results so + the time dimension is preserved. Used when a single GRIB2 blob carries + many timesteps (e.g. a month of ERA5 data fetched in one CDS request). + + Each entry is `%{datetime: DateTime.t() | nil, var: String.t(), + level: String.t(), values: %{{lat, lon} => float}}`. + """ + def extract_grid_messages(grib_binary, match_pattern, grid_spec) do + if available?() do + extract_messages_with_wgrib2(grib_binary, match_pattern, grid_spec) + else + {:error, :wgrib2_not_available} + end + end + + defp extract_messages_with_wgrib2(grib_binary, match_pattern, grid_spec) do + %{ + lon_start: lon_start, + lon_count: lon_count, + lon_step: lon_step, + lat_start: lat_start, + lat_count: lat_count, + lat_step: lat_step + } = grid_spec + + wgrib2_lon_start = normalize_lon(lon_start) + lon_spec = "#{wgrib2_lon_start}:#{lon_count}:#{lon_step}" + lat_spec = "#{lat_start}:#{lat_count}:#{lat_step}" + + tmp_grib = Path.join(System.tmp_dir!(), "era5_#{System.unique_integer([:positive])}.grib2") + tmp_bin = tmp_grib <> ".lola.bin" + + try do + File.write!(tmp_grib, grib_binary) + + args = [tmp_grib, "-match", match_pattern, "-lola", lon_spec, lat_spec, tmp_bin, "bin"] + + case System.cmd(wgrib2_path(), args, stderr_to_stdout: true) do + {output, 0} -> + messages = parse_wgrib2_inventory(output) + + case File.read(tmp_bin) do + {:ok, bin_data} -> + {:ok, build_messages_per_message(bin_data, messages, grid_spec)} + + {:error, :enoent} -> + {:ok, []} + + {:error, reason} -> + {:error, "Failed to read wgrib2 output: #{inspect(reason)}"} + end + + {output, exit_code} -> + {:error, "wgrib2 failed (exit #{exit_code}): #{String.slice(output, 0, 200)}"} + end + after + File.rm(tmp_grib) + File.rm(tmp_bin) + end + end + + defp build_messages_per_message(bin_data, messages, grid_spec) do + %{lon_count: nx, lat_count: ny} = grid_spec + points_per_message = nx * ny + bytes_per_message = points_per_message * 4 + + messages + |> Enum.with_index() + |> Enum.flat_map(fn {msg, msg_idx} -> + offset = msg_idx * bytes_per_message + + if offset + bytes_per_message <= byte_size(bin_data) do + chunk = binary_part(bin_data, offset, bytes_per_message) + values = extract_message_values(chunk, grid_spec) + + [Map.put(msg, :values, values)] + else + [] + end + end) + end + + defp extract_message_values(chunk, %{ + lon_count: nx, + lat_count: ny, + lon_start: lon_start, + lon_step: lon_step, + lat_start: lat_start, + lat_step: lat_step + }) do + Enum.reduce(0..(ny - 1), %{}, fn j, outer -> + lat = Float.round(lat_start + j * lat_step, 3) + + Enum.reduce(0..(nx - 1), outer, fn i, inner -> + offset = (j * nx + i) * 4 + <<_::binary-size(offset), value::float-little-32, _::binary>> = chunk + + if value > @undefined_value / 2 do + inner + else + lon = Float.round(denormalize_lon(lon_start + i * lon_step), 3) + Map.put(inner, {lat, lon}, value) + end + end) + end) + end + defp parse_lola_binary(bin_data, messages, grid_spec) do %{lon_count: nx, lat_count: ny, lon_start: lon_start, lon_step: lon_step, lat_start: lat_start, lat_step: lat_step} = grid_spec diff --git a/lib/microwaveprop/workers/era5_fetch_worker.ex b/lib/microwaveprop/workers/era5_fetch_worker.ex index eb74ab47..10846a35 100644 --- a/lib/microwaveprop/workers/era5_fetch_worker.ex +++ b/lib/microwaveprop/workers/era5_fetch_worker.ex @@ -1,12 +1,14 @@ defmodule Microwaveprop.Workers.Era5FetchWorker do @moduledoc """ - Fetches ERA5 reanalysis profiles for contacts where HRRR data is unavailable. - Primarily for pre-2014 contacts that predate HRRR availability. + Thin router that ensures an ERA5 profile is available for a single + `{lat, lon, valid_time}`. On a cache miss it enqueues the much more + efficient `Era5MonthBatchWorker` for the month-tile that contains the + requested point, rather than hitting the CDS API point-by-point. - Jobs are unique on `{lat, lon, valid_time}` — if another fetch for the same - grid point and hour is already queued, scheduled, running, or retrying, a - duplicate insert is collapsed onto the existing job so we never hit the - Copernicus CDS API twice for the same data. + Jobs are unique on `{lat, lon, valid_time}` so duplicate enrichment + requests collapse. The enqueued month-batch worker is itself unique on + `(year, month, tile_lat, tile_lon)` so the first enqueue wins and later + ones are no-ops. """ use Oban.Worker, queue: :era5, @@ -17,15 +19,16 @@ defmodule Microwaveprop.Workers.Era5FetchWorker do ] alias Microwaveprop.Repo - alias Microwaveprop.Weather.Era5Client + alias Microwaveprop.Weather.Era5BatchClient alias Microwaveprop.Weather.Era5Profile + alias Microwaveprop.Workers.Era5MonthBatchWorker require Logger @impl Oban.Worker def backoff(%Oban.Job{attempt: attempt}) do - # ERA5 CDS API can be slow; generous backoff - min(300 * Integer.pow(2, attempt - 1), _one_day = 86_400) + # Enqueueing the batch is cheap; retry fast so we stop blocking downstream. + min(30 * Integer.pow(2, attempt - 1), _one_hour = 3_600) end @impl Oban.Worker @@ -36,27 +39,24 @@ defmodule Microwaveprop.Workers.Era5FetchWorker do rlon = Float.round(lon * 4) / 4 if has_era5_profile?(rlat, rlon, valid_time) do - Logger.debug("ERA5: profile already exists for #{rlat},#{rlon} @ #{valid_time_str}") :ok else - Logger.info("ERA5: fetching profile for #{rlat},#{rlon} @ #{valid_time_str}") + {tile_lat, tile_lon} = Era5BatchClient.tile_for_point(rlat, rlon) - case Era5Client.fetch_profile(lat, lon, valid_time) do - {:ok, attrs} -> - %Era5Profile{} - |> Era5Profile.changeset(attrs) - |> Repo.insert( - on_conflict: :nothing, - conflict_target: [:lat, :lon, :valid_time] - ) + Logger.info( + "ERA5: enqueueing month batch for #{valid_time.year}-#{valid_time.month} tile #{tile_lat},#{tile_lon} (triggered by #{rlat},#{rlon} @ #{valid_time_str})" + ) - Logger.info("ERA5: stored profile for #{rlat},#{rlon} @ #{valid_time_str}") - :ok + %{ + year: valid_time.year, + month: valid_time.month, + tile_lat: tile_lat, + tile_lon: tile_lon + } + |> Era5MonthBatchWorker.new() + |> Oban.insert() - {:error, reason} -> - Logger.warning("ERA5: failed for #{rlat},#{rlon} @ #{valid_time_str}: #{inspect(reason)}") - {:error, reason} - end + :ok end end diff --git a/lib/microwaveprop/workers/era5_month_batch_worker.ex b/lib/microwaveprop/workers/era5_month_batch_worker.ex new file mode 100644 index 00000000..910928e7 --- /dev/null +++ b/lib/microwaveprop/workers/era5_month_batch_worker.ex @@ -0,0 +1,64 @@ +defmodule Microwaveprop.Workers.Era5MonthBatchWorker do + @moduledoc """ + Fetches one month of ERA5 data for a 2° × 2° tile in a single CDS request + and bulk-inserts all resulting hourly profiles into `era5_profiles`. + + This exists because fetching ERA5 per point-hour is tragically slow: each + CDS request goes through a full submit → poll → assemble → download cycle + whose overhead dwarfs the data transfer. Grouping by month+tile turns + thousands of CDS jobs into a handful, and once a tile-month is cached in + the DB every downstream query for a point inside that window is a local + lookup. + + Uniqueness on the full arg set means Oban collapses duplicate enqueues + for the same tile-month — multiple `Era5FetchWorker` requests for the + same region collapse into a single batch. + """ + use Oban.Worker, + queue: :era5, + max_attempts: 3, + unique: [ + period: :infinity, + states: [:available, :scheduled, :executing, :retryable], + keys: [:year, :month, :tile_lat, :tile_lon] + ] + + alias Microwaveprop.Weather.Era5BatchClient + + require Logger + + @impl Oban.Worker + def backoff(%Oban.Job{attempt: attempt}) do + # A full month fetch can sit in the CDS queue for ages; back off generously. + min(600 * Integer.pow(2, attempt - 1), _one_day = 86_400) + end + + @impl Oban.Worker + def perform(%Oban.Job{args: args}) do + params = %{ + year: fetch_int!(args, "year"), + month: fetch_int!(args, "month"), + tile_lat: fetch_int!(args, "tile_lat"), + tile_lon: fetch_int!(args, "tile_lon") + } + + case Era5BatchClient.fetch_month_into_db(params) do + {:ok, _count} -> + :ok + + {:error, reason} -> + Logger.warning( + "ERA5Batch: #{params.year}-#{params.month} tile #{params.tile_lat},#{params.tile_lon} failed: #{inspect(reason)}" + ) + + {:error, reason} + end + end + + defp fetch_int!(args, key) do + case Map.fetch!(args, key) do + n when is_integer(n) -> n + n when is_binary(n) -> String.to_integer(n) + end + end +end diff --git a/lib/mix/tasks/era5_backfill.ex b/lib/mix/tasks/era5_backfill.ex index 3441064d..66a93529 100644 --- a/lib/mix/tasks/era5_backfill.ex +++ b/lib/mix/tasks/era5_backfill.ex @@ -1,12 +1,18 @@ defmodule Mix.Tasks.Era5Backfill do - @shortdoc "Enqueue ERA5 fetch jobs for contacts without HRRR data" + @shortdoc "Enqueue ERA5 month-batch jobs for contacts without HRRR data" @moduledoc """ - Finds contacts where HRRR status is 'unavailable' (pre-2014 or missing from archive) - and enqueues ERA5 fetch jobs for their path points. + Finds contacts where HRRR status is `:unavailable` (pre-2014 or missing + from the archive) and enqueues the month-tile batch worker for every + tile-month their path points need. Usage: - mix era5_backfill # enqueue all (default limit 1000) - mix era5_backfill 5000 # enqueue up to 5000 + mix era5_backfill # scan up to 1000 contacts + mix era5_backfill 5000 # scan up to 5000 contacts + + Enqueues `Era5MonthBatchWorker` jobs deduplicated by + `(year, month, tile_lat, tile_lon)`. The per-point `Era5FetchWorker` + path still works for one-off enrichment, but this task goes straight + to the batch worker so backfills don't pay the per-point enqueue cost. """ use Mix.Task @@ -15,8 +21,8 @@ defmodule Mix.Tasks.Era5Backfill do alias Microwaveprop.Radio alias Microwaveprop.Radio.Contact alias Microwaveprop.Repo - alias Microwaveprop.Weather - alias Microwaveprop.Workers.Era5FetchWorker + alias Microwaveprop.Weather.Era5BatchClient + alias Microwaveprop.Workers.Era5MonthBatchWorker @impl Mix.Task def run(args) do @@ -37,40 +43,39 @@ defmodule Mix.Tasks.Era5Backfill do Mix.shell().info("Found #{length(contacts)} contacts needing ERA5 data") - jobs = + batches = contacts |> Enum.flat_map(fn contact -> + valid_time = %{DateTime.truncate(contact.qso_timestamp, :second) | minute: 0, second: 0} + contact |> Radio.contact_path_points() |> Enum.map(fn {lat, lon} -> - rlat = Float.round(lat * 4) / 4 - rlon = Float.round(lon * 4) / 4 - valid_time = %{DateTime.truncate(contact.qso_timestamp, :second) | minute: 0, second: 0} + {tile_lat, tile_lon} = Era5BatchClient.tile_for_point(lat, lon) - if Weather.find_nearest_era5(rlat, rlon, valid_time) do - nil - else - Era5FetchWorker.new(%{ - "lat" => rlat, - "lon" => rlon, - "valid_time" => DateTime.to_iso8601(valid_time) - }) - end + %{ + year: valid_time.year, + month: valid_time.month, + tile_lat: tile_lat, + tile_lon: tile_lon + } end) - |> Enum.reject(&is_nil/1) end) - |> Enum.uniq_by(fn changeset -> changeset.changes.args end) + |> Enum.uniq() - if jobs == [] do - Mix.shell().info("No ERA5 jobs needed (all data already exists)") + if batches == [] do + Mix.shell().info("No ERA5 batches to enqueue") else - # Use Oban.insert/1 so the worker's unique constraint collapses - # duplicate (lat, lon, valid_time) jobs that may already be in flight - # from a previous backfill run. Oban OSS insert_all does not check - # uniqueness. - Enum.each(jobs, &Oban.insert/1) + # Use Oban.insert/1 (not insert_all) so the worker's unique + # constraint collapses duplicate tile-month jobs against anything + # already queued. + Enum.each(batches, fn args -> + args + |> Era5MonthBatchWorker.new() + |> Oban.insert() + end) - Mix.shell().info("Enqueued #{length(jobs)} ERA5 fetch jobs") + Mix.shell().info("Enqueued #{length(batches)} ERA5 month-tile batches") end end end diff --git a/test/microwaveprop/weather/era5_batch_client_test.exs b/test/microwaveprop/weather/era5_batch_client_test.exs new file mode 100644 index 00000000..9197c796 --- /dev/null +++ b/test/microwaveprop/weather/era5_batch_client_test.exs @@ -0,0 +1,78 @@ +defmodule Microwaveprop.Weather.Era5BatchClientTest do + use Microwaveprop.DataCase, async: true + + alias Microwaveprop.Weather.Era5BatchClient + alias Microwaveprop.Weather.Era5Profile + + describe "tile_for_point/2" do + test "floors positive coordinates to a 2° lattice" do + assert {32, 96} = Era5BatchClient.tile_for_point(32.75, 97.5) + assert {32, 96} = Era5BatchClient.tile_for_point(33.99, 97.99) + assert {34, 98} = Era5BatchClient.tile_for_point(34.0, 98.0) + end + + test "floors negative longitudes toward -infinity" do + assert {32, -98} = Era5BatchClient.tile_for_point(32.75, -97.25) + assert {32, -98} = Era5BatchClient.tile_for_point(33.0, -97.01) + assert {32, -100} = Era5BatchClient.tile_for_point(33.0, -99.0) + end + + test "floors negative latitudes toward -infinity" do + assert {-2, 0} = Era5BatchClient.tile_for_point(-1.25, 0.5) + assert {-34, -60} = Era5BatchClient.tile_for_point(-33.5, -59.5) + end + + test "edges land on the tile to the right/north" do + # 30.0 exactly → tile 30 (south/west edge) + assert {30, 96} = Era5BatchClient.tile_for_point(30.0, 97.5) + end + end + + describe "fetch_month_into_db/1 idempotency" do + test "returns {:ok, 0} when the month-tile already has any profile" do + # Seed one profile inside the 2014-03, tile (32, -98) window. + %Era5Profile{} + |> Era5Profile.changeset(%{ + lat: 32.5, + lon: -97.25, + valid_time: ~U[2014-03-15 12:00:00Z], + profile: [] + }) + |> Repo.insert!() + + # No CDS API key configured in tests → if the cache check fails, the + # function would return an ERA5_CDS_API_KEY error. {:ok, 0} means the + # short-circuit path was taken. + assert {:ok, 0} = + Era5BatchClient.fetch_month_into_db(%{ + year: 2014, + month: 3, + tile_lat: 32, + tile_lon: -98 + }) + end + + test "a profile outside the tile window does not count as cached" do + # This profile is in tile (30, -100), not (32, -98). + %Era5Profile{} + |> Era5Profile.changeset(%{ + lat: 30.5, + lon: -99.25, + valid_time: ~U[2014-03-15 12:00:00Z], + profile: [] + }) + |> Repo.insert!() + + # The (32, -98) tile is still uncached, so fetch_month_into_db tries + # to reach CDS and fails with missing API key, which bubbles up + # (rather than :ok, 0). + assert {:error, _} = + Era5BatchClient.fetch_month_into_db(%{ + year: 2014, + month: 3, + tile_lat: 32, + tile_lon: -98 + }) + end + end +end diff --git a/test/microwaveprop/workers/era5_fetch_worker_test.exs b/test/microwaveprop/workers/era5_fetch_worker_test.exs index 9b0c3987..a6a3dfcf 100644 --- a/test/microwaveprop/workers/era5_fetch_worker_test.exs +++ b/test/microwaveprop/workers/era5_fetch_worker_test.exs @@ -1,7 +1,48 @@ defmodule Microwaveprop.Workers.Era5FetchWorkerTest do use Microwaveprop.DataCase + use Oban.Testing, repo: Microwaveprop.Repo + alias Microwaveprop.Weather.Era5Profile alias Microwaveprop.Workers.Era5FetchWorker + alias Microwaveprop.Workers.Era5MonthBatchWorker + + describe "perform/1" do + test "enqueues a month-tile batch worker when no profile is cached" do + args = %{"lat" => 32.75, "lon" => -97.25, "valid_time" => "2014-01-15T12:00:00Z"} + + Oban.Testing.with_testing_mode(:manual, fn -> + assert :ok = Era5FetchWorker.perform(%Oban.Job{args: args}) + + jobs = all_enqueued(worker: Era5MonthBatchWorker) + assert [job] = jobs + assert job.args["year"] == 2014 + assert job.args["month"] == 1 + # 32.75 → tile 32 (floor to even degree), -97.25 → tile -98 + assert job.args["tile_lat"] == 32 + assert job.args["tile_lon"] == -98 + end) + end + + test "is a no-op when the profile is already cached" do + valid_time = ~U[2014-01-15 12:00:00Z] + + %Era5Profile{} + |> Era5Profile.changeset(%{ + lat: 32.75, + lon: -97.25, + valid_time: valid_time, + profile: [] + }) + |> Repo.insert!() + + args = %{"lat" => 32.75, "lon" => -97.25, "valid_time" => DateTime.to_iso8601(valid_time)} + + Oban.Testing.with_testing_mode(:manual, fn -> + assert :ok = Era5FetchWorker.perform(%Oban.Job{args: args}) + assert [] = all_enqueued(worker: Era5MonthBatchWorker) + end) + end + end describe "unique constraint" do test "deduplicates identical (lat, lon, valid_time) jobs across inserts" do diff --git a/test/microwaveprop/workers/era5_month_batch_worker_test.exs b/test/microwaveprop/workers/era5_month_batch_worker_test.exs new file mode 100644 index 00000000..a36ba94a --- /dev/null +++ b/test/microwaveprop/workers/era5_month_batch_worker_test.exs @@ -0,0 +1,43 @@ +defmodule Microwaveprop.Workers.Era5MonthBatchWorkerTest do + use Microwaveprop.DataCase + + alias Microwaveprop.Workers.Era5MonthBatchWorker + + describe "unique constraint" do + test "collapses identical (year, month, tile_lat, tile_lon) jobs" do + args = %{year: 2014, month: 3, tile_lat: 32, tile_lon: -98} + + Oban.Testing.with_testing_mode(:manual, fn -> + {:ok, job1} = Oban.insert(Era5MonthBatchWorker.new(args)) + {:ok, job2} = Oban.insert(Era5MonthBatchWorker.new(args)) + + assert job1.id == job2.id + assert Repo.aggregate(Oban.Job, :count, :id) == 1 + end) + end + + test "different months for the same tile are distinct jobs" do + Oban.Testing.with_testing_mode(:manual, fn -> + {:ok, j1} = + Oban.insert(Era5MonthBatchWorker.new(%{year: 2014, month: 3, tile_lat: 32, tile_lon: -98})) + + {:ok, j2} = + Oban.insert(Era5MonthBatchWorker.new(%{year: 2014, month: 4, tile_lat: 32, tile_lon: -98})) + + assert j1.id != j2.id + end) + end + + test "different tiles in the same month are distinct jobs" do + Oban.Testing.with_testing_mode(:manual, fn -> + {:ok, j1} = + Oban.insert(Era5MonthBatchWorker.new(%{year: 2014, month: 3, tile_lat: 32, tile_lon: -98})) + + {:ok, j2} = + Oban.insert(Era5MonthBatchWorker.new(%{year: 2014, month: 3, tile_lat: 30, tile_lon: -98})) + + assert j1.id != j2.id + end) + end + end +end