From f668b457cbe98c226eea7a7bcdc0757b2e105914 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 5 May 2026 16:14:43 -0500 Subject: [PATCH] feat(pskr): enqueue hrrr_fetch_tasks for cells missing HRRR data MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The CalibrationSampler joined PSKR midpoints against hrrr_profiles via nearest-within-window, but post-Phase-3 Stream C nothing bulk-populates the grid — hrrr_profiles only sees per-QSO point fetches from the Rust hrrr-point-worker. Result: 0 / 16,042 prod samples had HRRR data. Producer pattern: when a cell has no profile in the lookahead window, enqueue a row into hrrr_fetch_tasks at the cell's HRRR-grid-rounded midpoint. The Rust worker drains it; the next 5-min rolling-window pass UPSERTs the same sample with non-NULL weather fields. Idempotent with the existing on-conflict design — no extra worker, no churn on cells already covered, multiple bands sharing a midpoint dedupe to a single fetch via Enum.uniq + jsonb point-union in HrrrPointEnqueuer. --- lib/microwaveprop/pskr/calibration_sampler.ex | 47 +++++++++++++++++-- .../pskr/calibration_sampler_test.exs | 41 ++++++++++++++++ 2 files changed, 84 insertions(+), 4 deletions(-) diff --git a/lib/microwaveprop/pskr/calibration_sampler.ex b/lib/microwaveprop/pskr/calibration_sampler.ex index 9384b8e1..9e6deb42 100644 --- a/lib/microwaveprop/pskr/calibration_sampler.ex +++ b/lib/microwaveprop/pskr/calibration_sampler.ex @@ -31,6 +31,8 @@ defmodule Microwaveprop.Pskr.CalibrationSampler do alias Microwaveprop.Pskr.SpotHourly alias Microwaveprop.Repo alias Microwaveprop.SpaceWeather + alias Microwaveprop.Weather + alias Microwaveprop.Weather.HrrrPointEnqueuer alias Microwaveprop.Weather.HrrrProfile require Logger @@ -58,12 +60,18 @@ defmodule Microwaveprop.Pskr.CalibrationSampler do else hrrr_index = build_hrrr_index(hour_utc) kp = fetch_kp(hour_utc) + cells = bucket_by_cell(spots) - samples = - spots - |> bucket_by_cell() - |> Enum.map(&build_sample(&1, hour_utc, hrrr_index, kp)) + # Producer side of the two-pass HRRR join: any cell whose + # midpoint has no profile in the lookahead window gets a + # `hrrr_fetch_tasks` row enqueued for the Rust hrrr-point-worker + # to drain. The next rolling-window cron pass (every 5 min) will + # find the landed profile and UPSERT the sample with non-NULL + # weather fields. Existing cells with HRRR data already in + # memory are skipped — no churn against the queue table. + enqueue_missing_hrrr(cells, hour_utc, hrrr_index) + samples = Enum.map(cells, &build_sample(&1, hour_utc, hrrr_index, kp)) upsert(samples) end end @@ -134,6 +142,37 @@ defmodule Microwaveprop.Pskr.CalibrationSampler do end end + # ── HRRR fetch enqueue (producer side) ────────────────────────── + + # For every PSKR cell that has no HRRR profile in the lookahead + # window, queue a point fetch at the cell midpoint. Dedup on + # `(rounded_hrrr_lat, rounded_hrrr_lon)` so multiple bands sharing + # a midpoint enqueue once, and let `HrrrPointEnqueuer.enqueue/1` + # union the points into the existing `hrrr_fetch_tasks` row for + # this hour rather than racing to insert duplicates. + defp enqueue_missing_hrrr(cells, hour_utc, hrrr_index) do + points = + cells + |> Enum.flat_map(fn {{_band, lat, lon}, _spots} -> + if nearest_hrrr(lat, lon, hrrr_index) do + [] + else + [Weather.round_to_hrrr_grid(lat, lon)] + end + end) + |> Enum.uniq() + + case points do + [] -> + :ok + + pts -> + {:ok, _} = HrrrPointEnqueuer.enqueue(%{hour_utc => pts}) + + Logger.info("Pskr.CalibrationSampler: enqueued #{length(pts)} HRRR points for #{hour_utc}") + end + end + # ── Kp lookup ─────────────────────────────────────────────────── # Single Kp value applies cluster-wide for the hour. We prefer the diff --git a/test/microwaveprop/pskr/calibration_sampler_test.exs b/test/microwaveprop/pskr/calibration_sampler_test.exs index 63dfb240..dd04ad1c 100644 --- a/test/microwaveprop/pskr/calibration_sampler_test.exs +++ b/test/microwaveprop/pskr/calibration_sampler_test.exs @@ -1,6 +1,8 @@ defmodule Microwaveprop.Pskr.CalibrationSamplerTest do use Microwaveprop.DataCase, async: true + import Ecto.Query + alias Microwaveprop.Pskr.CalibrationSample alias Microwaveprop.Pskr.CalibrationSampler alias Microwaveprop.Pskr.SpotHourly @@ -133,6 +135,45 @@ defmodule Microwaveprop.Pskr.CalibrationSamplerTest do assert sample.hrrr_ducting_detected == nil end + test "enqueues an hrrr_fetch_tasks row when a cell has no nearby HRRR profile" do + insert_spot!(%{midpoint_lat: 33.0, midpoint_lon: -97.0}) + + assert CalibrationSampler.build_for_hour(@hour) == 1 + + [task] = + Repo.all(from(t in "hrrr_fetch_tasks", select: %{points: t.points, status: t.status, valid_time: t.valid_time})) + + assert task.status == "queued" + assert task.valid_time == DateTime.to_naive(@hour) + # The cell midpoint, snapped to HRRR grid (0.01° rounding). + assert task.points == [%{"lat" => 33.0, "lon" => -97.0}] + end + + test "does not enqueue an hrrr_fetch_tasks row when an HRRR profile already covers the cell" do + insert_spot!(%{}) + insert_hrrr!(%{lat: 33.0, lon: -97.0}) + + CalibrationSampler.build_for_hour(@hour) + + assert Repo.aggregate(from(t in "hrrr_fetch_tasks"), :count) == 0 + end + + test "deduplicates HRRR enqueue points across bands sharing a midpoint" do + insert_spot!(%{band: "10000", midpoint_lat: 33.0, midpoint_lon: -97.0}) + insert_spot!(%{band: "24000", midpoint_lat: 33.0, midpoint_lon: -97.0, sender_grid: "EM13"}) + insert_spot!(%{band: "10000", midpoint_lat: 35.0, midpoint_lon: -97.0, sender_grid: "EM15"}) + + CalibrationSampler.build_for_hour(@hour) + + [task] = Repo.all(from(t in "hrrr_fetch_tasks", select: %{points: t.points})) + # Two unique midpoints (33.0,-97.0) and (35.0,-97.0); the second band at the + # first midpoint must collapse, otherwise the Rust worker double-fetches the + # same lat/lon for one valid_time. + assert length(task.points) == 2 + sorted = Enum.sort_by(task.points, & &1["lat"]) + assert sorted == [%{"lat" => 33.0, "lon" => -97.0}, %{"lat" => 35.0, "lon" => -97.0}] + end + test "stamps the latest Kp from SpaceWeather on every sample in the hour" do insert_spot!(%{}) insert_hrrr!(%{lat: 33.0, lon: -97.0})