Graham McIntire
7fb340bc35
Aliases: add module aliases for 9 nested module references Apply: replace apply/3 with direct module attribute calls Line length: break 1 long spec line Refactoring: extract helpers to reduce complexity and nesting in show.ex, radio.ex, weather workers, terrain, duct detection, backfill dashboard, contact map, and mix tasks
796 lines
25 KiB
Elixir
796 lines
25 KiB
Elixir
defmodule Microwaveprop.Weather do
|
|
@moduledoc false
|
|
|
|
import Ecto.Query
|
|
|
|
alias Microwaveprop.Repo
|
|
alias Microwaveprop.Weather.Era5Profile
|
|
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
|
|
Repo.one(
|
|
from(h in HrrrProfile,
|
|
where: h.is_grid_point == true,
|
|
select: max(h.valid_time)
|
|
)
|
|
)
|
|
end
|
|
|
|
@spec latest_weather_grid(map()) :: [map()]
|
|
def latest_weather_grid(bounds) do
|
|
case latest_grid_valid_time() do
|
|
nil ->
|
|
[]
|
|
|
|
latest_vt ->
|
|
from(h in HrrrProfile,
|
|
where:
|
|
h.valid_time == ^latest_vt and
|
|
h.is_grid_point == true and
|
|
h.lat >= ^bounds["south"] and h.lat <= ^bounds["north"] and
|
|
h.lon >= ^bounds["west"] and h.lon <= ^bounds["east"],
|
|
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
|
|
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)
|
|
|
|
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
|