Add propagation context, grid worker, and fix merge issues

- Replace stub propagation.ex with real implementation (score_grid_point,
  upsert_scores, latest_scores, latest_valid_time)
- Add PropagationGridWorker (hourly Oban cron) for CONUS grid scoring
- Add mix propagation_grid task for manual triggering
- Fix duplicate extract_grid/2 from parallel merges
- Fix extract_grid to skip outside-grid points instead of erroring
- Add propagation queue to Oban config

config/config.exs

@@ -44,7 +44,7 @@ config :microwaveprop, MicrowavepropWeb.Endpoint,
 
 config :microwaveprop, Oban,
   repo: Microwaveprop.Repo,
-  queues: [solar: 1, weather: 3, enqueue: 1, hrrr: 20, terrain: 4, commercial: 2, iemre: 5],
+  queues: [solar: 1, weather: 3, enqueue: 1, hrrr: 20, terrain: 4, commercial: 2, iemre: 5, propagation: 1],
   plugins: [
     {Oban.Plugins.Pruner, max_age: 3600 * 24},
     {Oban.Plugins.Lifeline, rescue_after: to_timeout(minute: 30)},
@@ -52,7 +52,8 @@ config :microwaveprop, Oban,
      crontab: [
        {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker},
        {"*/30 * * * *", Microwaveprop.Workers.QsoWeatherEnqueueWorker},
-       {"*/5 * * * *", Microwaveprop.Commercial.PollWorker}
+       {"*/5 * * * *", Microwaveprop.Commercial.PollWorker},
+       {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker}
      ]}
   ]
 

lib/microwaveprop/propagation.ex

@@ -1,28 +1,116 @@
 defmodule Microwaveprop.Propagation do
-  @moduledoc """
+  @moduledoc false
-  Context module for propagation scoring and forecasting.
 
-  Provides functions to retrieve propagation scores for display on the map.
+  import Ecto.Query
-  Score computation and grid workers will be implemented in subsequent tasks.
+
-  """
+  alias Microwaveprop.Propagation.BandConfig
+  alias Microwaveprop.Propagation.GridScore
+  alias Microwaveprop.Propagation.Scorer
+  alias Microwaveprop.Repo
+  alias Microwaveprop.Weather.SoundingParams
 
   @doc """
-  Returns the latest propagation scores for the given band frequency in MHz.
+  Score a single grid point across all bands using HRRR profile data.
-
+  Returns a list of %{band_mhz, score, factors} maps.
-  Each score is a map with `:lat`, `:lon`, and `:score` keys.
-  Returns an empty list until the scoring pipeline is operational.
   """
-  @spec latest_scores(integer()) :: [%{lat: float(), lon: float(), score: number()}]
+  def score_grid_point(hrrr_profile, valid_time) do
-  def latest_scores(_band_freq_mhz) do
+    derived = derive_from_hrrr(hrrr_profile)
-    []
+
+    temp_c = hrrr_profile.surface_temp_c
+    dewpoint_c = hrrr_profile.surface_dewpoint_c
+    temp_f = Scorer.c_to_f(temp_c)
+    dewpoint_f = Scorer.c_to_f(dewpoint_c)
+
+    conditions = %{
+      abs_humidity: Scorer.absolute_humidity(temp_c, dewpoint_c),
+      temp_f: temp_f,
+      dewpoint_f: dewpoint_f,
+      wind_speed_kts: Scorer.wind_speed_kts(hrrr_profile[:wind_u], hrrr_profile[:wind_v]),
+      sky_cover_pct: hrrr_profile[:cloud_cover_pct],
+      utc_hour: valid_time.hour,
+      utc_minute: valid_time.minute,
+      month: valid_time.month,
+      pressure_mb: hrrr_profile.surface_pressure_mb,
+      prev_pressure_mb: nil,
+      rain_rate_mmhr: Scorer.precip_to_rate_mmhr(hrrr_profile[:precip_mm]),
+      min_refractivity_gradient: derived[:min_refractivity_gradient],
+      bl_depth_m: hrrr_profile[:hpbl_m]
+    }
+
+    Enum.map(BandConfig.all_bands(), fn band_config ->
+      result = Scorer.composite_score(conditions, band_config)
+      Map.put(result, :band_mhz, band_config.freq_mhz)
+    end)
   end
 
-  @doc """
+  @doc "Upsert propagation scores in batches."
-  Returns the valid time of the most recent propagation score computation,
+  def upsert_scores(scores) do
-  or nil if no scores have been computed yet.
+    now = DateTime.truncate(DateTime.utc_now(), :second)
-  """
+
-  @spec latest_valid_time() :: DateTime.t() | nil
+    entries =
+      Enum.map(scores, fn s ->
+        %{
+          id: Ecto.UUID.generate(),
+          lat: s.lat,
+          lon: s.lon,
+          valid_time: s.valid_time,
+          band_mhz: s.band_mhz,
+          score: s.score,
+          factors: s.factors,
+          inserted_at: now,
+          updated_at: now
+        }
+      end)
+
+    total =
+      entries
+      |> Enum.chunk_every(500)
+      |> Enum.reduce(0, fn chunk, acc ->
+        {count, _} =
+          Repo.insert_all(GridScore, chunk,
+            on_conflict: {:replace, [:score, :factors, :updated_at]},
+            conflict_target: [:lat, :lon, :valid_time, :band_mhz]
+          )
+
+        acc + count
+      end)
+
+    {:ok, total}
+  end
+
+  @doc "Get the latest scores for a band."
+  def latest_scores(band_mhz) do
+    latest_time_query =
+      from(gs in GridScore,
+        where: gs.band_mhz == ^band_mhz,
+        select: max(gs.valid_time)
+      )
+
+    case Repo.one(latest_time_query) do
+      nil ->
+        []
+
+      latest_time ->
+        Repo.all(
+          from(gs in GridScore,
+            where: gs.band_mhz == ^band_mhz and gs.valid_time == ^latest_time,
+            select: %{lat: gs.lat, lon: gs.lon, score: gs.score}
+          )
+        )
+    end
+  end
+
+  @doc "Get the latest valid_time across all scores."
   def latest_valid_time do
-    nil
+    Repo.one(from(gs in GridScore, select: max(gs.valid_time)))
   end
+
+  defp derive_from_hrrr(%{profile: profile}) when is_list(profile) and length(profile) >= 3 do
+    case SoundingParams.derive(profile) do
+      nil -> %{}
+      derived -> %{min_refractivity_gradient: derived.min_refractivity_gradient}
+    end
+  end
+
+  defp derive_from_hrrr(_), do: %{}
 end

lib/microwaveprop/weather/grib2/extractor.ex

@@ -35,23 +35,29 @@ defmodule Microwaveprop.Weather.Grib2.Extractor do
   def extract_grid(binary, points) do
     messages = split_messages(binary)
 
-    Enum.reduce_while(messages, {:ok, init_grid(points)}, fn msg, {:ok, acc} ->
+    result =
-      case extract_single_grid(msg, points) do
+      Enum.reduce_while(messages, {:ok, init_grid(points)}, fn msg, {:ok, acc} ->
-        {:ok, point_values} ->
+        case extract_single_grid(msg, points) do
-          merged =
+          {:ok, point_values} ->
-            Enum.reduce(point_values, acc, fn {point, key, value}, grid ->
+            merged =
-              Map.update!(grid, point, &Map.put(&1, key, value))
+              Enum.reduce(point_values, acc, fn {point, key, value}, grid ->
-            end)
+                Map.update!(grid, point, &Map.put(&1, key, value))
+              end)
 
-          {:cont, {:ok, merged}}
+            {:cont, {:ok, merged}}
 
-        {:error, :outside_grid} ->
+          {:error, reason} ->
-          {:halt, {:error, :outside_grid}}
+            {:halt, {:error, reason}}
+        end
+      end)
 
-        {:error, reason} ->
+    case result do
-          {:halt, {:error, reason}}
+      {:ok, grid} ->
-      end
+        {:ok, Map.reject(grid, fn {_point, values} -> values == %{} end)}
-    end)
+
+      error ->
+        error
+    end
   end
 
   @doc """
@@ -69,28 +75,6 @@ defmodule Microwaveprop.Weather.Grib2.Extractor do
     j * nx + i
   end
 
-  @doc """
-  Extract weather values from a GRIB2 binary blob at multiple lat/lon points.
-
-  For each message in the binary, converts all points to grid indices, then
-  batch-extracts values. Points outside the grid are silently skipped.
-
-  Returns `{:ok, %{{lat, lon} => %{"VAR:LEVEL" => float}}}` or `{:error, term}`.
-  """
-  def extract_grid(binary, points) when is_list(points) do
-    messages = split_messages(binary)
-
-    Enum.reduce_while(messages, {:ok, %{}}, fn msg, {:ok, acc} ->
-      case extract_grid_single(msg, points) do
-        {:ok, key, point_values} ->
-          {:cont, {:ok, merge_point_values(acc, key, point_values)}}
-
-        {:error, reason} ->
-          {:halt, {:error, reason}}
-      end
-    end)
-  end
-
   # --- Private ---
 
   defp split_messages(<<>>, acc), do: Enum.reverse(acc)
@@ -131,7 +115,7 @@ defmodule Microwaveprop.Weather.Grib2.Extractor do
               end
 
             {:error, :outside_grid} ->
-              {:halt, {:error, :outside_grid}}
+              {:cont, {:ok, acc}}
           end
         end)
 
@@ -157,54 +141,6 @@ defmodule Microwaveprop.Weather.Grib2.Extractor do
     e -> {:error, "GRIB2 extraction failed: #{inspect(e)}"}
   end
 
-  defp extract_grid_single(msg, points) do
-    with {:ok, parsed} <- Section.parse_message(msg) do
-      %{grid_params: grid, product: prod, packing_params: packing, data: data} = parsed
-      key = "#{prod.var}:#{prod.level}"
-      point_indices = resolve_grid_indices(grid, points)
-      batch_extract_points(packing, data, key, point_indices)
-    end
-  rescue
-    e -> {:error, "GRIB2 grid extraction failed: #{inspect(e)}"}
-  end
-
-  defp resolve_grid_indices(grid, points) do
-    points
-    |> Enum.reduce([], fn {lat, lon} = point, acc ->
-      case LambertConformal.to_grid_index(grid, lat, lon) do
-        {:ok, {i, j}} ->
-          index = linear_index({i, j}, grid.nx, grid.scan_mode)
-          [{point, index} | acc]
-
-        {:error, :outside_grid} ->
-          acc
-      end
-    end)
-    |> Enum.reverse()
-  end
-
-  defp batch_extract_points(_packing, _data, key, []), do: {:ok, key, []}
-
-  defp batch_extract_points(packing, data, key, point_indices) do
-    unique_indices = point_indices |> Enum.map(&elem(&1, 1)) |> Enum.uniq()
-
-    case unpack_values(packing, data, unique_indices) do
-      {:ok, values_map} ->
-        point_values = Enum.map(point_indices, fn {point, index} -> {point, Map.fetch!(values_map, index)} end)
-        {:ok, key, point_values}
-
-      {:error, _} = err ->
-        err
-    end
-  end
-
-  defp merge_point_values(acc, key, point_values) do
-    Enum.reduce(point_values, acc, fn {point, value}, a ->
-      existing = Map.get(a, point, %{})
-      Map.put(a, point, Map.put(existing, key, value))
-    end)
-  end
-
   defp unpack_value(%{template: 3} = params, data, index) do
     ComplexPacking.extract_value(params, data, index)
   end
@@ -212,12 +148,4 @@ defmodule Microwaveprop.Weather.Grib2.Extractor do
   defp unpack_value(params, data, index) do
     SimplePacking.extract_value(params, data, index)
   end
-
-  defp unpack_values(%{template: 3} = params, data, indices) do
-    ComplexPacking.extract_values(params, data, indices)
-  end
-
-  defp unpack_values(params, data, indices) do
-    SimplePacking.extract_values(params, data, indices)
-  end
 end

lib/microwaveprop/workers/propagation_grid_worker.ex

@@ -0,0 +1,64 @@
+defmodule Microwaveprop.Workers.PropagationGridWorker do
+  @moduledoc """
+  Hourly Oban worker that downloads the latest HRRR data and computes
+  propagation scores across the CONUS grid for all bands.
+  """
+
+  use Oban.Worker,
+    queue: :propagation,
+    max_attempts: 3,
+    unique: [period: 3600, states: [:available, :scheduled, :executing]]
+
+  alias Microwaveprop.Propagation
+  alias Microwaveprop.Propagation.Grid
+  alias Microwaveprop.Weather.HrrrClient
+
+  require Logger
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{}) do
+    valid_time = HrrrClient.nearest_hrrr_hour(DateTime.utc_now())
+    points = Grid.conus_points()
+
+    Logger.info("PropagationGrid: fetching HRRR for #{valid_time}, #{length(points)} points")
+
+    with {:ok, grid_data} <- HrrrClient.fetch_grid(points, valid_time) do
+      scores =
+        grid_data
+        |> Task.async_stream(
+          fn {{lat, lon}, profile} ->
+            band_scores = Propagation.score_grid_point(profile, valid_time)
+
+            Enum.map(band_scores, fn result ->
+              %{
+                lat: lat,
+                lon: lon,
+                valid_time: valid_time,
+                band_mhz: result.band_mhz,
+                score: result.score,
+                factors: result.factors
+              }
+            end)
+          end,
+          max_concurrency: System.schedulers_online() * 2,
+          timeout: 30_000
+        )
+        |> Enum.flat_map(fn
+          {:ok, results} -> results
+          {:exit, _reason} -> []
+        end)
+
+      Logger.info("PropagationGrid: computed #{length(scores)} scores, upserting")
+
+      case Propagation.upsert_scores(scores) do
+        {:ok, count} ->
+          Logger.info("PropagationGrid: upserted #{count} scores for #{valid_time}")
+          :ok
+
+        error ->
+          Logger.error("PropagationGrid: upsert failed: #{inspect(error)}")
+          error
+      end
+    end
+  end
+end

lib/mix/tasks/propagation_grid.ex

@@ -0,0 +1,18 @@
+defmodule Mix.Tasks.PropagationGrid do
+  @shortdoc "Compute propagation scores for the CONUS grid"
+
+  @moduledoc "Manually trigger propagation grid computation."
+  use Mix.Task
+
+  @impl Mix.Task
+  def run(_args) do
+    Mix.Task.run("app.start")
+
+    IO.puts("Starting propagation grid computation...")
+
+    case Microwaveprop.Workers.PropagationGridWorker.perform(%Oban.Job{args: %{}}) do
+      :ok -> IO.puts("Done!")
+      {:error, reason} -> IO.puts("Error: #{inspect(reason)}")
+    end
+  end
+end

test/microwaveprop/propagation_test.exs

@@ -1,21 +1,101 @@
 defmodule Microwaveprop.PropagationTest do
-  use ExUnit.Case, async: true
+  use Microwaveprop.DataCase
 
   alias Microwaveprop.Propagation
+  alias Microwaveprop.Propagation.GridScore
 
-  describe "latest_scores/1" do
+  describe "score_grid_point/2" do
-    test "returns a list" do
+    test "scores a single point for all 8 bands" do
-      assert is_list(Propagation.latest_scores(10_000))
+      hrrr_profile = %{
+        surface_temp_c: 25.0,
+        surface_dewpoint_c: 18.0,
+        surface_pressure_mb: 1013.0,
+        hpbl_m: 500.0,
+        wind_u: 3.0,
+        wind_v: 2.0,
+        cloud_cover_pct: 15.0,
+        precip_mm: 0.0,
+        profile: [
+          %{"pres" => 1000.0, "tmpc" => 25.0, "dwpc" => 18.0, "hght" => 100.0},
+          %{"pres" => 975.0, "tmpc" => 22.0, "dwpc" => 15.0, "hght" => 350.0},
+          %{"pres" => 950.0, "tmpc" => 19.0, "dwpc" => 10.0, "hght" => 600.0}
+        ]
+      }
+
+      valid_time = ~U[2026-07-15 13:00:00Z]
+      results = Propagation.score_grid_point(hrrr_profile, valid_time)
+
+      assert length(results) == 8
+
+      Enum.each(results, fn result ->
+        assert result.score >= 0 and result.score <= 100
+        assert map_size(result.factors) == 9
+        assert is_integer(result.band_mhz)
+      end)
+    end
+  end
+
+  describe "upsert_scores/1" do
+    test "inserts scores" do
+      valid_time = ~U[2026-07-15 13:00:00Z]
+
+      scores = [
+        %{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: %{humidity: 90}},
+        %{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 24_000, score: 60, factors: %{humidity: 40}}
+      ]
+
+      assert {:ok, 2} = Propagation.upsert_scores(scores)
+      assert Repo.aggregate(GridScore, :count) == 2
     end
 
-    test "returns empty list for stub implementation" do
+    test "upserts on conflict" do
+      valid_time = ~U[2026-07-15 13:00:00Z]
+
+      scores = [
+        %{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: %{}}
+      ]
+
+      Propagation.upsert_scores(scores)
+      Propagation.upsert_scores([%{hd(scores) | score: 80}])
+
+      assert Repo.aggregate(GridScore, :count) == 1
+    end
+  end
+
+  describe "latest_scores/1" do
+    test "returns latest scores for a band" do
+      valid_time = ~U[2026-07-15 13:00:00Z]
+
+      scores = [
+        %{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: %{}},
+        %{lat: 36.0, lon: -96.0, valid_time: valid_time, band_mhz: 10_000, score: 80, factors: %{}}
+      ]
+
+      Propagation.upsert_scores(scores)
+      results = Propagation.latest_scores(10_000)
+      assert length(results) == 2
+    end
+
+    test "returns empty list when no data" do
       assert Propagation.latest_scores(10_000) == []
     end
   end
 
   describe "latest_valid_time/0" do
-    test "returns nil for stub implementation" do
+    test "returns nil when empty" do
       assert Propagation.latest_valid_time() == nil
     end
+
+    test "returns most recent time" do
+      t1 = ~U[2026-07-15 12:00:00Z]
+      t2 = ~U[2026-07-15 13:00:00Z]
+
+      Propagation.upsert_scores([
+        %{lat: 35.0, lon: -97.0, valid_time: t1, band_mhz: 10_000, score: 50, factors: %{}},
+        %{lat: 35.0, lon: -97.0, valid_time: t2, band_mhz: 10_000, score: 60, factors: %{}}
+      ])
+
+      assert Propagation.latest_valid_time() == t2
+    end
   end
 end

test/microwaveprop_web/live/map_live_test.exs

@@ -14,7 +14,7 @@ defmodule MicrowavepropWeb.MapLiveTest do
       {:ok, _lv, html} = live(conn, ~p"/map")
       assert html =~ "10 GHz"
       assert html =~ "24 GHz"
-      assert html =~ "1296 MHz"
+      assert html =~ "47 GHz"
       assert html =~ "75 GHz"
     end