Graham McIntire
a283ad9c66
QSO enrichment now groups all path points by HRRR hour and creates one batch job per hour instead of one job per point. The batch job downloads the GRIB2 data once and extracts all needed points from the same binary. Legacy single-point jobs are still supported for backward compatibility.
126 lines
4.1 KiB
Elixir
126 lines
4.1 KiB
Elixir
defmodule Microwaveprop.Workers.HrrrFetchWorker do
|
|
@moduledoc false
|
|
use Oban.Worker, queue: :hrrr, max_attempts: 20
|
|
|
|
alias Microwaveprop.Weather
|
|
alias Microwaveprop.Weather.HrrrClient
|
|
alias Microwaveprop.Weather.SoundingParams
|
|
|
|
require Logger
|
|
|
|
@impl Oban.Worker
|
|
def backoff(%Oban.Job{attempt: attempt}) do
|
|
# Exponential backoff: 2m, 4m, 8m, 16m, 32m, 1h, 2h, 4h, 6h, 6h, ...
|
|
min(120 * Integer.pow(2, attempt - 1), _six_hours = 21_600)
|
|
end
|
|
|
|
@impl Oban.Worker
|
|
def perform(%Oban.Job{args: %{"points" => points, "valid_time" => valid_time_str}}) do
|
|
# Batch mode: download GRIB2 once, extract multiple points
|
|
{:ok, valid_time, _} = DateTime.from_iso8601(valid_time_str)
|
|
|
|
point_tuples =
|
|
points
|
|
|> Enum.map(fn %{"lat" => lat, "lon" => lon} -> {lat, lon} end)
|
|
|> Enum.reject(fn {lat, lon} -> Weather.has_hrrr_profile?(lat, lon, valid_time) end)
|
|
|
|
if point_tuples == [] do
|
|
Logger.info("HRRR batch: all #{length(points)} points already exist for #{valid_time_str}")
|
|
:ok
|
|
else
|
|
Logger.info("HRRR batch: fetching #{length(point_tuples)} points for #{valid_time_str}")
|
|
|
|
case HrrrClient.fetch_grid(point_tuples, valid_time) do
|
|
{:ok, grid_data} ->
|
|
Enum.each(grid_data, fn {{lat, lon}, data} ->
|
|
store_profile(lat, lon, valid_time, data)
|
|
end)
|
|
|
|
Logger.info("HRRR batch: saved #{map_size(grid_data)} profiles for #{valid_time_str}")
|
|
:ok
|
|
|
|
{:error, reason} ->
|
|
handle_error(reason, "batch @ #{valid_time_str}")
|
|
end
|
|
end
|
|
end
|
|
|
|
def perform(%Oban.Job{args: %{"lat" => raw_lat, "lon" => raw_lon, "valid_time" => valid_time_str}}) do
|
|
# Legacy single-point mode
|
|
{:ok, valid_time, _} = DateTime.from_iso8601(valid_time_str)
|
|
{lat, lon} = Weather.round_to_hrrr_grid(raw_lat, raw_lon)
|
|
|
|
if Weather.has_hrrr_profile?(lat, lon, valid_time) do
|
|
Logger.info("HRRR profile already exists for #{lat},#{lon} @ #{valid_time_str}")
|
|
:ok
|
|
else
|
|
Logger.info("Fetching HRRR profile for #{lat},#{lon} @ #{valid_time_str}")
|
|
|
|
case HrrrClient.fetch_profile(lat, lon, valid_time) do
|
|
{:ok, data} ->
|
|
store_profile(lat, lon, valid_time, data)
|
|
:ok
|
|
|
|
{:error, reason} ->
|
|
handle_error(reason, "#{lat},#{lon} @ #{valid_time_str}")
|
|
end
|
|
end
|
|
end
|
|
|
|
defp store_profile(lat, lon, valid_time, data) do
|
|
params = SoundingParams.derive(data.profile)
|
|
|
|
attrs =
|
|
maybe_add_derived_params(
|
|
%{
|
|
valid_time: valid_time,
|
|
lat: lat,
|
|
lon: lon,
|
|
run_time: data.run_time,
|
|
profile: data.profile,
|
|
hpbl_m: data.hpbl_m,
|
|
pwat_mm: data.pwat_mm,
|
|
surface_temp_c: data.surface_temp_c,
|
|
surface_dewpoint_c: data.surface_dewpoint_c,
|
|
surface_pressure_mb: data.surface_pressure_mb
|
|
},
|
|
params
|
|
)
|
|
|
|
Weather.upsert_hrrr_profile(attrs)
|
|
end
|
|
|
|
defp handle_error(reason, label) do
|
|
if transient_failure?(reason) do
|
|
Logger.error("HRRR transient error for #{label}: #{inspect(reason)}")
|
|
{:error, reason}
|
|
else
|
|
Logger.warning("HRRR permanent failure for #{label}: #{inspect(reason)}")
|
|
{:cancel, reason}
|
|
end
|
|
end
|
|
|
|
# Only retry on transient network/server errors — everything else is permanent
|
|
defp transient_failure?(%{__exception__: true}), do: true
|
|
defp transient_failure?("HRRR idx HTTP " <> status), do: server_error?(status)
|
|
defp transient_failure?("HRRR grib HTTP " <> status), do: server_error?(status)
|
|
defp transient_failure?(_), do: false
|
|
|
|
defp server_error?(status) do
|
|
case Integer.parse(status) do
|
|
{code, _} when code in [429, 500, 502, 503, 504] -> true
|
|
_ -> false
|
|
end
|
|
end
|
|
|
|
defp maybe_add_derived_params(attrs, nil), do: attrs
|
|
|
|
defp maybe_add_derived_params(attrs, params) do
|
|
Map.merge(attrs, %{
|
|
surface_refractivity: params.surface_refractivity,
|
|
min_refractivity_gradient: params.min_refractivity_gradient,
|
|
ducting_detected: params.ducting_detected,
|
|
duct_characteristics: params.duct_characteristics
|
|
})
|
|
end
|
|
end
|