145 lines
5 KiB
Elixir
145 lines
5 KiB
Elixir
defmodule Microwaveprop.Workers.WeatherFetchWorker do
|
|
@moduledoc false
|
|
use Oban.Worker,
|
|
queue: :weather,
|
|
max_attempts: 20,
|
|
unique: [period: 300, states: [:available, :scheduled, :executing, :retryable]]
|
|
|
|
require Logger
|
|
|
|
alias Microwaveprop.Repo
|
|
alias Microwaveprop.Weather
|
|
alias Microwaveprop.Weather.IemClient
|
|
alias Microwaveprop.Weather.SoundingParams
|
|
alias Microwaveprop.Weather.Station
|
|
|
|
@impl Oban.Worker
|
|
def backoff(%Oban.Job{attempt: attempt}) do
|
|
min(120 * Integer.pow(2, attempt - 1), _six_hours = 21_600)
|
|
end
|
|
|
|
@impl Oban.Worker
|
|
def perform(%Oban.Job{args: %{"fetch_type" => "asos"} = args}) do
|
|
%{
|
|
"station_id" => station_id,
|
|
"station_code" => station_code,
|
|
"start_dt" => start_dt_str,
|
|
"end_dt" => end_dt_str
|
|
} = args
|
|
|
|
{:ok, start_dt, _} = DateTime.from_iso8601(start_dt_str)
|
|
{:ok, end_dt, _} = DateTime.from_iso8601(end_dt_str)
|
|
|
|
case Repo.get(Station, station_id) do
|
|
nil ->
|
|
Logger.warning("WeatherFetch ASOS: station #{station_id} not found, skipping")
|
|
:ok
|
|
|
|
station ->
|
|
if Weather.has_surface_observations?(station.id, start_dt, end_dt) do
|
|
Logger.debug("WeatherFetch ASOS: #{station_code} already has obs for #{start_dt_str}..#{end_dt_str}")
|
|
:ok
|
|
else
|
|
Logger.info("WeatherFetch ASOS: fetching #{station_code} for #{start_dt_str}..#{end_dt_str}")
|
|
|
|
case IemClient.fetch_asos(station_code, start_dt, end_dt) do
|
|
{:ok, rows} ->
|
|
count = Enum.count(rows, & &1.observed_at)
|
|
|
|
Enum.each(rows, fn row ->
|
|
if row.observed_at, do: Weather.upsert_surface_observation(station, row)
|
|
end)
|
|
|
|
Logger.info("WeatherFetch ASOS: #{station_code} ingested #{count} observations")
|
|
|
|
Phoenix.PubSub.broadcast(
|
|
Microwaveprop.PubSub,
|
|
"contact_enrichment:weather",
|
|
{:weather_ready, %{station_id: station_id}}
|
|
)
|
|
|
|
:ok
|
|
|
|
{:error, reason} ->
|
|
Logger.warning("WeatherFetch ASOS: #{station_code} failed: #{inspect(reason)}")
|
|
{:error, reason}
|
|
end
|
|
end
|
|
end
|
|
end
|
|
|
|
def perform(%Oban.Job{args: %{"fetch_type" => "raob"} = args}) do
|
|
%{
|
|
"station_id" => station_id,
|
|
"station_code" => station_code,
|
|
"sounding_time" => sounding_time_str
|
|
} = args
|
|
|
|
{:ok, sounding_time, _} = DateTime.from_iso8601(sounding_time_str)
|
|
|
|
case Repo.get(Station, station_id) do
|
|
nil ->
|
|
Logger.warning("WeatherFetch RAOB: station #{station_id} not found, skipping")
|
|
:ok
|
|
|
|
station ->
|
|
if Weather.has_sounding?(station.id, sounding_time) do
|
|
Logger.debug("WeatherFetch RAOB: #{station_code} already has sounding for #{sounding_time_str}")
|
|
:ok
|
|
else
|
|
Logger.info("WeatherFetch RAOB: fetching #{station_code} for #{sounding_time_str}")
|
|
|
|
case IemClient.fetch_raob(station_code, sounding_time) do
|
|
{:ok, [parsed | _]} ->
|
|
params = SoundingParams.derive(parsed.profile)
|
|
|
|
sounding_attrs =
|
|
if params do
|
|
%{
|
|
observed_at: parsed.observed_at,
|
|
profile: parsed.profile,
|
|
level_count: params.level_count,
|
|
surface_pressure_mb: params.surface_pressure_mb,
|
|
surface_temp_c: params.surface_temp_c,
|
|
surface_dewpoint_c: params.surface_dewpoint_c,
|
|
surface_refractivity: params.surface_refractivity,
|
|
min_refractivity_gradient: params.min_refractivity_gradient,
|
|
boundary_layer_depth_m: params.boundary_layer_depth_m,
|
|
precipitable_water_mm: params.precipitable_water_mm,
|
|
k_index: params.k_index,
|
|
lifted_index: params.lifted_index,
|
|
ducting_detected: params.ducting_detected,
|
|
duct_characteristics: params.duct_characteristics
|
|
}
|
|
else
|
|
%{
|
|
observed_at: parsed.observed_at,
|
|
profile: parsed.profile,
|
|
level_count: length(parsed.profile)
|
|
}
|
|
end
|
|
|
|
Weather.upsert_sounding(station, sounding_attrs)
|
|
|
|
Logger.info("WeatherFetch RAOB: #{station_code} ingested sounding with #{length(parsed.profile)} levels")
|
|
|
|
Phoenix.PubSub.broadcast(
|
|
Microwaveprop.PubSub,
|
|
"contact_enrichment:weather",
|
|
{:weather_ready, %{station_id: station_id}}
|
|
)
|
|
|
|
:ok
|
|
|
|
{:ok, []} ->
|
|
Logger.info("WeatherFetch RAOB: #{station_code} returned no data for #{sounding_time_str}")
|
|
:ok
|
|
|
|
{:error, reason} ->
|
|
Logger.warning("WeatherFetch RAOB: #{station_code} failed: #{inspect(reason)}")
|
|
{:error, reason}
|
|
end
|
|
end
|
|
end
|
|
end
|
|
end
|