Graham McIntire
7a2b1f292c
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.
56 lines
1.8 KiB
Elixir
56 lines
1.8 KiB
Elixir
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
|