From 4aa23ae67fcc17860e142d39e8cd4c787538798c Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Wed, 15 Apr 2026 18:49:02 -0500 Subject: [PATCH] ContactWeatherEnqueueWorker: dispatch NarrFetchWorker for historical backfill MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Swap the body of era5_jobs_for_contact/1 to enqueue NarrFetchWorker instead of Era5FetchWorker, and snap qso_timestamp via NarrClient.snap_to_analysis_hour/1 so it lands on a 3-hourly NARR analysis slot (00/03/06/09/12/15/18/21 UTC) — NarrFetchWorker raises on anything else. The :era5 type key, the era5_jobs_for_contact function name, and the era5_status field are kept as historical artifacts — they still refer to the era5_profiles table, which is reused 1:1 by NARR. A schema rename is a follow-up PR. The CDS Era5* workers (Era5FetchWorker, Era5SubmitWorker, Era5PollWorker, Era5MonthBatchWorker, Era5BatchClient) are now unreachable from the live submission flow. They stay in tree until the deletion follow-up PR. --- .../workers/contact_weather_enqueue_worker.ex | 12 ++- .../contact_weather_enqueue_worker_test.exs | 102 ++++++++++++++++++ 2 files changed, 111 insertions(+), 3 deletions(-) diff --git a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex index 09304439..94bf9051 100644 --- a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex +++ b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex @@ -5,9 +5,10 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do alias Microwaveprop.Radio alias Microwaveprop.Weather alias Microwaveprop.Weather.HrrrClient - alias Microwaveprop.Workers.Era5FetchWorker + alias Microwaveprop.Weather.NarrClient alias Microwaveprop.Workers.HrrrFetchWorker alias Microwaveprop.Workers.IemreFetchWorker + alias Microwaveprop.Workers.NarrFetchWorker alias Microwaveprop.Workers.TerrainProfileWorker alias Microwaveprop.Workers.WeatherFetchWorker @@ -178,10 +179,15 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do |> Enum.uniq_by(fn changeset -> changeset.changes.args end) end + # The function name is a historical artifact: the `:era5` type key still + # refers to the `era5_profiles` table (reused 1:1), but this now dispatches + # `NarrFetchWorker` against NCEP NARR — the CDS/ERA5 pipeline is gone. defp era5_jobs_for_contact(%{pos1: nil}), do: [] defp era5_jobs_for_contact(contact) do - valid_time = %{DateTime.truncate(contact.qso_timestamp, :second) | minute: 0, second: 0} + # NARR analyses are 3-hourly (00/03/06/09/12/15/18/21 UTC). NarrFetchWorker + # rejects anything else, so snap here in the caller. + valid_time = NarrClient.snap_to_analysis_hour(contact.qso_timestamp) contact |> Radio.contact_path_points() @@ -193,7 +199,7 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do [] else [ - Era5FetchWorker.new(%{ + NarrFetchWorker.new(%{ "lat" => rlat, "lon" => rlon, "valid_time" => DateTime.to_iso8601(valid_time) diff --git a/test/microwaveprop/workers/contact_weather_enqueue_worker_test.exs b/test/microwaveprop/workers/contact_weather_enqueue_worker_test.exs index d2670757..67467fd2 100644 --- a/test/microwaveprop/workers/contact_weather_enqueue_worker_test.exs +++ b/test/microwaveprop/workers/contact_weather_enqueue_worker_test.exs @@ -4,6 +4,7 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorkerTest do alias Microwaveprop.Radio.Contact alias Microwaveprop.Terrain.ElevationClient alias Microwaveprop.Weather + alias Microwaveprop.Weather.Era5Profile alias Microwaveprop.Weather.HrrrClient alias Microwaveprop.Weather.IemClient alias Microwaveprop.Workers.ContactWeatherEnqueueWorker @@ -561,6 +562,107 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorkerTest do end end + describe "build_era5_jobs/1" do + test "builds NarrFetchWorker jobs for each rounded path point" do + contact = + create_contact(%{ + qso_timestamp: ~U[2010-06-15 13:42:00Z], + pos1: %{"lat" => 32.907, "lon" => -97.038}, + pos2: nil + }) + + jobs = ContactWeatherEnqueueWorker.build_era5_jobs([contact]) + + assert length(jobs) == 1 + job = hd(jobs) + assert job.changes.worker == "Microwaveprop.Workers.NarrFetchWorker" + # 32.907 → 33.0 and -97.038 → -97.0 after 0.25° snap + assert job.changes.args["lat"] == 33.0 + assert job.changes.args["lon"] == -97.0 + end + + test "snaps valid_time to the nearest 3-hourly NARR analysis slot" do + contact = + create_contact(%{ + qso_timestamp: ~U[2010-06-15 13:42:00Z], + pos1: %{"lat" => 32.907, "lon" => -97.038}, + pos2: nil + }) + + jobs = ContactWeatherEnqueueWorker.build_era5_jobs([contact]) + + job = hd(jobs) + # 13:42 UTC snaps down to the 12:00 UTC NARR analysis slot. + assert job.changes.args["valid_time"] == "2010-06-15T12:00:00Z" + end + + test "builds one job per distinct rounded path point" do + contact = + create_contact(%{ + qso_timestamp: ~U[2010-06-15 12:00:00Z], + pos1: %{"lat" => 32.907, "lon" => -97.038}, + pos2: %{"lat" => 30.3, "lon" => -97.7} + }) + + jobs = ContactWeatherEnqueueWorker.build_era5_jobs([contact]) + + # pos1, midpoint, pos2 round to three distinct 0.25° grid cells. + assert length(jobs) == 3 + assert Enum.all?(jobs, &(&1.changes.worker == "Microwaveprop.Workers.NarrFetchWorker")) + end + + test "skips path points that already have an era5_profile row" do + contact = + create_contact(%{ + qso_timestamp: ~U[2010-06-15 13:42:00Z], + pos1: %{"lat" => 32.907, "lon" => -97.038}, + pos2: nil + }) + + # pos1 (32.907, -97.038) snaps to the (33.0, -97.0) 0.25° grid cell + %Era5Profile{} + |> Era5Profile.changeset(%{ + lat: 33.0, + lon: -97.0, + valid_time: ~U[2010-06-15 12:00:00Z], + profile: [] + }) + |> Repo.insert!() + + jobs = ContactWeatherEnqueueWorker.build_era5_jobs([contact]) + assert jobs == [] + end + + test "skips QSOs without pos1" do + contact = create_contact(%{pos1: nil}) + assert ContactWeatherEnqueueWorker.build_era5_jobs([contact]) == [] + end + + test "deduplicates jobs with identical args across contacts" do + q1 = + create_contact(%{ + station1: "A1", + qso_timestamp: ~U[2010-06-15 12:10:00Z], + pos1: %{"lat" => 32.907, "lon" => -97.038}, + pos2: nil + }) + + q2 = + create_contact(%{ + station1: "A2", + qso_timestamp: ~U[2010-06-15 13:42:00Z], + pos1: %{"lat" => 32.907, "lon" => -97.038}, + pos2: nil + }) + + jobs = ContactWeatherEnqueueWorker.build_era5_jobs([q1, q2]) + + # Both snap to 12:00Z and round to the same 0.25° grid cell → 1 job. + assert length(jobs) == 1 + assert hd(jobs).changes.worker == "Microwaveprop.Workers.NarrFetchWorker" + end + end + describe "enqueue_for_contact/1" do setup do Req.Test.stub(IemClient, fn conn ->