Graham McIntire
6476bc8045
The Era5FetchWorker now declares an Oban unique constraint on
{lat, lon, valid_time} for any job in available/scheduled/executing/
retryable, so two backfill runs targeting the same contact grid point
can no longer spawn parallel CDS requests for the same hour.
Because Oban OSS insert_all doesn't honor unique, ERA5 jobs are now
routed through Oban.insert/1 from ContactWeatherEnqueueWorker and the
era5_backfill mix task. Other worker types still use insert_all.
80 lines
2.5 KiB
Elixir
80 lines
2.5 KiB
Elixir
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.
|
|
|
|
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.
|
|
"""
|
|
use Oban.Worker,
|
|
queue: :era5,
|
|
max_attempts: 5,
|
|
unique: [
|
|
period: :infinity,
|
|
states: [:available, :scheduled, :executing, :retryable]
|
|
]
|
|
|
|
alias Microwaveprop.Repo
|
|
alias Microwaveprop.Weather.Era5Client
|
|
alias Microwaveprop.Weather.Era5Profile
|
|
|
|
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)
|
|
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
|
|
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}")
|
|
|
|
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: stored profile for #{rlat},#{rlon} @ #{valid_time_str}")
|
|
:ok
|
|
|
|
{:error, reason} ->
|
|
Logger.warning("ERA5: failed for #{rlat},#{rlon} @ #{valid_time_str}: #{inspect(reason)}")
|
|
{:error, reason}
|
|
end
|
|
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
|