From 434fca2ad1a15d91a821885154ab030e354c1b71 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 14 Apr 2026 11:12:51 -0500 Subject: [PATCH] Show forecast-hour progress on the map pipeline chip MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PropagationGridWorker now broadcasts a {:propagation_pipeline_progress, %{forecast_hour:, valid_time:}} message on the propagation:pipeline PubSub topic at the start of each process_forecast_hour/4 call. MapLive subscribes on mount, stashes the last message in the new :pipeline_progress assign, and clears it on the refresh tick whenever PipelineStatus flips out of :running. The chip component consults a new PipelineStatus.running_label/1 helper that formats the payload as "Updating propagation · now" for f00 and "Updating propagation · +Nh" for f01-f18, falling back to the base running label until the first progress message arrives. Covered by unit tests on the formatter and an integration test in MapLiveTest that seeds an executing grid job, broadcasts progress, and asserts the rendered chip. --- .../propagation/pipeline_status.ex | 22 ++++++-- .../workers/propagation_grid_worker.ex | 8 +++ lib/microwaveprop_web/live/map_live.ex | 52 ++++++++++++++++--- .../propagation/pipeline_status_test.exs | 20 +++++++ test/microwaveprop_web/live/map_live_test.exs | 46 +++++++++++++++- 5 files changed, 135 insertions(+), 13 deletions(-) diff --git a/lib/microwaveprop/propagation/pipeline_status.ex b/lib/microwaveprop/propagation/pipeline_status.ex index 13a923a4..cffe5751 100644 --- a/lib/microwaveprop/propagation/pipeline_status.ex +++ b/lib/microwaveprop/propagation/pipeline_status.ex @@ -36,13 +36,27 @@ defmodule Microwaveprop.Propagation.PipelineStatus do last_update_at: DateTime.t() | nil } + @doc """ + Format a forecast-progress payload (as broadcast by + `PropagationGridWorker`) into a short chip label for the map. + + Returns `nil` when the payload has no `forecast_hour` so callers can + fall back to the base running label from `current/0`. + """ + @spec running_label(map() | nil) :: String.t() | nil + def running_label(%{forecast_hour: 0}), do: "Updating propagation · now" + + def running_label(%{forecast_hour: fh}) when is_integer(fh) and fh > 0, do: "Updating propagation · +#{fh}h" + + def running_label(_), do: nil + @spec current() :: t() def current do case running_worker() do worker when is_binary(worker) -> %{ state: :running, - label: running_label(worker), + label: base_running_label(worker), last_update_at: latest_completed_at() } @@ -97,9 +111,9 @@ defmodule Microwaveprop.Propagation.PipelineStatus do end end - defp running_label(@grid_worker), do: "Updating propagation (HRRR run)" - defp running_label(@asos_worker), do: "Updating propagation (ASOS nudge)" - defp running_label(_), do: "Updating propagation" + defp base_running_label(@grid_worker), do: "Updating propagation (HRRR run)" + defp base_running_label(@asos_worker), do: "Updating propagation (ASOS nudge)" + defp base_running_label(_), do: "Updating propagation" defp format_age(0), do: "just now" defp format_age(1), do: "1m" diff --git a/lib/microwaveprop/workers/propagation_grid_worker.ex b/lib/microwaveprop/workers/propagation_grid_worker.ex index 5405accb..549e2104 100644 --- a/lib/microwaveprop/workers/propagation_grid_worker.ex +++ b/lib/microwaveprop/workers/propagation_grid_worker.ex @@ -84,6 +84,14 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do defp process_forecast_hour(points, run_time, forecast_hour, valid_time) do label = "f#{String.pad_leading(Integer.to_string(forecast_hour), 2, "0")}" + # Emit progress so the map page's pipeline status chip can show + # which forecast hour is currently being scored ("now", "+1h", …). + Phoenix.PubSub.broadcast( + Microwaveprop.PubSub, + "propagation:pipeline", + {:propagation_pipeline_progress, %{forecast_hour: forecast_hour, valid_time: valid_time}} + ) + case timed(label, fn -> HrrrClient.fetch_grid(points, run_time, forecast_hour: forecast_hour) end) do diff --git a/lib/microwaveprop_web/live/map_live.ex b/lib/microwaveprop_web/live/map_live.ex index 6e720a27..20b8ce62 100644 --- a/lib/microwaveprop_web/live/map_live.ex +++ b/lib/microwaveprop_web/live/map_live.ex @@ -24,6 +24,7 @@ defmodule MicrowavepropWeb.MapLive do def mount(_params, session, socket) do if connected?(socket) do Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "propagation:updated") + Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "propagation:pipeline") Process.send_after(self(), :refresh_pipeline_status, @pipeline_status_refresh_ms) end @@ -31,6 +32,7 @@ defmodule MicrowavepropWeb.MapLive do socket |> assign(:initial_utc_clock, Calendar.strftime(DateTime.utc_now(), "%H:%M UTC")) |> assign(:pipeline_status, PipelineStatus.current()) + |> assign(:pipeline_progress, nil) # LiveStash only persists the keys passed to `stash_assigns/2` # (selected_band + selected_time). On reconnect the recovered socket has @@ -283,7 +285,24 @@ defmodule MicrowavepropWeb.MapLive do def handle_info(:refresh_pipeline_status, socket) do Process.send_after(self(), :refresh_pipeline_status, @pipeline_status_refresh_ms) - {:noreply, assign(socket, :pipeline_status, PipelineStatus.current())} + + new_status = PipelineStatus.current() + + progress = + if new_status.state == :running do + socket.assigns[:pipeline_progress] + end + + socket = + socket + |> assign(:pipeline_status, new_status) + |> assign(:pipeline_progress, progress) + + {:noreply, socket} + end + + def handle_info({:propagation_pipeline_progress, progress}, socket) do + {:noreply, assign(socket, :pipeline_progress, progress)} end @impl true @@ -362,14 +381,17 @@ defmodule MicrowavepropWeb.MapLive do attr :id, :string, required: true attr :status, :map, required: true + attr :progress, :any, default: nil defp pipeline_status_chip(assigns) do + assigns = assign(assigns, :display_label, chip_display_label(assigns.status, assigns.progress)) + ~H"""
<%= case @status.state do %> <% :running -> %> @@ -377,21 +399,27 @@ defmodule MicrowavepropWeb.MapLive do name="hero-arrow-path" class="size-3.5 shrink-0 mt-0.5 text-info motion-safe:animate-spin" /> - {@status.label} + {@display_label} <% :idle -> %> - {@status.label} + {@display_label} <% :stale -> %> - {@status.label} + {@display_label} <% :unknown -> %> - {@status.label} + {@display_label} <% end %>
""" end + defp chip_display_label(%{state: :running} = status, progress) do + PipelineStatus.running_label(progress) || status.label + end + + defp chip_display_label(status, _progress), do: status.label + @impl true def render(assigns) do ~H""" @@ -453,7 +481,11 @@ defmodule MicrowavepropWeb.MapLive do - <.pipeline_status_chip id="pipeline-status-mobile" status={@pipeline_status} /> + <.pipeline_status_chip + id="pipeline-status-mobile" + status={@pipeline_status} + progress={@pipeline_progress} + />
@@ -723,7 +755,11 @@ defmodule MicrowavepropWeb.MapLive do
- <.pipeline_status_chip id="pipeline-status-desktop" status={@pipeline_status} /> + <.pipeline_status_chip + id="pipeline-status-desktop" + status={@pipeline_status} + progress={@pipeline_progress} + />
diff --git a/test/microwaveprop/propagation/pipeline_status_test.exs b/test/microwaveprop/propagation/pipeline_status_test.exs index 47397eff..29b9979c 100644 --- a/test/microwaveprop/propagation/pipeline_status_test.exs +++ b/test/microwaveprop/propagation/pipeline_status_test.exs @@ -140,4 +140,24 @@ defmodule Microwaveprop.Propagation.PipelineStatusTest do assert status.state == :unknown end end + + describe "running_label/1" do + test "forecast hour 0 renders as 'now'" do + assert PipelineStatus.running_label(%{forecast_hour: 0}) == + "Updating propagation · now" + end + + test "positive forecast hours render as +Nh" do + assert PipelineStatus.running_label(%{forecast_hour: 1}) == + "Updating propagation · +1h" + + assert PipelineStatus.running_label(%{forecast_hour: 18}) == + "Updating propagation · +18h" + end + + test "returns nil when no forecast hour is provided" do + assert PipelineStatus.running_label(nil) == nil + assert PipelineStatus.running_label(%{}) == nil + end + end end diff --git a/test/microwaveprop_web/live/map_live_test.exs b/test/microwaveprop_web/live/map_live_test.exs index 2c6336f4..63a03e29 100644 --- a/test/microwaveprop_web/live/map_live_test.exs +++ b/test/microwaveprop_web/live/map_live_test.exs @@ -1,5 +1,5 @@ defmodule MicrowavepropWeb.MapLiveTest do - use MicrowavepropWeb.ConnCase, async: true + use MicrowavepropWeb.ConnCase, async: false import Phoenix.LiveViewTest @@ -107,5 +107,49 @@ defmodule MicrowavepropWeb.MapLiveTest do # With a clean test DB no pipeline workers have ever run. assert html =~ ~s(data-pipeline-state="unknown") end + + test "shows the current forecast hour while PropagationGridWorker runs", %{conn: conn} do + # Put a grid-worker row in executing state so PipelineStatus.current/0 + # reports :running. Bypassing Oban's normal insert keeps the SQL + # sandbox from running the worker inline. + now = DateTime.utc_now() + + Microwaveprop.Repo.insert!( + struct!(Oban.Job, %{ + state: "executing", + queue: "propagation", + worker: "Microwaveprop.Workers.PropagationGridWorker", + args: %{}, + attempt: 1, + max_attempts: 3, + inserted_at: now, + scheduled_at: now, + attempted_at: now + }) + ) + + {:ok, lv, html} = live(conn, ~p"/map") + # Before any progress message arrives the chip uses the base label. + assert html =~ "Updating propagation" + assert html =~ ~s(data-pipeline-state="running") + + Phoenix.PubSub.broadcast( + Microwaveprop.PubSub, + "propagation:pipeline", + {:propagation_pipeline_progress, %{forecast_hour: 3, valid_time: ~U[2026-04-14 18:00:00Z]}} + ) + + html = render(lv) + assert html =~ "+3h" + + Phoenix.PubSub.broadcast( + Microwaveprop.PubSub, + "propagation:pipeline", + {:propagation_pipeline_progress, %{forecast_hour: 0, valid_time: ~U[2026-04-14 15:00:00Z]}} + ) + + html = render(lv) + assert html =~ "now" + end end end