diff --git a/config/config.exs b/config/config.exs index bc9535c3..4ebe929a 100644 --- a/config/config.exs +++ b/config/config.exs @@ -44,12 +44,13 @@ config :microwaveprop, MicrowavepropWeb.Endpoint, config :microwaveprop, Oban, repo: Microwaveprop.Repo, - queues: [solar: 1, weather: 3], + queues: [solar: 1, weather: 3, enqueue: 1], plugins: [ {Oban.Plugins.Pruner, max_age: 3600 * 24}, {Oban.Plugins.Cron, crontab: [ - {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker} + {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker}, + {"0 */4 * * *", Microwaveprop.Workers.QsoWeatherEnqueueWorker} ]} ] diff --git a/lib/microwaveprop/application.ex b/lib/microwaveprop/application.ex index fcf21a92..c1c8902a 100644 --- a/lib/microwaveprop/application.ex +++ b/lib/microwaveprop/application.ex @@ -7,6 +7,8 @@ defmodule Microwaveprop.Application do @impl true def start(_type, _args) do + migrate() + children = [ MicrowavepropWeb.Telemetry, Microwaveprop.Repo, @@ -25,6 +27,10 @@ defmodule Microwaveprop.Application do # Tell Phoenix to update the endpoint configuration # whenever the application is updated. + defp migrate do + Microwaveprop.Release.migrate() + end + @impl true def config_change(changed, _new, removed) do MicrowavepropWeb.Endpoint.config_change(changed, removed) diff --git a/lib/microwaveprop/radio.ex b/lib/microwaveprop/radio.ex index 2706ef47..5d3cd947 100644 --- a/lib/microwaveprop/radio.ex +++ b/lib/microwaveprop/radio.ex @@ -43,6 +43,20 @@ defmodule Microwaveprop.Radio do {field, dir} end + def unprocessed_qsos(limit \\ 500) do + Qso + |> where([q], q.weather_queued == false and not is_nil(q.pos1)) + |> order_by([q], asc: q.qso_timestamp) + |> limit(^limit) + |> Repo.all() + end + + def mark_weather_queued!(qso_ids) do + Qso + |> where([q], q.id in ^qso_ids) + |> Repo.update_all(set: [weather_queued: true]) + end + def get_qso!(id) do Repo.get!(Qso, id) end diff --git a/lib/microwaveprop/radio/qso.ex b/lib/microwaveprop/radio/qso.ex index 67b1cb35..06f5b3f1 100644 --- a/lib/microwaveprop/radio/qso.ex +++ b/lib/microwaveprop/radio/qso.ex @@ -18,6 +18,7 @@ defmodule Microwaveprop.Radio.Qso do field :mode, :string field :band, :decimal field :distance_km, :decimal + field :weather_queued, :boolean, default: false timestamps(type: :utc_datetime) end diff --git a/lib/microwaveprop/weather.ex b/lib/microwaveprop/weather.ex index 6169ef3f..61fab3f4 100644 --- a/lib/microwaveprop/weather.ex +++ b/lib/microwaveprop/weather.ex @@ -91,6 +91,39 @@ defmodule Microwaveprop.Weather do |> MapSet.new() end + def nearby_stations(lat, lon, station_type, radius_km) do + dlat = radius_km / @km_per_deg_lat + dlon = radius_km / (@km_per_deg_lat * :math.cos(lat * :math.pi() / 180)) + + Station + |> where([s], s.station_type == ^station_type) + |> where( + [s], + s.lat >= ^(lat - dlat) and s.lat <= ^(lat + dlat) and + s.lon >= ^(lon - dlon) and s.lon <= ^(lon + dlon) + ) + |> Repo.all() + end + + def sounding_times_around(dt) do + date = DateTime.to_date(dt) + + times = + if dt.hour < 12 do + [ + DateTime.new!(Date.add(date, -1), ~T[12:00:00], "Etc/UTC"), + DateTime.new!(date, ~T[00:00:00], "Etc/UTC") + ] + else + [ + DateTime.new!(date, ~T[00:00:00], "Etc/UTC"), + DateTime.new!(date, ~T[12:00:00], "Etc/UTC") + ] + end + + Enum.uniq(times) + end + def weather_for_qso(qso_params, opts \\ []) do lat = qso_params[:lat] || qso_params.lat lon = qso_params[:lon] || qso_params.lon diff --git a/lib/microwaveprop/workers/qso_weather_enqueue_worker.ex b/lib/microwaveprop/workers/qso_weather_enqueue_worker.ex new file mode 100644 index 00000000..ffecc3dd --- /dev/null +++ b/lib/microwaveprop/workers/qso_weather_enqueue_worker.ex @@ -0,0 +1,77 @@ +defmodule Microwaveprop.Workers.QsoWeatherEnqueueWorker do + @moduledoc false + use Oban.Worker, queue: :enqueue, max_attempts: 3 + + alias Microwaveprop.Radio + alias Microwaveprop.Weather + alias Microwaveprop.Workers.WeatherFetchWorker + + @asos_radius_km 150 + @sounding_radius_km 300 + + @impl Oban.Worker + def perform(%Oban.Job{}) do + qsos = Radio.unprocessed_qsos() + + if qsos != [] do + jobs = build_weather_jobs(qsos) + + if jobs != [] do + Oban.insert_all(jobs) + end + + qso_ids = Enum.map(qsos, & &1.id) + Radio.mark_weather_queued!(qso_ids) + end + + :ok + end + + def build_weather_jobs(qsos) do + qsos + |> Enum.flat_map(&jobs_for_qso/1) + |> Enum.uniq_by(fn changeset -> changeset.changes.args end) + end + + defp jobs_for_qso(qso) do + lat = qso.pos1["lat"] + lon = qso.pos1["lon"] || qso.pos1["lng"] + + asos_jobs = build_asos_jobs(lat, lon, qso.qso_timestamp) + raob_jobs = build_raob_jobs(lat, lon, qso.qso_timestamp) + + asos_jobs ++ raob_jobs + end + + defp build_asos_jobs(lat, lon, timestamp) do + start_dt = DateTime.add(timestamp, -2 * 3600, :second) + end_dt = DateTime.add(timestamp, 2 * 3600, :second) + + lat + |> Weather.nearby_stations(lon, "asos", @asos_radius_km) + |> Enum.map(fn station -> + WeatherFetchWorker.new(%{ + "fetch_type" => "asos", + "station_id" => station.id, + "station_code" => station.station_code, + "start_dt" => DateTime.to_iso8601(start_dt), + "end_dt" => DateTime.to_iso8601(end_dt) + }) + end) + end + + defp build_raob_jobs(lat, lon, timestamp) do + sounding_times = Weather.sounding_times_around(timestamp) + + stations = Weather.nearby_stations(lat, lon, "sounding", @sounding_radius_km) + + for station <- stations, sounding_time <- sounding_times do + WeatherFetchWorker.new(%{ + "fetch_type" => "raob", + "station_id" => station.id, + "station_code" => station.station_code, + "sounding_time" => DateTime.to_iso8601(sounding_time) + }) + end + end +end diff --git a/priv/repo/migrations/20260329190247_add_weather_queued_to_qsos.exs b/priv/repo/migrations/20260329190247_add_weather_queued_to_qsos.exs new file mode 100644 index 00000000..0a771cf6 --- /dev/null +++ b/priv/repo/migrations/20260329190247_add_weather_queued_to_qsos.exs @@ -0,0 +1,15 @@ +defmodule Microwaveprop.Repo.Migrations.AddWeatherQueuedToQsos do + use Ecto.Migration + + def change do + alter table(:qsos) do + add :weather_queued, :boolean, default: false, null: false + end + + # Backfill existing rows — they were already processed by the import script + execute "UPDATE qsos SET weather_queued = true", "SELECT 1" + + # Partial index for fast lookup of unprocessed QSOs + create index(:qsos, [:weather_queued], where: "weather_queued = false") + end +end diff --git a/test/microwaveprop/radio_test.exs b/test/microwaveprop/radio_test.exs index b4749294..d66baefd 100644 --- a/test/microwaveprop/radio_test.exs +++ b/test/microwaveprop/radio_test.exs @@ -100,6 +100,68 @@ defmodule Microwaveprop.RadioTest do end end + describe "unprocessed_qsos/1" do + test "returns QSOs where weather_queued is false and pos1 is not nil" do + q1 = create_qso(%{station1: "W5XD", qso_timestamp: ~U[2026-03-28 12:00:00Z]}) + _q2 = create_qso(%{station1: "K5TR", pos1: nil, qso_timestamp: ~U[2026-03-28 13:00:00Z]}) + + results = Radio.unprocessed_qsos() + ids = Enum.map(results, & &1.id) + assert q1.id in ids + end + + test "excludes QSOs already marked as weather_queued" do + q = create_qso() + Radio.mark_weather_queued!([q.id]) + + assert Radio.unprocessed_qsos() == [] + end + + test "orders by qso_timestamp ascending" do + q_late = create_qso(%{station1: "LATE", qso_timestamp: ~U[2026-03-28 20:00:00Z]}) + q_early = create_qso(%{station1: "EARLY", qso_timestamp: ~U[2026-03-28 10:00:00Z]}) + + results = Radio.unprocessed_qsos() + ids = Enum.map(results, & &1.id) + assert ids == [q_early.id, q_late.id] + end + + test "respects limit parameter" do + for i <- 1..5 do + ts = DateTime.add(~U[2026-01-01 00:00:00Z], i * 3600, :second) + create_qso(%{qso_timestamp: ts}) + end + + assert length(Radio.unprocessed_qsos(3)) == 3 + end + end + + describe "mark_weather_queued!/1" do + test "sets weather_queued to true for given IDs" do + q1 = create_qso(%{station1: "A1A"}) + q2 = create_qso(%{station1: "B2B"}) + + Radio.mark_weather_queued!([q1.id, q2.id]) + + assert Repo.get!(Qso, q1.id).weather_queued == true + assert Repo.get!(Qso, q2.id).weather_queued == true + end + + test "does not affect other QSOs" do + q1 = create_qso(%{station1: "A1A"}) + q2 = create_qso(%{station1: "B2B"}) + + Radio.mark_weather_queued!([q1.id]) + + assert Repo.get!(Qso, q1.id).weather_queued == true + assert Repo.get!(Qso, q2.id).weather_queued == false + end + + test "handles empty list" do + assert Radio.mark_weather_queued!([]) == {0, nil} + end + end + describe "get_qso!/1" do test "returns a QSO by ID" do qso = create_qso() diff --git a/test/microwaveprop/weather_test.exs b/test/microwaveprop/weather_test.exs index 658b6cc4..6c96f613 100644 --- a/test/microwaveprop/weather_test.exs +++ b/test/microwaveprop/weather_test.exs @@ -278,6 +278,97 @@ defmodule Microwaveprop.WeatherTest do end end + describe "nearby_stations/4" do + test "returns stations within radius" do + {:ok, _near} = + Weather.find_or_create_station(%{ + station_code: "KDFW", + station_type: "asos", + name: "DFW Airport", + lat: 32.90, + lon: -97.04 + }) + + {:ok, _far} = + Weather.find_or_create_station(%{ + station_code: "KORD", + station_type: "asos", + name: "Chicago O'Hare", + lat: 41.98, + lon: -87.90 + }) + + stations = Weather.nearby_stations(32.85, -97.05, "asos", 50) + codes = Enum.map(stations, & &1.station_code) + assert "KDFW" in codes + refute "KORD" in codes + end + + test "filters by station type" do + {:ok, _asos} = + Weather.find_or_create_station(%{ + station_code: "KDFW", + station_type: "asos", + name: "DFW Airport", + lat: 32.90, + lon: -97.04 + }) + + {:ok, _sounding} = + Weather.find_or_create_station(%{ + station_code: "FWD", + station_type: "sounding", + name: "Fort Worth", + lat: 32.83, + lon: -97.30 + }) + + stations = Weather.nearby_stations(32.85, -97.10, "sounding", 100) + codes = Enum.map(stations, & &1.station_code) + assert "FWD" in codes + refute "KDFW" in codes + end + + test "returns empty list when no stations in radius" do + {:ok, _} = + Weather.find_or_create_station(%{ + station_code: "KDFW", + station_type: "asos", + name: "DFW Airport", + lat: 32.90, + lon: -97.04 + }) + + assert Weather.nearby_stations(45.0, -80.0, "asos", 50) == [] + end + end + + describe "sounding_times_around/1" do + test "returns bracketing 00Z and 12Z for afternoon UTC" do + dt = ~U[2026-03-28 18:00:00Z] + times = Weather.sounding_times_around(dt) + assert times == [~U[2026-03-28 00:00:00Z], ~U[2026-03-28 12:00:00Z]] + end + + test "returns bracketing times for morning UTC" do + dt = ~U[2026-03-28 06:00:00Z] + times = Weather.sounding_times_around(dt) + assert times == [~U[2026-03-27 12:00:00Z], ~U[2026-03-28 00:00:00Z]] + end + + test "returns single time at exactly 00Z" do + dt = ~U[2026-03-28 00:00:00Z] + times = Weather.sounding_times_around(dt) + assert times == [~U[2026-03-27 12:00:00Z], ~U[2026-03-28 00:00:00Z]] + end + + test "returns single time at exactly 12Z" do + dt = ~U[2026-03-28 12:00:00Z] + times = Weather.sounding_times_around(dt) + assert times == [~U[2026-03-28 00:00:00Z], ~U[2026-03-28 12:00:00Z]] + end + end + describe "get_solar_index/1" do test "returns the solar index record for a given date" do {:ok, record} = Weather.upsert_solar_index(%{date: ~D[2024-06-15], sfi: 150.2, sunspot_number: 120}) diff --git a/test/microwaveprop/workers/qso_weather_enqueue_worker_test.exs b/test/microwaveprop/workers/qso_weather_enqueue_worker_test.exs new file mode 100644 index 00000000..964b8959 --- /dev/null +++ b/test/microwaveprop/workers/qso_weather_enqueue_worker_test.exs @@ -0,0 +1,161 @@ +defmodule Microwaveprop.Workers.QsoWeatherEnqueueWorkerTest do + use Microwaveprop.DataCase, async: true + + alias Microwaveprop.Radio.Qso + alias Microwaveprop.Weather + alias Microwaveprop.Weather.IemClient + alias Microwaveprop.Workers.QsoWeatherEnqueueWorker + + @qso_attrs %{ + station1: "W5XD", + station2: "K5TR", + qso_timestamp: ~U[2026-03-28 18:00:00Z], + mode: "CW", + band: Decimal.new("1296"), + grid1: "EM12", + grid2: "EM00", + pos1: %{"lat" => 32.9, "lon" => -97.0}, + pos2: %{"lat" => 30.3, "lon" => -97.7}, + distance_km: Decimal.new("295") + } + + defp create_qso(attrs \\ %{}) do + {:ok, qso} = + %Qso{} + |> Qso.changeset(Map.merge(@qso_attrs, attrs)) + |> Repo.insert() + + qso + end + + defp create_asos_station(code, lat, lon) do + {:ok, station} = + Weather.find_or_create_station(%{ + station_code: code, + station_type: "asos", + name: "#{code} Station", + lat: lat, + lon: lon + }) + + station + end + + defp create_sounding_station(code, lat, lon) do + {:ok, station} = + Weather.find_or_create_station(%{ + station_code: code, + station_type: "sounding", + name: "#{code} Sounding", + lat: lat, + lon: lon + }) + + station + end + + describe "build_weather_jobs/1" do + test "builds ASOS jobs for nearby stations" do + station = create_asos_station("KDFW", 32.90, -97.04) + qso = create_qso() + + jobs = QsoWeatherEnqueueWorker.build_weather_jobs([qso]) + + asos_jobs = + Enum.filter(jobs, fn j -> + j.changes.args["fetch_type"] == "asos" + end) + + assert length(asos_jobs) >= 1 + + job = hd(asos_jobs) + assert job.changes.args["station_id"] == station.id + assert job.changes.args["station_code"] == "KDFW" + assert job.changes.args["start_dt"] + assert job.changes.args["end_dt"] + end + + test "builds RAOB jobs for nearby sounding stations" do + station = create_sounding_station("FWD", 32.83, -97.30) + qso = create_qso() + + jobs = QsoWeatherEnqueueWorker.build_weather_jobs([qso]) + + raob_jobs = + Enum.filter(jobs, fn j -> + j.changes.args["fetch_type"] == "raob" + end) + + assert length(raob_jobs) >= 1 + + job = hd(raob_jobs) + assert job.changes.args["station_id"] == station.id + assert job.changes.args["station_code"] == "FWD" + assert job.changes.args["sounding_time"] + end + + test "deduplicates jobs with identical args" do + _station = create_asos_station("KDFW", 32.90, -97.04) + + # Two QSOs at the exact same time produce identical args → deduplicated + q1 = create_qso(%{station1: "A1", qso_timestamp: ~U[2026-03-28 18:00:00Z]}) + q2 = create_qso(%{station1: "A2", qso_timestamp: ~U[2026-03-28 18:00:00Z]}) + + jobs = QsoWeatherEnqueueWorker.build_weather_jobs([q1, q2]) + + asos_jobs = + Enum.filter(jobs, fn j -> + j.changes.args["fetch_type"] == "asos" + end) + + assert length(asos_jobs) == 1 + end + + test "returns empty list when no stations nearby" do + qso = create_qso(%{pos1: %{"lat" => 60.0, "lon" => -150.0}}) + + assert QsoWeatherEnqueueWorker.build_weather_jobs([qso]) == [] + end + end + + describe "perform/1" do + setup do + # Stub IEM client so inline Oban doesn't blow up when executing enqueued jobs + Req.Test.stub(IemClient, fn conn -> + case conn.request_path do + "/cgi-bin/request/asos.py" -> Req.Test.text(conn, "#DEBUG,\nstation,valid,tmpf\n") + _ -> Req.Test.json(conn, %{"profiles" => []}) + end + end) + + :ok + end + + test "enqueues weather jobs and marks QSOs as queued" do + _station = create_asos_station("KDFW", 32.90, -97.04) + qso = create_qso() + + assert qso.weather_queued == false + + assert :ok = QsoWeatherEnqueueWorker.perform(%Oban.Job{args: %{}}) + + updated = Repo.get!(Qso, qso.id) + assert updated.weather_queued == true + end + + test "returns :ok when no unprocessed QSOs exist" do + assert :ok = QsoWeatherEnqueueWorker.perform(%Oban.Job{args: %{}}) + end + + test "skips QSOs without positions" do + _station = create_asos_station("KDFW", 32.90, -97.04) + _qso = create_qso(%{pos1: nil}) + + assert :ok = QsoWeatherEnqueueWorker.perform(%Oban.Job{args: %{}}) + + # QSO without pos1 should not be marked as queued + all_qsos = Repo.all(Qso) + refute Enum.any?(all_qsos, & &1.weather_queued) + end + end +end