defmodule Microwaveprop.Propagation do @moduledoc false import Ecto.Query alias Microwaveprop.Propagation.BandConfig alias Microwaveprop.Propagation.RunTiming alias Microwaveprop.Propagation.Scorer alias Microwaveprop.Propagation.ScoreStore alias Microwaveprop.Repo # ── ML Model Lifecycle ────────────────────────────────────────────── alias Microwaveprop.Weather.SoundingParams require Logger @ml_key :propagation_ml @ml_module Microwaveprop.Propagation.Model @doc """ Loads the ML model from disk, compiles the predict function, and caches both in persistent_term. No-op if the model file doesn't exist or ML deps unavailable. """ @spec load_ml_model() :: :ok def load_ml_model do if Code.ensure_loaded?(@ml_module) do # credo:disable-for-next-line Credo.Check.Refactor.Apply case apply(@ml_module, :load, []) do {:ok, params} -> # credo:disable-for-next-line Credo.Check.Refactor.Apply predict_fn = apply(@ml_module, :compile_predict, []) :persistent_term.put(@ml_key, {predict_fn, params}) Logger.info("PropagationML: model loaded and compiled") :ok :error -> Logger.info("PropagationML: no model file found, using algorithm scorer only") :ok end else Logger.info("PropagationML: ML dependencies not available") # ── Scoring ───────────────────────────────────────────────────────── :ok end end @doc "Returns cached {predict_fn, params} tuple, or nil if not loaded." @spec ml_model() :: {function(), term()} | nil def ml_model do :persistent_term.get(@ml_key, nil) end @doc """ Score a single grid point across all bands using HRRR profile data. Uses ML model if loaded, falls back to algorithm scorer. Returns a list of %{band_mhz, score, factors} maps. """ @spec score_grid_point(map(), DateTime.t(), float(), float()) :: [%{band_mhz: non_neg_integer(), score: non_neg_integer(), factors: map()}] def score_grid_point(hrrr_profile, valid_time, latitude, longitude) do derived = derive_from_hrrr(hrrr_profile) temp_c = hrrr_profile.surface_temp_c dewpoint_c = hrrr_profile.surface_dewpoint_c # Skip points with missing or physically impossible surface data if is_nil(temp_c) or is_nil(dewpoint_c) or temp_c < -80 or temp_c > 60 or dewpoint_c < -80 or dewpoint_c > 50 do [] else score_grid_point_with_data(hrrr_profile, valid_time, temp_c, dewpoint_c, derived, latitude, longitude) end end defp score_grid_point_with_data(hrrr_profile, valid_time, temp_c, dewpoint_c, derived, latitude, longitude) do # Algorithm is the primary scorer — always used for the map score. # ML score stored in factors as :ml_score for comparison/analysis. score_with_algorithm(hrrr_profile, valid_time, temp_c, dewpoint_c, derived, latitude, longitude) end defp score_with_algorithm(hrrr_profile, valid_time, temp_c, dewpoint_c, derived, latitude, longitude) do 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, latitude: latitude, longitude: longitude, pressure_mb: hrrr_profile.surface_pressure_mb, prev_pressure_mb: nil, rain_rate_mmhr: merged_rain_rate(hrrr_profile), min_refractivity_gradient: hrrr_profile[:native_min_gradient] || derived[:min_refractivity_gradient], bl_depth_m: hrrr_profile[:hpbl_m], pwat_mm: hrrr_profile[:pwat_mm], best_duct_band_ghz: hrrr_profile[:best_duct_freq_ghz] || hrrr_profile[:best_duct_band_ghz], bulk_richardson: hrrr_profile[:bulk_richardson] } # Hoist the four band-invariant factors out of the 17-band inner # loop. time_of_day / sky / wind / pressure depend on conditions # alone, not the band — precomputing once per point drops ~30% of # the scoring wall time on the hourly chain. conditions = Map.merge(conditions, Scorer.precompute_band_invariants(conditions)) duct_info = if hrrr_profile[:duct_count] && hrrr_profile[:duct_count] > 0 do %{ duct_count: hrrr_profile[:duct_count], best_duct_freq_ghz: hrrr_profile[:best_duct_freq_ghz], max_duct_thickness_m: hrrr_profile[:max_duct_thickness_m], ducts: hrrr_profile[:ducts] || [] } end link_degradation = hrrr_profile[:commercial_link_degradation] # Kp is grid-point-invariant within a cycle. Callers populate # `:kp_index` once per scoring run (see SpaceWeather.latest_kp/0) # so we never query the DB inside the per-point band loop. Nil # means no aurora boost — the same fallback that quiet # geomagnetic conditions produce. kp_index = hrrr_profile[:kp_index] Enum.map(BandConfig.all_bands(), fn band_config -> conditions |> Scorer.composite_score(band_config) |> finalize_band_result(band_config, link_degradation, kp_index, duct_info) end) end # Apply terminal boosts and stash diagnostic factor entries. Split # out of `score_with_algorithm/7` to keep its cyclomatic complexity # within credo's per-function limit; this helper owns the # band-level conditional plumbing instead. defp finalize_band_result(result, band_config, link_degradation, kp_index, duct_info) do boosted_score = result.score |> maybe_apply_link_boost(link_degradation) |> Scorer.aurora_boost(kp_index, band_config) result |> Map.put(:score, boosted_score) |> Map.put(:band_mhz, band_config.freq_mhz) |> maybe_put_factor(:commercial_link_degradation, link_degradation) |> maybe_put_kp_factor(kp_index, band_config) |> maybe_put_factor(:duct_info, duct_info) end defp maybe_apply_link_boost(score, nil), do: score defp maybe_apply_link_boost(score, link_degradation), do: Scorer.commercial_link_boost(score, link_degradation) defp maybe_put_factor(result, _key, nil), do: result defp maybe_put_factor(result, key, value), do: put_in(result, [:factors, key], value) defp maybe_put_kp_factor(result, nil, _band_config), do: result defp maybe_put_kp_factor(result, _kp, %{freq_mhz: f}) when f > 432, do: result defp maybe_put_kp_factor(result, kp, _band_config), do: put_in(result, [:factors, :kp_index], kp) # Pick the heavier of HRRR's hourly accumulation-derived rate and NEXRAD's # reflectivity-derived rate. NEXRAD catches fast-moving convective cells that # fall between HRRR hourly analyses; HRRR catches broad stratiform rain that # NEXRAD reports as low dBZ. Taking max lets either source trigger the rain # penalty without double-counting. defp merged_rain_rate(hrrr_profile) do hrrr_rate = Scorer.precip_to_rate_mmhr(hrrr_profile[:precip_mm]) nexrad_rate = Scorer.dbz_to_rain_rate_mmhr(hrrr_profile[:nexrad_max_reflectivity_dbz]) max(hrrr_rate, nexrad_rate) end defdelegate replace_scores(scores, valid_time), to: ScoreStore defdelegate prune_old_scores(), to: ScoreStore defdelegate retain_scores_window(run_time), to: ScoreStore defdelegate available_valid_times(band_mhz), to: ScoreStore # ── Delegates to ScoreStore (file I/O + cache) ────────────────────── defdelegate hot_cache_window(), to: ScoreStore defdelegate scores_at(band_mhz, valid_time, bounds \\ nil), to: ScoreStore defdelegate scores_at_fresh(band_mhz, valid_time, bounds \\ nil), to: ScoreStore defdelegate warm_cache_and_broadcast(band_mhz, valid_time), to: ScoreStore defdelegate latest_scores(band_mhz, bounds \\ nil), to: ScoreStore defdelegate point_forecast(band_mhz, lat, lon), to: ScoreStore defdelegate point_detail(band_mhz, lat, lon, valid_time \\ nil), to: ScoreStore defdelegate latest_valid_time(), to: ScoreStore # ── Scoring helpers called from ScoreStore ────────────────────────── defdelegate latest_valid_time(band_mhz), to: ScoreStore @doc """ Rebuild the factor breakdown for a clicked grid cell by rescoring the persisted HRRR profile. Made public so ScoreStore can call it after reading the profile from disk. """ @spec factors_from_profile(non_neg_integer(), DateTime.t(), map(), float(), float()) :: map() def factors_from_profile(band_mhz, valid_time, profile, lat, lon) do profile |> Map.put(:kp_index, current_kp_index()) |> score_grid_point(valid_time, lat, lon) |> Enum.find(fn r -> r.band_mhz == band_mhz end) |> case do %{factors: factors} when is_map(factors) -> factors _ -> %{} end end # Latest geomagnetic Kp from SWPC, used by the aurora boost. Returns # nil if SpaceWeather hasn't been ingested yet (boost falls back to # quiet-conditions no-op). Prefer the integer `kp_index` over # `estimated_kp` so the threshold edges in `Scorer.aurora_boost/3` # are deterministic during the 3-hour window between official Kp # publications. defp current_kp_index do case Microwaveprop.SpaceWeather.latest_kp() do %{kp_index: kp} when is_integer(kp) -> kp %{estimated_kp: kp} when is_number(kp) -> trunc(kp) _ -> nil end # ── Run timings ───────────────────────────────────────────────────── end @doc """ Record wall-clock duration for a single forecast-hour chain step. Called by `PropagationGridWorker` at the end of every step (success or failure) so the timing history survives pod restarts and can be inspected later to see which steps are slow or flaky. """ @spec record_run_timing(map()) :: {:ok, RunTiming.t()} | {:error, Ecto.Changeset.t()} def record_run_timing(attrs) do %RunTiming{} |> RunTiming.changeset(attrs) |> Repo.insert() end @doc """ List the most-recently-started run-timing rows, newest first. Defaults to 100 rows; pass `:limit` to override. """ @spec list_recent_run_timings(keyword()) :: [RunTiming.t()] def list_recent_run_timings(opts \\ []) do limit = Keyword.get(opts, :limit, 100) RunTiming |> order_by(desc: :started_at) |> limit(^limit) |> Repo.all() end # ── Derived factors ───────────────────────────────────────────────── # Prefer the persisted scalar — `hrrr_profiles` already stored this at # ingestion time and AsosAdjustmentWorker loads 92k rows per tick without # the JSONB `profile` column to avoid a Jason.decode! storm on the DB pool. defp derive_from_hrrr(%{min_refractivity_gradient: grad}) when is_number(grad) do %{min_refractivity_gradient: grad * 1.0} end defp derive_from_hrrr(%{profile: [_, _, _ | _] = profile}) do case SoundingParams.derive(profile) do nil -> %{} derived -> %{min_refractivity_gradient: derived.min_refractivity_gradient} end end defp derive_from_hrrr(_), do: %{} end