Graham McIntire
8637253fda
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.
80 lines
2.4 KiB
Elixir
80 lines
2.4 KiB
Elixir
defmodule Microwaveprop.Workers.Era5FetchWorker do
|
|
@moduledoc """
|
|
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}` 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,
|
|
max_attempts: 5,
|
|
unique: [
|
|
period: :infinity,
|
|
states: [:available, :scheduled, :executing, :retryable]
|
|
]
|
|
|
|
alias Microwaveprop.Repo
|
|
alias Microwaveprop.Weather.Era5BatchClient
|
|
alias Microwaveprop.Weather.Era5Profile
|
|
alias Microwaveprop.Workers.Era5MonthBatchWorker
|
|
|
|
require Logger
|
|
|
|
@impl Oban.Worker
|
|
def backoff(%Oban.Job{attempt: attempt}) do
|
|
# 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
|
|
def perform(%Oban.Job{args: %{"lat" => lat, "lon" => lon, "valid_time" => valid_time_str}}) do
|
|
{:ok, valid_time, _} = DateTime.from_iso8601(valid_time_str)
|
|
|
|
rlat = Float.round(lat * 4) / 4
|
|
rlon = Float.round(lon * 4) / 4
|
|
|
|
if has_era5_profile?(rlat, rlon, valid_time) do
|
|
:ok
|
|
else
|
|
{tile_lat, tile_lon} = Era5BatchClient.tile_for_point(rlat, rlon)
|
|
|
|
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})"
|
|
)
|
|
|
|
%{
|
|
year: valid_time.year,
|
|
month: valid_time.month,
|
|
tile_lat: tile_lat,
|
|
tile_lon: tile_lon
|
|
}
|
|
|> Era5MonthBatchWorker.new()
|
|
|> Oban.insert()
|
|
|
|
:ok
|
|
end
|
|
end
|
|
|
|
defp has_era5_profile?(lat, lon, valid_time) do
|
|
import Ecto.Query
|
|
|
|
dlat = 0.13
|
|
dlon = 0.13
|
|
time_start = DateTime.add(valid_time, -1800, :second)
|
|
time_end = DateTime.add(valid_time, 1800, :second)
|
|
|
|
Era5Profile
|
|
|> where(
|
|
[p],
|
|
p.lat >= ^(lat - dlat) and p.lat <= ^(lat + dlat) and
|
|
p.lon >= ^(lon - dlon) and p.lon <= ^(lon + dlon) and
|
|
p.valid_time >= ^time_start and p.valid_time <= ^time_end
|
|
)
|
|
|> Repo.exists?()
|
|
end
|
|
end
|