144 lines
4.7 KiB
Elixir
144 lines
4.7 KiB
Elixir
defmodule Microwaveprop.Workers.HrrrFetchWorker do
|
|
@moduledoc false
|
|
use Oban.Worker, queue: :hrrr, max_attempts: 20, unique: [period: 300, states: [:available, :scheduled, :executing, :retryable]]
|
|
|
|
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} ->
|
|
profiles =
|
|
Enum.map(grid_data, fn {{lat, lon}, data} ->
|
|
build_profile_attrs(lat, lon, valid_time, data)
|
|
end)
|
|
|
|
Weather.upsert_hrrr_profiles_batch(profiles)
|
|
|
|
Enum.each(grid_data, fn {{lat, lon}, _data} ->
|
|
broadcast_enrichment(lat, lon, valid_time)
|
|
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)
|
|
broadcast_enrichment(lat, lon, valid_time)
|
|
:ok
|
|
|
|
{:error, reason} ->
|
|
handle_error(reason, "#{lat},#{lon} @ #{valid_time_str}")
|
|
end
|
|
end
|
|
end
|
|
|
|
defp store_profile(lat, lon, valid_time, data) do
|
|
attrs = build_profile_attrs(lat, lon, valid_time, data)
|
|
Weather.upsert_hrrr_profile(attrs)
|
|
end
|
|
|
|
defp build_profile_attrs(lat, lon, valid_time, data) do
|
|
params = SoundingParams.derive(data.profile)
|
|
|
|
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
|
|
)
|
|
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
|
|
|
|
defp broadcast_enrichment(lat, lon, valid_time) do
|
|
Phoenix.PubSub.broadcast(
|
|
Microwaveprop.PubSub,
|
|
"contact_enrichment:hrrr",
|
|
{:hrrr_ready, %{lat: lat, lon: lon, valid_time: valid_time}}
|
|
)
|
|
end
|
|
end
|