Batch ERA5 fetches by month and 2° tile
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.
This commit is contained in:
parent
4ec8aa568f
commit
8637253fda
9 changed files with 744 additions and 60 deletions
298
lib/microwaveprop/weather/era5_batch_client.ex
Normal file
298
lib/microwaveprop/weather/era5_batch_client.ex
Normal file
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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(<<y::binary-4, m::binary-2, d::binary-2, h::binary-2, rest::binary>>) do
|
||||
{mi, s} =
|
||||
case rest do
|
||||
<<mm::binary-2, ss::binary-2>> -> {mm, ss}
|
||||
<<mm::binary-2>> -> {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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
64
lib/microwaveprop/workers/era5_month_batch_worker.ex
Normal file
64
lib/microwaveprop/workers/era5_month_batch_worker.ex
Normal file
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
|
|||
78
test/microwaveprop/weather/era5_batch_client_test.exs
Normal file
78
test/microwaveprop/weather/era5_batch_client_test.exs
Normal file
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
|
|||
43
test/microwaveprop/workers/era5_month_batch_worker_test.exs
Normal file
43
test/microwaveprop/workers/era5_month_batch_worker_test.exs
Normal file
|
|
@ -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
|
||||
Loading…
Add table
Reference in a new issue