feat(weather): derive hrrr_profiles scalars from existing profile JSONB

The Rust hrrr_points worker now persists surface_refractivity /
min_refractivity_gradient / ducting_detected at upsert time (previous
commit), but the ~80k rows written before that fix have the scalar
columns NULL even though the pressure-level `profile` JSONB is
populated. Re-fetching them would re-pull tens of GB of HRRR grids
we already parsed.

Weather.backfill_hrrr_scalars/1 streams rows with
`surface_refractivity IS NULL`, runs the existing
SoundingParams.derive/1 against the stored profile, and writes the
scalars back — no network, no GRIB decode. Batched (500 rows per
tick, 1000 tick cap) to keep transactions short on the Turing Pi 2
PG node.

Meant to be run once via `bin/microwaveprop rpc` after deploy.
Idempotent; the WHERE clause excludes already-backfilled rows.
This commit is contained in:
Graham McIntire 2026-04-24 14:08:54 -05:00
parent 3f2d97735e
commit a6c5c4037d
No known key found for this signature in database
GPG key ID: F4ABF488E6029E59

View file

@ -385,6 +385,75 @@ defmodule Microwaveprop.Weather do
{:ok, n}
end
@doc """
Derive `surface_refractivity` / `min_refractivity_gradient` /
`ducting_detected` for every `hrrr_profiles` row where those columns
are NULL but the pressure-level `profile` JSONB is populated.
Required one-off after the Rust hrrr_points worker shipped without
deriving these scalars existing rows have the raw levels but the
DB scalars are NULL. Idempotent; rows with scalars already set are
skipped by the WHERE clause.
Batches by `limit` rows per pass to keep transactions short on the
Turing Pi 2 Postgres node. Returns the total number of rows advanced.
"""
@spec backfill_hrrr_scalars(keyword()) :: non_neg_integer()
def backfill_hrrr_scalars(opts \\ []) do
batch = Keyword.get(opts, :batch_size, 500)
max_batches = Keyword.get(opts, :max_batches, 1_000)
Enum.reduce_while(1..max_batches, 0, fn _, total ->
case backfill_hrrr_batch(batch) do
0 -> {:halt, total}
n -> {:cont, total + n}
end
end)
end
defp backfill_hrrr_batch(limit) do
rows =
HrrrProfile
|> where([h], is_nil(h.surface_refractivity))
|> limit(^limit)
|> select([h], %{id: h.id, profile: h.profile})
|> Repo.all()
updates =
Enum.flat_map(rows, fn %{id: id, profile: profile} ->
case SoundingParams.derive(profile || []) do
%{} = derived ->
[
%{
id: id,
surface_refractivity: Map.get(derived, :surface_refractivity),
min_refractivity_gradient: Map.get(derived, :min_refractivity_gradient),
ducting_detected: Map.get(derived, :ducting_detected, false),
duct_characteristics: Map.get(derived, :duct_characteristics)
}
]
_ ->
[]
end
end)
Enum.each(updates, fn u ->
HrrrProfile
|> where([h], h.id == ^u.id)
|> Repo.update_all(
set: [
surface_refractivity: u.surface_refractivity,
min_refractivity_gradient: u.min_refractivity_gradient,
ducting_detected: u.ducting_detected,
duct_characteristics: u.duct_characteristics,
updated_at: DateTime.truncate(DateTime.utc_now(), :second)
]
)
end)
length(updates)
end
@doc """
True if the station already has at least one surface observation
anywhere within the given UTC date. Used by the `asos_day` worker