4ac606e682
Close 6 cleanup gaps on the shared /data NFS mount:
- HRDPS .hrdps.prop score files: parse_valid_time regex now matches
the .hrdps.prop extension, so these files are found and pruned
- HRDPS scalar dirs (weather_scalars/{iso}.hrdps/): strip .hrdps suffix
before DateTime.from_iso8601 in list_valid_time_dirs
- ScalarFile: prune_older_than was defined but never called from
Propagation.prune_old_scores - now called alongside Scores/Profiles
- Orphan .tmp.* files: sweep all three base dirs (scores/profiles/scalars)
for .tmp.* files left by crashed atomic-write processes
- Building cache (/data/buildings/): add MsFootprints.prune_older_than
with 30-day mtime cutoff
- Canopy cache (/data/canopy/ + staging/): add Canopy.prune_older_than
with 30-day cutoff, cleanup empty staging dir
All cleanup is cutoff-based (older than X time) so if the prune worker
is missed for any period it catches up fully on the next run.
PropagationPruneWorker updated to call all new cleanup functions.
719 lines
26 KiB
Elixir
719 lines
26 KiB
Elixir
defmodule Microwaveprop.Propagation do
|
||
@moduledoc false
|
||
|
||
import Ecto.Query
|
||
|
||
alias Microwaveprop.Propagation.BandConfig
|
||
alias Microwaveprop.Propagation.Grid
|
||
alias Microwaveprop.Propagation.ProfilesFile
|
||
alias Microwaveprop.Propagation.RunTiming
|
||
alias Microwaveprop.Propagation.ScoreCache
|
||
alias Microwaveprop.Propagation.Scorer
|
||
alias Microwaveprop.Propagation.ScoresFile
|
||
alias Microwaveprop.Repo
|
||
alias Microwaveprop.Weather.ScalarFile
|
||
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
|
||
case @ml_module.load() do
|
||
{:ok, params} ->
|
||
predict_fn = @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")
|
||
: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
|
||
|
||
@doc """
|
||
Replace every propagation score for `valid_time` with `scores`.
|
||
|
||
Used by `PropagationGridWorker` on the hot path. Scores are written
|
||
as binary files on disk via `ScoresFile.write!/3`, one file per
|
||
band.
|
||
|
||
Consumes `scores` in a single streaming pass that folds each score
|
||
straight into a per-band accumulator. Previously this function ran
|
||
`Enum.to_list/1` followed by `Enum.group_by/2`, which held two full
|
||
copies of the ~460k-entry grid (list + grouped list) in memory at
|
||
once — the hot path's largest transient spike after native-duct
|
||
merge. The single-pass reduce keeps only one copy and buys back
|
||
~100 MB of headroom per forecast-hour step.
|
||
"""
|
||
@spec replace_scores(Enumerable.t(), DateTime.t()) :: {:ok, non_neg_integer()} | {:error, term()}
|
||
def replace_scores(scores, %DateTime{} = valid_time) do
|
||
do_replace_scores(scores, valid_time)
|
||
end
|
||
|
||
defp do_replace_scores(scores, valid_time) do
|
||
# Pure grouping phase runs outside the telemetry span — typically
|
||
# <10ms on small result sets, and the span's two dispatches
|
||
# (~100µs each) would otherwise dominate. The span now wraps only
|
||
# the per-band writes, which is where the actual DB cost lives.
|
||
{per_band, total} =
|
||
Enum.reduce(scores, {%{}, 0}, fn score, {acc, count} ->
|
||
{Map.update(acc, score.band_mhz, [score], &[score | &1]), count + 1}
|
||
end)
|
||
|
||
Microwaveprop.Instrument.span(
|
||
[:db, :replace_scores],
|
||
%{valid_time: valid_time},
|
||
fn ->
|
||
Enum.each(per_band, fn {band_mhz, band_scores} ->
|
||
try do
|
||
ScoresFile.write!(band_mhz, valid_time, band_scores)
|
||
rescue
|
||
e ->
|
||
Logger.warning("Propagation: ScoresFile write failed for band=#{band_mhz} vt=#{valid_time}: #{inspect(e)}")
|
||
end
|
||
end)
|
||
|
||
{:ok, total}
|
||
end
|
||
)
|
||
end
|
||
|
||
@doc """
|
||
Remove score files with valid_times older than 3 hours. Called on
|
||
a cron by `Microwaveprop.Workers.PropagationPruneWorker`.
|
||
|
||
The cutoff sits one hour beyond HRRR's ~2h publish lag: the hourly
|
||
seeder picks `run_time = now - 2h`, so the f00 analysis file is
|
||
written at valid_time = now - 2h. A 2h cutoff deletes it within
|
||
minutes; a 3h cutoff keeps it alive until the next hourly run
|
||
supersedes it.
|
||
"""
|
||
@spec prune_old_scores() :: :ok
|
||
def prune_old_scores do
|
||
cutoff = DateTime.add(DateTime.utc_now(), -3, :hour)
|
||
file_deleted = ScoresFile.prune_older_than(cutoff)
|
||
profiles_deleted = ProfilesFile.prune_older_than(cutoff)
|
||
scalar_deleted = ScalarFile.prune_older_than(cutoff)
|
||
|
||
total = file_deleted + profiles_deleted + scalar_deleted
|
||
|
||
if total > 0 do
|
||
Logger.info(
|
||
"PropagationScores: pruned #{file_deleted} old score files + " <>
|
||
"#{profiles_deleted} profile files + #{scalar_deleted} scalar dirs " <>
|
||
"(before #{cutoff})"
|
||
)
|
||
end
|
||
|
||
# Sweep orphaned .tmp.* files left by crashed atomic-write processes
|
||
tmp_deleted =
|
||
Enum.reduce([ScoresFile.base_dir(), ProfilesFile.base_dir(), ScalarFile.base_dir()], 0, &sweep_tmp_dir/2)
|
||
|
||
if tmp_deleted > 0 do
|
||
Logger.info("PropagationScores: swept #{tmp_deleted} orphaned .tmp files")
|
||
end
|
||
|
||
:ok
|
||
end
|
||
|
||
defp sweep_tmp_dir(dir, acc) do
|
||
case File.ls(dir) do
|
||
{:ok, entries} ->
|
||
entries
|
||
|> Enum.filter(&String.contains?(&1, ".tmp."))
|
||
|> Enum.reduce(acc, fn entry, acc ->
|
||
_ = File.rm_rf(Path.join(dir, entry))
|
||
acc + 1
|
||
end)
|
||
|
||
_ ->
|
||
acc
|
||
end
|
||
end
|
||
|
||
@doc """
|
||
Returns distinct valid_times for a band, ordered ascending. Always
|
||
reads from the on-disk `ScoresFile` store — the `ScoreCache` only
|
||
holds whatever hours have been fetched or broadcast, which can be a
|
||
partial view of what's actually on disk, so using it as the source
|
||
of truth for the timeline makes new forecast hours invisible until
|
||
the cache happens to catch up. Filters out times more than 1 hour in
|
||
the past, but always includes the most recent valid_time so there's
|
||
always data to display.
|
||
"""
|
||
# HRRR forecast horizon: f00..f48 covers the next 48 hours from cycle
|
||
# time (f01-f18 hourly, f21-f48 3-hourly). Anything beyond that in the
|
||
# score store is a leftover from a stale cycle and clutters the
|
||
# timeline without adding information.
|
||
@hrrr_forecast_horizon_hours 48
|
||
|
||
@spec available_valid_times(non_neg_integer()) :: [DateTime.t()]
|
||
def available_valid_times(band_mhz) do
|
||
{past_cutoff, future_cutoff} = hot_cache_window()
|
||
|
||
case ScoresFile.list_valid_times(band_mhz) do
|
||
[] -> []
|
||
times -> filter_or_latest(times, past_cutoff, future_cutoff)
|
||
end
|
||
end
|
||
|
||
@doc """
|
||
The active forecast window for the `/map` UI: one hour in the past
|
||
through HRRR's 48-hour forecast horizon. Used by `NotifyListener` to
|
||
bound ETS growth — long-horizon GEFS `.prop` files on disk must not
|
||
balloon `propagation_score_cache` past the memory the UI actually
|
||
reads.
|
||
"""
|
||
@spec hot_cache_window() :: {DateTime.t(), DateTime.t()}
|
||
def hot_cache_window do
|
||
now = DateTime.utc_now()
|
||
past = DateTime.add(now, -3600, :second)
|
||
future = DateTime.add(now, @hrrr_forecast_horizon_hours * 3600, :second)
|
||
{past, future}
|
||
end
|
||
|
||
defp filter_or_latest(times, past_cutoff, future_cutoff) do
|
||
fresh =
|
||
Enum.filter(times, fn t ->
|
||
DateTime.compare(t, past_cutoff) != :lt and DateTime.compare(t, future_cutoff) != :gt
|
||
end)
|
||
|
||
if fresh == [] do
|
||
[Enum.max(times, DateTime)]
|
||
else
|
||
fresh
|
||
end
|
||
end
|
||
|
||
@doc """
|
||
Get scores for a band at a specific valid_time, optionally within a bounding box.
|
||
If valid_time is nil, uses the earliest available (current analysis hour).
|
||
Excludes factors for performance.
|
||
"""
|
||
@spec scores_at(non_neg_integer(), DateTime.t() | nil, %{optional(String.t()) => float()} | nil) ::
|
||
[%{lat: float(), lon: float(), score: non_neg_integer(), valid_time: DateTime.t()}]
|
||
def scores_at(band_mhz, valid_time, bounds \\ nil) do
|
||
time = valid_time || earliest_valid_time(band_mhz)
|
||
|
||
case time do
|
||
nil -> []
|
||
_ -> scores_at_fetch(band_mhz, time, bounds)
|
||
end
|
||
end
|
||
|
||
# Cache-hit path is the map's most frequent LiveView call (~every
|
||
# pan + click). A wrapping Instrument.span fires 2 telemetry handler
|
||
# dispatches that dominate the ~10µs ETS lookup — skip the span on
|
||
# hits and rely on the cheap hit/miss counter for the cache-ratio
|
||
# panel. The miss path still wraps the disk read where duration is
|
||
# the meaningful signal.
|
||
defp scores_at_fetch(band_mhz, time, bounds) do
|
||
case ScoreCache.fetch_bounds(band_mhz, time, bounds) do
|
||
{:ok, scores} ->
|
||
:telemetry.execute([:microwaveprop, :propagation, :scores_at, :cache], %{}, %{hit: true})
|
||
Enum.map(scores, &Map.put(&1, :valid_time, time))
|
||
|
||
:miss ->
|
||
:telemetry.execute([:microwaveprop, :propagation, :scores_at, :cache], %{}, %{hit: false})
|
||
|
||
Microwaveprop.Instrument.span([:propagation, :scores_at], %{band_mhz: band_mhz}, fn ->
|
||
read_from_disk_and_cache(band_mhz, time, bounds)
|
||
end)
|
||
end
|
||
end
|
||
|
||
@doc """
|
||
Variant of `scores_at/3` that always reads from the `.prop` file on
|
||
disk and overwrites the cache entry, rather than returning whatever
|
||
the cache happens to hold. Use from update paths (the map's
|
||
`propagation_updated` handler) where the underlying file has just
|
||
been rewritten but the cache may still contain the previous chain's
|
||
scores because of the race between `propagation:cache` fan-out and
|
||
`propagation:updated` delivery.
|
||
"""
|
||
@spec scores_at_fresh(non_neg_integer(), DateTime.t(), %{optional(String.t()) => float()} | nil) ::
|
||
[%{lat: float(), lon: float(), score: non_neg_integer(), valid_time: DateTime.t()}]
|
||
def scores_at_fresh(band_mhz, %DateTime{} = valid_time, bounds \\ nil) do
|
||
read_from_disk_and_cache(band_mhz, valid_time, bounds)
|
||
end
|
||
|
||
defp read_from_disk_and_cache(band_mhz, time, bounds) do
|
||
full = ScoresFile.read_bounds(band_mhz, time)
|
||
ScoreCache.put(band_mhz, time, full)
|
||
|
||
full
|
||
|> filter_bounds(bounds)
|
||
|> Enum.map(&Map.put(&1, :valid_time, time))
|
||
end
|
||
|
||
@doc """
|
||
Load the full North America score set for `{band_mhz, valid_time}` from
|
||
the on-disk binary files (HRRR `.prop` + HRDPS `.hrdps.prop`, merged) and
|
||
broadcast it to every `ScoreCache` in the cluster. Called from
|
||
`PropagationGridWorker` after each forecast hour so all pods have a warm
|
||
cache by the time clients begin requesting the new hour.
|
||
|
||
Returns `{:error, :enoent}` only when neither file exists. A single
|
||
missing file (HRDPS pre-cycle, HRRR briefly absent) is OK — the cache
|
||
warms with whichever side is available.
|
||
"""
|
||
@spec warm_cache_and_broadcast(non_neg_integer(), DateTime.t()) ::
|
||
:ok | {:error, :enoent | :invalid_format}
|
||
def warm_cache_and_broadcast(band_mhz, valid_time) do
|
||
hrrr =
|
||
case ScoresFile.read(band_mhz, valid_time) do
|
||
{:ok, payload} -> ScoresFile.extract_points(payload, nil)
|
||
_ -> nil
|
||
end
|
||
|
||
hrdps =
|
||
case ScoresFile.read_hrdps(band_mhz, valid_time) do
|
||
{:ok, payload} -> ScoresFile.extract_points(payload, nil)
|
||
_ -> nil
|
||
end
|
||
|
||
case {hrrr, hrdps} do
|
||
{nil, nil} ->
|
||
{:error, :enoent}
|
||
|
||
{h, c} ->
|
||
ScoreCache.broadcast_put(band_mhz, valid_time, (h || []) ++ (c || []))
|
||
:ok
|
||
end
|
||
end
|
||
|
||
defp filter_bounds(scores, nil), do: scores
|
||
|
||
defp filter_bounds(scores, %{"south" => s, "north" => n, "west" => w, "east" => e}) do
|
||
Enum.filter(scores, fn %{lat: lat, lon: lon} ->
|
||
lat >= s and lat <= n and lon >= w and lon <= e
|
||
end)
|
||
end
|
||
|
||
@doc "Get the latest scores for a band (alias for scores_at with earliest valid_time)."
|
||
@spec latest_scores(non_neg_integer(), %{optional(String.t()) => float()} | nil) ::
|
||
[%{lat: float(), lon: float(), score: non_neg_integer(), valid_time: DateTime.t()}]
|
||
def latest_scores(band_mhz, bounds \\ nil) do
|
||
scores_at(band_mhz, nil, bounds)
|
||
end
|
||
|
||
defp earliest_valid_time(band_mhz) do
|
||
case ScoresFile.list_valid_times(band_mhz) do
|
||
[earliest | _] -> earliest
|
||
[] -> nil
|
||
end
|
||
end
|
||
|
||
@doc "Get scores across all forecast hours for a single grid point (for sparkline)."
|
||
@spec point_forecast(non_neg_integer(), float(), float()) ::
|
||
[%{valid_time: DateTime.t(), score: non_neg_integer()}]
|
||
def point_forecast(band_mhz, lat, lon) do
|
||
Microwaveprop.Instrument.span([:propagation, :point_forecast], %{band_mhz: band_mhz}, fn ->
|
||
{snapped_lat, snapped_lon} = snap_to_grid(lat, lon)
|
||
now = DateTime.utc_now()
|
||
|
||
# Use the on-disk .prop file list as the authoritative timeline so
|
||
# the chart never falls behind the main-map timeline (which also
|
||
# reads the disk). The cache is still consulted per-hour for a
|
||
# fast score lookup; a miss falls through to the file.
|
||
# Fan the per-hour disk lookups across 4 tasks. Each ScoresFile
|
||
# read_point is an NFS stat + pread of ~100 bytes (keyed byte at
|
||
# row*cols+col), so the ceiling is NFS RTT × number of hours —
|
||
# sequential ran ~4–5× the wall time of the slowest read.
|
||
band_mhz
|
||
|> ScoresFile.list_valid_times()
|
||
|> forecast_window(now)
|
||
|> Task.async_stream(
|
||
&point_forecast_entry(band_mhz, &1, snapped_lat, snapped_lon),
|
||
max_concurrency: 4,
|
||
ordered: true,
|
||
timeout: 5_000
|
||
)
|
||
|> Enum.flat_map(fn
|
||
{:ok, nil} ->
|
||
[]
|
||
|
||
{:ok, entry} ->
|
||
[entry]
|
||
|
||
{:exit, reason} ->
|
||
Logger.error(
|
||
"Propagation.point_forecast async lookup failed: band_mhz=#{band_mhz} lat=#{snapped_lat} lon=#{snapped_lon} reason=#{inspect(reason)}"
|
||
)
|
||
|
||
[]
|
||
end)
|
||
end)
|
||
end
|
||
|
||
defp point_forecast_entry(band_mhz, valid_time, lat, lon) do
|
||
case ScoreCache.fetch_point(band_mhz, valid_time, lat, lon) do
|
||
{:ok, score} ->
|
||
%{valid_time: valid_time, score: score}
|
||
|
||
:miss ->
|
||
case ScoresFile.read_point(band_mhz, valid_time, lat, lon) do
|
||
nil -> nil
|
||
score -> %{valid_time: valid_time, score: score}
|
||
end
|
||
end
|
||
end
|
||
|
||
# Select the set of valid_times the forecast chart should render.
|
||
# Mirrors `available_valid_times`: keep everything from one hour
|
||
# before now onward so the most recent analysis hour (typically
|
||
# ~30–60 min behind wall clock due to HRRR publishing lag) sits at
|
||
# the left edge of the chart as "now". When every hour on disk is
|
||
# older than that cutoff, fall back to just the newest entry so the
|
||
# chart can still render a single data point.
|
||
defp forecast_window([], _now), do: []
|
||
|
||
defp forecast_window(times, now) do
|
||
past_cutoff = DateTime.add(now, -3600, :second)
|
||
future_cutoff = DateTime.add(now, @hrrr_forecast_horizon_hours * 3600, :second)
|
||
filter_or_latest(times, past_cutoff, future_cutoff)
|
||
end
|
||
|
||
defp snap_to_grid(lat, lon) do
|
||
step = Grid.step()
|
||
{Float.round(Float.round(lat / step) * step, 3), Float.round(Float.round(lon / step) * step, 3)}
|
||
end
|
||
|
||
@doc """
|
||
Get the full score and factors for a specific grid point, snapped to
|
||
the nearest grid cell.
|
||
|
||
`:profile_source` describes where the factor breakdown came from:
|
||
|
||
* `:exact` — rescored from this `valid_time`'s own profile file.
|
||
* `{:fallback, fallback_valid_time}` — the requested hour's profile
|
||
was missing, so we rescored from the most recent analysis profile
|
||
within the lookback window. Treat as approximate.
|
||
* `:unavailable` — no profile available; `factors` is `%{}`.
|
||
"""
|
||
@spec point_detail(non_neg_integer(), float(), float(), DateTime.t() | nil) ::
|
||
%{
|
||
lat: float(),
|
||
lon: float(),
|
||
score: non_neg_integer(),
|
||
factors: map(),
|
||
profile_source: :exact | {:fallback, DateTime.t()} | :unavailable,
|
||
valid_time: DateTime.t()
|
||
}
|
||
| nil
|
||
def point_detail(band_mhz, lat, lon, valid_time \\ nil) do
|
||
Microwaveprop.Instrument.span([:propagation, :point_detail], %{band_mhz: band_mhz}, fn ->
|
||
do_point_detail(band_mhz, lat, lon, valid_time)
|
||
end)
|
||
end
|
||
|
||
defp do_point_detail(band_mhz, lat, lon, valid_time) do
|
||
{snapped_lat, snapped_lon} = snap_to_grid(lat, lon)
|
||
time = valid_time || latest_valid_time(band_mhz)
|
||
|
||
case time do
|
||
nil ->
|
||
nil
|
||
|
||
_ ->
|
||
case ScoresFile.read_point(band_mhz, time, snapped_lat, snapped_lon) do
|
||
nil ->
|
||
nil
|
||
|
||
score ->
|
||
{factors, source} = factors_for(band_mhz, time, snapped_lat, snapped_lon)
|
||
|
||
%{
|
||
lat: snapped_lat,
|
||
lon: snapped_lon,
|
||
score: score,
|
||
factors: factors,
|
||
profile_source: source,
|
||
valid_time: time
|
||
}
|
||
end
|
||
end
|
||
end
|
||
|
||
# Rebuild the factor breakdown for a clicked grid cell by rescoring
|
||
# the persisted HRRR profile.
|
||
#
|
||
# The Rust pipeline (`prop_grid_rs`) writes a per-cell profile file
|
||
# for every chain step (f00..f18 since Phase 2 cutover), so the
|
||
# `:exact` branch covers a healthy production state. The fallback to
|
||
# a recent analysis profile remains as a safety net for missed
|
||
# cycles or partial chain runs — when it kicks in we tag the result
|
||
# `{:fallback, profile_valid_time}` so the UI can label the
|
||
# breakdown as approximated rather than silently misrepresent it.
|
||
@fallback_profile_lookback_hours 24
|
||
@spec factors_for(non_neg_integer(), DateTime.t(), float(), float()) ::
|
||
{map(), :exact | {:fallback, DateTime.t()} | :unavailable}
|
||
defp factors_for(band_mhz, valid_time, lat, lon) do
|
||
case ProfilesFile.read_point(valid_time, lat, lon) do
|
||
nil ->
|
||
factors_from_fallback_profile(band_mhz, valid_time, lat, lon)
|
||
|
||
profile ->
|
||
{factors_from_profile(band_mhz, valid_time, profile, lat, lon), :exact}
|
||
end
|
||
end
|
||
|
||
defp factors_from_fallback_profile(band_mhz, valid_time, lat, lon) do
|
||
case latest_profile_time_within_lookback(valid_time) do
|
||
nil ->
|
||
{%{}, :unavailable}
|
||
|
||
fallback_time ->
|
||
case ProfilesFile.read_point(fallback_time, lat, lon) do
|
||
nil ->
|
||
{%{}, :unavailable}
|
||
|
||
profile ->
|
||
{factors_from_profile(band_mhz, fallback_time, profile, lat, lon), {:fallback, fallback_time}}
|
||
end
|
||
end
|
||
end
|
||
|
||
defp 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
|
||
end
|
||
|
||
defp latest_profile_time_within_lookback(%DateTime{} = valid_time) do
|
||
lookback_cutoff = DateTime.add(valid_time, -@fallback_profile_lookback_hours * 3600, :second)
|
||
|
||
ProfilesFile.list_valid_times()
|
||
|> Enum.filter(fn t ->
|
||
DateTime.compare(t, valid_time) != :gt and DateTime.compare(t, lookback_cutoff) != :lt
|
||
end)
|
||
|> case do
|
||
[] -> nil
|
||
past -> Enum.max(past, DateTime)
|
||
end
|
||
end
|
||
|
||
@doc "Get the latest valid_time across all bands."
|
||
@spec latest_valid_time() :: DateTime.t() | nil
|
||
def latest_valid_time do
|
||
ScoresFile.latest_valid_time()
|
||
end
|
||
|
||
@doc "Get the latest valid_time for a specific band."
|
||
@spec latest_valid_time(non_neg_integer()) :: DateTime.t() | nil
|
||
def latest_valid_time(band_mhz) do
|
||
case ScoresFile.list_valid_times(band_mhz) do
|
||
[] -> nil
|
||
times -> Enum.max(times, DateTime)
|
||
end
|
||
end
|
||
|
||
## Run timings
|
||
|
||
@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
|
||
|
||
# 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}) 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
|