Add more caching to make the map feel instant
- ScoreCache stores {band, valid_time} as %{{lat, lon} => score} map so
point lookups are O(1); adds fetch_point/4 and valid_times/1
- available_valid_times/1 reads directly from ScoreCache when warm,
falls back to DB on cold start
- point_forecast/3 iterates cached valid_times and uses fetch_point/4
instead of hitting the DB per click
- NexradCache: node-local ETS cache of decoded n0q PNG pixel buffers
keyed by 5-minute rounded timestamp; skips ~1-5s HTTP+decode on
concurrent/repeat clicks within the same window
- MapLive: start_async the rain_scatter fetch so point_detail renders
immediately with a pending marker; push rain_scatter_update when
NEXRAD resolves
- MapLive: preload all 18 remaining forecast hours for the current
viewport after mount/band change/propagation_updated; client caches
them and renders timeline scrubs instantly without a server roundtrip.
Adds set_selected_time event for fast-path state sync.
- Propagation map JS: forecastCache map + drawScatterMarkers helper,
timeline click uses preloaded cache when available
This commit is contained in:
parent
8b407c87ee
commit
250709a1b2
10 changed files with 485 additions and 54 deletions
|
|
@ -88,7 +88,7 @@ interface PointDetail {
|
|||
extended_range_km: number
|
||||
exceptional_range_km: number
|
||||
factors: Record<string, number> & { duct_info?: DuctInfo }
|
||||
rain_scatter?: RainScatter
|
||||
rain_scatter?: RainScatter | "pending"
|
||||
forecast?: ForecastPoint[]
|
||||
[key: string]: unknown
|
||||
}
|
||||
|
|
@ -141,6 +141,7 @@ interface PropagationMapHook extends ViewHook {
|
|||
timelineEl: HTMLElement | null
|
||||
timelineData: TimelineEntry[]
|
||||
selectedTime: string | null
|
||||
forecastCache: Map<string, ScorePoint[]>
|
||||
|
||||
showDetailPanel(this: PropagationMapHook): void
|
||||
hideDetailPanel(this: PropagationMapHook): void
|
||||
|
|
@ -148,6 +149,7 @@ interface PropagationMapHook extends ViewHook {
|
|||
interpolateScore(this: PropagationMapHook, lat: number, lon: number, step: number): number | null
|
||||
scoreColorRGB(this: PropagationMapHook, score: number): RGB
|
||||
lookupPoint(this: PropagationMapHook, lat: number, lon: number): (ScorePoint & BandInfo) | null
|
||||
drawScatterMarkers(this: PropagationMapHook, scatter: RainScatter): void
|
||||
renderTimeline(this: PropagationMapHook): void
|
||||
sendBounds(this: PropagationMapHook): void
|
||||
}
|
||||
|
|
@ -563,12 +565,24 @@ function buildPopupHTML(detail: PointDetail, viewshedLoading: boolean): string {
|
|||
${explanations}
|
||||
</div>
|
||||
${detail.factors.duct_info ? buildDuctInfoHTML(detail.factors.duct_info) : ""}
|
||||
${detail.rain_scatter && detail.rain_scatter.cells.length > 0 ? buildScatterHTML(detail.rain_scatter) : ""}
|
||||
${buildScatterBlock(detail.rain_scatter)}
|
||||
${detail.forecast ? buildForecastSvg(detail.forecast) : ""}
|
||||
${viewshedStatus}
|
||||
</div>`
|
||||
}
|
||||
|
||||
function buildScatterBlock(scatter: RainScatter | "pending" | undefined): string {
|
||||
if (scatter === "pending") {
|
||||
return `
|
||||
<div style="padding:4px 10px 6px;border-top:1px solid rgba(255,255,255,0.15);">
|
||||
<div style="font-size:10px;font-weight:700;opacity:0.5;margin-bottom:3px;text-transform:uppercase;letter-spacing:0.5px;">Rain Scatter</div>
|
||||
<div style="font-size:11px;opacity:0.6;">Checking NEXRAD\u2026</div>
|
||||
</div>`
|
||||
}
|
||||
if (scatter && scatter.cells.length > 0) return buildScatterHTML(scatter)
|
||||
return ""
|
||||
}
|
||||
|
||||
function buildScatterHTML(scatter: RainScatter): string {
|
||||
const cls = scatter.classification
|
||||
const cells = scatter.cells
|
||||
|
|
@ -631,6 +645,7 @@ export const PropagationMap: Record<string, unknown> & {
|
|||
interpolateScore(this: PropagationMapHook, lat: number, lon: number, step: number): number | null
|
||||
scoreColorRGB(this: PropagationMapHook, score: number): RGB
|
||||
lookupPoint(this: PropagationMapHook, lat: number, lon: number): (ScorePoint & BandInfo) | null
|
||||
drawScatterMarkers(this: PropagationMapHook, scatter: RainScatter): void
|
||||
renderTimeline(this: PropagationMapHook): void
|
||||
sendBounds(this: PropagationMapHook): void
|
||||
destroyed(this: PropagationMapHook): void
|
||||
|
|
@ -659,6 +674,7 @@ export const PropagationMap: Record<string, unknown> & {
|
|||
|
||||
this.scoreOverlay = null
|
||||
this.scores = []
|
||||
this.forecastCache = new Map()
|
||||
|
||||
this.colorScale = [
|
||||
{ min: 80, r: 0, g: 255, b: 163 },
|
||||
|
|
@ -677,6 +693,7 @@ export const PropagationMap: Record<string, unknown> & {
|
|||
this.handleEvent("update_scores", ({ scores }: { scores: ScorePoint[] }) => {
|
||||
topbar.hide()
|
||||
this.renderScores(scores)
|
||||
if (this.selectedTime) this.forecastCache.set(this.selectedTime, scores)
|
||||
|
||||
// If a point was selected, refresh its detail with the new data
|
||||
if (this.clickedLatLng && this.detailPanel && this.detailPanel.style.display !== "none") {
|
||||
|
|
@ -684,6 +701,14 @@ export const PropagationMap: Record<string, unknown> & {
|
|||
}
|
||||
})
|
||||
|
||||
// Preloaded forecast hours arrive after mount/band change — stash them so
|
||||
// future timeline scrubs can render instantly from the local cache.
|
||||
this.handleEvent("preload_forecast", ({ hours }: { hours: { time: string; scores: ScorePoint[] }[] }) => {
|
||||
for (const h of hours) {
|
||||
this.forecastCache.set(h.time, h.scores)
|
||||
}
|
||||
})
|
||||
|
||||
this.bandInfo = JSON.parse(this.el.dataset.bandInfo || "{}")
|
||||
this.rangeCircles = L.layerGroup().addTo(this.map)
|
||||
|
||||
|
|
@ -772,6 +797,15 @@ export const PropagationMap: Record<string, unknown> & {
|
|||
}
|
||||
this.lastDetail = null
|
||||
|
||||
this.handleEvent("rain_scatter_update", ({ rain_scatter }: { lat: number; lon: number; rain_scatter: RainScatter }) => {
|
||||
if (!this.lastDetail || !this.detailPanel) return
|
||||
this.lastDetail.rain_scatter = rain_scatter
|
||||
this.detailPanel.innerHTML = buildPopupHTML(this.lastDetail, this.viewshedLoading)
|
||||
if (rain_scatter && rain_scatter.cells && rain_scatter.cells.length > 0) {
|
||||
this.drawScatterMarkers(rain_scatter)
|
||||
}
|
||||
})
|
||||
|
||||
this.handleEvent("point_detail", (detail: PointDetail) => {
|
||||
if (detail && this.detailPanel) {
|
||||
const merged: PointDetail = { ...detail, ...this.bandInfo }
|
||||
|
|
@ -785,21 +819,8 @@ export const PropagationMap: Record<string, unknown> & {
|
|||
} else {
|
||||
this.scatterMarkers = L.layerGroup().addTo(this.map)
|
||||
}
|
||||
if (detail.rain_scatter && detail.rain_scatter.cells.length > 0) {
|
||||
for (const c of detail.rain_scatter.cells) {
|
||||
const opacity = Math.min(0.9, Math.max(0.3, (c.scatter_db + 30) / 30))
|
||||
const color = c.dbz >= 45 ? "#dc2626" : c.dbz >= 35 ? "#ea580c" : "#ca8a04"
|
||||
const marker = L.circleMarker([c.lat, c.lon], {
|
||||
radius: Math.min(12, Math.max(4, c.dbz / 5)),
|
||||
color: color,
|
||||
fillColor: color,
|
||||
fillOpacity: opacity * 0.5,
|
||||
weight: 1.5,
|
||||
opacity: opacity,
|
||||
interactive: false
|
||||
})
|
||||
this.scatterMarkers!.addLayer(marker)
|
||||
}
|
||||
if (detail.rain_scatter && detail.rain_scatter !== "pending" && detail.rain_scatter.cells.length > 0) {
|
||||
this.drawScatterMarkers(detail.rain_scatter)
|
||||
}
|
||||
|
||||
// Draw propagation reach polygon based on contiguous good cells
|
||||
|
|
@ -1083,6 +1104,25 @@ export const PropagationMap: Record<string, unknown> & {
|
|||
return { lat: rlat, lon: rlon, score, ...this.bandInfo }
|
||||
},
|
||||
|
||||
drawScatterMarkers(this: PropagationMapHook, scatter: RainScatter) {
|
||||
if (!this.scatterMarkers) this.scatterMarkers = L.layerGroup().addTo(this.map)
|
||||
this.scatterMarkers.clearLayers()
|
||||
for (const c of scatter.cells) {
|
||||
const opacity = Math.min(0.9, Math.max(0.3, (c.scatter_db + 30) / 30))
|
||||
const color = c.dbz >= 45 ? "#dc2626" : c.dbz >= 35 ? "#ea580c" : "#ca8a04"
|
||||
const marker = L.circleMarker([c.lat, c.lon], {
|
||||
radius: Math.min(12, Math.max(4, c.dbz / 5)),
|
||||
color,
|
||||
fillColor: color,
|
||||
fillOpacity: opacity * 0.5,
|
||||
weight: 1.5,
|
||||
opacity,
|
||||
interactive: false
|
||||
})
|
||||
this.scatterMarkers.addLayer(marker)
|
||||
}
|
||||
},
|
||||
|
||||
renderTimeline(this: PropagationMapHook) {
|
||||
if (!this.timelineEl || this.timelineData.length === 0) {
|
||||
if (this.timelineEl) this.timelineEl.style.display = "none"
|
||||
|
|
@ -1171,16 +1211,31 @@ export const PropagationMap: Record<string, unknown> & {
|
|||
// Attach click handlers
|
||||
this.timelineEl.querySelectorAll<HTMLButtonElement>("button[data-time]").forEach(btn => {
|
||||
btn.addEventListener("click", () => {
|
||||
this.selectedTime = btn.dataset.time!
|
||||
const time = btn.dataset.time!
|
||||
this.selectedTime = time
|
||||
this.renderTimeline()
|
||||
topbar.show()
|
||||
this.pushEvent("select_time", { time: btn.dataset.time! })
|
||||
|
||||
const cached = this.forecastCache.get(time)
|
||||
if (cached) {
|
||||
// Fast path — render instantly from preloaded cache, just inform
|
||||
// the server of the new selected time for state tracking.
|
||||
this.renderScores(cached)
|
||||
this.pushEvent("set_selected_time", { time })
|
||||
} else {
|
||||
// Cache miss — fall back to server roundtrip via the existing
|
||||
// update_scores path.
|
||||
topbar.show()
|
||||
this.pushEvent("select_time", { time })
|
||||
}
|
||||
})
|
||||
})
|
||||
},
|
||||
|
||||
sendBounds(this: PropagationMapHook) {
|
||||
topbar.show()
|
||||
// Bounds changed — preloaded forecast hours are now for stale viewport.
|
||||
// Drop them so timeline scrubs re-fetch for the new viewport.
|
||||
this.forecastCache.clear()
|
||||
// Pad bounds by ~0.5° so Leaflet edge tiles always have data to render
|
||||
const b = this.map.getBounds().pad(0.1)
|
||||
this.pushEvent("map_bounds", {
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@ defmodule Microwaveprop.Application do
|
|||
{Cluster.Supervisor, [topologies, [name: Microwaveprop.ClusterSupervisor]]},
|
||||
{Phoenix.PubSub, name: Microwaveprop.PubSub},
|
||||
Microwaveprop.Propagation.ScoreCache,
|
||||
Microwaveprop.Weather.NexradCache,
|
||||
{Oban, Application.fetch_env!(:microwaveprop, Oban)},
|
||||
Microwaveprop.RepoListener,
|
||||
# Start to serve requests, typically the last entry
|
||||
|
|
|
|||
|
|
@ -202,6 +202,17 @@ defmodule Microwaveprop.Propagation do
|
|||
def available_valid_times(band_mhz) do
|
||||
cutoff = DateTime.add(DateTime.utc_now(), -3600, :second)
|
||||
|
||||
case ScoreCache.valid_times(band_mhz) do
|
||||
[] -> available_valid_times_from_db(band_mhz, cutoff)
|
||||
cached -> filter_fresh(cached, cutoff)
|
||||
end
|
||||
end
|
||||
|
||||
defp filter_fresh(times, cutoff) do
|
||||
Enum.filter(times, fn t -> DateTime.compare(t, cutoff) != :lt end)
|
||||
end
|
||||
|
||||
defp available_valid_times_from_db(band_mhz, cutoff) do
|
||||
times =
|
||||
Repo.all(
|
||||
from(gs in GridScore,
|
||||
|
|
@ -302,17 +313,34 @@ defmodule Microwaveprop.Propagation do
|
|||
@spec point_forecast(non_neg_integer(), float(), float()) ::
|
||||
[%{valid_time: DateTime.t(), score: non_neg_integer()}]
|
||||
def point_forecast(band_mhz, lat, lon) do
|
||||
step = Grid.step()
|
||||
snapped_lat = Float.round(Float.round(lat / step) * step, 3)
|
||||
snapped_lon = Float.round(Float.round(lon / step) * step, 3)
|
||||
{snapped_lat, snapped_lon} = snap_to_grid(lat, lon)
|
||||
now = DateTime.utc_now()
|
||||
|
||||
case ScoreCache.valid_times(band_mhz) do
|
||||
[] -> point_forecast_from_db(band_mhz, snapped_lat, snapped_lon, now)
|
||||
cached -> point_forecast_from_cache(band_mhz, snapped_lat, snapped_lon, now, cached)
|
||||
end
|
||||
end
|
||||
|
||||
defp point_forecast_from_cache(band_mhz, lat, lon, now, cached_times) do
|
||||
cached_times
|
||||
|> Enum.filter(fn t -> DateTime.compare(t, now) != :lt end)
|
||||
|> Enum.map(fn t ->
|
||||
case ScoreCache.fetch_point(band_mhz, t, lat, lon) do
|
||||
{:ok, score} -> %{valid_time: t, score: score}
|
||||
:miss -> nil
|
||||
end
|
||||
end)
|
||||
|> Enum.reject(&is_nil/1)
|
||||
end
|
||||
|
||||
defp point_forecast_from_db(band_mhz, lat, lon, now) do
|
||||
Repo.all(
|
||||
from(gs in GridScore,
|
||||
where:
|
||||
gs.band_mhz == ^band_mhz and
|
||||
gs.lat == ^snapped_lat and
|
||||
gs.lon == ^snapped_lon and
|
||||
gs.lat == ^lat and
|
||||
gs.lon == ^lon and
|
||||
gs.valid_time >= ^now,
|
||||
select: %{valid_time: gs.valid_time, score: gs.score},
|
||||
order_by: [asc: gs.valid_time]
|
||||
|
|
@ -320,6 +348,12 @@ defmodule Microwaveprop.Propagation do
|
|||
)
|
||||
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 nearest grid."
|
||||
@spec point_detail(non_neg_integer(), float(), float(), DateTime.t() | nil) ::
|
||||
%{
|
||||
|
|
|
|||
|
|
@ -7,6 +7,9 @@ defmodule Microwaveprop.Propagation.ScoreCache do
|
|||
on a cache miss. `broadcast_put/3` fans the payload out across the cluster on
|
||||
the `"propagation:cache"` PubSub topic so every BEAM node stays in sync with
|
||||
a single compute.
|
||||
|
||||
Internally each cache entry stores a `%{{lat, lon} => score}` map so that
|
||||
per-point lookups (used by `point_forecast/3` and `point_detail/4`) are O(1).
|
||||
"""
|
||||
use GenServer
|
||||
|
||||
|
|
@ -29,7 +32,7 @@ defmodule Microwaveprop.Propagation.ScoreCache do
|
|||
@spec fetch(non_neg_integer(), DateTime.t()) :: {:ok, [score()]} | :miss
|
||||
def fetch(band_mhz, valid_time) do
|
||||
case :ets.lookup(@table, {band_mhz, valid_time}) do
|
||||
[{_, scores}] -> {:ok, scores}
|
||||
[{_, grid}] -> {:ok, grid_to_list(grid)}
|
||||
[] -> :miss
|
||||
end
|
||||
end
|
||||
|
|
@ -37,15 +40,37 @@ defmodule Microwaveprop.Propagation.ScoreCache do
|
|||
@spec fetch_bounds(non_neg_integer(), DateTime.t(), bounds() | nil) ::
|
||||
{:ok, [score()]} | :miss
|
||||
def fetch_bounds(band_mhz, valid_time, bounds) do
|
||||
case fetch(band_mhz, valid_time) do
|
||||
{:ok, scores} -> {:ok, filter_bounds(scores, bounds)}
|
||||
:miss -> :miss
|
||||
case :ets.lookup(@table, {band_mhz, valid_time}) do
|
||||
[{_, grid}] -> {:ok, grid_to_filtered_list(grid, bounds)}
|
||||
[] -> :miss
|
||||
end
|
||||
end
|
||||
|
||||
@doc """
|
||||
Look up the score for a single `{lat, lon}` point at `{band_mhz, valid_time}`.
|
||||
Returns `{:ok, score}` on hit or `:miss` on cache miss. The coordinates are
|
||||
used as-is (no snap-to-grid) — callers that need the nearest grid cell should
|
||||
snap first.
|
||||
"""
|
||||
@spec fetch_point(non_neg_integer(), DateTime.t(), float(), float()) ::
|
||||
{:ok, non_neg_integer()} | :miss
|
||||
def fetch_point(band_mhz, valid_time, lat, lon) do
|
||||
case :ets.lookup(@table, {band_mhz, valid_time}) do
|
||||
[{_, grid}] ->
|
||||
case Map.get(grid, {lat, lon}) do
|
||||
nil -> :miss
|
||||
score -> {:ok, score}
|
||||
end
|
||||
|
||||
[] ->
|
||||
:miss
|
||||
end
|
||||
end
|
||||
|
||||
@spec put(non_neg_integer(), DateTime.t(), [score()]) :: :ok
|
||||
def put(band_mhz, valid_time, scores) do
|
||||
:ets.insert(@table, {{band_mhz, valid_time}, scores})
|
||||
grid = list_to_grid(scores)
|
||||
:ets.insert(@table, {{band_mhz, valid_time}, grid})
|
||||
:ok
|
||||
end
|
||||
|
||||
|
|
@ -62,6 +87,16 @@ defmodule Microwaveprop.Propagation.ScoreCache do
|
|||
:ok
|
||||
end
|
||||
|
||||
@doc "Returns the sorted list of valid_times cached for `band_mhz`."
|
||||
@spec valid_times(non_neg_integer()) :: [DateTime.t()]
|
||||
def valid_times(band_mhz) do
|
||||
match_spec = [{{{band_mhz, :"$1"}, :_}, [], [:"$1"]}]
|
||||
|
||||
@table
|
||||
|> :ets.select(match_spec)
|
||||
|> Enum.sort({:asc, DateTime})
|
||||
end
|
||||
|
||||
@spec prune_older_than(DateTime.t()) :: non_neg_integer()
|
||||
def prune_older_than(cutoff) do
|
||||
match_spec = [
|
||||
|
|
@ -105,11 +140,21 @@ defmodule Microwaveprop.Propagation.ScoreCache do
|
|||
|
||||
# ---------- Internal ----------
|
||||
|
||||
defp filter_bounds(scores, nil), do: scores
|
||||
defp list_to_grid(scores) do
|
||||
Map.new(scores, fn %{lat: lat, lon: lon, score: score} -> {{lat, lon}, score} end)
|
||||
end
|
||||
|
||||
defp filter_bounds(scores, %{"south" => s, "north" => n, "west" => w, "east" => e}) do
|
||||
Enum.filter(scores, fn %{lat: lat, lon: lon} ->
|
||||
defp grid_to_list(grid) do
|
||||
Enum.map(grid, fn {{lat, lon}, score} -> %{lat: lat, lon: lon, score: score} end)
|
||||
end
|
||||
|
||||
defp grid_to_filtered_list(grid, nil), do: grid_to_list(grid)
|
||||
|
||||
defp grid_to_filtered_list(grid, %{"south" => s, "north" => n, "west" => w, "east" => e}) do
|
||||
grid
|
||||
|> Enum.filter(fn {{lat, lon}, _score} ->
|
||||
lat >= s and lat <= n and lon >= w and lon <= e
|
||||
end)
|
||||
|> Enum.map(fn {{lat, lon}, score} -> %{lat: lat, lon: lon, score: score} end)
|
||||
end
|
||||
end
|
||||
|
|
|
|||
55
lib/microwaveprop/weather/nexrad_cache.ex
Normal file
55
lib/microwaveprop/weather/nexrad_cache.ex
Normal file
|
|
@ -0,0 +1,55 @@
|
|||
defmodule Microwaveprop.Weather.NexradCache do
|
||||
@moduledoc """
|
||||
Node-local ETS cache of decoded NEXRAD n0q composite reflectivity frames.
|
||||
Keyed by a 5-minute rounded timestamp. Stores the raw pixel buffer + image
|
||||
width so per-point rain-cell extraction can skip the HTTP fetch + PNG decode
|
||||
(which take 1-5 seconds for a ~5 MB CONUS-wide image).
|
||||
|
||||
The cache is populated by `Microwaveprop.Weather.NexradClient.fetch_rain_cells/4`
|
||||
on its first call per 5-min window, then reused by every concurrent click
|
||||
until the window rolls over.
|
||||
"""
|
||||
use GenServer
|
||||
|
||||
@table :nexrad_frame_cache
|
||||
|
||||
@type pixels :: binary()
|
||||
@type width :: pos_integer()
|
||||
|
||||
@spec start_link(keyword()) :: GenServer.on_start()
|
||||
def start_link(opts) do
|
||||
GenServer.start_link(__MODULE__, opts, name: __MODULE__)
|
||||
end
|
||||
|
||||
@spec fetch(DateTime.t()) :: {:ok, pixels(), width()} | :miss
|
||||
def fetch(rounded_ts) do
|
||||
case :ets.lookup(@table, rounded_ts) do
|
||||
[{_, pixels, width}] -> {:ok, pixels, width}
|
||||
[] -> :miss
|
||||
end
|
||||
end
|
||||
|
||||
@spec put(DateTime.t(), pixels(), width()) :: :ok
|
||||
def put(rounded_ts, pixels, width) do
|
||||
:ets.insert(@table, {rounded_ts, pixels, width})
|
||||
:ok
|
||||
end
|
||||
|
||||
@spec prune_older_than(DateTime.t()) :: non_neg_integer()
|
||||
def prune_older_than(cutoff_ts) do
|
||||
match_spec = [{{:"$1", :_, :_}, [{:<, :"$1", {:const, cutoff_ts}}], [true]}]
|
||||
:ets.select_delete(@table, match_spec)
|
||||
end
|
||||
|
||||
@spec clear() :: :ok
|
||||
def clear do
|
||||
:ets.delete_all_objects(@table)
|
||||
:ok
|
||||
end
|
||||
|
||||
@impl true
|
||||
def init(_opts) do
|
||||
:ets.new(@table, [:set, :named_table, :public, read_concurrency: true])
|
||||
{:ok, %{}}
|
||||
end
|
||||
end
|
||||
|
|
@ -58,37 +58,48 @@ defmodule Microwaveprop.Weather.NexradClient do
|
|||
"#{@base_url}/#{date_path}/GIS/uscomp/n0q_#{file_ts}.png"
|
||||
end
|
||||
|
||||
alias Microwaveprop.Weather.NexradCache
|
||||
|
||||
@doc """
|
||||
Fetch the latest n0q frame and extract all rain cells with reflectivity
|
||||
above `min_dbz` within `radius_km` of the given point. Returns
|
||||
`{:ok, [{lat, lon, dbz}]}` or `{:error, reason}`.
|
||||
|
||||
Samples every ~5 km (10 pixels) within the bounding box for efficiency.
|
||||
The decoded PNG pixel buffer is cached in `NexradCache` for the current
|
||||
5-minute window, so concurrent clicks and repeat lookups skip the HTTP +
|
||||
PNG decode entirely.
|
||||
"""
|
||||
@spec fetch_rain_cells(float(), float(), float(), float()) :: {:ok, [{float(), float(), float()}]} | {:error, term()}
|
||||
def fetch_rain_cells(lat, lon, radius_km \\ 300.0, min_dbz \\ 25.0) do
|
||||
now = DateTime.utc_now()
|
||||
rounded = round_to_5min(now)
|
||||
url = frame_url(rounded)
|
||||
|
||||
case NexradCache.fetch(rounded) do
|
||||
{:ok, pixels, width} ->
|
||||
{:ok, extract_rain_cells(pixels, width, lat, lon, radius_km, min_dbz)}
|
||||
|
||||
:miss ->
|
||||
case fetch_and_decode_frame(rounded) do
|
||||
{:ok, pixels, width} ->
|
||||
NexradCache.put(rounded, pixels, width)
|
||||
NexradCache.prune_older_than(DateTime.add(rounded, -600, :second))
|
||||
{:ok, extract_rain_cells(pixels, width, lat, lon, radius_km, min_dbz)}
|
||||
|
||||
error ->
|
||||
error
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
defp fetch_and_decode_frame(rounded) do
|
||||
url = frame_url(rounded)
|
||||
req_opts = Application.get_env(:microwaveprop, :nexrad_req_options, [])
|
||||
|
||||
case Req.get(url, [receive_timeout: 60_000, retry: false] ++ req_opts) do
|
||||
{:ok, %{status: 200, body: body}} when is_binary(body) ->
|
||||
case decode_png_to_pixels(body) do
|
||||
{:ok, pixels, width} ->
|
||||
cells = extract_rain_cells(pixels, width, lat, lon, radius_km, min_dbz)
|
||||
{:ok, cells}
|
||||
|
||||
{:error, reason} ->
|
||||
{:error, reason}
|
||||
end
|
||||
|
||||
{:ok, %{status: status}} ->
|
||||
{:error, "NEXRAD n0q HTTP #{status}"}
|
||||
|
||||
{:error, reason} ->
|
||||
{:error, reason}
|
||||
{:ok, %{status: 200, body: body}} when is_binary(body) -> decode_png_to_pixels(body)
|
||||
{:ok, %{status: status}} -> {:error, "NEXRAD n0q HTTP #{status}"}
|
||||
{:error, reason} -> {:error, reason}
|
||||
end
|
||||
end
|
||||
|
||||
|
|
|
|||
|
|
@ -32,6 +32,8 @@ defmodule MicrowavepropWeb.MapLive do
|
|||
bounds = bounds_around(center)
|
||||
initial_scores = Propagation.scores_at(@default_band, selected_time, bounds)
|
||||
|
||||
if connected?(socket), do: send(self(), :preload_forecast)
|
||||
|
||||
{:ok,
|
||||
assign(socket,
|
||||
page_title: "Propagation Prediction Map",
|
||||
|
|
@ -75,6 +77,8 @@ defmodule MicrowavepropWeb.MapLive do
|
|||
selected_time = closest_to_now(valid_times)
|
||||
scores = Propagation.scores_at(band, selected_time, socket.assigns.bounds)
|
||||
|
||||
send(self(), :preload_forecast)
|
||||
|
||||
socket =
|
||||
socket
|
||||
|> assign(selected_band: band, valid_times: valid_times, selected_time: selected_time)
|
||||
|
|
@ -103,6 +107,16 @@ defmodule MicrowavepropWeb.MapLive do
|
|||
end
|
||||
end
|
||||
|
||||
# Fast path: the client already has the preloaded hour's scores in its
|
||||
# local cache and only needs the server to remember which time is selected
|
||||
# (used later for point_detail and forecast state on reconnect).
|
||||
def handle_event("set_selected_time", %{"time" => time_str}, socket) do
|
||||
case DateTime.from_iso8601(time_str) do
|
||||
{:ok, time, _} -> {:noreply, assign(socket, :selected_time, time)}
|
||||
_ -> {:noreply, socket}
|
||||
end
|
||||
end
|
||||
|
||||
def handle_event("set_antenna_height", %{"height_ft" => value}, socket) do
|
||||
height =
|
||||
case Integer.parse(value) do
|
||||
|
|
@ -134,10 +148,22 @@ defmodule MicrowavepropWeb.MapLive do
|
|||
%{time: DateTime.to_iso8601(f.valid_time), score: f.score}
|
||||
end)
|
||||
|
||||
scatter = fetch_rain_scatter(lat, lon, band)
|
||||
# Push the detail immediately with rain_scatter as :pending — NEXRAD
|
||||
# fetching is slow on a cold 5-min window so we don't want to block the
|
||||
# LiveView process. The :rain_scatter async task pushes its own event.
|
||||
payload =
|
||||
if detail,
|
||||
do: detail |> Map.put(:forecast, forecast_data) |> Map.put(:rain_scatter, :pending),
|
||||
else: %{}
|
||||
|
||||
payload = if detail, do: detail |> Map.put(:forecast, forecast_data) |> Map.put(:rain_scatter, scatter), else: %{}
|
||||
{:noreply, push_event(socket, "point_detail", payload)}
|
||||
socket =
|
||||
socket
|
||||
|> push_event("point_detail", payload)
|
||||
|> start_async({:rain_scatter, lat, lon, band}, fn ->
|
||||
fetch_rain_scatter(lat, lon, band)
|
||||
end)
|
||||
|
||||
{:noreply, socket}
|
||||
end
|
||||
|
||||
def handle_event("map_bounds", bounds, socket) do
|
||||
|
|
@ -179,6 +205,21 @@ defmodule MicrowavepropWeb.MapLive do
|
|||
end
|
||||
|
||||
@impl true
|
||||
def handle_info(:preload_forecast, socket) do
|
||||
band = socket.assigns.selected_band
|
||||
bounds = socket.assigns.bounds
|
||||
selected = socket.assigns.selected_time
|
||||
|
||||
hours =
|
||||
socket.assigns.valid_times
|
||||
|> Enum.reject(&(selected && DateTime.compare(&1, selected) == :eq))
|
||||
|> Enum.map(fn t ->
|
||||
%{time: DateTime.to_iso8601(t), scores: Propagation.scores_at(band, t, bounds)}
|
||||
end)
|
||||
|
||||
{:noreply, push_event(socket, "preload_forecast", %{hours: hours})}
|
||||
end
|
||||
|
||||
def handle_info({:propagation_updated, _valid_times}, socket) do
|
||||
band = socket.assigns.selected_band
|
||||
valid_times = Propagation.available_valid_times(band)
|
||||
|
|
@ -192,6 +233,8 @@ defmodule MicrowavepropWeb.MapLive do
|
|||
|
||||
scores = Propagation.scores_at(band, selected_time, socket.assigns.bounds)
|
||||
|
||||
send(self(), :preload_forecast)
|
||||
|
||||
socket =
|
||||
socket
|
||||
|> assign(valid_times: valid_times, selected_time: selected_time)
|
||||
|
|
@ -220,6 +263,15 @@ defmodule MicrowavepropWeb.MapLive do
|
|||
{:noreply, socket}
|
||||
end
|
||||
|
||||
def handle_async({:rain_scatter, lat, lon, _band}, {:ok, scatter}, socket) do
|
||||
{:noreply, push_event(socket, "rain_scatter_update", %{lat: lat, lon: lon, rain_scatter: scatter})}
|
||||
end
|
||||
|
||||
def handle_async({:rain_scatter, _lat, _lon, _band}, {:exit, reason}, socket) do
|
||||
Logger.warning("Rain scatter fetch failed: #{inspect(reason)}")
|
||||
{:noreply, socket}
|
||||
end
|
||||
|
||||
defp push_timeline(socket) do
|
||||
times =
|
||||
Enum.map(socket.assigns.valid_times, fn t ->
|
||||
|
|
|
|||
|
|
@ -16,7 +16,8 @@ defmodule Microwaveprop.Propagation.ScoreCacheTest do
|
|||
test "returns cached scores after put" do
|
||||
scores = [%{lat: 32.0, lon: -97.0, score: 75}]
|
||||
ScoreCache.put(10_000, ~U[2026-04-12 12:00:00Z], scores)
|
||||
assert {:ok, ^scores} = ScoreCache.fetch(10_000, ~U[2026-04-12 12:00:00Z])
|
||||
assert {:ok, [%{lat: 32.0, lon: -97.0, score: 75}]} =
|
||||
ScoreCache.fetch(10_000, ~U[2026-04-12 12:00:00Z])
|
||||
end
|
||||
|
||||
test "isolates entries by band" do
|
||||
|
|
@ -94,12 +95,64 @@ defmodule Microwaveprop.Propagation.ScoreCacheTest do
|
|||
end
|
||||
end
|
||||
|
||||
describe "fetch_point/4" do
|
||||
test "returns :miss when nothing cached" do
|
||||
assert ScoreCache.fetch_point(10_000, ~U[2026-04-12 12:00:00Z], 32.0, -97.0) == :miss
|
||||
end
|
||||
|
||||
test "returns :miss when the point is not in the cached grid" do
|
||||
ScoreCache.put(10_000, ~U[2026-04-12 12:00:00Z], [
|
||||
%{lat: 32.0, lon: -97.0, score: 75}
|
||||
])
|
||||
|
||||
assert ScoreCache.fetch_point(10_000, ~U[2026-04-12 12:00:00Z], 40.0, -74.0) == :miss
|
||||
end
|
||||
|
||||
test "returns the score for a cached point" do
|
||||
ScoreCache.put(10_000, ~U[2026-04-12 12:00:00Z], [
|
||||
%{lat: 32.0, lon: -97.0, score: 75},
|
||||
%{lat: 33.0, lon: -97.0, score: 80}
|
||||
])
|
||||
|
||||
assert {:ok, 75} = ScoreCache.fetch_point(10_000, ~U[2026-04-12 12:00:00Z], 32.0, -97.0)
|
||||
assert {:ok, 80} = ScoreCache.fetch_point(10_000, ~U[2026-04-12 12:00:00Z], 33.0, -97.0)
|
||||
end
|
||||
end
|
||||
|
||||
describe "valid_times/1" do
|
||||
test "returns empty list when nothing cached" do
|
||||
assert ScoreCache.valid_times(10_000) == []
|
||||
end
|
||||
|
||||
test "returns sorted valid_times for a band" do
|
||||
ScoreCache.put(10_000, ~U[2026-04-12 14:00:00Z], [])
|
||||
ScoreCache.put(10_000, ~U[2026-04-12 12:00:00Z], [])
|
||||
ScoreCache.put(10_000, ~U[2026-04-12 13:00:00Z], [])
|
||||
|
||||
assert ScoreCache.valid_times(10_000) == [
|
||||
~U[2026-04-12 12:00:00Z],
|
||||
~U[2026-04-12 13:00:00Z],
|
||||
~U[2026-04-12 14:00:00Z]
|
||||
]
|
||||
end
|
||||
|
||||
test "only returns times for the requested band" do
|
||||
ScoreCache.put(10_000, ~U[2026-04-12 12:00:00Z], [])
|
||||
ScoreCache.put(24_000, ~U[2026-04-12 13:00:00Z], [])
|
||||
|
||||
assert ScoreCache.valid_times(10_000) == [~U[2026-04-12 12:00:00Z]]
|
||||
assert ScoreCache.valid_times(24_000) == [~U[2026-04-12 13:00:00Z]]
|
||||
end
|
||||
end
|
||||
|
||||
describe "broadcast_put/3" do
|
||||
test "populates the cache on the local node" do
|
||||
scores = [%{lat: 32.0, lon: -97.0, score: 80}]
|
||||
ScoreCache.broadcast_put(10_000, ~U[2026-04-12 12:00:00Z], scores)
|
||||
ScoreCache.sync()
|
||||
assert {:ok, ^scores} = ScoreCache.fetch(10_000, ~U[2026-04-12 12:00:00Z])
|
||||
|
||||
assert {:ok, [%{lat: 32.0, lon: -97.0, score: 80}]} =
|
||||
ScoreCache.fetch(10_000, ~U[2026-04-12 12:00:00Z])
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -124,6 +124,84 @@ defmodule Microwaveprop.PropagationTest do
|
|||
end
|
||||
end
|
||||
|
||||
describe "point_forecast/3 with cache" do
|
||||
test "returns forecast timeline from cached scores when warm" do
|
||||
t1 = DateTime.truncate(DateTime.add(DateTime.utc_now(), 3600, :second), :second)
|
||||
t2 = DateTime.add(t1, 3600, :second)
|
||||
|
||||
ScoreCache.put(10_000, t1, [%{lat: 32.875, lon: -97.0, score: 70}])
|
||||
ScoreCache.put(10_000, t2, [%{lat: 32.875, lon: -97.0, score: 75}])
|
||||
|
||||
assert Repo.aggregate(GridScore, :count) == 0
|
||||
|
||||
result = Propagation.point_forecast(10_000, 32.875, -97.0)
|
||||
|
||||
assert [
|
||||
%{valid_time: ^t1, score: 70},
|
||||
%{valid_time: ^t2, score: 75}
|
||||
] = result
|
||||
end
|
||||
|
||||
test "filters past valid_times out" do
|
||||
past = DateTime.add(DateTime.utc_now(), -3600, :second) |> DateTime.truncate(:second)
|
||||
future = DateTime.add(DateTime.utc_now(), 3600, :second) |> DateTime.truncate(:second)
|
||||
|
||||
ScoreCache.put(10_000, past, [%{lat: 32.875, lon: -97.0, score: 60}])
|
||||
ScoreCache.put(10_000, future, [%{lat: 32.875, lon: -97.0, score: 80}])
|
||||
|
||||
result = Propagation.point_forecast(10_000, 32.875, -97.0)
|
||||
assert [%{valid_time: ^future, score: 80}] = result
|
||||
end
|
||||
|
||||
test "snaps coordinates to the nearest grid point" do
|
||||
valid_time = DateTime.truncate(DateTime.add(DateTime.utc_now(), 3600, :second), :second)
|
||||
ScoreCache.put(10_000, valid_time, [%{lat: 32.875, lon: -97.0, score: 65}])
|
||||
|
||||
# Off-grid query snaps to 32.875, -97.0
|
||||
result = Propagation.point_forecast(10_000, 32.88, -96.98)
|
||||
assert [%{valid_time: ^valid_time, score: 65}] = result
|
||||
end
|
||||
|
||||
test "returns empty list when no cached or DB data for the point" do
|
||||
assert Propagation.point_forecast(10_000, 32.875, -97.0) == []
|
||||
end
|
||||
end
|
||||
|
||||
describe "available_valid_times/1 with cache" do
|
||||
test "returns cached valid_times without hitting the DB when cache is warm" do
|
||||
t1 = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
t2 = DateTime.add(t1, 3600, :second)
|
||||
|
||||
ScoreCache.put(10_000, t1, [])
|
||||
ScoreCache.put(10_000, t2, [])
|
||||
|
||||
assert Repo.aggregate(GridScore, :count) == 0
|
||||
assert Propagation.available_valid_times(10_000) == [t1, t2]
|
||||
end
|
||||
|
||||
test "filters cached times older than the 1-hour cutoff" do
|
||||
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
stale = DateTime.add(now, -7200, :second)
|
||||
fresh = DateTime.add(now, 1800, :second)
|
||||
|
||||
ScoreCache.put(10_000, stale, [])
|
||||
ScoreCache.put(10_000, fresh, [])
|
||||
|
||||
assert Propagation.available_valid_times(10_000) == [fresh]
|
||||
end
|
||||
|
||||
test "falls back to DB on cold cache" do
|
||||
valid_time = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
|
||||
Propagation.upsert_scores([
|
||||
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: %{}}
|
||||
])
|
||||
|
||||
assert ScoreCache.valid_times(10_000) == []
|
||||
assert Propagation.available_valid_times(10_000) == [valid_time]
|
||||
end
|
||||
end
|
||||
|
||||
describe "warm_cache_and_broadcast/2" do
|
||||
test "loads scores from DB into the cache" do
|
||||
valid_time = ~U[2026-07-15 13:00:00Z]
|
||||
|
|
|
|||
47
test/microwaveprop/weather/nexrad_cache_test.exs
Normal file
47
test/microwaveprop/weather/nexrad_cache_test.exs
Normal file
|
|
@ -0,0 +1,47 @@
|
|||
defmodule Microwaveprop.Weather.NexradCacheTest do
|
||||
use ExUnit.Case, async: false
|
||||
|
||||
alias Microwaveprop.Weather.NexradCache
|
||||
|
||||
setup do
|
||||
NexradCache.clear()
|
||||
:ok
|
||||
end
|
||||
|
||||
describe "fetch/1" do
|
||||
test "returns :miss when nothing is cached" do
|
||||
assert NexradCache.fetch(~U[2026-04-12 12:00:00Z]) == :miss
|
||||
end
|
||||
|
||||
test "returns cached pixels + width after put" do
|
||||
NexradCache.put(~U[2026-04-12 12:00:00Z], <<1, 2, 3, 4>>, 2)
|
||||
assert {:ok, <<1, 2, 3, 4>>, 2} = NexradCache.fetch(~U[2026-04-12 12:00:00Z])
|
||||
end
|
||||
|
||||
test "isolates entries by timestamp" do
|
||||
NexradCache.put(~U[2026-04-12 12:00:00Z], <<1>>, 1)
|
||||
NexradCache.put(~U[2026-04-12 12:05:00Z], <<2>>, 1)
|
||||
|
||||
assert {:ok, <<1>>, 1} = NexradCache.fetch(~U[2026-04-12 12:00:00Z])
|
||||
assert {:ok, <<2>>, 1} = NexradCache.fetch(~U[2026-04-12 12:05:00Z])
|
||||
end
|
||||
end
|
||||
|
||||
describe "prune_older_than/1" do
|
||||
test "removes entries with a timestamp strictly before the cutoff" do
|
||||
NexradCache.put(~U[2026-04-12 11:50:00Z], <<1>>, 1)
|
||||
NexradCache.put(~U[2026-04-12 12:00:00Z], <<2>>, 1)
|
||||
|
||||
NexradCache.prune_older_than(~U[2026-04-12 11:55:00Z])
|
||||
|
||||
assert NexradCache.fetch(~U[2026-04-12 11:50:00Z]) == :miss
|
||||
assert {:ok, <<2>>, 1} = NexradCache.fetch(~U[2026-04-12 12:00:00Z])
|
||||
end
|
||||
|
||||
test "keeps entries equal to the cutoff" do
|
||||
NexradCache.put(~U[2026-04-12 12:00:00Z], <<1>>, 1)
|
||||
NexradCache.prune_older_than(~U[2026-04-12 12:00:00Z])
|
||||
assert {:ok, <<1>>, 1} = NexradCache.fetch(~U[2026-04-12 12:00:00Z])
|
||||
end
|
||||
end
|
||||
end
|
||||
Loading…
Add table
Reference in a new issue