Graham McIntire
581955bd69
Some checks failed
Build and Push / Build and Push Docker Image (push) Has been cancelled
Create shared ContactsFixtures module with globally-unique qso_timestamp seconds to prevent unique-constraint violations when async test modules run in parallel and insert contacts with identical dedup-key columns. The qso_timestamp column is timestamp(0), so microsecond offsets were truncated — use System.unique_integer monotonic seconds instead. Also make count-asserting tests resilient to sandbox-leaked contacts from prior tests by using >= assertions or status-based checks rather than exact counts. Includes automated DateTime.add → DateTime.shift migration from mix format.
99 lines
2.9 KiB
Elixir
99 lines
2.9 KiB
Elixir
defmodule Microwaveprop.Workers.NexradWorker do
|
|
@moduledoc """
|
|
Fetches one n0q NEXRAD composite frame and stores per-point
|
|
observations in `nexrad_observations`.
|
|
|
|
Jobs are unique on `{year, month, day, hour, minute}` so backfill
|
|
sweeps that enqueue duplicate timestamps collapse automatically.
|
|
"""
|
|
|
|
use Oban.Pro.Worker,
|
|
queue: :nexrad,
|
|
max_attempts: 3,
|
|
unique: [
|
|
period: :infinity,
|
|
states: :incomplete,
|
|
keys: [:year, :month, :day, :hour, :minute]
|
|
]
|
|
|
|
import Ecto.Query
|
|
|
|
alias Microwaveprop.Radio.Contact
|
|
alias Microwaveprop.Repo
|
|
alias Microwaveprop.Weather.NexradClient
|
|
alias Microwaveprop.Weather.NexradObservation
|
|
|
|
require Logger
|
|
|
|
@impl Oban.Pro.Worker
|
|
def process(%Oban.Job{args: args}) do
|
|
%{"year" => year, "month" => month, "day" => day, "hour" => hour} = args
|
|
minute = Map.get(args, "minute", 0)
|
|
{:ok, date} = Date.new(year, month, day)
|
|
{:ok, time} = Time.new(hour, minute, 0)
|
|
{:ok, timestamp} = DateTime.new(date, time, "Etc/UTC")
|
|
|
|
points = points_of_interest_for_time(timestamp)
|
|
|
|
if points == [] do
|
|
Logger.info("NexradWorker: no points for #{timestamp}, skipping")
|
|
:ok
|
|
else
|
|
fetch_and_upsert(timestamp, points)
|
|
end
|
|
end
|
|
|
|
@doc false
|
|
@spec points_of_interest_for_time(DateTime.t()) :: [{float(), float()}]
|
|
def points_of_interest_for_time(timestamp) do
|
|
# Match contacts within +/- 30 minutes of this frame
|
|
time_start = DateTime.shift(timestamp, minute: -30)
|
|
time_end = DateTime.shift(timestamp, minute: 30)
|
|
|
|
Contact
|
|
|> where([c], not is_nil(c.pos1))
|
|
|> where([c], c.qso_timestamp >= ^time_start and c.qso_timestamp <= ^time_end)
|
|
|> select([c], c.pos1)
|
|
|> Repo.all()
|
|
|> Enum.flat_map(fn pos ->
|
|
case {pos["lat"], pos["lon"]} do
|
|
{lat, lon} when is_number(lat) and is_number(lon) -> [{snap(lat), snap(lon)}]
|
|
_ -> []
|
|
end
|
|
end)
|
|
|> Enum.uniq()
|
|
end
|
|
|
|
defp snap(x), do: Float.round(x * 1.0, 3)
|
|
|
|
defp fetch_and_upsert(timestamp, points) do
|
|
case NexradClient.fetch_frame(timestamp, points) do
|
|
{:ok, observations} ->
|
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
|
|
|
rows =
|
|
Enum.map(observations, fn obs ->
|
|
obs
|
|
|> Map.put(:id, Ecto.UUID.generate())
|
|
|> Map.put(:inserted_at, now)
|
|
|> Map.put(:updated_at, now)
|
|
end)
|
|
|
|
{inserted, _} =
|
|
Repo.insert_all(
|
|
NexradObservation,
|
|
rows,
|
|
on_conflict:
|
|
{:replace, ~w(mean_reflectivity_dbz max_reflectivity_dbz texture_variance pixel_count updated_at)a},
|
|
conflict_target: [:lat, :lon, :observed_at]
|
|
)
|
|
|
|
Logger.info("NexradWorker: upserted #{inserted} observations for #{timestamp}")
|
|
:ok
|
|
|
|
{:error, reason} ->
|
|
Logger.warning("NexradWorker failed for #{timestamp}: #{inspect(reason)}")
|
|
{:error, reason}
|
|
end
|
|
end
|
|
end
|