Graham McIntire
725a7e6bda
Two related UI changes driven by the same data:
* Move the selected-layer description out of the sidebar and into a
dedicated top-right overlay on the map. The sidebar kept the same
layer buttons but dropped the descriptive text; the new overlay
shows group + layer name + description in one card. Mobile still
shows the description in the collapsible panel since it has no
top-right free real estate.
* Add a forecast-hour timeline bar at the bottom of the map, matching
the propagation map. WeatherMapLive enumerates ProfilesFile on mount
(the grid worker has been persisting f00..f18 on disk since
commit 07ffcf5), pushes data-valid-times for the JS hook to render
as buttons, and handles a new select_time event by reading the
per-hour ProfilesFile on demand via Weather.weather_grid_at/2.
weather_grid_at/2 deliberately skips the GridCache write path for
non-latest hours — caching 18 × 92k rows would add ~300 MB per pod.
A ~2 MB ETF decode per scrub is fast enough for a click.
Subscribes to propagation:pipeline so the timeline picks up new
forecast hours live as they land, without waiting for the full chain
to finish.
1116 lines
36 KiB
Elixir
1116 lines
36 KiB
Elixir
defmodule Microwaveprop.Weather do
|
||
@moduledoc false
|
||
|
||
import Ecto.Query
|
||
|
||
alias Microwaveprop.Propagation.ProfilesFile
|
||
alias Microwaveprop.Repo
|
||
alias Microwaveprop.Weather.Era5Profile
|
||
alias Microwaveprop.Weather.GridCache
|
||
alias Microwaveprop.Weather.HrrrNativeProfile
|
||
alias Microwaveprop.Weather.HrrrProfile
|
||
alias Microwaveprop.Weather.IemClient
|
||
alias Microwaveprop.Weather.IemreObservation
|
||
alias Microwaveprop.Weather.Metar5minObservation
|
||
alias Microwaveprop.Weather.RtmaObservation
|
||
alias Microwaveprop.Weather.SolarIndex
|
||
alias Microwaveprop.Weather.Sounding
|
||
alias Microwaveprop.Weather.SoundingParams
|
||
alias Microwaveprop.Weather.Station
|
||
alias Microwaveprop.Weather.SurfaceObservation
|
||
alias Microwaveprop.Weather.WeatherLayers
|
||
|
||
require Logger
|
||
|
||
# Approximate km per degree latitude
|
||
@km_per_deg_lat 111.0
|
||
|
||
@spec find_or_create_station(map()) :: {:ok, Station.t()} | {:error, Ecto.Changeset.t()}
|
||
def find_or_create_station(attrs) do
|
||
code = attrs[:station_code] || attrs["station_code"]
|
||
type = attrs[:station_type] || attrs["station_type"]
|
||
|
||
if code && type do
|
||
case Repo.get_by(Station, station_code: code, station_type: type) do
|
||
nil ->
|
||
%Station{}
|
||
|> Station.changeset(attrs)
|
||
|> Repo.insert()
|
||
|
||
station ->
|
||
{:ok, station}
|
||
end
|
||
else
|
||
%Station{}
|
||
|> Station.changeset(attrs)
|
||
|> Repo.insert()
|
||
end
|
||
end
|
||
|
||
@spec upsert_surface_observation(Station.t(), map()) :: {:ok, SurfaceObservation.t()} | {:error, Ecto.Changeset.t()}
|
||
def upsert_surface_observation(%Station{} = station, attrs) do
|
||
attrs = Map.put(attrs, :station_id, station.id)
|
||
|
||
%SurfaceObservation{}
|
||
|> SurfaceObservation.changeset(attrs)
|
||
|> Repo.insert(
|
||
on_conflict:
|
||
from(s in SurfaceObservation,
|
||
update: [
|
||
set: [
|
||
temp_f: fragment("EXCLUDED.temp_f"),
|
||
dewpoint_f: fragment("EXCLUDED.dewpoint_f"),
|
||
relative_humidity: fragment("EXCLUDED.relative_humidity"),
|
||
wind_speed_kts: fragment("EXCLUDED.wind_speed_kts"),
|
||
sea_level_pressure_mb: fragment("EXCLUDED.sea_level_pressure_mb"),
|
||
sky_condition: fragment("EXCLUDED.sky_condition"),
|
||
precip_1h_in: fragment("EXCLUDED.precip_1h_in"),
|
||
wx_codes: fragment("EXCLUDED.wx_codes"),
|
||
updated_at: fragment("EXCLUDED.updated_at")
|
||
]
|
||
],
|
||
where:
|
||
s.temp_f != fragment("EXCLUDED.temp_f") or
|
||
s.dewpoint_f != fragment("EXCLUDED.dewpoint_f") or
|
||
s.relative_humidity != fragment("EXCLUDED.relative_humidity") or
|
||
s.wind_speed_kts != fragment("EXCLUDED.wind_speed_kts") or
|
||
s.sea_level_pressure_mb != fragment("EXCLUDED.sea_level_pressure_mb")
|
||
),
|
||
conflict_target: [:station_id, :observed_at],
|
||
returning: true,
|
||
stale_error_field: :id
|
||
)
|
||
end
|
||
|
||
@spec upsert_sounding(Station.t(), map()) :: {:ok, Sounding.t()} | {:error, Ecto.Changeset.t()}
|
||
def upsert_sounding(%Station{} = station, attrs) do
|
||
attrs = Map.put(attrs, :station_id, station.id)
|
||
|
||
%Sounding{}
|
||
|> Sounding.changeset(attrs)
|
||
|> Repo.insert(
|
||
on_conflict:
|
||
from(s in Sounding,
|
||
update: [
|
||
set: [
|
||
profile: fragment("EXCLUDED.profile"),
|
||
level_count: fragment("EXCLUDED.level_count"),
|
||
surface_pressure_mb: fragment("EXCLUDED.surface_pressure_mb"),
|
||
surface_temp_c: fragment("EXCLUDED.surface_temp_c"),
|
||
surface_dewpoint_c: fragment("EXCLUDED.surface_dewpoint_c"),
|
||
surface_refractivity: fragment("EXCLUDED.surface_refractivity"),
|
||
min_refractivity_gradient: fragment("EXCLUDED.min_refractivity_gradient"),
|
||
boundary_layer_depth_m: fragment("EXCLUDED.boundary_layer_depth_m"),
|
||
precipitable_water_mm: fragment("EXCLUDED.precipitable_water_mm"),
|
||
k_index: fragment("EXCLUDED.k_index"),
|
||
lifted_index: fragment("EXCLUDED.lifted_index"),
|
||
ducting_detected: fragment("EXCLUDED.ducting_detected"),
|
||
duct_characteristics: fragment("EXCLUDED.duct_characteristics"),
|
||
updated_at: fragment("EXCLUDED.updated_at")
|
||
]
|
||
],
|
||
where:
|
||
s.level_count != fragment("EXCLUDED.level_count") or
|
||
s.surface_temp_c != fragment("EXCLUDED.surface_temp_c") or
|
||
s.surface_refractivity != fragment("EXCLUDED.surface_refractivity")
|
||
),
|
||
conflict_target: [:station_id, :observed_at],
|
||
returning: true,
|
||
stale_error_field: :id
|
||
)
|
||
end
|
||
|
||
@spec upsert_solar_index(map()) :: {:ok, SolarIndex.t()} | {:error, Ecto.Changeset.t()}
|
||
def upsert_solar_index(attrs) do
|
||
%SolarIndex{}
|
||
|> SolarIndex.changeset(attrs)
|
||
|> Repo.insert(
|
||
on_conflict:
|
||
from(s in SolarIndex,
|
||
update: [
|
||
set: [
|
||
sfi: fragment("EXCLUDED.sfi"),
|
||
sfi_adjusted: fragment("EXCLUDED.sfi_adjusted"),
|
||
sunspot_number: fragment("EXCLUDED.sunspot_number"),
|
||
ap_index: fragment("EXCLUDED.ap_index"),
|
||
kp_values: fragment("EXCLUDED.kp_values"),
|
||
updated_at: fragment("EXCLUDED.updated_at")
|
||
]
|
||
],
|
||
where:
|
||
s.sfi != fragment("EXCLUDED.sfi") or
|
||
s.ap_index != fragment("EXCLUDED.ap_index")
|
||
),
|
||
conflict_target: [:date],
|
||
returning: true,
|
||
stale_error_field: :id
|
||
)
|
||
end
|
||
|
||
@spec has_surface_observations?(Ecto.UUID.t(), DateTime.t(), DateTime.t()) :: boolean()
|
||
def has_surface_observations?(station_id, start_dt, end_dt) do
|
||
SurfaceObservation
|
||
|> where([o], o.station_id == ^station_id)
|
||
|> where([o], o.observed_at >= ^start_dt and o.observed_at <= ^end_dt)
|
||
|> Repo.exists?()
|
||
end
|
||
|
||
@doc "Returns a MapSet of station_ids that have surface observations in the time window."
|
||
@spec station_ids_with_surface_observations([Ecto.UUID.t()], DateTime.t(), DateTime.t()) :: MapSet.t(Ecto.UUID.t())
|
||
def station_ids_with_surface_observations(station_ids, start_dt, end_dt) do
|
||
SurfaceObservation
|
||
|> where([o], o.station_id in ^station_ids)
|
||
|> where([o], o.observed_at >= ^start_dt and o.observed_at <= ^end_dt)
|
||
|> select([o], o.station_id)
|
||
|> distinct(true)
|
||
|> Repo.all()
|
||
|> MapSet.new()
|
||
end
|
||
|
||
@spec has_sounding?(Ecto.UUID.t(), DateTime.t()) :: boolean()
|
||
def has_sounding?(station_id, observed_at) do
|
||
Sounding
|
||
|> where([s], s.station_id == ^station_id and s.observed_at == ^observed_at)
|
||
|> Repo.exists?()
|
||
end
|
||
|
||
@doc "Returns a MapSet of {station_id, observed_at} tuples that have soundings."
|
||
@spec station_ids_with_soundings([Ecto.UUID.t()], [DateTime.t()]) :: MapSet.t({Ecto.UUID.t(), DateTime.t()})
|
||
def station_ids_with_soundings(station_ids, sounding_times) do
|
||
Sounding
|
||
|> where([s], s.station_id in ^station_ids and s.observed_at in ^sounding_times)
|
||
|> select([s], {s.station_id, s.observed_at})
|
||
|> distinct(true)
|
||
|> Repo.all()
|
||
|> MapSet.new()
|
||
end
|
||
|
||
@doc "Batch upsert solar indices using insert_all in chunks of 500."
|
||
@spec upsert_solar_indices_batch([map()]) :: non_neg_integer()
|
||
def upsert_solar_indices_batch(records) do
|
||
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||
|
||
records
|
||
|> Enum.chunk_every(500)
|
||
|> Enum.reduce(0, fn chunk, acc ->
|
||
entries =
|
||
Enum.map(chunk, fn attrs ->
|
||
%{
|
||
id: Ecto.UUID.generate(),
|
||
date: attrs[:date] || attrs.date,
|
||
sfi: attrs[:sfi],
|
||
sfi_adjusted: attrs[:sfi_adjusted],
|
||
sunspot_number: attrs[:sunspot_number],
|
||
ap_index: attrs[:ap_index],
|
||
kp_values: attrs[:kp_values],
|
||
inserted_at: now,
|
||
updated_at: now
|
||
}
|
||
end)
|
||
|
||
{count, _} =
|
||
Repo.insert_all(SolarIndex, entries,
|
||
on_conflict:
|
||
from(s in SolarIndex,
|
||
update: [
|
||
set: [
|
||
sfi: fragment("EXCLUDED.sfi"),
|
||
sfi_adjusted: fragment("EXCLUDED.sfi_adjusted"),
|
||
sunspot_number: fragment("EXCLUDED.sunspot_number"),
|
||
ap_index: fragment("EXCLUDED.ap_index"),
|
||
kp_values: fragment("EXCLUDED.kp_values"),
|
||
updated_at: fragment("EXCLUDED.updated_at")
|
||
]
|
||
],
|
||
where:
|
||
s.sfi != fragment("EXCLUDED.sfi") or
|
||
s.ap_index != fragment("EXCLUDED.ap_index")
|
||
),
|
||
conflict_target: [:date]
|
||
)
|
||
|
||
acc + count
|
||
end)
|
||
end
|
||
|
||
@spec get_solar_index(Date.t()) :: SolarIndex.t() | nil
|
||
def get_solar_index(date) do
|
||
Repo.get_by(SolarIndex, date: date)
|
||
end
|
||
|
||
@spec existing_solar_dates() :: MapSet.t(Date.t())
|
||
def existing_solar_dates do
|
||
SolarIndex
|
||
|> select([s], s.date)
|
||
|> Repo.all()
|
||
|> MapSet.new()
|
||
end
|
||
|
||
@spec sync_stations!() :: :ok
|
||
def sync_stations! do
|
||
asos =
|
||
for s <-
|
||
~w(AK AL AR AZ CA CO CT DE FL GA HI IA ID IL IN KS KY LA MA MD ME MI MN MO MS MT NC ND NE NH NJ NM NV NY OH OK OR PA RI SC SD TN TX UT VA VT WA WI WV WY),
|
||
do: "#{s}_ASOS"
|
||
|
||
for network <- asos ++ ["RAOB"] do
|
||
sync_network(network)
|
||
Process.sleep(200)
|
||
end
|
||
|
||
count = Repo.aggregate(Station, :count)
|
||
Logger.info("Weather stations sync complete: #{count} total")
|
||
:ok
|
||
end
|
||
|
||
defp sync_network(network) do
|
||
type = if String.contains?(network, "ASOS"), do: "asos", else: "sounding"
|
||
|
||
case IemClient.fetch_network(network) do
|
||
{:ok, stations} ->
|
||
for s <- stations do
|
||
%Station{}
|
||
|> Station.changeset(Map.put(s, :station_type, type))
|
||
|> Repo.insert(on_conflict: :nothing, conflict_target: [:station_code, :station_type])
|
||
end
|
||
|
||
Logger.info("Synced #{length(stations)} stations from #{network}")
|
||
|
||
{:error, e} ->
|
||
Logger.warning("Failed to sync #{network}: #{inspect(e)}")
|
||
end
|
||
end
|
||
|
||
@spec nearby_stations(float(), float(), String.t(), number()) :: [Station.t()]
|
||
def nearby_stations(lat, lon, station_type, radius_km) do
|
||
dlat = radius_km / @km_per_deg_lat
|
||
dlon = radius_km / (@km_per_deg_lat * :math.cos(lat * :math.pi() / 180))
|
||
|
||
Station
|
||
|> where([s], s.station_type == ^station_type)
|
||
|> where(
|
||
[s],
|
||
s.lat >= ^(lat - dlat) and s.lat <= ^(lat + dlat) and
|
||
s.lon >= ^(lon - dlon) and s.lon <= ^(lon + dlon)
|
||
)
|
||
|> Repo.all()
|
||
end
|
||
|
||
@spec sounding_times_around(DateTime.t()) :: [DateTime.t()]
|
||
def sounding_times_around(dt) do
|
||
date = DateTime.to_date(dt)
|
||
|
||
times =
|
||
if dt.hour < 12 do
|
||
[
|
||
DateTime.new!(Date.add(date, -1), ~T[12:00:00], "Etc/UTC"),
|
||
DateTime.new!(date, ~T[00:00:00], "Etc/UTC")
|
||
]
|
||
else
|
||
[
|
||
DateTime.new!(date, ~T[00:00:00], "Etc/UTC"),
|
||
DateTime.new!(date, ~T[12:00:00], "Etc/UTC")
|
||
]
|
||
end
|
||
|
||
Enum.uniq(times)
|
||
end
|
||
|
||
@spec weather_for_contact(map(), keyword()) :: %{
|
||
surface_observations: [SurfaceObservation.t()],
|
||
soundings: [Sounding.t()]
|
||
}
|
||
def weather_for_contact(contact_params, opts \\ []) do
|
||
lat = contact_params[:lat] || contact_params.lat
|
||
lon = contact_params[:lon] || contact_params.lon
|
||
timestamp = contact_params[:timestamp] || contact_params.timestamp
|
||
|
||
radius_km = Keyword.get(opts, :radius_km, 150)
|
||
time_window_hours = Keyword.get(opts, :time_window_hours, 6)
|
||
|
||
# Bounding box in degrees
|
||
dlat = radius_km / @km_per_deg_lat
|
||
dlon = radius_km / (@km_per_deg_lat * :math.cos(lat * :math.pi() / 180))
|
||
|
||
time_start = DateTime.add(timestamp, -time_window_hours * 3600, :second)
|
||
time_end = DateTime.add(timestamp, time_window_hours * 3600, :second)
|
||
|
||
station_ids =
|
||
Station
|
||
|> where(
|
||
[s],
|
||
s.lat >= ^(lat - dlat) and s.lat <= ^(lat + dlat) and
|
||
s.lon >= ^(lon - dlon) and s.lon <= ^(lon + dlon)
|
||
)
|
||
|> select([s], s.id)
|
||
|
||
surface_observations =
|
||
SurfaceObservation
|
||
|> where([o], o.station_id in subquery(station_ids))
|
||
|> where([o], o.observed_at >= ^time_start and o.observed_at <= ^time_end)
|
||
|> preload(:station)
|
||
|> Repo.all()
|
||
|
||
soundings =
|
||
Sounding
|
||
|> where([s], s.station_id in subquery(station_ids))
|
||
|> where([s], s.observed_at >= ^time_start and s.observed_at <= ^time_end)
|
||
|> preload(:station)
|
||
|> Repo.all()
|
||
|
||
%{surface_observations: surface_observations, soundings: soundings}
|
||
end
|
||
|
||
@spec latest_grid_valid_time() :: DateTime.t() | nil
|
||
def latest_grid_valid_time do
|
||
cond do
|
||
vt = GridCache.latest_valid_time() -> vt
|
||
vt = ProfilesFile.latest_valid_time() -> vt
|
||
true -> latest_grid_valid_time_from_db()
|
||
end
|
||
end
|
||
|
||
# Last-resort fallback for historical data sitting in the legacy
|
||
# hrrr_profiles table. PropagationGridWorker no longer writes
|
||
# grid-point rows there, so in steady state this returns nil and
|
||
# the ProfilesFile fallback above is the real source of truth.
|
||
defp latest_grid_valid_time_from_db do
|
||
Repo.one(
|
||
from(h in HrrrProfile,
|
||
where: h.is_grid_point == true,
|
||
select: max(h.valid_time)
|
||
)
|
||
)
|
||
end
|
||
|
||
@doc """
|
||
Cache-only read for the /weather LiveView mount hot path. Returns whatever
|
||
is in `GridCache` for the latest valid_time and fires a deduped background
|
||
fill task on a miss instead of blocking. The async task broadcasts
|
||
`weather:updated` when done, triggering every connected LiveView to refresh.
|
||
|
||
Callers that genuinely need synchronous data (tests, scripts) should use
|
||
`load_weather_grid/1` instead.
|
||
"""
|
||
@spec latest_weather_grid(map()) :: [map()]
|
||
def latest_weather_grid(bounds) do
|
||
case latest_grid_valid_time() do
|
||
nil ->
|
||
[]
|
||
|
||
latest_vt ->
|
||
case GridCache.fetch_bounds(latest_vt, bounds) do
|
||
{:ok, rows} ->
|
||
rows
|
||
|
||
:miss ->
|
||
kickoff_async_grid_fill(latest_vt)
|
||
[]
|
||
end
|
||
end
|
||
end
|
||
|
||
@doc """
|
||
Synchronous cache-or-DB read. Blocks for several seconds on a cold cache
|
||
because `load_weather_grid_from_db/1` parses 176k rows with JSONB columns.
|
||
Used by tests and by `weather_point_detail/3` fallbacks. LiveView callers
|
||
should prefer `latest_weather_grid/1`.
|
||
"""
|
||
@spec load_weather_grid(map()) :: [map()]
|
||
def load_weather_grid(bounds) do
|
||
case latest_grid_valid_time() do
|
||
nil ->
|
||
[]
|
||
|
||
latest_vt ->
|
||
case GridCache.fetch_bounds(latest_vt, bounds) do
|
||
{:ok, rows} ->
|
||
rows
|
||
|
||
:miss ->
|
||
full = load_grid_rows_for(latest_vt)
|
||
GridCache.put(latest_vt, full)
|
||
filter_weather_bounds(full, bounds)
|
||
end
|
||
end
|
||
end
|
||
|
||
@doc """
|
||
All persisted weather valid_times sorted ascending. The grid worker
|
||
writes a ProfilesFile for every forecast hour (f00..f18), so this
|
||
enumerates the 19-entry forecast timeline the /weather page renders
|
||
at the bottom of the map.
|
||
"""
|
||
@spec available_weather_valid_times() :: [DateTime.t()]
|
||
def available_weather_valid_times do
|
||
ProfilesFile.list_valid_times()
|
||
end
|
||
|
||
@doc """
|
||
Read the weather grid for a specific `valid_time` and bounds. Like
|
||
`load_weather_grid/1` but takes the valid_time explicitly so the
|
||
timeline can scrub to any forecast hour, not just the analysis hour.
|
||
Returns `[]` if no profile file exists for that valid_time.
|
||
|
||
Deliberately does NOT write forecast-hour grids back into `GridCache`
|
||
on a miss: caching 18 forecast hours × 92k points would add ~300 MB
|
||
per pod. The ProfilesFile read is a single ~2 MB ETF decode per scrub,
|
||
which is fast enough for a user click.
|
||
"""
|
||
@spec weather_grid_at(DateTime.t(), map()) :: [map()]
|
||
def weather_grid_at(%DateTime{} = valid_time, bounds) do
|
||
case GridCache.fetch_bounds(valid_time, bounds) do
|
||
{:ok, rows} ->
|
||
rows
|
||
|
||
:miss ->
|
||
case ProfilesFile.read(valid_time) do
|
||
{:ok, grid_data} ->
|
||
grid_data
|
||
|> build_grid_cache_rows(valid_time)
|
||
|> filter_weather_bounds(bounds)
|
||
|
||
{:error, _} ->
|
||
[]
|
||
end
|
||
end
|
||
end
|
||
|
||
# Build derived GridCache rows for a valid_time from whichever
|
||
# source has data: the persisted ProfilesFile first (hot path in
|
||
# steady state), then the legacy hrrr_profiles table (historical
|
||
# data only).
|
||
defp load_grid_rows_for(valid_time) do
|
||
case ProfilesFile.read(valid_time) do
|
||
{:ok, grid_data} -> build_grid_cache_rows(grid_data, valid_time)
|
||
{:error, _} -> load_weather_grid_from_db(valid_time)
|
||
end
|
||
end
|
||
|
||
defp filter_weather_bounds(rows, nil), do: rows
|
||
|
||
defp filter_weather_bounds(rows, %{"south" => s, "north" => n, "west" => w, "east" => e}) do
|
||
Enum.filter(rows, fn %{lat: lat, lon: lon} ->
|
||
lat >= s and lat <= n and lon >= w and lon <= e
|
||
end)
|
||
end
|
||
|
||
defp kickoff_async_grid_fill(valid_time) do
|
||
if GridCache.claim_fill(valid_time) do
|
||
Task.start(fn ->
|
||
try do
|
||
Logger.info("Weather.grid_cache async fill starting for #{valid_time}")
|
||
warm_grid_cache_and_broadcast(valid_time)
|
||
|
||
Phoenix.PubSub.broadcast(
|
||
Microwaveprop.PubSub,
|
||
"weather:updated",
|
||
{:weather_updated, valid_time}
|
||
)
|
||
|
||
Logger.info("Weather.grid_cache async fill complete for #{valid_time}")
|
||
rescue
|
||
e ->
|
||
Logger.error("Weather.grid_cache async fill failed: #{inspect(e)}")
|
||
after
|
||
GridCache.release_fill(valid_time)
|
||
end
|
||
end)
|
||
end
|
||
|
||
:ok
|
||
end
|
||
|
||
defp load_weather_grid_from_db(latest_vt) do
|
||
from(h in HrrrProfile,
|
||
where: h.valid_time == ^latest_vt and h.is_grid_point == true,
|
||
select: %{
|
||
lat: h.lat,
|
||
lon: h.lon,
|
||
valid_time: h.valid_time,
|
||
temperature: h.surface_temp_c,
|
||
dewpoint_depression: fragment("? - ?", h.surface_temp_c, h.surface_dewpoint_c),
|
||
bl_height: h.hpbl_m,
|
||
pwat: h.pwat_mm,
|
||
refractivity_gradient: h.min_refractivity_gradient,
|
||
ducting: h.ducting_detected,
|
||
surface_pressure_mb: h.surface_pressure_mb,
|
||
surface_temp_c: h.surface_temp_c,
|
||
surface_dewpoint_c: h.surface_dewpoint_c,
|
||
surface_refractivity: h.surface_refractivity,
|
||
profile: h.profile,
|
||
duct_characteristics: h.duct_characteristics
|
||
}
|
||
)
|
||
|> Repo.all(timeout: 60_000)
|
||
|> Enum.map(&derive_and_clean/1)
|
||
end
|
||
|
||
@doc """
|
||
Eagerly populate the `GridCache` with the full CONUS weather grid for
|
||
`valid_time` and broadcast it to every node in the cluster. Used by the cold
|
||
cache fill path (`kickoff_async_grid_fill/1`) — prefer
|
||
`build_grid_cache_rows/2` inside `PropagationGridWorker`, which already has
|
||
the in-memory grid data and avoids the ~20s JSONB round trip.
|
||
"""
|
||
@spec warm_grid_cache_and_broadcast(DateTime.t()) :: :ok
|
||
def warm_grid_cache_and_broadcast(valid_time) do
|
||
rows = load_grid_rows_for(valid_time)
|
||
GridCache.broadcast_put(valid_time, rows)
|
||
:ok
|
||
end
|
||
|
||
@doc """
|
||
Warm the local `GridCache` from the latest persisted `ProfilesFile`
|
||
on pod startup. Makes `/weather` usable immediately after a deploy
|
||
instead of waiting for the next hourly PropagationGridWorker run.
|
||
Local put only — every node reads the same NFS mount so no need to
|
||
broadcast.
|
||
"""
|
||
@spec warm_grid_cache_from_latest_profile() :: :ok
|
||
def warm_grid_cache_from_latest_profile do
|
||
case ProfilesFile.latest_valid_time() do
|
||
nil ->
|
||
:ok
|
||
|
||
valid_time ->
|
||
try do
|
||
rows = load_grid_rows_for(valid_time)
|
||
GridCache.put(valid_time, rows)
|
||
Logger.info("Weather: warmed GridCache from ProfilesFile for #{valid_time} (#{length(rows)} rows)")
|
||
rescue
|
||
e ->
|
||
Logger.warning("Weather: ProfilesFile warm failed: #{inspect(e)}")
|
||
end
|
||
|
||
:ok
|
||
end
|
||
end
|
||
|
||
@doc """
|
||
Build derived weather grid cache rows directly from the in-memory HRRR
|
||
`grid_data` that `PropagationGridWorker` already has after fetching a
|
||
forecast hour. Avoids the 15s+ DB round trip that `load_weather_grid_from_db/1`
|
||
pays to SELECT + JSONB-decode 92k hrrr_profiles rows — the source of the
|
||
crash loop that stopped forecast hours from being written.
|
||
|
||
`grid_data` is `%{{lat, lon} => profile_map}` as produced by
|
||
`HrrrClient.fetch_grid/3` (optionally enriched via native duct merge). The
|
||
output matches the shape of `load_weather_grid_from_db/1` after
|
||
`derive_and_clean/1`.
|
||
"""
|
||
@spec build_grid_cache_rows(%{{float(), float()} => map()}, DateTime.t()) :: [map()]
|
||
def build_grid_cache_rows(grid_data, valid_time) do
|
||
Enum.flat_map(grid_data, fn {{lat, lon}, profile} ->
|
||
build_grid_cache_row(lat, lon, profile, valid_time)
|
||
end)
|
||
end
|
||
|
||
defp build_grid_cache_row(lat, lon, profile, valid_time) do
|
||
temp_c = profile[:surface_temp_c]
|
||
|
||
if is_nil(temp_c) or temp_c < -80 or temp_c > 60 do
|
||
[]
|
||
else
|
||
sounding = derive_sounding(profile[:profile])
|
||
|
||
row = %{
|
||
lat: lat,
|
||
lon: lon,
|
||
valid_time: valid_time,
|
||
temperature: temp_c,
|
||
dewpoint_depression: depression(temp_c, profile[:surface_dewpoint_c]),
|
||
bl_height: profile[:hpbl_m],
|
||
pwat: profile[:pwat_mm],
|
||
refractivity_gradient: sounding[:min_refractivity_gradient],
|
||
ducting: sounding[:ducting_detected],
|
||
surface_pressure_mb: profile[:surface_pressure_mb],
|
||
surface_temp_c: temp_c,
|
||
surface_dewpoint_c: profile[:surface_dewpoint_c],
|
||
surface_refractivity: sounding[:surface_refractivity],
|
||
profile: profile[:profile] || [],
|
||
duct_characteristics: sounding[:duct_characteristics]
|
||
}
|
||
|
||
[derive_and_clean(row)]
|
||
end
|
||
end
|
||
|
||
defp derive_sounding(profile) when is_list(profile) and length(profile) >= 3 do
|
||
SoundingParams.derive(profile) || %{}
|
||
end
|
||
|
||
defp derive_sounding(_), do: %{}
|
||
|
||
defp depression(nil, _), do: nil
|
||
defp depression(_, nil), do: nil
|
||
defp depression(t, d), do: t - d
|
||
|
||
@spec weather_point_detail(float(), float(), DateTime.t()) :: map() | nil
|
||
def weather_point_detail(lat, lon, valid_time) do
|
||
step = 0.125
|
||
snapped_lat = Float.round(Float.round(lat / step) * step, 3)
|
||
snapped_lon = Float.round(Float.round(lon / step) * step, 3)
|
||
|
||
case GridCache.fetch_point(valid_time, snapped_lat, snapped_lon) do
|
||
{:ok, row} ->
|
||
row
|
||
|
||
:miss ->
|
||
case weather_point_detail_from_profiles(valid_time, snapped_lat, snapped_lon) do
|
||
nil -> weather_point_detail_from_db(valid_time, snapped_lat, snapped_lon)
|
||
row -> row
|
||
end
|
||
end
|
||
end
|
||
|
||
# Derive a single GridCache-shaped row from a persisted ProfilesFile
|
||
# entry for `(valid_time, lat, lon)`. Returns nil when the file
|
||
# doesn't exist or the point has no profile.
|
||
defp weather_point_detail_from_profiles(valid_time, snapped_lat, snapped_lon) do
|
||
case ProfilesFile.read_point(valid_time, snapped_lat, snapped_lon) do
|
||
nil ->
|
||
nil
|
||
|
||
profile ->
|
||
case build_grid_cache_rows(%{{snapped_lat, snapped_lon} => profile}, valid_time) do
|
||
[row] -> row
|
||
_ -> nil
|
||
end
|
||
end
|
||
end
|
||
|
||
defp weather_point_detail_from_db(valid_time, snapped_lat, snapped_lon) do
|
||
from(h in HrrrProfile,
|
||
where: h.lat == ^snapped_lat and h.lon == ^snapped_lon and h.valid_time == ^valid_time,
|
||
select: %{
|
||
lat: h.lat,
|
||
lon: h.lon,
|
||
valid_time: h.valid_time,
|
||
temperature: h.surface_temp_c,
|
||
dewpoint_depression: fragment("? - ?", h.surface_temp_c, h.surface_dewpoint_c),
|
||
bl_height: h.hpbl_m,
|
||
pwat: h.pwat_mm,
|
||
refractivity_gradient: h.min_refractivity_gradient,
|
||
ducting: h.ducting_detected,
|
||
surface_pressure_mb: h.surface_pressure_mb,
|
||
surface_temp_c: h.surface_temp_c,
|
||
surface_dewpoint_c: h.surface_dewpoint_c,
|
||
surface_refractivity: h.surface_refractivity,
|
||
profile: h.profile,
|
||
duct_characteristics: h.duct_characteristics
|
||
}
|
||
)
|
||
|> Repo.one()
|
||
|> then(fn
|
||
nil -> nil
|
||
row -> derive_and_clean(row)
|
||
end)
|
||
end
|
||
|
||
defp derive_and_clean(row) do
|
||
derived = WeatherLayers.derive(row)
|
||
|
||
row
|
||
|> Map.merge(derived)
|
||
|> Map.drop([:profile, :duct_characteristics, :surface_temp_c, :surface_dewpoint_c])
|
||
end
|
||
|
||
@spec upsert_hrrr_profile(map()) :: {:ok, HrrrProfile.t()} | {:error, Ecto.Changeset.t()}
|
||
def upsert_hrrr_profile(attrs) do
|
||
%HrrrProfile{}
|
||
|> HrrrProfile.changeset(attrs)
|
||
|> Repo.insert(
|
||
on_conflict: :nothing,
|
||
conflict_target: [:lat, :lon, :valid_time]
|
||
)
|
||
end
|
||
|
||
@spec upsert_hrrr_profiles_batch([map()], keyword()) :: {non_neg_integer(), nil}
|
||
def upsert_hrrr_profiles_batch(profiles, _opts \\ []) do
|
||
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||
|
||
profiles
|
||
|> Enum.chunk_every(500)
|
||
|> Enum.reduce({0, nil}, fn chunk, {total_count, _} ->
|
||
entries =
|
||
Enum.map(chunk, fn attrs ->
|
||
is_gp =
|
||
rem(round(attrs.lat * 1000), 125) == 0 and
|
||
rem(round(attrs.lon * 1000), 125) == 0
|
||
|
||
Map.merge(attrs, %{
|
||
id: Ecto.UUID.generate(),
|
||
is_grid_point: is_gp,
|
||
inserted_at: now,
|
||
updated_at: now
|
||
})
|
||
end)
|
||
|
||
{count, rows} =
|
||
Repo.insert_all(HrrrProfile, entries,
|
||
on_conflict: :nothing,
|
||
conflict_target: [:lat, :lon, :valid_time]
|
||
)
|
||
|
||
{total_count + count, rows}
|
||
end)
|
||
end
|
||
|
||
@spec has_hrrr_profile?(float(), float(), DateTime.t()) :: boolean()
|
||
def has_hrrr_profile?(lat, lon, valid_time) do
|
||
dlat = 0.07
|
||
dlon = 0.07
|
||
|
||
HrrrProfile
|
||
|> where(
|
||
[h],
|
||
h.lat >= ^(lat - dlat) and h.lat <= ^(lat + dlat) and
|
||
h.lon >= ^(lon - dlon) and h.lon <= ^(lon + dlon) and
|
||
h.valid_time == ^valid_time
|
||
)
|
||
|> Repo.exists?()
|
||
end
|
||
|
||
@spec hrrr_for_contact(map()) :: HrrrProfile.t() | nil
|
||
def hrrr_for_contact(%{pos1: nil}), do: nil
|
||
|
||
def hrrr_for_contact(contact) do
|
||
lat = contact.pos1["lat"]
|
||
lon = contact.pos1["lon"] || contact.pos1["lng"]
|
||
|
||
if lat && lon do
|
||
find_nearest_hrrr(lat, lon, contact.qso_timestamp)
|
||
end
|
||
end
|
||
|
||
@spec find_nearest_hrrr(float(), float(), DateTime.t()) :: HrrrProfile.t() | nil
|
||
def find_nearest_hrrr(lat, lon, timestamp) do
|
||
dlat = 0.07
|
||
dlon = 0.07
|
||
time_start = DateTime.add(timestamp, -3600, :second)
|
||
time_end = DateTime.add(timestamp, 3600, :second)
|
||
|
||
HrrrProfile
|
||
|> where(
|
||
[h],
|
||
h.lat >= ^(lat - dlat) and h.lat <= ^(lat + dlat) and
|
||
h.lon >= ^(lon - dlon) and h.lon <= ^(lon + dlon) and
|
||
h.valid_time >= ^time_start and h.valid_time <= ^time_end
|
||
)
|
||
|> order_by([h],
|
||
asc:
|
||
fragment(
|
||
"ABS(? - ?) + ABS(? - ?) + ABS(EXTRACT(EPOCH FROM ? - ?))",
|
||
h.lat,
|
||
^lat,
|
||
h.lon,
|
||
^lon,
|
||
h.valid_time,
|
||
^timestamp
|
||
)
|
||
)
|
||
|> limit(1)
|
||
|> Repo.one()
|
||
end
|
||
|
||
@spec hrrr_profiles_for_path(map()) :: [HrrrProfile.t()]
|
||
def hrrr_profiles_for_path(%{pos1: nil}), do: []
|
||
|
||
def hrrr_profiles_for_path(contact) do
|
||
contact
|
||
|> Microwaveprop.Radio.contact_path_points()
|
||
|> Enum.map(fn {lat, lon} -> find_nearest_hrrr(lat, lon, contact.qso_timestamp) end)
|
||
|> Enum.reject(&is_nil/1)
|
||
end
|
||
|
||
@spec find_nearest_native_profile(float(), float(), DateTime.t()) ::
|
||
HrrrNativeProfile.t() | nil
|
||
def find_nearest_native_profile(lat, lon, timestamp) do
|
||
dlat = 0.07
|
||
dlon = 0.07
|
||
time_start = DateTime.add(timestamp, -3600, :second)
|
||
time_end = DateTime.add(timestamp, 3600, :second)
|
||
|
||
HrrrNativeProfile
|
||
|> where(
|
||
[n],
|
||
n.lat >= ^(lat - dlat) and n.lat <= ^(lat + dlat) and
|
||
n.lon >= ^(lon - dlon) and n.lon <= ^(lon + dlon) and
|
||
n.valid_time >= ^time_start and n.valid_time <= ^time_end
|
||
)
|
||
|> order_by([n],
|
||
asc:
|
||
fragment(
|
||
"ABS(? - ?) + ABS(? - ?) + ABS(EXTRACT(EPOCH FROM ? - ?))",
|
||
n.lat,
|
||
^lat,
|
||
n.lon,
|
||
^lon,
|
||
n.valid_time,
|
||
^timestamp
|
||
)
|
||
)
|
||
|> limit(1)
|
||
|> Repo.one()
|
||
end
|
||
|
||
@doc """
|
||
Find the best available atmospheric profile for a contact.
|
||
Tries HRRR first (3 km, hourly), falls back to ERA5 (0.25°, hourly).
|
||
Returns the profile struct or nil.
|
||
"""
|
||
@spec best_profile_for_contact(map()) :: HrrrProfile.t() | Era5Profile.t() | nil
|
||
def best_profile_for_contact(contact) do
|
||
hrrr_for_contact(contact) || era5_for_contact(contact)
|
||
end
|
||
|
||
@doc "Find all atmospheric profiles along a contact's path, from any source."
|
||
@spec profiles_along_path(map()) :: [HrrrProfile.t() | Era5Profile.t()]
|
||
def profiles_along_path(contact) do
|
||
hrrr_path = hrrr_profiles_for_path(contact)
|
||
|
||
if hrrr_path == [] do
|
||
era5_profiles_for_path(contact)
|
||
else
|
||
hrrr_path
|
||
end
|
||
end
|
||
|
||
@spec era5_for_contact(map()) :: Era5Profile.t() | nil
|
||
def era5_for_contact(%{pos1: nil}), do: nil
|
||
|
||
def era5_for_contact(contact) do
|
||
lat = contact.pos1["lat"]
|
||
lon = contact.pos1["lon"] || contact.pos1["lng"]
|
||
|
||
if lat && lon do
|
||
find_nearest_era5(lat, lon, contact.qso_timestamp)
|
||
end
|
||
end
|
||
|
||
@spec era5_profiles_for_path(map()) :: [Era5Profile.t()]
|
||
def era5_profiles_for_path(%{pos1: nil}), do: []
|
||
|
||
def era5_profiles_for_path(contact) do
|
||
contact
|
||
|> Microwaveprop.Radio.contact_path_points()
|
||
|> Enum.map(fn {lat, lon} -> find_nearest_era5(lat, lon, contact.qso_timestamp) end)
|
||
|> Enum.reject(&is_nil/1)
|
||
end
|
||
|
||
@spec find_nearest_era5(float(), float(), DateTime.t()) :: Era5Profile.t() | nil
|
||
def find_nearest_era5(lat, lon, timestamp) do
|
||
dlat = 0.15
|
||
dlon = 0.15
|
||
time_start = DateTime.add(timestamp, -1800, :second)
|
||
time_end = DateTime.add(timestamp, 1800, :second)
|
||
|
||
Era5Profile
|
||
|> where(
|
||
[p],
|
||
p.lat >= ^(lat - dlat) and p.lat <= ^(lat + dlat) and
|
||
p.lon >= ^(lon - dlon) and p.lon <= ^(lon + dlon) and
|
||
p.valid_time >= ^time_start and p.valid_time <= ^time_end
|
||
)
|
||
|> order_by([p],
|
||
asc:
|
||
fragment(
|
||
"ABS(? - ?) + ABS(? - ?)",
|
||
p.lat,
|
||
^lat,
|
||
p.lon,
|
||
^lon
|
||
)
|
||
)
|
||
|> limit(1)
|
||
|> Repo.one()
|
||
end
|
||
|
||
@spec find_nearest_rtma(float(), float(), DateTime.t()) :: RtmaObservation.t() | nil
|
||
def find_nearest_rtma(lat, lon, timestamp) do
|
||
dlat = 0.05
|
||
dlon = 0.05
|
||
time_start = DateTime.add(timestamp, -900, :second)
|
||
time_end = DateTime.add(timestamp, 900, :second)
|
||
|
||
RtmaObservation
|
||
|> where(
|
||
[o],
|
||
o.lat >= ^(lat - dlat) and o.lat <= ^(lat + dlat) and
|
||
o.lon >= ^(lon - dlon) and o.lon <= ^(lon + dlon) and
|
||
o.valid_time >= ^time_start and o.valid_time <= ^time_end
|
||
)
|
||
|> order_by([o],
|
||
asc:
|
||
fragment(
|
||
"ABS(? - ?) + ABS(? - ?) + ABS(EXTRACT(EPOCH FROM ? - ?))",
|
||
o.lat,
|
||
^lat,
|
||
o.lon,
|
||
^lon,
|
||
o.valid_time,
|
||
^timestamp
|
||
)
|
||
)
|
||
|> limit(1)
|
||
|> Repo.one()
|
||
end
|
||
|
||
@spec round_to_hrrr_grid(float(), float()) :: {float(), float()}
|
||
def round_to_hrrr_grid(lat, lon) do
|
||
{Float.round(lat / 1.0, 2), Float.round(lon / 1.0, 2)}
|
||
end
|
||
|
||
@doc """
|
||
Delete every `is_grid_point = true` row from `hrrr_profiles`, regardless of
|
||
age. Grid-point profiles are historical — the propagation grid now lives in
|
||
`/data/scores` binary files, nothing reads `is_grid_point = true` rows
|
||
anymore, and forecast runs no longer write them. This purge walks each
|
||
partition directly so a single DELETE can't scan the whole parent table.
|
||
Preserves QSO-linked rows (`is_grid_point = false`), which remain the data
|
||
path for contact enrichment and the `/path` calculator.
|
||
"""
|
||
@spec purge_grid_point_profiles() :: non_neg_integer()
|
||
def purge_grid_point_profiles do
|
||
deleted =
|
||
Enum.reduce(hrrr_profile_partitions(), 0, fn partition, acc ->
|
||
%{num_rows: n} =
|
||
Repo.query!(
|
||
~s(DELETE FROM "#{partition}" WHERE is_grid_point = true),
|
||
[],
|
||
timeout: 600_000
|
||
)
|
||
|
||
if n > 0 do
|
||
require Logger
|
||
|
||
Logger.info("Purged #{n} grid-point rows from #{partition}")
|
||
end
|
||
|
||
acc + n
|
||
end)
|
||
|
||
deleted
|
||
end
|
||
|
||
@spec hrrr_profile_partitions() :: [String.t()]
|
||
defp hrrr_profile_partitions do
|
||
{:ok, %{rows: rows}} =
|
||
Repo.query(
|
||
"""
|
||
SELECT child.relname
|
||
FROM pg_inherits
|
||
JOIN pg_class parent ON pg_inherits.inhparent = parent.oid
|
||
JOIN pg_class child ON pg_inherits.inhrelid = child.oid
|
||
WHERE parent.relname = 'hrrr_profiles'
|
||
ORDER BY child.relname
|
||
""",
|
||
[],
|
||
timeout: 60_000
|
||
)
|
||
|
||
Enum.map(rows, fn [name] -> name end)
|
||
end
|
||
|
||
@spec round_to_iemre_grid(float(), float()) :: {float(), float()}
|
||
def round_to_iemre_grid(lat, lon) do
|
||
{Float.round(lat * 8) / 8, Float.round(lon * 8) / 8}
|
||
end
|
||
|
||
@spec upsert_iemre_observation(map()) :: {:ok, IemreObservation.t()} | {:error, Ecto.Changeset.t()}
|
||
def upsert_iemre_observation(attrs) do
|
||
%IemreObservation{}
|
||
|> IemreObservation.changeset(attrs)
|
||
|> Repo.insert(
|
||
on_conflict: :nothing,
|
||
conflict_target: [:lat, :lon, :date]
|
||
)
|
||
end
|
||
|
||
@spec has_iemre_observation?(float(), float(), Date.t()) :: boolean()
|
||
def has_iemre_observation?(lat, lon, date) do
|
||
IemreObservation
|
||
|> where([i], i.lat == ^lat and i.lon == ^lon and i.date == ^date)
|
||
|> Repo.exists?()
|
||
end
|
||
|
||
@spec iemre_for_contact(map()) :: IemreObservation.t() | nil
|
||
def iemre_for_contact(%{pos1: nil}), do: nil
|
||
|
||
def iemre_for_contact(contact) do
|
||
lat = contact.pos1["lat"]
|
||
lon = contact.pos1["lon"] || contact.pos1["lng"]
|
||
|
||
if lat && lon do
|
||
find_nearest_iemre(lat, lon, contact.qso_timestamp)
|
||
end
|
||
end
|
||
|
||
@spec find_nearest_iemre(float(), float(), DateTime.t()) :: IemreObservation.t() | nil
|
||
def find_nearest_iemre(lat, lon, timestamp) do
|
||
{rlat, rlon} = round_to_iemre_grid(lat, lon)
|
||
date = DateTime.to_date(timestamp)
|
||
|
||
IemreObservation
|
||
|> where([i], i.lat == ^rlat and i.lon == ^rlon and i.date == ^date)
|
||
|> Repo.one()
|
||
end
|
||
|
||
@spec iemre_for_path(map()) :: [IemreObservation.t()]
|
||
def iemre_for_path(%{pos1: nil}), do: []
|
||
|
||
def iemre_for_path(contact) do
|
||
contact
|
||
|> Microwaveprop.Radio.contact_path_points()
|
||
|> Enum.map(fn {lat, lon} -> find_nearest_iemre(lat, lon, contact.qso_timestamp) end)
|
||
|> Enum.reject(&is_nil/1)
|
||
end
|
||
|
||
@doc """
|
||
Find the nearest surface observation to a given (lat, lon, time),
|
||
preferring 5-minute METAR data when available, falling back to the
|
||
hourly `surface_observations` table.
|
||
|
||
Returns a map with `:temp_f`, `:dewpoint_f`, `:wind_speed_kts`,
|
||
`:observed_at`, etc. — the same shape regardless of which table the
|
||
data came from. Returns `nil` if neither source has data.
|
||
"""
|
||
@spec recent_surface_obs(float(), float(), DateTime.t()) :: Metar5minObservation.t() | SurfaceObservation.t() | nil
|
||
def recent_surface_obs(lat, lon, timestamp) do
|
||
dlat = 0.5
|
||
dlon = 0.5
|
||
time_start = DateTime.add(timestamp, -1800, :second)
|
||
time_end = DateTime.add(timestamp, 1800, :second)
|
||
|
||
station_ids =
|
||
Station
|
||
|> where(
|
||
[s],
|
||
s.lat >= ^(lat - dlat) and s.lat <= ^(lat + dlat) and
|
||
s.lon >= ^(lon - dlon) and s.lon <= ^(lon + dlon)
|
||
)
|
||
|> select([s], s.id)
|
||
|
||
# Try 5-min first
|
||
metar_5min =
|
||
Metar5minObservation
|
||
|> where([o], o.station_id in subquery(station_ids))
|
||
|> where([o], o.observed_at >= ^time_start and o.observed_at <= ^time_end)
|
||
|> order_by([o], asc: fragment("ABS(EXTRACT(EPOCH FROM ? - ?))", o.observed_at, ^timestamp))
|
||
|> limit(1)
|
||
|> Repo.one()
|
||
|
||
if metar_5min do
|
||
metar_5min
|
||
else
|
||
# Fall back to hourly
|
||
SurfaceObservation
|
||
|> where([o], o.station_id in subquery(station_ids))
|
||
|> where([o], o.observed_at >= ^time_start and o.observed_at <= ^time_end)
|
||
|> order_by([o], asc: fragment("ABS(EXTRACT(EPOCH FROM ? - ?))", o.observed_at, ^timestamp))
|
||
|> limit(1)
|
||
|> Repo.one()
|
||
end
|
||
end
|
||
end
|