From e0d99006083303165b9ffd378e89d239de114291 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 31 Mar 2026 09:10:12 -0500 Subject: [PATCH] Add real-time ASOS adjustments between hourly HRRR updates New AsosAdjustmentWorker runs every 10 minutes: - Fetches latest ASOS observations from all ~2900 US stations via IEM bulk currents API (parallel fetch across 51 state networks) - For each grid point within 75km of a reporting station, re-scores using fresh ASOS data (temp, dewpoint, wind, sky, pressure, precip) with HRRR refractivity gradient from the last hourly computation - Pushes updated scores to the map via PubSub Also stores HRRR profiles in the database during grid computation so the data persists for reference and ASOS blending. --- config/config.exs | 3 +- lib/microwaveprop/weather.ex | 18 ++ lib/microwaveprop/weather/iem_client.ex | 61 +++++ .../workers/asos_adjustment_worker.ex | 220 ++++++++++++++++++ .../workers/propagation_grid_worker.ex | 54 +++++ 5 files changed, 355 insertions(+), 1 deletion(-) create mode 100644 lib/microwaveprop/workers/asos_adjustment_worker.ex diff --git a/config/config.exs b/config/config.exs index 04feb6b1..0aa83cd0 100644 --- a/config/config.exs +++ b/config/config.exs @@ -53,7 +53,8 @@ config :microwaveprop, Oban, {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker}, {"*/30 * * * *", Microwaveprop.Workers.QsoWeatherEnqueueWorker}, {"*/5 * * * *", Microwaveprop.Commercial.PollWorker}, - {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker} + {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker}, + {"*/10 * * * *", Microwaveprop.Workers.AsosAdjustmentWorker} ]} ] diff --git a/lib/microwaveprop/weather.ex b/lib/microwaveprop/weather.ex index 19d76e16..4bd393e7 100644 --- a/lib/microwaveprop/weather.ex +++ b/lib/microwaveprop/weather.ex @@ -177,6 +177,24 @@ defmodule Microwaveprop.Weather do ) end + def upsert_hrrr_profiles_batch(profiles) do + now = DateTime.truncate(DateTime.utc_now(), :second) + + entries = + Enum.map(profiles, fn attrs -> + Map.merge(attrs, %{ + id: Ecto.UUID.generate(), + inserted_at: now, + updated_at: now + }) + end) + + Repo.insert_all(HrrrProfile, entries, + on_conflict: {:replace_all_except, [:id, :inserted_at]}, + conflict_target: [:lat, :lon, :valid_time] + ) + end + def has_hrrr_profile?(lat, lon, valid_time) do {rlat, rlon} = round_to_hrrr_grid(lat, lon) diff --git a/lib/microwaveprop/weather/iem_client.ex b/lib/microwaveprop/weather/iem_client.ex index 3a508d83..0c6f2a01 100644 --- a/lib/microwaveprop/weather/iem_client.ex +++ b/lib/microwaveprop/weather/iem_client.ex @@ -112,6 +112,67 @@ defmodule Microwaveprop.Weather.IemClient do base + jitter end + @doc """ + Fetch current ASOS observations for all stations in the given state networks. + Returns `{:ok, [%{station_code, lat, lon, temp_f, dewpoint_f, ...}]}`. + """ + def fetch_current_asos(state_networks) do + results = + state_networks + |> Task.async_stream( + fn network -> + url = "#{@iem_base}/api/1/currents.json?network=#{network}" + + case Req.get(url, req_options()) do + {:ok, %{status: 200, body: %{"data" => data}}} -> {:ok, data} + {:ok, %{status: status}} -> {:error, "IEM currents HTTP #{status}"} + {:error, reason} -> {:error, reason} + end + end, + max_concurrency: 10, + timeout: 30_000 + ) + |> Enum.flat_map(fn + {:ok, {:ok, data}} -> data + _ -> [] + end) + + observations = + Enum.flat_map(results, fn obs -> + # Skip stations with missing core data + if obs["tmpf"] && obs["dwpf"] && obs["lat"] && obs["lon"] do + [ + %{ + station_code: obs["station"], + lat: obs["lat"], + lon: obs["lon"], + utc_valid: parse_utc_valid(obs["utc_valid"]), + temp_f: obs["tmpf"], + dewpoint_f: obs["dwpf"], + wind_speed_kts: obs["sknt"], + sky_condition: obs["skyc1"], + sea_level_pressure_mb: obs["mslp"], + altimeter_setting: obs["alti"], + precip_1h_in: obs["phour"] + } + ] + else + [] + end + end) + + {:ok, observations} + end + + defp parse_utc_valid(nil), do: nil + + defp parse_utc_valid(str) when is_binary(str) do + case DateTime.from_iso8601(str) do + {:ok, dt, _} -> DateTime.truncate(dt, :second) + _ -> nil + end + end + # --- Parsers --- def parse_iemre_json(json) when is_map(json) do diff --git a/lib/microwaveprop/workers/asos_adjustment_worker.ex b/lib/microwaveprop/workers/asos_adjustment_worker.ex new file mode 100644 index 00000000..b818f956 --- /dev/null +++ b/lib/microwaveprop/workers/asos_adjustment_worker.ex @@ -0,0 +1,220 @@ +defmodule Microwaveprop.Workers.AsosAdjustmentWorker do + @moduledoc """ + Runs every 10 minutes to fetch the latest ASOS observations and re-score + grid points near reporting stations. This provides fresher data between + the hourly HRRR grid computations. + + For each grid point, finds the nearest ASOS station within 75 km. + If the station reported within the last 30 minutes, re-scores using + the ASOS observation for temp/dewpoint/wind/sky/pressure/precip and + the HRRR refractivity gradient from the last grid computation. + """ + + use Oban.Worker, queue: :propagation, max_attempts: 3 + + alias Microwaveprop.Propagation + alias Microwaveprop.Propagation.BandConfig + alias Microwaveprop.Propagation.Grid + alias Microwaveprop.Propagation.Scorer + alias Microwaveprop.Radio + alias Microwaveprop.Weather.IemClient + + require Logger + + @state_networks ~w( + AL_ASOS AK_ASOS AZ_ASOS AR_ASOS CA_ASOS CO_ASOS CT_ASOS DE_ASOS + FL_ASOS GA_ASOS HI_ASOS ID_ASOS IL_ASOS IN_ASOS IA_ASOS KS_ASOS + KY_ASOS LA_ASOS ME_ASOS MD_ASOS MA_ASOS MI_ASOS MN_ASOS MS_ASOS + MO_ASOS MT_ASOS NE_ASOS NV_ASOS NH_ASOS NJ_ASOS NM_ASOS NY_ASOS + NC_ASOS ND_ASOS OH_ASOS OK_ASOS OR_ASOS PA_ASOS RI_ASOS SC_ASOS + SD_ASOS TN_ASOS TX_ASOS UT_ASOS VT_ASOS VA_ASOS WA_ASOS WV_ASOS + WI_ASOS WY_ASOS DC_ASOS + ) + + # Only adjust grid points within this distance of an ASOS station + @max_station_distance_km 75 + # Only use observations newer than this + @max_obs_age_seconds 1800 + + @impl Oban.Worker + def perform(%Oban.Job{}) do + now = DateTime.utc_now() + cutoff = DateTime.add(now, -@max_obs_age_seconds, :second) + + Logger.info("AsosAdjustment: fetching current ASOS observations") + + with {:ok, observations} <- IemClient.fetch_current_asos(@state_networks) do + # Filter to recent observations + fresh = + Enum.filter(observations, fn obs -> + obs.utc_valid && DateTime.after?(obs.utc_valid, cutoff) + end) + + Logger.info("AsosAdjustment: #{length(fresh)} fresh observations from #{length(observations)} total") + + if fresh == [] do + :ok + else + valid_time = DateTime.truncate(now, :second) + scores = score_grid_from_asos(fresh, valid_time) + + Logger.info("AsosAdjustment: computed #{length(scores)} adjusted scores") + + case Propagation.upsert_scores(scores) do + {:ok, count} -> + Logger.info("AsosAdjustment: upserted #{count} scores") + + Phoenix.PubSub.broadcast( + Microwaveprop.PubSub, + "propagation:updated", + {:propagation_updated, valid_time} + ) + + :ok + + error -> + Logger.error("AsosAdjustment: upsert failed: #{inspect(error)}") + error + end + end + end + end + + defp score_grid_from_asos(observations, valid_time) do + # Build a spatial index of observations + obs_list = + Enum.map(observations, fn obs -> + %{ + lat: obs.lat, + lon: obs.lon, + temp_f: obs.temp_f, + dewpoint_f: obs.dewpoint_f, + wind_speed_kts: obs.wind_speed_kts, + sky_condition: obs.sky_condition, + sea_level_pressure_mb: pressure_from_obs(obs), + precip_1h_in: obs.precip_1h_in + } + end) + + # Get the last HRRR refractivity data for blending + hrrr_refractivity = load_hrrr_refractivity() + + # For each grid point, find nearest station and score + Grid.conus_points() + |> Task.async_stream( + fn {lat, lon} = point -> + nearest = find_nearest_station(obs_list, lat, lon) + + if nearest do + score_point_from_obs(point, nearest, valid_time, hrrr_refractivity) + end + end, + max_concurrency: System.schedulers_online() * 2, + timeout: 10_000 + ) + |> Enum.flat_map(fn + {:ok, nil} -> [] + {:ok, scores} -> scores + {:exit, _} -> [] + end) + end + + defp find_nearest_station(obs_list, lat, lon) do + obs_list + |> Enum.map(fn obs -> + dist = Radio.haversine_km(lat, lon, obs.lat, obs.lon) + {dist, obs} + end) + |> Enum.filter(fn {dist, _} -> dist <= @max_station_distance_km end) + |> Enum.min_by(fn {dist, _} -> dist end, fn -> nil end) + |> case do + nil -> nil + {_dist, obs} -> obs + end + end + + defp score_point_from_obs({lat, lon}, obs, valid_time, hrrr_refractivity) do + temp_c = Scorer.f_to_c(obs.temp_f) + dewpoint_c = Scorer.f_to_c(obs.dewpoint_f) + + if is_nil(temp_c) or is_nil(dewpoint_c) do + [] + else + # Get HRRR refractivity for this grid point (from last hourly computation) + refractivity = Map.get(hrrr_refractivity, {lat, lon}) + + sky_pct = sky_condition_to_pct(obs.sky_condition) + rain_rate = precip_to_rate(obs.precip_1h_in) + + conditions = %{ + abs_humidity: Scorer.absolute_humidity(temp_c, dewpoint_c), + temp_f: obs.temp_f, + dewpoint_f: obs.dewpoint_f, + wind_speed_kts: obs.wind_speed_kts, + sky_cover_pct: sky_pct, + utc_hour: valid_time.hour, + utc_minute: valid_time.minute, + month: valid_time.month, + pressure_mb: obs.sea_level_pressure_mb, + prev_pressure_mb: nil, + rain_rate_mmhr: rain_rate, + min_refractivity_gradient: refractivity, + bl_depth_m: nil + } + + Enum.map(BandConfig.all_bands(), fn band -> + result = Scorer.composite_score(conditions, band) + + %{ + lat: lat, + lon: lon, + valid_time: valid_time, + band_mhz: band.freq_mhz, + score: result.score, + factors: result.factors + } + end) + end + end + + # Load the refractivity gradient values from the last HRRR grid computation + defp load_hrrr_refractivity do + case Propagation.latest_valid_time() do + nil -> + %{} + + _time -> + # Use the 10 GHz band (arbitrary — refractivity factor is the same for all bands) + 10_000 + |> Propagation.latest_scores() + |> Enum.reduce(%{}, fn %{lat: lat, lon: lon} = _score, acc -> + # We don't store refractivity separately, so use nil (neutral score) + Map.put(acc, {lat, lon}, nil) + end) + end + end + + defp sky_condition_to_pct(nil), do: nil + defp sky_condition_to_pct("CLR"), do: 0.0 + defp sky_condition_to_pct("SKC"), do: 0.0 + defp sky_condition_to_pct("FEW"), do: 18.0 + defp sky_condition_to_pct("SCT"), do: 40.0 + defp sky_condition_to_pct("BKN"), do: 75.0 + defp sky_condition_to_pct("OVC"), do: 100.0 + defp sky_condition_to_pct("VV"), do: 100.0 + defp sky_condition_to_pct(_), do: nil + + defp pressure_from_obs(%{sea_level_pressure_mb: mslp}) when is_number(mslp), do: mslp + + defp pressure_from_obs(%{altimeter_setting: alti}) when is_number(alti) do + # Convert altimeter setting (inHg) to millibars + alti * 33.8639 + end + + defp pressure_from_obs(_), do: nil + + # Convert hourly precip accumulation (inches) to rain rate (mm/hr) + defp precip_to_rate(nil), do: 0.0 + defp precip_to_rate(inches) when inches <= 0, do: 0.0 + defp precip_to_rate(inches), do: inches * 25.4 +end diff --git a/lib/microwaveprop/workers/propagation_grid_worker.ex b/lib/microwaveprop/workers/propagation_grid_worker.ex index f322aeb8..3ce66f55 100644 --- a/lib/microwaveprop/workers/propagation_grid_worker.ex +++ b/lib/microwaveprop/workers/propagation_grid_worker.ex @@ -10,7 +10,9 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do alias Microwaveprop.Propagation alias Microwaveprop.Propagation.Grid + alias Microwaveprop.Weather alias Microwaveprop.Weather.HrrrClient + alias Microwaveprop.Weather.SoundingParams require Logger @@ -29,6 +31,9 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do result = with {:ok, grid_data} <- HrrrClient.fetch_grid(points, valid_time) do + # Store HRRR profiles in the database for future reference + store_hrrr_profiles(grid_data, valid_time) + scores = compute_scores(grid_data, valid_time) Logger.info("PropagationGrid: computed #{length(scores)} scores, upserting") @@ -51,6 +56,55 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do result end + defp store_hrrr_profiles(grid_data, valid_time) do + stored = + Enum.flat_map(grid_data, fn {{lat, lon}, profile} -> + if profile.surface_temp_c do + params = + if is_list(profile.profile) and length(profile.profile) >= 3, + do: SoundingParams.derive(profile.profile) + + attrs = %{ + valid_time: valid_time, + lat: lat, + lon: lon, + run_time: profile.run_time, + profile: profile.profile || [], + hpbl_m: profile.hpbl_m, + pwat_mm: profile.pwat_mm, + surface_temp_c: profile.surface_temp_c, + surface_dewpoint_c: profile.surface_dewpoint_c, + surface_pressure_mb: profile.surface_pressure_mb + } + + attrs = + if params do + Map.merge(attrs, %{ + surface_refractivity: params.surface_refractivity, + min_refractivity_gradient: params.min_refractivity_gradient, + ducting_detected: params.ducting_detected, + duct_characteristics: params.duct_characteristics + }) + else + attrs + end + + [attrs] + else + [] + end + end) + + # Batch upsert — skip duplicates + stored + |> Enum.chunk_every(500) + |> Enum.each(fn chunk -> + Weather.upsert_hrrr_profiles_batch(chunk) + end) + + Logger.info("PropagationGrid: stored #{length(stored)} HRRR profiles") + end + defp compute_scores(grid_data, valid_time) do grid_data |> Task.async_stream(