diff --git a/lib/microwaveprop/weather.ex b/lib/microwaveprop/weather.ex index cd005add..4be15ba5 100644 --- a/lib/microwaveprop/weather.ex +++ b/lib/microwaveprop/weather.ex @@ -362,6 +362,31 @@ defmodule Microwaveprop.Weather do %{surface_observations: surface_observations, soundings: soundings} end + @sounding_search_radii_km [150, 300, 600, 1000] + + @doc """ + Search for soundings in widening radii around the given location, stopping + at the first radius that returns any. Returns `%{soundings, radius_km, + exhausted}` where `exhausted: true` means the widest radius also came up + empty — the caller should trigger a fetch for missing data at that point. + """ + @spec soundings_with_widening_radius(map()) :: %{ + soundings: [Sounding.t()], + radius_km: pos_integer(), + exhausted: boolean() + } + def soundings_with_widening_radius(params) do + Enum.reduce_while(@sounding_search_radii_km, nil, fn radius_km, _acc -> + result = weather_for_contact(params, radius_km: radius_km) + + if result.soundings == [] do + {:cont, %{soundings: [], radius_km: radius_km, exhausted: true}} + else + {:halt, %{soundings: result.soundings, radius_km: radius_km, exhausted: false}} + end + end) + end + @spec latest_grid_valid_time() :: DateTime.t() | nil def latest_grid_valid_time do cond do diff --git a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex index cd881ed0..23c487ee 100644 --- a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex +++ b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex @@ -391,23 +391,74 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do end defp build_raob_jobs(lat, lon, timestamp) do + build_raob_jobs(lat, lon, timestamp, @sounding_radius_km, nil) + end + + defp build_raob_jobs(lat, lon, timestamp, radius_km, contact_id) do sounding_times = Weather.sounding_times_around(timestamp) - stations = Weather.nearby_stations(lat, lon, "sounding", @sounding_radius_km) + stations = Weather.nearby_stations(lat, lon, "sounding", radius_km) station_ids = Enum.map(stations, & &1.id) covered = Weather.station_ids_with_soundings(station_ids, sounding_times) for station <- stations, sounding_time <- sounding_times, not MapSet.member?(covered, {station.id, sounding_time}) do - WeatherFetchWorker.new(%{ - "fetch_type" => "raob", - "station_id" => station.id, - "station_code" => station.station_code, - "sounding_time" => DateTime.to_iso8601(sounding_time) - }) + args = + Map.merge( + %{ + "fetch_type" => "raob", + "station_id" => station.id, + "station_code" => station.station_code, + "sounding_time" => DateTime.to_iso8601(sounding_time) + }, + if(contact_id, do: %{"contact_id" => contact_id}, else: %{}) + ) + + WeatherFetchWorker.new(args) end end + @doc """ + Enqueue RAOB fetch jobs for the given contact at a progressively wider + radius until the radius yields at least one sounding station. Passes + the contact id through so `WeatherFetchWorker` can PubSub the contact + detail page when each sounding lands. + + Returns `{:ok, %{radius_km, jobs_enqueued}}`, or `{:error, :no_stations}` + when even 1000 km finds nothing. + """ + @spec enqueue_raob_fetch_with_widening(Microwaveprop.Radio.Contact.t()) :: + {:ok, %{radius_km: pos_integer(), jobs_enqueued: non_neg_integer()}} + | {:error, :no_stations} + | {:error, :no_coords} + def enqueue_raob_fetch_with_widening(contact) do + case Radio.contact_path_points(contact) do + [] -> + {:error, :no_coords} + + [{lat, lon} | _] -> + do_widening_raob(lat, lon, contact) + end + end + + defp do_widening_raob(lat, lon, contact) do + Enum.reduce_while([150, 300, 600, 1000], {:error, :no_stations}, fn radius, _acc -> + stations = Weather.nearby_stations(lat, lon, "sounding", radius) + + if stations == [] do + {:cont, {:error, :no_stations}} + else + {:halt, enqueue_raob_at_radius(lat, lon, contact, radius)} + end + end) + end + + defp enqueue_raob_at_radius(lat, lon, contact, radius) do + jobs = build_raob_jobs(lat, lon, contact.qso_timestamp, radius, contact.id) + if jobs != [], do: insert_all_chunked(jobs) + {:ok, %{radius_km: radius, jobs_enqueued: length(jobs)}} + end + defp insert_all_chunked(jobs) do jobs |> Enum.chunk_every(@insert_batch_size) diff --git a/lib/microwaveprop/workers/weather_fetch_worker.ex b/lib/microwaveprop/workers/weather_fetch_worker.ex index 3a98c72c..5edbb2fe 100644 --- a/lib/microwaveprop/workers/weather_fetch_worker.ex +++ b/lib/microwaveprop/workers/weather_fetch_worker.ex @@ -48,6 +48,8 @@ defmodule Microwaveprop.Workers.WeatherFetchWorker do "sounding_time" => sounding_time_str } = args + contact_id = Map.get(args, "contact_id") + {:ok, sounding_time, _} = DateTime.from_iso8601(sounding_time_str) case Repo.get(Station, station_id) do @@ -56,7 +58,7 @@ defmodule Microwaveprop.Workers.WeatherFetchWorker do :ok station -> - fetch_raob(station, station_id, station_code, sounding_time, sounding_time_str) + fetch_raob(station, station_id, station_code, sounding_time, sounding_time_str, contact_id) end end @@ -113,24 +115,26 @@ defmodule Microwaveprop.Workers.WeatherFetchWorker do # -- RAOB helpers -- - defp fetch_raob(station, station_id, station_code, sounding_time, sounding_time_str) do + defp fetch_raob(station, station_id, station_code, sounding_time, sounding_time_str, contact_id) do if Weather.has_sounding?(station.id, sounding_time) do Logger.debug("WeatherFetch RAOB: #{station_code} already has sounding for #{sounding_time_str}") + broadcast_sounding_fetched(station_id, contact_id) :ok else Logger.info("WeatherFetch RAOB: fetching #{station_code} for #{sounding_time_str}") - do_fetch_raob(station, station_id, station_code, sounding_time, sounding_time_str) + do_fetch_raob(station, station_id, station_code, sounding_time, sounding_time_str, contact_id) end end - defp do_fetch_raob(station, station_id, station_code, sounding_time, sounding_time_str) do + defp do_fetch_raob(station, station_id, station_code, sounding_time, sounding_time_str, contact_id) do case IemClient.fetch_raob(station_code, sounding_time) do {:ok, [parsed | _]} -> - ingest_raob(station, station_id, station_code, parsed) + ingest_raob(station, station_id, station_code, parsed, contact_id) {:ok, []} -> Weather.upsert_sounding(station, %{observed_at: sounding_time, profile: [], level_count: 0}) Logger.info("WeatherFetch RAOB: #{station_code} no data for #{sounding_time_str}, stored stub") + broadcast_sounding_fetched(station_id, contact_id) :ok {:error, reason} -> @@ -139,18 +143,30 @@ defmodule Microwaveprop.Workers.WeatherFetchWorker do end end - defp ingest_raob(station, station_id, station_code, parsed) do + defp ingest_raob(station, station_id, station_code, parsed, contact_id) do sounding_attrs = build_sounding_attrs(parsed) Weather.upsert_sounding(station, sounding_attrs) Logger.info("WeatherFetch RAOB: #{station_code} ingested sounding with #{length(parsed.profile)} levels") + broadcast_sounding_fetched(station_id, contact_id) + + :ok + end + + defp broadcast_sounding_fetched(station_id, contact_id) do Phoenix.PubSub.broadcast( Microwaveprop.PubSub, "contact_enrichment:weather", {:weather_ready, %{station_id: station_id}} ) - :ok + if contact_id do + Phoenix.PubSub.broadcast( + Microwaveprop.PubSub, + "contact_enrichment:#{contact_id}", + {:sounding_fetched, %{station_id: station_id, contact_id: contact_id}} + ) + end end defp build_sounding_attrs(parsed) do diff --git a/lib/microwaveprop_web/live/contact_live/show.ex b/lib/microwaveprop_web/live/contact_live/show.ex index ed2b356f..b71b8d31 100644 --- a/lib/microwaveprop_web/live/contact_live/show.ex +++ b/lib/microwaveprop_web/live/contact_live/show.ex @@ -42,6 +42,8 @@ defmodule MicrowavepropWeb.ContactLive.Show do can_enqueue: can_enqueue, surface_observations: [], soundings: [], + sounding_search_radius_km: nil, + sounding_search_pending?: false, solar: nil, hrrr: nil, hrrr_path: [], @@ -364,9 +366,12 @@ defmodule MicrowavepropWeb.ContactLive.Show do |> assign( contact: contact, surface_observations: weather.surface_observations, - soundings: weather.soundings + soundings: weather.soundings, + sounding_search_radius_km: Map.get(weather, :sounding_search_radius_km), + sounding_search_pending?: false ) |> mark_hydrated(:weather) + |> maybe_kick_off_sounding_fetch(weather) |> refresh_computed() {:noreply, socket} @@ -447,6 +452,26 @@ defmodule MicrowavepropWeb.ContactLive.Show do end @impl true + def handle_info({:sounding_fetched, %{contact_id: contact_id}}, socket) do + if socket.assigns.contact.id == contact_id do + contact = socket.assigns.contact + weather = load_weather(contact) + + socket = + socket + |> assign( + soundings: weather.soundings, + sounding_search_radius_km: weather.sounding_search_radius_km, + sounding_search_pending?: weather.soundings == [] + ) + |> refresh_computed() + + {:noreply, socket} + else + {:noreply, socket} + end + end + def handle_info({:terrain_ready, _contact_id}, socket) do contact = %{socket.assigns.contact | terrain_status: :complete} terrain = Terrain.get_terrain_profile(contact.id) @@ -464,6 +489,39 @@ defmodule MicrowavepropWeb.ContactLive.Show do )} end + # Soundings are sparse, so even the 1000 km widening search can come up + # empty for a given contact when no RAOB data has been ingested yet. + # When that happens, enqueue fetches for stations at the widest radius + # that has any; the WeatherFetchWorker broadcasts back via + # `contact_enrichment:` so this LiveView rehydrates when + # each sounding lands. + defp maybe_kick_off_sounding_fetch(socket, weather) do + contact = socket.assigns.contact + + cond do + not socket.assigns.can_enqueue -> + socket + + not Map.get(weather, :sounding_search_exhausted?, false) -> + socket + + is_nil(contact.pos1) -> + socket + + true -> + case ContactWeatherEnqueueWorker.enqueue_raob_fetch_with_widening(contact) do + {:ok, %{jobs_enqueued: jobs, radius_km: radius}} when jobs > 0 -> + assign(socket, + sounding_search_pending?: true, + sounding_search_radius_km: radius + ) + + _ -> + socket + end + end + end + defp maybe_enqueue_weather(weather, contact) do has_data = weather.surface_observations != [] or weather.soundings != [] @@ -582,7 +640,12 @@ defmodule MicrowavepropWeb.ContactLive.Show do fetch_weather_for_path(contact, lat1, lon1, lat2, lon2) :error -> - %{surface_observations: [], soundings: []} + %{ + surface_observations: [], + soundings: [], + sounding_search_exhausted?: false, + sounding_search_radius_km: nil + } end end @@ -611,9 +674,22 @@ defmodule MicrowavepropWeb.ContactLive.Show do radius_km: radius ) + # The surface-observation radius tracks the path geometry, but + # soundings are sparse (~120/day nationwide) so we widen out to + # 1000 km before giving up and letting the fetch-on-demand path + # (handle_async(:soundings_widen)) backfill from IEM. + sounding_search = + Weather.soundings_with_widening_radius(%{ + lat: mid_lat, + lon: mid_lon, + timestamp: contact.qso_timestamp + }) + %{ surface_observations: closest_observations(weather.surface_observations, mid_lat, mid_lon, 5), - soundings: weather.soundings + soundings: sounding_search.soundings, + sounding_search_exhausted?: sounding_search.exhausted, + sounding_search_radius_km: sounding_search.radius_km } end @@ -1149,13 +1225,18 @@ defmodule MicrowavepropWeb.ContactLive.Show do Loading soundings… + <% @sounding_search_pending? -> %> + + + Fetching soundings from IEM (search radius {@sounding_search_radius_km} km)… + <% @contact.weather_status == :queued -> %> Fetching sounding data from nearby stations {queue_info(@queue_counts, "weather")} <% true -> %> - No soundings found nearby. + No soundings found within 1000 km. <% end %> <% else %> diff --git a/test/microwaveprop/weather_test.exs b/test/microwaveprop/weather_test.exs index ff287d72..6056c6c6 100644 --- a/test/microwaveprop/weather_test.exs +++ b/test/microwaveprop/weather_test.exs @@ -203,6 +203,53 @@ defmodule Microwaveprop.WeatherTest do end end + describe "soundings_with_widening_radius/1" do + test "returns soundings at the first radius that finds any" do + # Station ~220 km east of the query point: outside the 150 km ring + # but inside the 300 km ring. + {:ok, far_station} = + Weather.find_or_create_station(%{ + station_code: "FAR", + station_type: "sounding", + wmo_number: 72_001, + name: "Far Site", + lat: 32.9, + lon: -94.6 + }) + + Weather.upsert_sounding(far_station, %{ + observed_at: ~U[2026-03-28 12:00:00Z], + profile: [%{"pres" => 1013, "hght" => 100, "tmpc" => 20, "dwpc" => 10}], + level_count: 1 + }) + + result = + Weather.soundings_with_widening_radius(%{ + lat: 32.9, + lon: -97.0, + timestamp: ~U[2026-03-28 14:00:00Z] + }) + + assert result.soundings != [] + # 130 km is outside 150 but inside 300 + assert result.radius_km == 300 + refute result.exhausted + end + + test "reports exhausted when no soundings exist within the widest radius" do + result = + Weather.soundings_with_widening_radius(%{ + lat: 32.9, + lon: -97.0, + timestamp: ~U[2026-03-28 14:00:00Z] + }) + + assert result.soundings == [] + assert result.exhausted + assert result.radius_km == 1000 + end + end + describe "has_surface_observations?/3" do test "returns true when observations exist in the time window" do {:ok, station} = Weather.find_or_create_station(@station_attrs)