perf+test: cache nearest_hrrr per cell, batch pg_inherits, +property tests
Two performance fixes the property tests pinned down: * CalibrationSampler.build_for_hour/1 walked every cell through nearest_hrrr/3 twice — once in the producer and again in build_sample. Now computed once into a (band, lat, lon) → hrrr_or_nil map, halving the O(cells × profiles) scan (~10M comparisons saved per fire at typical sizes). * PartitionManager.ensure_quarterly_partitions/2 issued one pg_inherits scan per (parent × lookahead). Refactored to one scan per parent with in-memory coverage check across all quarters. Property tests for both modules: * PartitionManager: tile-coverage invariant (no gaps, no overlaps), exact 3-month bounds, name format, multi-call idempotence, multi-parent independence — caught a real NaiveDateTime sort bug in the original test helpers. Default Enum.sort_by/2 falls back to term comparison on NaiveDateTime; now uses NaiveDateTime as the comparator module. * CalibrationSampler: enqueued points = unique missing midpoints, empty enqueue when fully covered, sample row count invariant under HRRR availability. Saved operational gotchas to CLAUDE.md (Oban leader election spans all app=prop pods, kubectl exec deploy/prop ambiguity, half-year partition coexistence, HRRR f000 publish lag). 3310 tests + 228 properties, 0 failures.
This commit is contained in:
parent
f78d4ccb9f
commit
7ee19c5b6a
6 changed files with 428 additions and 36 deletions
|
|
@ -231,3 +231,12 @@ User submits QSO → enqueue_for_qso() → weather/hrrr/terrain/iemre workers
|
|||
- Push to `github` remote for code. The `dokku` remote is obsolete — do not push there.
|
||||
- Production Oban config in `config/runtime.exs` — no backfill cron, enrichment triggered by QSO submission
|
||||
- Shared NFS mount `/data` in production container (server: node3 at `10.0.15.103:/data`). SRTM tiles live at `/data/srtm`.
|
||||
|
||||
### Operational gotchas
|
||||
- **`kubectl exec deploy/prop` may route to the backfill pod**: the `prop` and `prop-backfill` deployments share label selectors, so `deploy/prop` can attach to either. When diagnosing cron / leader-only behavior, identify the hot pod by name first: `kubectl -n prop get pods -o name | grep -E '^pod/prop-[a-z0-9]+-[a-z0-9]+$' | head -1`. The backfill pod has `PROP_ROLE=backfill` and `peer: false` on its Oban; running RPC there is misleading because plugins gated on leader (Cron, etc.) appear inert by design.
|
||||
- **Oban Pro `oban_peers` schema differs from OSS Oban**: there is no `leader` column. The single row in `oban_peers` IS the leader; columns are `name`, `node`, `started_at`, `expires_at`. Use `SELECT * FROM oban_peers` for diagnosis, never `WHERE leader = true`.
|
||||
- **All `app=prop` pods (hot + backfill) join the same libcluster cluster** and are eligible for Oban leader election. Backfill pods carry `peer: false` in `runtime.exs` to make them ineligible — without that gate a backfill pod can win election and run its empty crontab as the cluster cron, silently stopping every scheduled job until the next leader rotation. If crons stop, check the leader node first.
|
||||
- **HRRR `hrrr_profiles` partitions are quarterly EXCEPT for some half-year ones** (e.g. `hrrr_profiles_2027_01` covers H1 2027, not just Q1). Any partition-management code must pre-check coverage via `pg_inherits` + `pg_get_expr(child.relpartbound, child.oid)` before issuing CREATE — naïvely emitting a quarterly bound that a wider existing partition already contains will trip Postgres `42P17 invalid_object_definition`.
|
||||
- **HRRR f000 publishes ~50–65 min after each cycle hour**: any worker that fetches HRRR for the current hour before that window will get 404s and mark its task `failed`. The PSKR producer + 5-min rolling-window cron tolerates this — failed tasks are reset to `queued` on the next enqueue. Don't add aggressive retry/alerting for "failed" `hrrr_fetch_tasks` rows < 1h old.
|
||||
- **Do not query `pskr_calibration_samples` for HRRR coverage less than ~5 minutes after a sampler fire**: the sampler runs every 5 min and the producer/Rust-drain/re-pass loop takes one full cycle to close. Hour `t` is fully populated only after at least two sampler fires past hour `t+1`.
|
||||
- **PSKR weather backfill is a two-pass producer/consumer**, not a join: `Microwaveprop.Pskr.CalibrationSampler` enqueues `hrrr_fetch_tasks` for cells missing HRRR; the Rust `hrrr-point-rs` worker drains them into `hrrr_profiles`; the next sampler pass UPSERTs the same `pskr_calibration_samples` row with non-NULL HRRR fields. The conflict target `(hour_utc, band, midpoint_lat, midpoint_lon)` makes it idempotent — never add a "skip if already exists" guard, that breaks the two-pass loop.
|
||||
|
|
|
|||
|
|
@ -61,17 +61,28 @@ defmodule Microwaveprop.PartitionManager do
|
|||
today = Date.utc_today()
|
||||
quarters = quarter_starts(today, lookahead_quarters)
|
||||
|
||||
for parent <- parents, {q_start, q_end} <- quarters do
|
||||
ensure_partition(parent, q_start, q_end)
|
||||
# One pg_inherits scan per parent (not per parent × lookahead).
|
||||
# The result is a list of {lo, hi} NaiveDateTime tuples; coverage
|
||||
# checks then run in-memory.
|
||||
for parent <- parents, existing = existing_bounds(parent), {q_start, q_end} <- quarters do
|
||||
ensure_partition_with_bounds(parent, q_start, q_end, existing)
|
||||
end
|
||||
end
|
||||
|
||||
# Public for direct access from tests / mix tasks.
|
||||
# Public for direct access from tests / mix tasks. Issues its own
|
||||
# pg_inherits scan; prefer the bulk path through
|
||||
# `ensure_quarterly_partitions/2` to amortize it.
|
||||
@spec ensure_partition(String.t(), Date.t(), Date.t()) :: result()
|
||||
def ensure_partition(parent, %Date{} = q_start, %Date{} = q_end) do
|
||||
name = partition_name(parent, q_start)
|
||||
ensure_partition_with_bounds(parent, q_start, q_end, existing_bounds(parent))
|
||||
end
|
||||
|
||||
if covered?(parent, q_start, q_end) do
|
||||
defp ensure_partition_with_bounds(parent, q_start, q_end, existing) do
|
||||
name = partition_name(parent, q_start)
|
||||
q_start_dt = NaiveDateTime.new!(q_start, ~T[00:00:00])
|
||||
q_end_dt = NaiveDateTime.new!(q_end, ~T[00:00:00])
|
||||
|
||||
if covered_in_memory?(existing, q_start_dt, q_end_dt) do
|
||||
{parent, name, :exists}
|
||||
else
|
||||
Repo.query!("""
|
||||
|
|
@ -85,11 +96,14 @@ defmodule Microwaveprop.PartitionManager do
|
|||
end
|
||||
end
|
||||
|
||||
# Returns true when at least one existing partition of `parent`
|
||||
# fully contains the [q_start, q_end) range. Partial overlaps fall
|
||||
# through to CREATE, where Postgres raises a clear error rather
|
||||
# than letting the worker pretend success.
|
||||
defp covered?(parent, q_start, q_end) do
|
||||
# Pull every child partition's bound expression for `parent`. Each
|
||||
# row that matches the standard `FROM ('iso') TO ('iso')` shape is
|
||||
# parsed into a `{lo_naive, hi_naive}` tuple; rows that don't match
|
||||
# (DEFAULT partitions, MINVALUE/MAXVALUE, multi-column boundaries,
|
||||
# etc.) are dropped — they can't usefully participate in a
|
||||
# range-contains check anyway.
|
||||
@spec existing_bounds(String.t()) :: [{NaiveDateTime.t(), NaiveDateTime.t()}]
|
||||
defp existing_bounds(parent) do
|
||||
%Postgrex.Result{rows: rows} =
|
||||
Repo.query!(
|
||||
"""
|
||||
|
|
@ -102,23 +116,21 @@ defmodule Microwaveprop.PartitionManager do
|
|||
[parent]
|
||||
)
|
||||
|
||||
Enum.any?(rows, fn [bound] ->
|
||||
Enum.flat_map(rows, fn [bound] ->
|
||||
case Regex.run(~r/FROM \('([^']+)'\) TO \('([^']+)'\)/, bound) do
|
||||
[_, lo, hi] ->
|
||||
lo_dt = NaiveDateTime.from_iso8601!(lo)
|
||||
hi_dt = NaiveDateTime.from_iso8601!(hi)
|
||||
q_start_dt = NaiveDateTime.new!(q_start, ~T[00:00:00])
|
||||
q_end_dt = NaiveDateTime.new!(q_end, ~T[00:00:00])
|
||||
|
||||
NaiveDateTime.compare(lo_dt, q_start_dt) != :gt and
|
||||
NaiveDateTime.compare(hi_dt, q_end_dt) != :lt
|
||||
|
||||
_ ->
|
||||
false
|
||||
[_, lo, hi] -> [{NaiveDateTime.from_iso8601!(lo), NaiveDateTime.from_iso8601!(hi)}]
|
||||
_ -> []
|
||||
end
|
||||
end)
|
||||
end
|
||||
|
||||
defp covered_in_memory?(existing, q_start_dt, q_end_dt) do
|
||||
Enum.any?(existing, fn {lo, hi} ->
|
||||
NaiveDateTime.compare(lo, q_start_dt) != :gt and
|
||||
NaiveDateTime.compare(hi, q_end_dt) != :lt
|
||||
end)
|
||||
end
|
||||
|
||||
defp partition_name(parent, %Date{year: y, month: m}) do
|
||||
"#{parent}_#{y}_#{String.pad_leading(Integer.to_string(m), 2, "0")}"
|
||||
end
|
||||
|
|
|
|||
|
|
@ -44,6 +44,8 @@ defmodule Microwaveprop.Pskr.CalibrationSampler do
|
|||
@hrrr_lookahead_deg 0.07
|
||||
@hrrr_lookahead_seconds 3600
|
||||
|
||||
@typep cell_key :: {band :: String.t(), lat :: float(), lon :: float()}
|
||||
|
||||
@doc """
|
||||
Build calibration samples for one hour. Returns the upsert count.
|
||||
Returns `0` when no PSKR spots were captured for that hour — for
|
||||
|
|
@ -62,6 +64,16 @@ defmodule Microwaveprop.Pskr.CalibrationSampler do
|
|||
kp = fetch_kp(hour_utc)
|
||||
cells = bucket_by_cell(spots)
|
||||
|
||||
# Compute (band, lat, lon) → nearest_hrrr_or_nil ONCE. Both the
|
||||
# producer (enqueue_missing_hrrr) and the per-bucket sample
|
||||
# assembly need it, and `nearest_hrrr/3` is an O(profiles) scan;
|
||||
# walking the cell list twice was ~10M extra comparisons per
|
||||
# fire at typical sizes (~1k cells × ~5k profiles).
|
||||
cell_hrrr =
|
||||
Map.new(cells, fn {{_band, lat, lon} = key, _spots} ->
|
||||
{key, nearest_hrrr(lat, lon, hrrr_index)}
|
||||
end)
|
||||
|
||||
# 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
|
||||
|
|
@ -69,9 +81,9 @@ defmodule Microwaveprop.Pskr.CalibrationSampler do
|
|||
# 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)
|
||||
enqueue_missing_hrrr(cell_hrrr, hour_utc)
|
||||
|
||||
samples = Enum.map(cells, &build_sample(&1, hour_utc, hrrr_index, kp))
|
||||
samples = Enum.map(cells, &build_sample(&1, hour_utc, cell_hrrr, kp))
|
||||
upsert(samples)
|
||||
end
|
||||
end
|
||||
|
|
@ -150,16 +162,13 @@ defmodule Microwaveprop.Pskr.CalibrationSampler do
|
|||
# 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.
|
||||
@spec enqueue_missing_hrrr(map(), DateTime.t(), [map()]) :: :ok
|
||||
defp enqueue_missing_hrrr(cells, hour_utc, hrrr_index) do
|
||||
@spec enqueue_missing_hrrr(%{cell_key() => map() | nil}, DateTime.t()) :: :ok
|
||||
defp enqueue_missing_hrrr(cell_hrrr, hour_utc) 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
|
||||
cell_hrrr
|
||||
|> Enum.flat_map(fn
|
||||
{_key, %{}} -> []
|
||||
{{_band, lat, lon}, nil} -> [Weather.round_to_hrrr_grid(lat, lon)]
|
||||
end)
|
||||
|> Enum.uniq()
|
||||
|
||||
|
|
@ -190,8 +199,8 @@ defmodule Microwaveprop.Pskr.CalibrationSampler do
|
|||
|
||||
# ── Per-bucket sample assembly ──────────────────────────────────
|
||||
|
||||
defp build_sample({{band, lat, lon}, cell_spots}, hour_utc, hrrr_index, kp) do
|
||||
hrrr = nearest_hrrr(lat, lon, hrrr_index)
|
||||
defp build_sample({{band, lat, lon} = key, cell_spots}, hour_utc, cell_hrrr, kp) do
|
||||
hrrr = Map.get(cell_hrrr, key)
|
||||
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
|
||||
%{
|
||||
|
|
|
|||
151
test/microwaveprop/partition_manager_property_test.exs
Normal file
151
test/microwaveprop/partition_manager_property_test.exs
Normal file
|
|
@ -0,0 +1,151 @@
|
|||
defmodule Microwaveprop.PartitionManagerPropertyTest do
|
||||
@moduledoc """
|
||||
Properties of the partition tiling produced by
|
||||
`Microwaveprop.PartitionManager.ensure_quarterly_partitions/2`. The
|
||||
goal is to pin down the math invariants (no gaps, no overlaps,
|
||||
exact 3-month bounds, deterministic naming) so refactors of the
|
||||
internal quarter helpers can't silently break them.
|
||||
|
||||
All tests use a synthetic parent created inside the sandbox
|
||||
transaction so DDL rolls back cleanly when the test finishes.
|
||||
"""
|
||||
use Microwaveprop.DataCase, async: true
|
||||
use ExUnitProperties
|
||||
|
||||
alias Microwaveprop.PartitionManager
|
||||
alias Microwaveprop.Repo
|
||||
|
||||
defp synthetic_parent do
|
||||
"test_partitioned_#{System.unique_integer([:positive])}"
|
||||
end
|
||||
|
||||
defp create_partitioned_parent!(name) do
|
||||
Repo.query!("""
|
||||
CREATE TABLE public."#{name}" (
|
||||
valid_time timestamp without time zone NOT NULL,
|
||||
payload integer
|
||||
) PARTITION BY RANGE (valid_time)
|
||||
""")
|
||||
end
|
||||
|
||||
defp partitions_of(parent) do
|
||||
%Postgrex.Result{rows: rows} =
|
||||
Repo.query!(
|
||||
"""
|
||||
SELECT child.relname, pg_get_expr(child.relpartbound, child.oid)
|
||||
FROM pg_inherits
|
||||
JOIN pg_class parent ON pg_inherits.inhparent = parent.oid
|
||||
JOIN pg_class child ON pg_inherits.inhrelid = child.oid
|
||||
WHERE parent.relname = $1
|
||||
""",
|
||||
[parent]
|
||||
)
|
||||
|
||||
rows
|
||||
|> Enum.map(fn [name, bound] ->
|
||||
[_, lo, hi] = Regex.run(~r/FROM \('([^']+)'\) TO \('([^']+)'\)/, bound)
|
||||
{name, NaiveDateTime.from_iso8601!(lo), NaiveDateTime.from_iso8601!(hi)}
|
||||
end)
|
||||
# Default `Enum.sort_by/2` uses `Kernel.<=/2` which falls back to
|
||||
# struct-field order on NaiveDateTime — NOT chronological. Pass
|
||||
# NaiveDateTime as the comparator module so sort_by uses
|
||||
# NaiveDateTime.compare/2.
|
||||
|> Enum.sort_by(fn {_, lo, _} -> lo end, NaiveDateTime)
|
||||
end
|
||||
|
||||
property "partitions tile the lookahead range with no gaps and no overlaps" do
|
||||
check all(lookahead <- integer(0..8), max_runs: 25) do
|
||||
parent = synthetic_parent()
|
||||
create_partitioned_parent!(parent)
|
||||
|
||||
_ = PartitionManager.ensure_quarterly_partitions(lookahead, [parent])
|
||||
|
||||
partitions = partitions_of(parent)
|
||||
assert length(partitions) == lookahead + 1
|
||||
|
||||
partitions
|
||||
|> Enum.chunk_every(2, 1, :discard)
|
||||
|> Enum.each(fn [{name_a, _, hi_a} = a, {name_b, lo_b, _} = b] ->
|
||||
# Adjacent partitions must touch exactly: end of A == start of B.
|
||||
assert NaiveDateTime.compare(hi_a, lo_b) == :eq,
|
||||
"gap/overlap: #{name_a} #{inspect(a)} → #{name_b} #{inspect(b)}"
|
||||
end)
|
||||
end
|
||||
end
|
||||
|
||||
property "every partition spans exactly one quarter (89-92 days)" do
|
||||
check all(lookahead <- integer(0..6), max_runs: 15) do
|
||||
parent = synthetic_parent()
|
||||
create_partitioned_parent!(parent)
|
||||
|
||||
_ = PartitionManager.ensure_quarterly_partitions(lookahead, [parent])
|
||||
|
||||
Enum.each(partitions_of(parent), fn {_, lo, hi} ->
|
||||
days = NaiveDateTime.diff(hi, lo, :day)
|
||||
assert days in 89..92, "quarter spanned #{days} days"
|
||||
# Quarter starts on the 1st of Jan/Apr/Jul/Oct at midnight.
|
||||
assert lo.day == 1
|
||||
assert lo.hour == 0
|
||||
assert lo.minute == 0
|
||||
assert lo.second == 0
|
||||
assert lo.month in [1, 4, 7, 10]
|
||||
end)
|
||||
end
|
||||
end
|
||||
|
||||
property "partition names follow <parent>_YYYY_MM and are unique" do
|
||||
check all(lookahead <- integer(0..6), max_runs: 15) do
|
||||
parent = synthetic_parent()
|
||||
create_partitioned_parent!(parent)
|
||||
|
||||
_ = PartitionManager.ensure_quarterly_partitions(lookahead, [parent])
|
||||
names = Enum.map(partitions_of(parent), fn {n, _, _} -> n end)
|
||||
|
||||
Enum.each(names, fn n ->
|
||||
assert Regex.match?(~r/^#{parent}_\d{4}_(01|04|07|10)$/, n),
|
||||
"bad partition name: #{n}"
|
||||
end)
|
||||
|
||||
assert length(names) == length(Enum.uniq(names)), "duplicate partition names"
|
||||
end
|
||||
end
|
||||
|
||||
property "is idempotent — repeated calls are stable" do
|
||||
check all(
|
||||
lookahead <- integer(0..4),
|
||||
extra_runs <- integer(1..3),
|
||||
max_runs: 10
|
||||
) do
|
||||
parent = synthetic_parent()
|
||||
create_partitioned_parent!(parent)
|
||||
|
||||
_ = PartitionManager.ensure_quarterly_partitions(lookahead, [parent])
|
||||
first = partitions_of(parent)
|
||||
|
||||
Enum.each(1..extra_runs, fn _ ->
|
||||
results = PartitionManager.ensure_quarterly_partitions(lookahead, [parent])
|
||||
# Every entry on a re-run must be :exists — never :created.
|
||||
assert Enum.all?(results, fn {_, _, status} -> status == :exists end)
|
||||
end)
|
||||
|
||||
assert partitions_of(parent) == first
|
||||
end
|
||||
end
|
||||
|
||||
property "respects multi-parent calls (independent tiles per parent)" do
|
||||
check all(
|
||||
lookahead <- integer(0..3),
|
||||
n_parents <- integer(1..3),
|
||||
max_runs: 8
|
||||
) do
|
||||
parents = Enum.map(1..n_parents, fn _ -> synthetic_parent() end)
|
||||
Enum.each(parents, &create_partitioned_parent!/1)
|
||||
|
||||
_ = PartitionManager.ensure_quarterly_partitions(lookahead, parents)
|
||||
|
||||
Enum.each(parents, fn p ->
|
||||
assert length(partitions_of(p)) == lookahead + 1
|
||||
end)
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
@ -61,7 +61,9 @@ defmodule Microwaveprop.PartitionManagerTest do
|
|||
{NaiveDateTime.from_iso8601!(lo), NaiveDateTime.from_iso8601!(hi)}
|
||||
end)
|
||||
|
||||
sorted = Enum.sort_by(bounds, fn {lo, _} -> lo end)
|
||||
# Pass `NaiveDateTime` as the comparator module — default
|
||||
# `<=` falls back to term comparison, which is not chronological.
|
||||
sorted = Enum.sort_by(bounds, fn {lo, _} -> lo end, NaiveDateTime)
|
||||
|
||||
Enum.each(sorted, fn {lo, hi} ->
|
||||
# Quarter starts on the 1st of Jan/Apr/Jul/Oct at midnight.
|
||||
|
|
|
|||
209
test/microwaveprop/pskr/calibration_sampler_property_test.exs
Normal file
209
test/microwaveprop/pskr/calibration_sampler_property_test.exs
Normal file
|
|
@ -0,0 +1,209 @@
|
|||
defmodule Microwaveprop.Pskr.CalibrationSamplerPropertyTest do
|
||||
@moduledoc """
|
||||
Properties of the producer side of the PSKR calibration pipeline.
|
||||
These pin down the contract that future refactors of
|
||||
`enqueue_missing_hrrr/2` (currently inlined in
|
||||
`Microwaveprop.Pskr.CalibrationSampler.build_for_hour/1`) cannot
|
||||
silently break:
|
||||
|
||||
1. Cells with HRRR coverage never get re-enqueued (no churn on
|
||||
`hrrr_fetch_tasks`).
|
||||
2. Cells without HRRR coverage are *all* enqueued (or, for
|
||||
multiple bands sharing a midpoint, exactly one task per
|
||||
distinct rounded grid point).
|
||||
3. Calibration sample row count equals distinct cell count
|
||||
regardless of HRRR availability.
|
||||
"""
|
||||
use Microwaveprop.DataCase, async: true
|
||||
use ExUnitProperties
|
||||
|
||||
alias Microwaveprop.Pskr.CalibrationSample
|
||||
alias Microwaveprop.Pskr.CalibrationSampler
|
||||
alias Microwaveprop.Pskr.SpotHourly
|
||||
alias Microwaveprop.Repo
|
||||
alias Microwaveprop.Weather.HrrrProfile
|
||||
|
||||
@hour ~U[2026-05-04 18:00:00Z]
|
||||
|
||||
# ── Generators ──────────────────────────────────────────────────
|
||||
|
||||
defp band_gen, do: member_of(["10000", "24000", "47000"])
|
||||
|
||||
# CONUS-ish: 25-50N, 67-125W — within HRRR domain so the
|
||||
# round_to_hrrr_grid output is well-defined.
|
||||
defp lat_gen, do: float(min: 25.0, max: 49.5)
|
||||
defp lon_gen, do: float(min: -125.0, max: -67.0)
|
||||
|
||||
# Per-band cell. Coordinates stay in `bind` so all spots within a
|
||||
# generated cell share the same midpoint (otherwise bucket_by_cell
|
||||
# would split them).
|
||||
defp cell_spec_gen do
|
||||
gen all(
|
||||
band <- band_gen(),
|
||||
lat <- lat_gen(),
|
||||
lon <- lon_gen(),
|
||||
n_spots <- integer(1..3)
|
||||
) do
|
||||
%{band: band, lat: lat, lon: lon, n_spots: n_spots}
|
||||
end
|
||||
end
|
||||
|
||||
defp insert_spot!(attrs) do
|
||||
{:ok, _} =
|
||||
%SpotHourly{}
|
||||
|> SpotHourly.changeset(
|
||||
Map.merge(
|
||||
%{
|
||||
hour_utc: @hour,
|
||||
band: "10000",
|
||||
sender_grid: "EM00aa",
|
||||
receiver_grid: "DM00bb",
|
||||
spot_count: 1,
|
||||
distance_km: 100.0,
|
||||
modes: ["FT8"]
|
||||
},
|
||||
attrs
|
||||
)
|
||||
)
|
||||
|> Repo.insert()
|
||||
end
|
||||
|
||||
defp insert_hrrr!(lat, lon) do
|
||||
{:ok, _} =
|
||||
%HrrrProfile{}
|
||||
|> HrrrProfile.changeset(%{
|
||||
valid_time: @hour,
|
||||
lat: lat,
|
||||
lon: lon,
|
||||
run_time: @hour,
|
||||
surface_temp_c: 20.0,
|
||||
surface_dewpoint_c: 15.0,
|
||||
pwat_mm: 25.0,
|
||||
surface_pressure_mb: 1013.0,
|
||||
min_refractivity_gradient: -100.0,
|
||||
hpbl_m: 500.0,
|
||||
ducting_detected: false,
|
||||
profile: []
|
||||
})
|
||||
|> Repo.insert()
|
||||
end
|
||||
|
||||
defp seed_spots!(specs, base_callsign) do
|
||||
specs
|
||||
|> Enum.with_index()
|
||||
|> Enum.each(fn {spec, idx} ->
|
||||
Enum.each(1..spec.n_spots, fn s_idx ->
|
||||
insert_spot!(%{
|
||||
band: spec.band,
|
||||
midpoint_lat: spec.lat,
|
||||
midpoint_lon: spec.lon,
|
||||
# Each spot needs a distinct (band, sender, receiver) so
|
||||
# the unique index doesn't collapse them.
|
||||
sender_grid: "S#{base_callsign}#{idx}#{s_idx}",
|
||||
receiver_grid: "R#{base_callsign}#{idx}#{s_idx}"
|
||||
})
|
||||
end)
|
||||
end)
|
||||
end
|
||||
|
||||
defp task_count, do: Repo.aggregate(from(t in "hrrr_fetch_tasks"), :count)
|
||||
|
||||
defp task_points do
|
||||
case Repo.one(from(t in "hrrr_fetch_tasks", select: t.points)) do
|
||||
nil -> []
|
||||
pts -> Enum.map(pts, fn %{"lat" => la, "lon" => lo} -> {la, lo} end)
|
||||
end
|
||||
end
|
||||
|
||||
# 0.125° snap matches CalibrationSampler @grid_step_deg.
|
||||
defp snap_cell(lat), do: Float.round(lat / 0.125) * 0.125
|
||||
|
||||
defp expected_unique_cells(specs) do
|
||||
specs
|
||||
|> Enum.map(fn s -> {s.band, snap_cell(s.lat), snap_cell(s.lon)} end)
|
||||
|> Enum.uniq()
|
||||
end
|
||||
|
||||
# ── Properties ──────────────────────────────────────────────────
|
||||
|
||||
property "no HRRR profile inserted: every distinct cell midpoint is enqueued exactly once" do
|
||||
check all(specs <- list_of(cell_spec_gen(), min_length: 1, max_length: 6), max_runs: 15) do
|
||||
Repo.delete_all(SpotHourly)
|
||||
Repo.delete_all(CalibrationSample)
|
||||
Repo.delete_all(HrrrProfile)
|
||||
Repo.query!("DELETE FROM hrrr_fetch_tasks")
|
||||
seed_spots!(specs, "A")
|
||||
|
||||
CalibrationSampler.build_for_hour(@hour)
|
||||
|
||||
# One row per (band, snapped_cell) goes into samples.
|
||||
assert Repo.aggregate(CalibrationSample, :count) == length(expected_unique_cells(specs))
|
||||
|
||||
# Exactly one hrrr_fetch_tasks row (one valid_time = one row by
|
||||
# unique index). The point list inside it must equal the unique
|
||||
# set of cell midpoints rounded to the HRRR grid.
|
||||
assert task_count() == 1
|
||||
|
||||
expected_pts =
|
||||
specs
|
||||
|> Enum.map(fn s -> {snap_cell(s.lat), snap_cell(s.lon)} end)
|
||||
|> Enum.map(fn {la, lo} -> {Float.round(la, 2), Float.round(lo, 2)} end)
|
||||
|> Enum.uniq()
|
||||
|
||||
got = Enum.map(task_points(), fn {la, lo} -> {Float.round(la, 2), Float.round(lo, 2)} end)
|
||||
assert Enum.sort(got) == Enum.sort(expected_pts)
|
||||
end
|
||||
end
|
||||
|
||||
property "every cell covered by HRRR: nothing is enqueued" do
|
||||
check all(specs <- list_of(cell_spec_gen(), min_length: 1, max_length: 5), max_runs: 12) do
|
||||
Repo.delete_all(SpotHourly)
|
||||
Repo.delete_all(CalibrationSample)
|
||||
Repo.delete_all(HrrrProfile)
|
||||
Repo.query!("DELETE FROM hrrr_fetch_tasks")
|
||||
seed_spots!(specs, "B")
|
||||
|
||||
# Plant an HRRR profile near every snapped cell midpoint so
|
||||
# nearest_hrrr finds a match for all cells.
|
||||
specs
|
||||
|> Enum.map(fn s -> {snap_cell(s.lat), snap_cell(s.lon)} end)
|
||||
|> Enum.uniq()
|
||||
|> Enum.each(fn {la, lo} -> insert_hrrr!(la, lo) end)
|
||||
|
||||
CalibrationSampler.build_for_hour(@hour)
|
||||
|
||||
assert task_count() == 0
|
||||
# Samples still get written, just with non-NULL HRRR fields.
|
||||
assert Repo.aggregate(CalibrationSample, :count) == length(expected_unique_cells(specs))
|
||||
end
|
||||
end
|
||||
|
||||
property "sample row count is invariant under HRRR availability" do
|
||||
check all(specs <- list_of(cell_spec_gen(), min_length: 1, max_length: 5), max_runs: 12) do
|
||||
# No HRRR
|
||||
Repo.delete_all(SpotHourly)
|
||||
Repo.delete_all(CalibrationSample)
|
||||
Repo.delete_all(HrrrProfile)
|
||||
Repo.query!("DELETE FROM hrrr_fetch_tasks")
|
||||
seed_spots!(specs, "C")
|
||||
CalibrationSampler.build_for_hour(@hour)
|
||||
no_hrrr_count = Repo.aggregate(CalibrationSample, :count)
|
||||
|
||||
# Reset, plant HRRR, re-run.
|
||||
Repo.delete_all(SpotHourly)
|
||||
Repo.delete_all(CalibrationSample)
|
||||
Repo.query!("DELETE FROM hrrr_fetch_tasks")
|
||||
seed_spots!(specs, "D")
|
||||
|
||||
specs
|
||||
|> Enum.map(fn s -> {snap_cell(s.lat), snap_cell(s.lon)} end)
|
||||
|> Enum.uniq()
|
||||
|> Enum.each(fn {la, lo} -> insert_hrrr!(la, lo) end)
|
||||
|
||||
CalibrationSampler.build_for_hour(@hour)
|
||||
with_hrrr_count = Repo.aggregate(CalibrationSample, :count)
|
||||
|
||||
assert no_hrrr_count == with_hrrr_count
|
||||
end
|
||||
end
|
||||
end
|
||||
Loading…
Add table
Reference in a new issue