From a6c5c4037d53a451cc8ee6d383e27375c0c6ea88 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Fri, 24 Apr 2026 14:08:54 -0500 Subject: [PATCH] feat(weather): derive hrrr_profiles scalars from existing profile JSONB MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- lib/microwaveprop/weather.ex | 69 ++++++++++++++++++++++++++++++++++++ 1 file changed, 69 insertions(+) diff --git a/lib/microwaveprop/weather.ex b/lib/microwaveprop/weather.ex index 2e211f27..bab5bed9 100644 --- a/lib/microwaveprop/weather.ex +++ b/lib/microwaveprop/weather.ex @@ -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