79 lines
2.4 KiB
Elixir
79 lines
2.4 KiB
Elixir
defmodule Microwaveprop.Workers.PropagationGridWorker do
|
|
@moduledoc """
|
|
Hourly Oban worker that downloads the latest HRRR data and computes
|
|
propagation scores across the CONUS grid for all bands.
|
|
"""
|
|
|
|
use Oban.Worker,
|
|
queue: :propagation,
|
|
max_attempts: 3
|
|
|
|
alias Microwaveprop.Propagation
|
|
alias Microwaveprop.Propagation.Grid
|
|
alias Microwaveprop.Weather.HrrrClient
|
|
|
|
require Logger
|
|
|
|
# Pause these queues while the grid fetch runs so they don't compete for bandwidth
|
|
@pause_queues [:hrrr, :weather, :iemre, :terrain]
|
|
|
|
@impl Oban.Worker
|
|
def perform(%Oban.Job{}) do
|
|
# HRRR takes ~45min to publish after the hour. Use 2 hours ago to ensure availability.
|
|
two_hours_ago = DateTime.add(DateTime.utc_now(), -2, :hour)
|
|
valid_time = two_hours_ago |> HrrrClient.nearest_hrrr_hour() |> DateTime.truncate(:second)
|
|
points = Grid.conus_points()
|
|
|
|
Logger.info("PropagationGrid: pausing backfill queues, fetching HRRR for #{valid_time}, #{length(points)} points")
|
|
Enum.each(@pause_queues, &Oban.pause_queue(queue: &1))
|
|
|
|
result =
|
|
with {:ok, grid_data} <- HrrrClient.fetch_grid(points, valid_time) do
|
|
scores = compute_scores(grid_data, valid_time)
|
|
|
|
Logger.info("PropagationGrid: computed #{length(scores)} scores, upserting")
|
|
|
|
case Propagation.upsert_scores(scores) do
|
|
{:ok, count} ->
|
|
Logger.info("PropagationGrid: upserted #{count} scores for #{valid_time}")
|
|
Phoenix.PubSub.broadcast(Microwaveprop.PubSub, "propagation:updated", {:propagation_updated, valid_time})
|
|
:ok
|
|
|
|
error ->
|
|
Logger.error("PropagationGrid: upsert failed: #{inspect(error)}")
|
|
error
|
|
end
|
|
end
|
|
|
|
Logger.info("PropagationGrid: resuming backfill queues")
|
|
Enum.each(@pause_queues, &Oban.resume_queue(queue: &1))
|
|
|
|
result
|
|
end
|
|
|
|
defp compute_scores(grid_data, valid_time) do
|
|
grid_data
|
|
|> Task.async_stream(
|
|
fn {{lat, lon}, profile} ->
|
|
band_scores = Propagation.score_grid_point(profile, valid_time)
|
|
|
|
Enum.map(band_scores, fn r ->
|
|
%{
|
|
lat: lat,
|
|
lon: lon,
|
|
valid_time: valid_time,
|
|
band_mhz: r.band_mhz,
|
|
score: r.score,
|
|
factors: r.factors
|
|
}
|
|
end)
|
|
end,
|
|
max_concurrency: System.schedulers_online() * 2,
|
|
timeout: 30_000
|
|
)
|
|
|> Enum.flat_map(fn
|
|
{:ok, results} -> results
|
|
{:exit, _reason} -> []
|
|
end)
|
|
end
|
|
end
|