diff --git a/config/runtime.exs b/config/runtime.exs index 8fd05af5..e51abed8 100644 --- a/config/runtime.exs +++ b/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 = [ diff --git a/lib/microwaveprop/pskr/calibration_sample.ex b/lib/microwaveprop/pskr/calibration_sample.ex new file mode 100644 index 00000000..f9d56ded --- /dev/null +++ b/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 diff --git a/lib/microwaveprop/pskr/calibration_sampler.ex b/lib/microwaveprop/pskr/calibration_sampler.ex new file mode 100644 index 00000000..9384b8e1 --- /dev/null +++ b/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 diff --git a/lib/microwaveprop/workers/pskr_calibration_worker.ex b/lib/microwaveprop/workers/pskr_calibration_worker.ex new file mode 100644 index 00000000..246c286e --- /dev/null +++ b/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 diff --git a/priv/repo/migrations/20260504210756_create_pskr_calibration_samples.exs b/priv/repo/migrations/20260504210756_create_pskr_calibration_samples.exs new file mode 100644 index 00000000..ac4eb7be --- /dev/null +++ b/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 diff --git a/test/microwaveprop/pskr/calibration_sampler_test.exs b/test/microwaveprop/pskr/calibration_sampler_test.exs new file mode 100644 index 00000000..63dfb240 --- /dev/null +++ b/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 diff --git a/test/microwaveprop/workers/pskr_calibration_worker_test.exs b/test/microwaveprop/workers/pskr_calibration_worker_test.exs new file mode 100644 index 00000000..b5bc6dc2 --- /dev/null +++ b/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