Graham McIntire
a24ce027f4
The /weather LiveView was timing out on pod restart because latest_weather_grid/1 ran load_weather_grid_from_db/1 synchronously on cache miss — that query reads 176k hrrr_profiles rows with two huge JSONB columns (profile array + duct_characteristics), and JSONB decoding blocked the LiveView process past the 15-second pool ownership timeout. On cache miss latest_weather_grid/1 now returns empty immediately and kicks off a deduped background fill through GridCache.claim_fill/1 (a per-valid_time ETS lock so N concurrent mounts after restart don't each fire the slow query). When the fill completes it broadcasts weather:updated, and the handle_info in WeatherMapLive hits the warm cache and pushes rows over the socket. Added load_weather_grid/1 as a synchronous sibling for tests and any caller that genuinely needs inline data. Tests updated to use it and to clear GridCache in setup so ETS state doesn't leak between cases. Also untrack k8s/secret.yaml and add both it and k8s/*-secret.yaml to .gitignore — those manifests hold plaintext SMTP and database credentials that should not live in the repo. Secrets already in git history should be rotated separately.
898 lines
28 KiB
Elixir
898 lines
28 KiB
Elixir
defmodule Microwaveprop.Weather do
|
|
@moduledoc false
|
|
|
|
import Ecto.Query
|
|
|
|
alias Microwaveprop.Repo
|
|
alias Microwaveprop.Weather.Era5Profile
|
|
alias Microwaveprop.Weather.GridCache
|
|
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.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
|
|
case GridCache.latest_valid_time() do
|
|
%DateTime{} = vt ->
|
|
vt
|
|
|
|
nil ->
|
|
Repo.one(
|
|
from(h in HrrrProfile,
|
|
where: h.is_grid_point == true,
|
|
select: max(h.valid_time)
|
|
)
|
|
)
|
|
end
|
|
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_weather_grid_from_db(latest_vt)
|
|
GridCache.put(latest_vt, full)
|
|
filter_weather_bounds(full, bounds)
|
|
end
|
|
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()
|
|
|> 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. Called from
|
|
`PropagationGridWorker` after each hourly HRRR upsert so connected clients
|
|
see zero-DB pan/zoom latency on the `/weather` map.
|
|
"""
|
|
@spec warm_grid_cache_and_broadcast(DateTime.t()) :: :ok
|
|
def warm_grid_cache_and_broadcast(valid_time) do
|
|
rows = load_weather_grid_from_db(valid_time)
|
|
GridCache.broadcast_put(valid_time, rows)
|
|
:ok
|
|
end
|
|
|
|
@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 -> weather_point_detail_from_db(valid_time, snapped_lat, snapped_lon)
|
|
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
|
|
|
|
@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 HRRR grid profiles older than 24 hours that sit on the 0.125° grid.
|
|
Preserves QSO-linked profiles which are at arbitrary lat/lon positions.
|
|
With 19 forecast hours per run, this keeps ~2 runs worth of grid profiles.
|
|
"""
|
|
@spec prune_old_grid_profiles() :: non_neg_integer()
|
|
def prune_old_grid_profiles do
|
|
cutoff = DateTime.add(DateTime.utc_now(), -24, :hour)
|
|
# Bound the scan to recent partitions only (grid profiles are at most ~48h old).
|
|
# Without a lower bound, Postgres scans every partition back to 2016.
|
|
floor = DateTime.add(DateTime.utc_now(), -72, :hour)
|
|
|
|
# Only delete profiles on the 0.125° propagation grid, preserving QSO-linked
|
|
# profiles which sit at arbitrary lat/lon positions (HRRR's native ~3km grid).
|
|
%{num_rows: deleted} =
|
|
Repo.query!(
|
|
"""
|
|
DELETE FROM hrrr_profiles
|
|
WHERE valid_time >= $2 AND valid_time < $1
|
|
AND is_grid_point = true
|
|
""",
|
|
[cutoff, floor],
|
|
timeout: 300_000
|
|
)
|
|
|
|
if deleted > 0 do
|
|
require Logger
|
|
|
|
Logger.info("Pruned #{deleted} old HRRR grid profiles (valid_time < #{cutoff})")
|
|
end
|
|
|
|
deleted
|
|
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
|