feat(pskr): hourly calibration sampler joining spots × HRRR × Kp

PSK Reporter is now an always-on data feed, so the calibration
corpus that recalibration will eventually train against can grow
hourly from when the firehose started. One row per (hour, band,
0.125° midpoint cell) joins three sources:

  * `pskr_spots_hourly` — observed spot density (truth signal)
  * `hrrr_profiles` nearest match at the cell midpoint at hour
    boundary (atmospheric features: T, Td, PWAT, P, dN/dh, HPBL,
    ducting flag)
  * `geomagnetic_observations` latest Kp at the hour (space weather)

Predicted scores are intentionally NOT stored — they're a function
of the algorithm version under evaluation. Storing only features
keeps the corpus stable across every weight refit; the recalibrator
computes predictions on demand.

Components:

  * Migration `20260504210756_create_pskr_calibration_samples` —
    new table + unique index on (hour, band, midpoint_lat, lon),
    plus a midpoint spatial index on `pskr_spots_hourly` to keep
    the join cheap.
  * `Pskr.CalibrationSample` — schema mirror of the table with
    the same `(hour, band, midpoint_lat, lon)` unique constraint.
  * `Pskr.CalibrationSampler.build_for_hour/1` — pulls all spots
    for the hour, snaps midpoints to the propagation grid (0.125°),
    builds an in-memory HRRR index over a ±0.07°/±60 min window,
    and bulk-upserts samples. Idempotent — re-runs upsert.
  * `Workers.PskrCalibrationWorker` — Oban cron entry on the
    `:backfill_enqueue` queue. Default args target the previous
    full hour; explicit `hour_utc` arg reruns any hour.
  * Cron `25 * * * *` — fires past HRRR analysis publish window
    (~HH:50→HH:05) and PSKR aggregator's 60s flush.

Forward-only: PSKR has no historical archive, so the corpus only
grows from when the feed started. Recalibration weight refits
should wait for ~30 days / ~10k samples to cover diurnal and
synoptic variability.

Tests cover cell-snapping, HRRR feature joining, Kp stamping,
median-distance aggregation, mode dedup, idempotent reruns, and
the worker's default-hour and explicit-hour args (12 new tests,
all passing).

Backfill pipeline untouched — none of these changes feed into
contact enrichment.

config/runtime.exs

@@ -309,7 +309,14 @@ if config_env() == :prod do
        # cheap enough to run daily so the table self-heals if a
        # deploy clears it or new profiles need to fold into the means.
        # 02:30 UTC sits inside the quiet pre-NARR/HRRR ingestion gap.
-       {"30 2 * * *", Microwaveprop.Workers.AdminTaskWorker, args: %{"task" => "climatology"}}
+       {"30 2 * * *", Microwaveprop.Workers.AdminTaskWorker, args: %{"task" => "climatology"}},
+       # Build the PSKR calibration corpus one hour at a time. HH:25
+       # is past the HRRR analysis publish window (~HH:50 of the
+       # previous hour, surfaces by HH:05) and past PSKR's 60s
+       # aggregator flush, so the previous-hour sample has the
+       # atmospheric and signal data it needs. Idempotent UPSERT, so
+       # missed fires are safe to backfill manually.
+       {"25 * * * *", Microwaveprop.Workers.PskrCalibrationWorker}
      ]}
 
   base_plugins = [

lib/microwaveprop/pskr/calibration_sample.ex

@@ -0,0 +1,80 @@
+defmodule Microwaveprop.Pskr.CalibrationSample do
+  @moduledoc """
+  One sample for the PSKR-driven recalibration corpus: a (hour, band,
+  midpoint cell) bucket joining aggregate PSKR spot density with the
+  HRRR atmospheric state and SWPC geomagnetic state at that hour.
+
+  ## Why we don't store predicted scores
+
+  The corpus exists so the recalibrator can learn weights from
+  (atmospheric features → observed spot density). Predicted scores
+  are derived quantities that change every time the algorithm
+  changes; storing them locks the corpus to a single algorithm
+  version. Features are immutable — they describe what the
+  atmosphere actually was — so the same sample row stays useful
+  across every weight refit.
+
+  ## Cell granularity
+
+  `midpoint_lat`/`midpoint_lon` are snapped to the propagation grid
+  (currently 0.125°) by `Pskr.CalibrationSampler` before insert.
+  Two PSKR paths whose midpoints land in the same grid cell collapse
+  into a single row — the same logic that already drives
+  `pskr_spots_hourly` aggregation, applied at the spatial dimension.
+
+  ## Forward-only
+
+  PSK Reporter has no historical archive — the firehose is real-time
+  only. The corpus grows from when the feed first started and never
+  backfills further. Recalibration weight refits should only be
+  attempted once enough days have accumulated to cover diurnal and
+  synoptic variability (heuristic: ≥ 30 d, ≥ 10k samples).
+  """
+  use Ecto.Schema
+
+  import Ecto.Changeset
+
+  @primary_key {:id, :binary_id, autogenerate: true}
+  @foreign_key_type :binary_id
+
+  schema "pskr_calibration_samples" do
+    field :hour_utc, :utc_datetime
+    field :band, :string
+    field :midpoint_lat, :float
+    field :midpoint_lon, :float
+
+    field :spot_count, :integer, default: 0
+    field :distinct_paths, :integer, default: 0
+    field :median_distance_km, :float
+    field :modes, {:array, :string}, default: []
+
+    field :surface_temp_c, :float
+    field :surface_dewpoint_c, :float
+    field :pwat_mm, :float
+    field :surface_pressure_mb, :float
+    field :min_refractivity_gradient, :float
+    field :hpbl_m, :float
+    field :hrrr_ducting_detected, :boolean
+
+    field :kp_index, :integer
+
+    timestamps(type: :utc_datetime)
+  end
+
+  @type t :: %__MODULE__{}
+
+  @cast_fields ~w(hour_utc band midpoint_lat midpoint_lon spot_count distinct_paths
+                  median_distance_km modes surface_temp_c surface_dewpoint_c pwat_mm
+                  surface_pressure_mb min_refractivity_gradient hpbl_m hrrr_ducting_detected
+                  kp_index)a
+
+  @required_fields ~w(hour_utc band midpoint_lat midpoint_lon spot_count)a
+
+  @spec changeset(t(), map()) :: Ecto.Changeset.t()
+  def changeset(record, attrs) do
+    record
+    |> cast(attrs, @cast_fields)
+    |> validate_required(@required_fields)
+    |> unique_constraint([:hour_utc, :band, :midpoint_lat, :midpoint_lon])
+  end
+end

lib/microwaveprop/pskr/calibration_sampler.ex

@@ -0,0 +1,236 @@
+defmodule Microwaveprop.Pskr.CalibrationSampler do
+  @moduledoc """
+  Builds `Pskr.CalibrationSample` rows by joining PSKR spot density
+  with HRRR atmospheric state and SWPC geomagnetic state at a given
+  hour.
+
+  Workflow:
+
+  1. Pull every `pskr_spots_hourly` row for the target hour.
+  2. Snap each row's midpoint to the propagation grid (0.125°) and
+     bucket by `(band, snapped_lat, snapped_lon)`.
+  3. For each bucket, aggregate the spot stats (sum count, count
+     distinct paths, dedup modes, median distance).
+  4. For each bucket, look up the nearest `hrrr_profiles` row to the
+     midpoint at the target hour and copy the surface/refractivity
+     fields onto the sample. HRRR is pulled in one window query and
+     joined in-memory to keep the per-bucket cost a hashtable lookup.
+  5. Look up SWPC Kp once per run.
+  6. Bulk-upsert the result with `(hour_utc, band, midpoint_lat,
+     midpoint_lon)` as the conflict target.
+
+  The sampler is intentionally idempotent: re-running for the same
+  hour is a no-op except for the `updated_at` bump, so the worker
+  can safely overlap with manual reruns and the cron retry on
+  failure.
+  """
+
+  import Ecto.Query
+
+  alias Microwaveprop.Pskr.CalibrationSample
+  alias Microwaveprop.Pskr.SpotHourly
+  alias Microwaveprop.Repo
+  alias Microwaveprop.SpaceWeather
+  alias Microwaveprop.Weather.HrrrProfile
+
+  require Logger
+
+  @grid_step_deg 0.125
+  # HRRR domain is ~3 km native, so a ±0.07° (~7 km) lat/lon window
+  # plus ±60 min reliably finds at least one profile for any CONUS
+  # midpoint at most hours.
+  @hrrr_lookahead_deg 0.07
+  @hrrr_lookahead_seconds 3600
+
+  @doc """
+  Build calibration samples for one hour. Returns the upsert count.
+  Returns `0` when no PSKR spots were captured for that hour — for
+  example, during PSKR client downtime.
+  """
+  @spec build_for_hour(DateTime.t()) :: non_neg_integer()
+  def build_for_hour(%DateTime{} = hour_utc) do
+    hour_utc = align_to_hour(hour_utc)
+    spots = fetch_spots(hour_utc)
+
+    if spots == [] do
+      Logger.info("Pskr.CalibrationSampler: no spots for #{hour_utc} — skipping")
+      0
+    else
+      hrrr_index = build_hrrr_index(hour_utc)
+      kp = fetch_kp(hour_utc)
+
+      samples =
+        spots
+        |> bucket_by_cell()
+        |> Enum.map(&build_sample(&1, hour_utc, hrrr_index, kp))
+
+      upsert(samples)
+    end
+  end
+
+  # ── Spot fetch + cell bucketing ─────────────────────────────────
+
+  defp fetch_spots(hour_utc) do
+    Repo.all(
+      from(s in SpotHourly,
+        where: s.hour_utc == ^hour_utc and not is_nil(s.midpoint_lat) and not is_nil(s.midpoint_lon),
+        select: %{
+          band: s.band,
+          midpoint_lat: s.midpoint_lat,
+          midpoint_lon: s.midpoint_lon,
+          spot_count: s.spot_count,
+          distance_km: s.distance_km,
+          modes: s.modes
+        }
+      )
+    )
+  end
+
+  defp bucket_by_cell(spots) do
+    Enum.group_by(spots, fn s -> {s.band, snap(s.midpoint_lat), snap(s.midpoint_lon)} end)
+  end
+
+  defp snap(deg), do: Float.round(deg / @grid_step_deg) * @grid_step_deg
+
+  # ── HRRR feature lookup ─────────────────────────────────────────
+
+  defp build_hrrr_index(hour_utc) do
+    time_start = DateTime.add(hour_utc, -@hrrr_lookahead_seconds, :second)
+    time_end = DateTime.add(hour_utc, @hrrr_lookahead_seconds, :second)
+
+    Repo.all(
+      from(h in HrrrProfile,
+        where: h.valid_time >= ^time_start and h.valid_time <= ^time_end,
+        select: %{
+          lat: h.lat,
+          lon: h.lon,
+          surface_temp_c: h.surface_temp_c,
+          surface_dewpoint_c: h.surface_dewpoint_c,
+          pwat_mm: h.pwat_mm,
+          surface_pressure_mb: h.surface_pressure_mb,
+          min_refractivity_gradient: h.min_refractivity_gradient,
+          hpbl_m: h.hpbl_m,
+          ducting_detected: h.ducting_detected
+        }
+      )
+    )
+  end
+
+  # Pick the HRRR row whose lat/lon is closest to (lat, lon). Linear
+  # scan over the in-memory window — at the typical ~5k profile
+  # window size and ~1k cells per hour, this is faster than 1k DB
+  # round trips. Returns `nil` if no profile is in the window.
+  defp nearest_hrrr(_lat, _lon, []), do: nil
+
+  defp nearest_hrrr(lat, lon, profiles) do
+    candidates =
+      Enum.filter(profiles, fn p ->
+        abs(p.lat - lat) <= @hrrr_lookahead_deg and abs(p.lon - lon) <= @hrrr_lookahead_deg
+      end)
+
+    case candidates do
+      [] -> nil
+      list -> Enum.min_by(list, fn p -> abs(p.lat - lat) + abs(p.lon - lon) end)
+    end
+  end
+
+  # ── Kp lookup ───────────────────────────────────────────────────
+
+  # Single Kp value applies cluster-wide for the hour. We prefer the
+  # integer `kp_index` (the official 3-hour Kp) and fall back to the
+  # truncated `estimated_kp` when SWPC hasn't published the official
+  # value yet — same precedence as `Propagation.current_kp_index/0`.
+  defp fetch_kp(_hour_utc) do
+    case SpaceWeather.latest_kp() do
+      %{kp_index: kp} when is_integer(kp) -> kp
+      %{estimated_kp: kp} when is_number(kp) -> trunc(kp)
+      _ -> nil
+    end
+  end
+
+  # ── 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)
+    now = DateTime.truncate(DateTime.utc_now(), :second)
+
+    %{
+      id: Ecto.UUID.generate(),
+      hour_utc: hour_utc,
+      band: band,
+      midpoint_lat: lat,
+      midpoint_lon: lon,
+      spot_count: Enum.sum(Enum.map(cell_spots, & &1.spot_count)),
+      distinct_paths: length(cell_spots),
+      median_distance_km: median_distance(cell_spots),
+      modes: dedup_modes(cell_spots),
+      surface_temp_c: hrrr && hrrr.surface_temp_c,
+      surface_dewpoint_c: hrrr && hrrr.surface_dewpoint_c,
+      pwat_mm: hrrr && hrrr.pwat_mm,
+      surface_pressure_mb: hrrr && hrrr.surface_pressure_mb,
+      min_refractivity_gradient: hrrr && hrrr.min_refractivity_gradient,
+      hpbl_m: hrrr && hrrr.hpbl_m,
+      hrrr_ducting_detected: hrrr && hrrr.ducting_detected,
+      kp_index: kp,
+      inserted_at: now,
+      updated_at: now
+    }
+  end
+
+  defp median_distance(cell_spots) do
+    cell_spots
+    |> Enum.map(& &1.distance_km)
+    |> Enum.reject(&is_nil/1)
+    |> case do
+      [] ->
+        nil
+
+      list ->
+        sorted = Enum.sort(list)
+        n = length(sorted)
+        mid = div(n, 2)
+        if rem(n, 2) == 0, do: (Enum.at(sorted, mid - 1) + Enum.at(sorted, mid)) / 2, else: Enum.at(sorted, mid)
+    end
+  end
+
+  defp dedup_modes(cell_spots) do
+    cell_spots |> Enum.flat_map(& &1.modes) |> Enum.uniq()
+  end
+
+  # ── Upsert ──────────────────────────────────────────────────────
+
+  defp upsert([]), do: 0
+
+  defp upsert(samples) do
+    {count, _} =
+      Repo.insert_all(CalibrationSample, samples,
+        on_conflict:
+          from(c in CalibrationSample,
+            update: [
+              set: [
+                spot_count: fragment("EXCLUDED.spot_count"),
+                distinct_paths: fragment("EXCLUDED.distinct_paths"),
+                median_distance_km: fragment("EXCLUDED.median_distance_km"),
+                modes: fragment("EXCLUDED.modes"),
+                surface_temp_c: fragment("EXCLUDED.surface_temp_c"),
+                surface_dewpoint_c: fragment("EXCLUDED.surface_dewpoint_c"),
+                pwat_mm: fragment("EXCLUDED.pwat_mm"),
+                surface_pressure_mb: fragment("EXCLUDED.surface_pressure_mb"),
+                min_refractivity_gradient: fragment("EXCLUDED.min_refractivity_gradient"),
+                hpbl_m: fragment("EXCLUDED.hpbl_m"),
+                hrrr_ducting_detected: fragment("EXCLUDED.hrrr_ducting_detected"),
+                kp_index: fragment("EXCLUDED.kp_index"),
+                updated_at: fragment("EXCLUDED.updated_at")
+              ]
+            ]
+          ),
+        conflict_target: [:hour_utc, :band, :midpoint_lat, :midpoint_lon]
+      )
+
+    count
+  end
+
+  defp align_to_hour(%DateTime{} = dt) do
+    %{dt | minute: 0, second: 0, microsecond: {0, 0}}
+  end
+end

lib/microwaveprop/workers/pskr_calibration_worker.ex

@@ -0,0 +1,56 @@
+defmodule Microwaveprop.Workers.PskrCalibrationWorker do
+  @moduledoc """
+  Hourly cron that builds `Pskr.CalibrationSample` rows by joining
+  PSKR spot density with HRRR atmospheric state and SWPC space
+  weather at a given hour.
+
+  Default behaviour fires for the previous full hour — by HH:25 UTC
+  the HRRR cycle has analysis available and PSKR's hourly aggregator
+  has flushed its accumulator. Pass `"hour_utc"` in the args map to
+  rerun a specific hour:
+
+      Microwaveprop.Workers.PskrCalibrationWorker.new(%{"hour_utc" => "2026-05-04T18:00:00Z"})
+
+  The sampler is idempotent (UPSERT on `(hour_utc, band, midpoint_lat,
+  midpoint_lon)`) so reruns are safe.
+
+  Lives on the `:backfill_enqueue` queue rather than spawning a new
+  one — the workload is small (~1 query window, in-memory join,
+  bulk upsert) and cluster-wide single-slot is plenty for one fire
+  per hour.
+  """
+  use Oban.Worker,
+    queue: :backfill_enqueue,
+    max_attempts: 3,
+    unique: [
+      period: 3600,
+      states: [:available, :scheduled, :executing, :retryable]
+    ]
+
+  alias Microwaveprop.Pskr.CalibrationSampler
+
+  require Logger
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{args: args}) do
+    hour = parse_hour(args)
+    Logger.info("PskrCalibrationWorker: building samples for #{hour}")
+    count = CalibrationSampler.build_for_hour(hour)
+    Logger.info("PskrCalibrationWorker: upserted #{count} samples for #{hour}")
+    :ok
+  end
+
+  defp parse_hour(%{"hour_utc" => iso}) when is_binary(iso) do
+    {:ok, dt, _} = DateTime.from_iso8601(iso)
+    dt
+  end
+
+  defp parse_hour(_) do
+    # Default: the previous fully-completed hour.
+    DateTime.utc_now()
+    |> DateTime.add(-3600, :second)
+    |> Map.put(:minute, 0)
+    |> Map.put(:second, 0)
+    |> Map.put(:microsecond, {0, 0})
+  end
+end

priv/repo/migrations/20260504210756_create_pskr_calibration_samples.exs

@@ -0,0 +1,64 @@
+defmodule Microwaveprop.Repo.Migrations.CreatePskrCalibrationSamples do
+  use Ecto.Migration
+
+  # Calibration corpus joining PSKR spot density (truth signal) with
+  # HRRR atmospheric state and SWPC space weather (predictors). One
+  # row per (hour, band, midpoint cell). The recalibrator reads from
+  # this table and computes predicted scores on demand from the
+  # stored features — never store the prediction itself, since it
+  # depends on the algorithm version under evaluation.
+
+  def change do
+    create table(:pskr_calibration_samples, primary_key: false) do
+      add :id, :binary_id, primary_key: true
+      add :hour_utc, :utc_datetime, null: false
+      add :band, :string, null: false
+      add :midpoint_lat, :float, null: false
+      add :midpoint_lon, :float, null: false
+
+      # Aggregate PSKR signal at this cell.
+      add :spot_count, :integer, null: false, default: 0
+      add :distinct_paths, :integer, null: false, default: 0
+      add :median_distance_km, :float
+      add :modes, {:array, :string}, null: false, default: []
+
+      # HRRR atmospheric snapshot at midpoint at the hour boundary.
+      # Nullable because the join may miss for pre-2014 cells or
+      # cells outside the HRRR domain.
+      add :surface_temp_c, :float
+      add :surface_dewpoint_c, :float
+      add :pwat_mm, :float
+      add :surface_pressure_mb, :float
+      add :min_refractivity_gradient, :float
+      add :hpbl_m, :float
+      add :hrrr_ducting_detected, :boolean
+
+      # SWPC geomagnetic state at the hour boundary. Same Kp value
+      # applies to every cell in the same hour, but stored per-row so
+      # the recalibrator doesn't have to re-join SpaceWeather for
+      # every batch of samples.
+      add :kp_index, :integer
+
+      timestamps(type: :utc_datetime)
+    end
+
+    # Same uniqueness shape as `pskr_spots_hourly` so the upsert
+    # pattern translates 1:1. Cell granularity is set by the sampler
+    # (currently 0.125° to match the propagation grid).
+    create unique_index(:pskr_calibration_samples, [
+             :hour_utc,
+             :band,
+             :midpoint_lat,
+             :midpoint_lon
+           ])
+
+    create index(:pskr_calibration_samples, [:hour_utc])
+
+    # Make the spot-side spatial filter the sampler runs cheap.
+    # `pskr_spots_hourly` already has unique-index coverage on
+    # (hour_utc, band, sender_grid, receiver_grid) but no index that
+    # supports a midpoint range scan, which is the hot path for both
+    # the sampler join and any future map-side density queries.
+    create index(:pskr_spots_hourly, [:midpoint_lat, :midpoint_lon])
+  end
+end

test/microwaveprop/pskr/calibration_sampler_test.exs

@@ -0,0 +1,184 @@
+defmodule Microwaveprop.Pskr.CalibrationSamplerTest do
+  use Microwaveprop.DataCase, async: true
+
+  alias Microwaveprop.Pskr.CalibrationSample
+  alias Microwaveprop.Pskr.CalibrationSampler
+  alias Microwaveprop.Pskr.SpotHourly
+  alias Microwaveprop.Repo
+  alias Microwaveprop.SpaceWeather.GeomagneticObservation
+  alias Microwaveprop.Weather.HrrrProfile
+
+  @hour ~U[2026-05-04 18:00:00Z]
+
+  defp insert_spot!(attrs) do
+    {:ok, _} =
+      %SpotHourly{}
+      |> SpotHourly.changeset(
+        Map.merge(
+          %{
+            hour_utc: @hour,
+            band: "10000",
+            sender_grid: "EM12KL",
+            receiver_grid: "DM43ST",
+            spot_count: 1,
+            midpoint_lat: 33.0,
+            midpoint_lon: -97.0,
+            distance_km: 200.0,
+            modes: ["FT8"]
+          },
+          attrs
+        )
+      )
+      |> Repo.insert()
+  end
+
+  defp insert_hrrr!(attrs) do
+    {:ok, _} =
+      %HrrrProfile{}
+      |> HrrrProfile.changeset(
+        Map.merge(
+          %{
+            valid_time: @hour,
+            lat: 33.0,
+            lon: -97.0,
+            run_time: @hour,
+            surface_temp_c: 25.0,
+            surface_dewpoint_c: 18.0,
+            pwat_mm: 30.0,
+            surface_pressure_mb: 1013.0,
+            min_refractivity_gradient: -100.0,
+            hpbl_m: 500.0,
+            ducting_detected: false,
+            profile: []
+          },
+          attrs
+        )
+      )
+      |> Repo.insert()
+  end
+
+  defp insert_kp!(value) do
+    {:ok, _} =
+      %GeomagneticObservation{}
+      |> GeomagneticObservation.changeset(%{
+        valid_time: DateTime.add(@hour, 60, :second),
+        kp_index: value
+      })
+      |> Repo.insert()
+  end
+
+  describe "build_for_hour/1" do
+    test "skips and returns 0 when no spots exist for the hour" do
+      assert CalibrationSampler.build_for_hour(@hour) == 0
+      assert Repo.aggregate(CalibrationSample, :count) == 0
+    end
+
+    test "builds one sample per (band, snapped midpoint) cell" do
+      insert_spot!(%{midpoint_lat: 33.01, midpoint_lon: -97.02})
+      insert_spot!(%{midpoint_lat: 33.04, midpoint_lon: -97.03, sender_grid: "EM13", spot_count: 3})
+      # Distinct grid cell (>0.125° away)
+      insert_spot!(%{midpoint_lat: 35.0, midpoint_lon: -97.0, sender_grid: "EM15"})
+      insert_hrrr!(%{lat: 33.0, lon: -97.0})
+
+      assert CalibrationSampler.build_for_hour(@hour) == 2
+      [a, b] = CalibrationSample |> Repo.all() |> Enum.sort_by(& &1.midpoint_lat)
+      # Two near-midpoints collapsed; one distant cell stayed separate.
+      assert a.spot_count == 4
+      assert a.distinct_paths == 2
+      assert b.spot_count == 1
+    end
+
+    test "snaps midpoints to the 0.125° propagation grid" do
+      insert_spot!(%{midpoint_lat: 33.06, midpoint_lon: -97.07})
+      insert_hrrr!(%{lat: 33.0, lon: -97.0})
+
+      CalibrationSampler.build_for_hour(@hour)
+      [sample] = Repo.all(CalibrationSample)
+      # 33.06 → nearest 0.125° step → 33.0; -97.07 → -97.125
+      assert sample.midpoint_lat == 33.0
+      assert sample.midpoint_lon == -97.125
+    end
+
+    test "joins HRRR atmospheric features at the midpoint" do
+      insert_spot!(%{})
+
+      insert_hrrr!(%{
+        lat: 33.0,
+        lon: -97.0,
+        surface_temp_c: 22.5,
+        surface_dewpoint_c: 16.0,
+        pwat_mm: 28.0,
+        min_refractivity_gradient: -150.0,
+        hpbl_m: 420.0,
+        ducting_detected: true
+      })
+
+      CalibrationSampler.build_for_hour(@hour)
+      [sample] = Repo.all(CalibrationSample)
+      assert sample.surface_temp_c == 22.5
+      assert sample.surface_dewpoint_c == 16.0
+      assert sample.pwat_mm == 28.0
+      assert sample.min_refractivity_gradient == -150.0
+      assert sample.hpbl_m == 420.0
+      assert sample.hrrr_ducting_detected == true
+    end
+
+    test "leaves HRRR fields nil when no nearby profile exists" do
+      insert_spot!(%{})
+      # No HRRR profile inserted at all.
+      CalibrationSampler.build_for_hour(@hour)
+      [sample] = Repo.all(CalibrationSample)
+      assert sample.surface_temp_c == nil
+      assert sample.hpbl_m == nil
+      assert sample.hrrr_ducting_detected == nil
+    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})
+      insert_kp!(5)
+
+      CalibrationSampler.build_for_hour(@hour)
+      [sample] = Repo.all(CalibrationSample)
+      assert sample.kp_index == 5
+    end
+
+    test "computes median distance across the cell's path mix" do
+      insert_spot!(%{distance_km: 100.0, sender_grid: "EM10"})
+      insert_spot!(%{distance_km: 200.0, sender_grid: "EM11"})
+      insert_spot!(%{distance_km: 300.0, sender_grid: "EM12"})
+      insert_hrrr!(%{lat: 33.0, lon: -97.0})
+
+      CalibrationSampler.build_for_hour(@hour)
+      [sample] = Repo.all(CalibrationSample)
+      assert sample.median_distance_km == 200.0
+    end
+
+    test "deduplicates modes across the cell's spots" do
+      insert_spot!(%{modes: ["FT8"], sender_grid: "EM10"})
+      insert_spot!(%{modes: ["FT4", "FT8"], sender_grid: "EM11"})
+      insert_hrrr!(%{lat: 33.0, lon: -97.0})
+
+      CalibrationSampler.build_for_hour(@hour)
+      [sample] = Repo.all(CalibrationSample)
+      assert Enum.sort(sample.modes) == ["FT4", "FT8"]
+    end
+
+    test "is idempotent — re-running the same hour upserts the row" do
+      insert_spot!(%{})
+      insert_hrrr!(%{lat: 33.0, lon: -97.0})
+
+      CalibrationSampler.build_for_hour(@hour)
+      first_count = Repo.aggregate(CalibrationSample, :count)
+
+      # Mutate the spot count and re-run; the sample must update, not duplicate.
+      Repo.update_all(SpotHourly, set: [spot_count: 99])
+      CalibrationSampler.build_for_hour(@hour)
+      second_count = Repo.aggregate(CalibrationSample, :count)
+
+      assert first_count == second_count
+      [sample] = Repo.all(CalibrationSample)
+      assert sample.spot_count == 99
+    end
+  end
+end

test/microwaveprop/workers/pskr_calibration_worker_test.exs

@@ -0,0 +1,62 @@
+defmodule Microwaveprop.Workers.PskrCalibrationWorkerTest do
+  use Microwaveprop.DataCase, async: true
+  use Oban.Testing, repo: Microwaveprop.Repo
+
+  alias Microwaveprop.Pskr.CalibrationSample
+  alias Microwaveprop.Pskr.SpotHourly
+  alias Microwaveprop.Repo
+  alias Microwaveprop.Workers.PskrCalibrationWorker
+
+  defp insert_spot!(hour_utc) do
+    {:ok, _} =
+      %SpotHourly{}
+      |> SpotHourly.changeset(%{
+        hour_utc: hour_utc,
+        band: "10000",
+        sender_grid: "EM12KL",
+        receiver_grid: "DM43ST",
+        spot_count: 1,
+        midpoint_lat: 33.0,
+        midpoint_lon: -97.0,
+        distance_km: 200.0,
+        modes: ["FT8"]
+      })
+      |> Repo.insert()
+  end
+
+  test "perform/1 with explicit hour_utc samples that hour" do
+    hour = ~U[2026-05-04 18:00:00Z]
+    insert_spot!(hour)
+
+    assert :ok =
+             perform_job(PskrCalibrationWorker, %{"hour_utc" => DateTime.to_iso8601(hour)})
+
+    [sample] = Repo.all(CalibrationSample)
+    assert sample.hour_utc == hour
+  end
+
+  test "perform/1 with no args defaults to the previous full hour" do
+    # Pick the hour just before "now" — the same logic the worker
+    # uses internally — so the spot we insert lines up with what the
+    # default arg path will pull.
+    target =
+      DateTime.utc_now()
+      |> DateTime.add(-3600, :second)
+      |> Map.put(:minute, 0)
+      |> Map.put(:second, 0)
+      |> Map.put(:microsecond, {0, 0})
+
+    insert_spot!(target)
+    assert :ok = perform_job(PskrCalibrationWorker, %{})
+    assert Repo.aggregate(CalibrationSample, :count) == 1
+  end
+
+  test "perform/1 returns :ok even when no spots exist for the hour" do
+    assert :ok =
+             perform_job(PskrCalibrationWorker, %{
+               "hour_utc" => "2026-05-04T03:00:00Z"
+             })
+
+    assert Repo.aggregate(CalibrationSample, :count) == 0
+  end
+end