defmodule Microwaveprop.Weather.Soundings do @moduledoc false import Ecto.Query alias Microwaveprop.Repo alias Microwaveprop.Weather.Sounding alias Microwaveprop.Weather.Station alias Microwaveprop.Weather.SurfaceObservation # Approximate km per degree latitude @km_per_deg_lat 111.0 @sounding_search_radii_km [150, 300, 600, 1000] @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 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 @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.shift(timestamp, hour: -time_window_hours) time_end = DateTime.shift(timestamp, hour: time_window_hours) 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 @doc """ Search for soundings in widening radii around the given location, stopping at the first radius that returns any. Returns `%{soundings, radius_km, exhausted}` where `exhausted: true` means the widest radius also came up empty — the caller should trigger a fetch for missing data at that point. """ @spec soundings_with_widening_radius(map()) :: %{ soundings: [Sounding.t()], radius_km: pos_integer(), exhausted: boolean() } def soundings_with_widening_radius(params) do Enum.reduce_while(@sounding_search_radii_km, nil, fn radius_km, _acc -> result = weather_for_contact(params, radius_km: radius_km) if result.soundings == [] do {:cont, %{soundings: [], radius_km: radius_km, exhausted: true}} else {:halt, %{soundings: result.soundings, radius_km: radius_km, exhausted: false}} end end) end @doc """ Nearest sounding to (`lat`, `lon`) within `radius_km` km and a ±3-hour window around `timestamp`. Joins `weather_stations` to `soundings` so the caller gets the raw sounding row (station_id set, derived duct fields populated) back. Returns `{:ok, sounding}` or `{:error, :not_found}`. Use this in the path calculator to surface the nearest RAOB's `ducting_detected` flag as an independent check on HRRR's pressure-level duct signal, which under-reads thin surface ducts. """ @spec nearest_sounding_to(float(), float(), DateTime.t(), keyword()) :: {:ok, Sounding.t()} | {:error, :not_found} def nearest_sounding_to(lat, lon, timestamp, opts \\ []) do radius_km = Keyword.get(opts, :radius_km, 300) hours = Keyword.get(opts, :hours, 3) # 1 deg lat ≈ 111 km; lon scaled by cos(lat). dlat = radius_km / 111.0 dlon = radius_km / (111.0 * max(0.1, :math.cos(lat * :math.pi() / 180.0))) time_start = DateTime.shift(timestamp, hour: -hours) time_end = DateTime.shift(timestamp, hour: hours) from(s in Sounding, join: station in assoc(s, :station), where: station.lat >= ^(lat - dlat) and station.lat <= ^(lat + dlat) and station.lon >= ^(lon - dlon) and station.lon <= ^(lon + dlon) and s.observed_at >= ^time_start and s.observed_at <= ^time_end, order_by: fragment( "SQRT(POW(? - ?, 2) + POW(? - ?, 2)) + ABS(EXTRACT(EPOCH FROM ? - ?)) / 86400.0", station.lat, ^lat, station.lon, ^lon, s.observed_at, ^timestamp ), limit: 1, preload: [:station] ) |> Repo.one() |> case do nil -> {:error, :not_found} sounding -> {:ok, sounding} end end end