diff --git a/config/config.exs b/config/config.exs index d9f6d7dc..04feb6b1 100644 --- a/config/config.exs +++ b/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} ]} ] diff --git a/lib/microwaveprop/propagation.ex b/lib/microwaveprop/propagation.ex index 93ed9028..9db4318c 100644 --- a/lib/microwaveprop/propagation.ex +++ b/lib/microwaveprop/propagation.ex @@ -1,28 +1,116 @@ defmodule Microwaveprop.Propagation do - @moduledoc """ - Context module for propagation scoring and forecasting. + @moduledoc false - Provides functions to retrieve propagation scores for display on the map. - Score computation and grid workers will be implemented in subsequent tasks. - """ + import Ecto.Query + + 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. - - Each score is a map with `:lat`, `:lon`, and `:score` keys. - Returns an empty list until the scoring pipeline is operational. + Score a single grid point across all bands using HRRR profile data. + Returns a list of %{band_mhz, score, factors} maps. """ - @spec latest_scores(integer()) :: [%{lat: float(), lon: float(), score: number()}] - def latest_scores(_band_freq_mhz) do - [] + def score_grid_point(hrrr_profile, valid_time) 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 """ - Returns the valid time of the most recent propagation score computation, - or nil if no scores have been computed yet. - """ - @spec latest_valid_time() :: DateTime.t() | nil + @doc "Upsert propagation scores in batches." + def upsert_scores(scores) do + now = DateTime.truncate(DateTime.utc_now(), :second) + + 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 diff --git a/lib/microwaveprop/weather/grib2/extractor.ex b/lib/microwaveprop/weather/grib2/extractor.ex index d27dcd81..c4df96d6 100644 --- a/lib/microwaveprop/weather/grib2/extractor.ex +++ b/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} -> - case extract_single_grid(msg, points) do - {:ok, point_values} -> - merged = - Enum.reduce(point_values, acc, fn {point, key, value}, grid -> - Map.update!(grid, point, &Map.put(&1, key, value)) - end) + result = + Enum.reduce_while(messages, {:ok, init_grid(points)}, fn msg, {:ok, acc} -> + case extract_single_grid(msg, points) do + {:ok, point_values} -> + merged = + Enum.reduce(point_values, acc, fn {point, key, value}, grid -> + Map.update!(grid, point, &Map.put(&1, key, value)) + end) - {:cont, {:ok, merged}} + {:cont, {:ok, merged}} - {:error, :outside_grid} -> - {:halt, {:error, :outside_grid}} + {:error, reason} -> + {:halt, {:error, reason}} + end + end) - {:error, reason} -> - {:halt, {:error, reason}} - end - end) + case result do + {:ok, grid} -> + {:ok, Map.reject(grid, fn {_point, values} -> values == %{} 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 diff --git a/lib/microwaveprop/workers/propagation_grid_worker.ex b/lib/microwaveprop/workers/propagation_grid_worker.ex new file mode 100644 index 00000000..5c182055 --- /dev/null +++ b/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 diff --git a/lib/mix/tasks/propagation_grid.ex b/lib/mix/tasks/propagation_grid.ex new file mode 100644 index 00000000..76cbce55 --- /dev/null +++ b/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 diff --git a/test/microwaveprop/propagation_test.exs b/test/microwaveprop/propagation_test.exs index b1734291..da437f1b 100644 --- a/test/microwaveprop/propagation_test.exs +++ b/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 - test "returns a list" do - assert is_list(Propagation.latest_scores(10_000)) + describe "score_grid_point/2" do + test "scores a single point for all 8 bands" do + 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 diff --git a/test/microwaveprop_web/live/map_live_test.exs b/test/microwaveprop_web/live/map_live_test.exs index 498b49e3..b3678bd7 100644 --- a/test/microwaveprop_web/live/map_live_test.exs +++ b/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