feat(pskr): enqueue hrrr_fetch_tasks for cells missing HRRR data
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.
This commit is contained in:
parent
1650744e89
commit
f668b457cb
2 changed files with 84 additions and 4 deletions
|
|
@ -31,6 +31,8 @@ defmodule Microwaveprop.Pskr.CalibrationSampler do
|
||||||
alias Microwaveprop.Pskr.SpotHourly
|
alias Microwaveprop.Pskr.SpotHourly
|
||||||
alias Microwaveprop.Repo
|
alias Microwaveprop.Repo
|
||||||
alias Microwaveprop.SpaceWeather
|
alias Microwaveprop.SpaceWeather
|
||||||
|
alias Microwaveprop.Weather
|
||||||
|
alias Microwaveprop.Weather.HrrrPointEnqueuer
|
||||||
alias Microwaveprop.Weather.HrrrProfile
|
alias Microwaveprop.Weather.HrrrProfile
|
||||||
|
|
||||||
require Logger
|
require Logger
|
||||||
|
|
@ -58,12 +60,18 @@ defmodule Microwaveprop.Pskr.CalibrationSampler do
|
||||||
else
|
else
|
||||||
hrrr_index = build_hrrr_index(hour_utc)
|
hrrr_index = build_hrrr_index(hour_utc)
|
||||||
kp = fetch_kp(hour_utc)
|
kp = fetch_kp(hour_utc)
|
||||||
|
cells = bucket_by_cell(spots)
|
||||||
|
|
||||||
samples =
|
# Producer side of the two-pass HRRR join: any cell whose
|
||||||
spots
|
# midpoint has no profile in the lookahead window gets a
|
||||||
|> bucket_by_cell()
|
# `hrrr_fetch_tasks` row enqueued for the Rust hrrr-point-worker
|
||||||
|> Enum.map(&build_sample(&1, hour_utc, hrrr_index, kp))
|
# 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)
|
upsert(samples)
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
@ -134,6 +142,37 @@ defmodule Microwaveprop.Pskr.CalibrationSampler do
|
||||||
end
|
end
|
||||||
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 ───────────────────────────────────────────────────
|
# ── Kp lookup ───────────────────────────────────────────────────
|
||||||
|
|
||||||
# Single Kp value applies cluster-wide for the hour. We prefer the
|
# Single Kp value applies cluster-wide for the hour. We prefer the
|
||||||
|
|
|
||||||
|
|
@ -1,6 +1,8 @@
|
||||||
defmodule Microwaveprop.Pskr.CalibrationSamplerTest do
|
defmodule Microwaveprop.Pskr.CalibrationSamplerTest do
|
||||||
use Microwaveprop.DataCase, async: true
|
use Microwaveprop.DataCase, async: true
|
||||||
|
|
||||||
|
import Ecto.Query
|
||||||
|
|
||||||
alias Microwaveprop.Pskr.CalibrationSample
|
alias Microwaveprop.Pskr.CalibrationSample
|
||||||
alias Microwaveprop.Pskr.CalibrationSampler
|
alias Microwaveprop.Pskr.CalibrationSampler
|
||||||
alias Microwaveprop.Pskr.SpotHourly
|
alias Microwaveprop.Pskr.SpotHourly
|
||||||
|
|
@ -133,6 +135,45 @@ defmodule Microwaveprop.Pskr.CalibrationSamplerTest do
|
||||||
assert sample.hrrr_ducting_detected == nil
|
assert sample.hrrr_ducting_detected == nil
|
||||||
end
|
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
|
test "stamps the latest Kp from SpaceWeather on every sample in the hour" do
|
||||||
insert_spot!(%{})
|
insert_spot!(%{})
|
||||||
insert_hrrr!(%{lat: 33.0, lon: -97.0})
|
insert_hrrr!(%{lat: 33.0, lon: -97.0})
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue