diff --git a/lib/microwaveprop/commercial/poll_worker.ex b/lib/microwaveprop/commercial/poll_worker.ex index ad7f0b0f..ee6bae88 100644 --- a/lib/microwaveprop/commercial/poll_worker.ex +++ b/lib/microwaveprop/commercial/poll_worker.ex @@ -21,27 +21,32 @@ defmodule Microwaveprop.Commercial.PollWorker do def poll_and_record(links, poll_fn) do now = DateTime.utc_now() + Enum.each(links, &poll_link(&1, poll_fn, now)) + end - Enum.each(links, fn link -> - case poll_fn.(link.host, link.community, link.radio_type) do - {:ok, data} -> - attrs = - data - |> Map.put(:link_id, link.id) - |> Map.put(:sampled_at, now) + defp poll_link(link, poll_fn, now) do + case poll_fn.(link.host, link.community, link.radio_type) do + {:ok, data} -> + save_sample(link, data, now) - case Commercial.create_sample(attrs) do - {:ok, _sample} -> - Logger.info("Recorded sample for #{link.label}") + {:error, reason} -> + Logger.warning("SNMP poll failed for #{link.label} (#{link.host}): #{inspect(reason)}") + end + end - {:error, changeset} -> - Logger.warning("Failed to save sample for #{link.label}: #{inspect(changeset.errors)}") - end + defp save_sample(link, data, now) do + attrs = + data + |> Map.put(:link_id, link.id) + |> Map.put(:sampled_at, now) - {:error, reason} -> - Logger.warning("SNMP poll failed for #{link.label} (#{link.host}): #{inspect(reason)}") - end - end) + case Commercial.create_sample(attrs) do + {:ok, _sample} -> + Logger.info("Recorded sample for #{link.label}") + + {:error, changeset} -> + Logger.warning("Failed to save sample for #{link.label}: #{inspect(changeset.errors)}") + end end defp fetch_weather(links) do @@ -54,27 +59,33 @@ defmodule Microwaveprop.Commercial.PollWorker do now = DateTime.utc_now() start_dt = DateTime.add(now, -3600, :second) - Enum.each(stations, fn station_code -> - {:ok, station} = - Weather.find_or_create_station(%{ - station_code: station_code, - station_type: "asos", - name: station_code, - lat: 0.0, - lon: 0.0 - }) + Enum.each(stations, &fetch_station_weather(&1, start_dt, now)) + end - case IemClient.fetch_asos(station_code, start_dt, now) do - {:ok, rows} -> - Enum.each(rows, fn row -> - if row.observed_at, do: Weather.upsert_surface_observation(station, row) - end) + defp fetch_station_weather(station_code, start_dt, now) do + {:ok, station} = + Weather.find_or_create_station(%{ + station_code: station_code, + station_type: "asos", + name: station_code, + lat: 0.0, + lon: 0.0 + }) - Logger.info("Fetched #{length(rows)} ASOS observations for #{station_code}") + case IemClient.fetch_asos(station_code, start_dt, now) do + {:ok, rows} -> + ingest_asos(station, station_code, rows) - {:error, reason} -> - Logger.warning("ASOS fetch failed for #{station_code}: #{inspect(reason)}") - end - end) + {:error, reason} -> + Logger.warning("ASOS fetch failed for #{station_code}: #{inspect(reason)}") + end + end + + defp ingest_asos(station, station_code, rows) do + rows + |> Enum.filter(& &1.observed_at) + |> Enum.each(&Weather.upsert_surface_observation(station, &1)) + + Logger.info("Fetched #{length(rows)} ASOS observations for #{station_code}") end end diff --git a/lib/microwaveprop/geo.ex b/lib/microwaveprop/geo.ex index 09232851..b153fbf6 100644 --- a/lib/microwaveprop/geo.ex +++ b/lib/microwaveprop/geo.ex @@ -1,6 +1,8 @@ defmodule Microwaveprop.Geo do @moduledoc "Geodetic helper functions: distances, bearings, coordinate math." + alias Microwaveprop.Radio.Maidenhead + @earth_radius_km 6371.0 @doc "Great-circle distance between two points in km." @@ -31,7 +33,7 @@ defmodule Microwaveprop.Geo do @doc "Center of a 4-character Maidenhead grid square." @spec maidenhead_center(String.t()) :: {float(), float()} | nil def maidenhead_center(grid) when is_binary(grid) and byte_size(grid) >= 4 do - case Microwaveprop.Radio.Maidenhead.to_latlon(grid) do + case Maidenhead.to_latlon(grid) do {:ok, {lat, lon}} -> {lat, lon} :error -> nil end diff --git a/lib/microwaveprop/propagation.ex b/lib/microwaveprop/propagation.ex index e2c26bef..39311346 100644 --- a/lib/microwaveprop/propagation.ex +++ b/lib/microwaveprop/propagation.ex @@ -22,9 +22,9 @@ defmodule Microwaveprop.Propagation do @spec load_ml_model() :: :ok def load_ml_model do if Code.ensure_loaded?(@ml_module) do - case apply(@ml_module, :load, []) do + case @ml_module.load() do {:ok, params} -> - predict_fn = apply(@ml_module, :compile_predict, []) + predict_fn = @ml_module.compile_predict() :persistent_term.put(@ml_key, {predict_fn, params}) Logger.info("PropagationML: model loaded and compiled") :ok diff --git a/lib/microwaveprop/propagation/duct.ex b/lib/microwaveprop/propagation/duct.ex index ede9e7dd..88386c7f 100644 --- a/lib/microwaveprop/propagation/duct.ex +++ b/lib/microwaveprop/propagation/duct.ex @@ -69,36 +69,35 @@ defmodule Microwaveprop.Propagation.Duct do def detect_ducts(m_profile) do m_profile |> Enum.chunk_every(2, 1, :discard) - |> Enum.reduce({[], nil}, fn [{h1, m1}, {h2, m2}], {ducts, current_duct} -> - if m2 < m1 do - # M is decreasing — we're in a duct - case current_duct do - nil -> - {ducts, %{base_m: h1, base_m_val: m1, top_m: h2, min_m_val: m2}} - - duct -> - {ducts, %{duct | top_m: h2, min_m_val: min(duct.min_m_val, m2)}} - end - else - # M is increasing — close any open duct - case current_duct do - nil -> - {ducts, nil} - - duct -> - finished = finalize_duct(duct) - {[finished | ducts], nil} - end - end - end) - |> then(fn {ducts, current} -> - case current do - nil -> Enum.reverse(ducts) - duct -> Enum.reverse([finalize_duct(duct) | ducts]) - end + |> Enum.reduce({[], nil}, fn [{h1, m1}, {h2, m2}], acc -> + process_m_pair(acc, h1, m1, h2, m2) end) + |> close_trailing_duct() end + defp process_m_pair({ducts, current_duct}, h1, m1, h2, m2) when m2 < m1 do + # M is decreasing — we're in a duct + duct = extend_or_start_duct(current_duct, h1, m1, h2, m2) + {ducts, duct} + end + + defp process_m_pair({ducts, nil}, _h1, _m1, _h2, _m2), do: {ducts, nil} + + defp process_m_pair({ducts, duct}, _h1, _m1, _h2, _m2) do + {[finalize_duct(duct) | ducts], nil} + end + + defp extend_or_start_duct(nil, h1, m1, h2, m2) do + %{base_m: h1, base_m_val: m1, top_m: h2, min_m_val: m2} + end + + defp extend_or_start_duct(duct, _h1, _m1, h2, m2) do + %{duct | top_m: h2, min_m_val: min(duct.min_m_val, m2)} + end + + defp close_trailing_duct({ducts, nil}), do: Enum.reverse(ducts) + defp close_trailing_duct({ducts, duct}), do: Enum.reverse([finalize_duct(duct) | ducts]) + defp finalize_duct(%{base_m: base, top_m: top, base_m_val: m_base, min_m_val: m_min}) do %{ base_m: base, diff --git a/lib/microwaveprop/radio.ex b/lib/microwaveprop/radio.ex index e59244f6..7bd72b62 100644 --- a/lib/microwaveprop/radio.ex +++ b/lib/microwaveprop/radio.ex @@ -133,7 +133,11 @@ defmodule Microwaveprop.Radio do @enrichable_statuses [:pending, :failed] - @spec contacts_needing_enrichment(atom(), non_neg_integer(), (Ecto.Query.t() -> Ecto.Query.t()) | nil) :: [Contact.t()] + @spec contacts_needing_enrichment( + atom(), + non_neg_integer(), + (Ecto.Query.t() -> Ecto.Query.t()) | nil + ) :: [Contact.t()] def contacts_needing_enrichment(field, limit \\ 500, extra_filter \\ nil) do query = Contact @@ -191,25 +195,25 @@ defmodule Microwaveprop.Radio do end def contact_path_points(%{pos1: pos1, pos2: pos2}) do - lat1 = pos1["lat"] - lon1 = pos1["lon"] || pos1["lng"] - lat2 = pos2["lat"] - lon2 = pos2["lon"] || pos2["lng"] - - cond do - lat1 && lon1 && lat2 && lon2 -> - mid_lat = (lat1 + lat2) / 2 - mid_lon = (lon1 + lon2) / 2 - [{lat1, lon1}, {mid_lat, mid_lon}, {lat2, lon2}] - - lat1 && lon1 -> - [{lat1, lon1}] - - true -> - [] - end + build_path_points(extract_latlon(pos1), extract_latlon(pos2)) end + defp extract_latlon(%{"lat" => lat} = pos) when is_number(lat) do + lon = pos["lon"] || pos["lng"] + if lon, do: {lat, lon} + end + + defp extract_latlon(_), do: nil + + defp build_path_points({lat1, lon1}, {lat2, lon2}) do + mid_lat = (lat1 + lat2) / 2 + mid_lon = (lon1 + lon2) / 2 + [{lat1, lon1}, {mid_lat, mid_lon}, {lat2, lon2}] + end + + defp build_path_points({lat1, lon1}, _), do: [{lat1, lon1}] + defp build_path_points(_, _), do: [] + @earth_radius_km 6371.0 @spec haversine_km(number(), number(), number(), number()) :: float() @@ -620,45 +624,49 @@ defmodule Microwaveprop.Radio do @spec apply_edit_to_contact(Contact.t(), map()) :: Contact.t() def apply_edit_to_contact(contact, proposed_changes) do changes = build_contact_changes(contact, proposed_changes) - - grids_or_band_changed = - Map.has_key?(proposed_changes, "grid1") or - Map.has_key?(proposed_changes, "grid2") or - Map.has_key?(proposed_changes, "band") - - changes = - if grids_or_band_changed do - # Recompute positions from grids - new_grid1 = Map.get(proposed_changes, "grid1", contact.grid1) - new_grid2 = Map.get(proposed_changes, "grid2", contact.grid2) - - pos1 = resolve_grid_position(new_grid1) || contact.pos1 - pos2 = resolve_grid_position(new_grid2) || contact.pos2 - - distance = - if pos1 && pos2 do - lon1 = pos1["lon"] || pos1["lng"] - lon2 = pos2["lon"] || pos2["lng"] - pos1["lat"] |> haversine_km(lon1, pos2["lat"], lon2) |> round() |> Decimal.new() - end - - changes - |> Map.put(:pos1, pos1) - |> Map.put(:pos2, pos2) - |> Map.put(:distance_km, distance) - |> Map.put(:hrrr_status, :pending) - |> Map.put(:weather_status, :pending) - |> Map.put(:terrain_status, :pending) - |> Map.put(:iemre_status, :pending) - else - changes - end + changes = maybe_recompute_positions(changes, contact, proposed_changes) contact |> Ecto.Changeset.change(changes) |> Repo.update!() end + defp maybe_recompute_positions(changes, contact, proposed) do + grids_or_band_changed = + Map.has_key?(proposed, "grid1") or + Map.has_key?(proposed, "grid2") or + Map.has_key?(proposed, "band") + + if grids_or_band_changed do + recompute_positions(changes, contact, proposed) + else + changes + end + end + + defp recompute_positions(changes, contact, proposed) do + pos1 = resolve_grid_position(Map.get(proposed, "grid1", contact.grid1)) || contact.pos1 + pos2 = resolve_grid_position(Map.get(proposed, "grid2", contact.grid2)) || contact.pos2 + distance = compute_distance(pos1, pos2) + + changes + |> Map.put(:pos1, pos1) + |> Map.put(:pos2, pos2) + |> Map.put(:distance_km, distance) + |> Map.put(:hrrr_status, :pending) + |> Map.put(:weather_status, :pending) + |> Map.put(:terrain_status, :pending) + |> Map.put(:iemre_status, :pending) + end + + defp compute_distance(pos1, pos2) when not is_nil(pos1) and not is_nil(pos2) do + lon1 = pos1["lon"] || pos1["lng"] + lon2 = pos2["lon"] || pos2["lng"] + pos1["lat"] |> haversine_km(lon1, pos2["lat"], lon2) |> round() |> Decimal.new() + end + + defp compute_distance(_, _), do: nil + defp build_contact_changes(_contact, proposed) do Enum.reduce(proposed, %{}, fn {"band", val}, acc -> diff --git a/lib/microwaveprop/release.ex b/lib/microwaveprop/release.ex index 2aef471c..2cb0a2f9 100644 --- a/lib/microwaveprop/release.ex +++ b/lib/microwaveprop/release.ex @@ -17,6 +17,7 @@ defmodule Microwaveprop.Release do bin/microwaveprop eval "Microwaveprop.Release.scorer_diff(\"{\\\"humidity\\\":0.18,...}\")" """ alias Microwaveprop.Workers.AdminTaskWorker + alias Microwaveprop.Workers.HrrrNativeGridWorker @app :microwaveprop @@ -78,7 +79,7 @@ defmodule Microwaveprop.Release do for [year, month, day, hour, cnt] <- rows do args = %{year: year, month: month, day: day, hour: hour} - case Oban.insert(Microwaveprop.Workers.HrrrNativeGridWorker.new(args, unique: [period: :infinity])) do + case Oban.insert(HrrrNativeGridWorker.new(args, unique: [period: :infinity])) do {:ok, _} -> IO.puts(" #{year}-#{month}-#{day} #{hour}Z (#{cnt} contacts)") {:error, _} -> IO.puts(" #{year}-#{month}-#{day} #{hour}Z (skipped)") end diff --git a/lib/microwaveprop/terrain/srtm.ex b/lib/microwaveprop/terrain/srtm.ex index c77be730..54075dd5 100644 --- a/lib/microwaveprop/terrain/srtm.ex +++ b/lib/microwaveprop/terrain/srtm.ex @@ -61,20 +61,28 @@ defmodule Microwaveprop.Terrain.Srtm do read_elevation(fd, lat, lon) {:error, :enoent} -> - if opts[:download] && File.dir?(tiles_dir) do - case download_tile(lat, lon, tiles_dir) do - {:ok, _} -> - case :file.open(path, [:read, :binary, :raw]) do - {:ok, fd} -> read_elevation(fd, lat, lon) - {:error, _} -> {:error, :no_tile} - end + maybe_download_and_read(lat, lon, path, tiles_dir, opts) + end + end - {:error, _} -> - {:error, :no_tile} - end - else - {:error, :no_tile} + defp maybe_download_and_read(lat, lon, path, tiles_dir, opts) do + if opts[:download] && File.dir?(tiles_dir) do + download_and_read(lat, lon, path, tiles_dir) + else + {:error, :no_tile} + end + end + + defp download_and_read(lat, lon, path, tiles_dir) do + case download_tile(lat, lon, tiles_dir) do + {:ok, _} -> + case :file.open(path, [:read, :binary, :raw]) do + {:ok, fd} -> read_elevation(fd, lat, lon) + {:error, _} -> {:error, :no_tile} end + + {:error, _} -> + {:error, :no_tile} end end diff --git a/lib/microwaveprop/weather.ex b/lib/microwaveprop/weather.ex index b5d16f52..5b7a6c6d 100644 --- a/lib/microwaveprop/weather.ex +++ b/lib/microwaveprop/weather.ex @@ -250,22 +250,7 @@ defmodule Microwaveprop.Weather do do: "#{s}_ASOS" for network <- asos ++ ["RAOB"] 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 - + sync_network(network) Process.sleep(200) end @@ -274,6 +259,24 @@ defmodule Microwaveprop.Weather do :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 diff --git a/lib/microwaveprop/workers/admin_task_worker.ex b/lib/microwaveprop/workers/admin_task_worker.ex index 5ad07221..ae5f738a 100644 --- a/lib/microwaveprop/workers/admin_task_worker.ex +++ b/lib/microwaveprop/workers/admin_task_worker.ex @@ -20,6 +20,7 @@ defmodule Microwaveprop.Workers.AdminTaskWorker do alias Microwaveprop.Backtest.Features alias Microwaveprop.Propagation.Duct alias Microwaveprop.Propagation.Inversion + alias Microwaveprop.Propagation.Recalibrator alias Microwaveprop.Repo alias Microwaveprop.Weather.HrrrNativeProfile alias Microwaveprop.Weather.ThetaE @@ -191,7 +192,7 @@ defmodule Microwaveprop.Workers.AdminTaskWorker do Logger.info("AdminTask: running weight recalibration (sample=#{sample_size}, epochs=#{epochs}, lr=#{learning_rate})") result = - Microwaveprop.Propagation.Recalibrator.fit( + Recalibrator.fit( sample_size: sample_size, epochs: epochs, learning_rate: learning_rate diff --git a/lib/microwaveprop/workers/hrrr_native_grid_worker.ex b/lib/microwaveprop/workers/hrrr_native_grid_worker.ex index 04a3b1a0..7ec5cfb6 100644 --- a/lib/microwaveprop/workers/hrrr_native_grid_worker.ex +++ b/lib/microwaveprop/workers/hrrr_native_grid_worker.ex @@ -122,37 +122,7 @@ defmodule Microwaveprop.Workers.HrrrNativeGridWorker do :ok <- HrrrClient.download_grib_ranges_to_file(url, ranges, tmp_grib), _ = Logger.info("HRRR native downloaded to #{tmp_grib}, extracting #{length(points)} points..."), {:ok, profiles} <- HrrrNativeClient.extract_native_profiles_from_file(tmp_grib, points) do - rows = - Enum.flat_map(profiles, fn {{lat, lon}, profile} -> - if profile[:level_count] && profile.level_count > 0 do - now = DateTime.truncate(DateTime.utc_now(), :second) - - [ - Map.merge(profile, %{ - id: Ecto.UUID.generate(), - valid_time: valid_time, - run_time: valid_time, - lat: lat, - lon: lon, - inserted_at: now, - updated_at: now - }) - ] - else - [] - end - end) - - {inserted, _} = - Repo.insert_all( - HrrrNativeProfile, - rows, - on_conflict: {:replace_all_except, [:id, :inserted_at]}, - conflict_target: [:lat, :lon, :valid_time] - ) - - Logger.info("HrrrNativeGridWorker: upserted #{inserted} profiles for #{valid_time}") - :ok + upsert_native_profiles(profiles, valid_time) else {:error, reason} -> Logger.warning("HrrrNativeGridWorker failed for #{valid_time}: #{inspect(reason)}") @@ -160,6 +130,40 @@ defmodule Microwaveprop.Workers.HrrrNativeGridWorker do end end + defp upsert_native_profiles(profiles, valid_time) do + now = DateTime.truncate(DateTime.utc_now(), :second) + + rows = + Enum.flat_map(profiles, fn {{lat, lon}, profile} -> + if profile[:level_count] && profile.level_count > 0 do + [ + Map.merge(profile, %{ + id: Ecto.UUID.generate(), + valid_time: valid_time, + run_time: valid_time, + lat: lat, + lon: lon, + inserted_at: now, + updated_at: now + }) + ] + else + [] + end + end) + + {inserted, _} = + Repo.insert_all( + HrrrNativeProfile, + rows, + on_conflict: {:replace_all_except, [:id, :inserted_at]}, + conflict_target: [:lat, :lon, :valid_time] + ) + + Logger.info("HrrrNativeGridWorker: upserted #{inserted} profiles for #{valid_time}") + :ok + end + defp fetch_idx(url) do case Req.get(url, receive_timeout: 120_000) do {:ok, %{status: 200, body: body}} -> {:ok, body} diff --git a/lib/microwaveprop/workers/terrain_profile_worker.ex b/lib/microwaveprop/workers/terrain_profile_worker.ex index 353d4d94..27cdf845 100644 --- a/lib/microwaveprop/workers/terrain_profile_worker.ex +++ b/lib/microwaveprop/workers/terrain_profile_worker.ex @@ -22,66 +22,70 @@ defmodule Microwaveprop.Workers.TerrainProfileWorker do Radio.set_enrichment_status!([contact_id], :terrain_status, :complete) :ok else - contact = Radio.get_contact!(contact_id) - - with %{"lat" => lat1} <- contact.pos1, - lon1 when is_number(lon1) <- contact.pos1["lon"] || contact.pos1["lng"], - %{"lat" => lat2} <- contact.pos2, - lon2 when is_number(lon2) <- contact.pos2["lon"] || contact.pos2["lng"] do - dist_km = Decimal.to_float(contact.distance_km || Decimal.new(0)) - freq_ghz = Decimal.to_float(contact.band) / 1000 - - # Look up HRRR refractivity gradient for dynamic k-factor - k = lookup_k_factor(contact) - - case ElevationClient.fetch_elevation_profile(lat1, lon1, lat2, lon2, 64, download: true) do - {:ok, profile} -> - analysis = TerrainAnalysis.analyse(profile, dist_km, freq_ghz, k_factor: k) - - path_points = - Enum.map(profile, fn p -> - %{ - "lat" => p.lat, - "lon" => p.lon, - "d" => p.d, - "elev" => p.elev, - "dist_km" => p.dist_km - } - end) - - Terrain.upsert_terrain_profile(%{ - contact_id: contact_id, - sample_count: length(profile), - path_points: path_points, - max_elevation_m: analysis.max_elevation_m, - min_clearance_m: analysis.min_clearance_m, - diffraction_db: analysis.diffraction_db, - fresnel_hit_count: analysis.fresnel_hit_count, - obstructed_count: analysis.obstructed_count, - verdict: analysis.verdict - }) - - Radio.set_enrichment_status!([contact_id], :terrain_status, :complete) - - Phoenix.PubSub.broadcast( - Microwaveprop.PubSub, - "contact_enrichment:#{contact_id}", - {:terrain_ready, contact_id} - ) - - :ok - - {:error, reason} -> - {:error, reason} - end - else - _ -> - Radio.set_enrichment_status!([contact_id], :terrain_status, :unavailable) - :ok - end + analyse_terrain(contact_id) end end + defp analyse_terrain(contact_id) do + contact = Radio.get_contact!(contact_id) + + with %{"lat" => lat1} <- contact.pos1, + lon1 when is_number(lon1) <- contact.pos1["lon"] || contact.pos1["lng"], + %{"lat" => lat2} <- contact.pos2, + lon2 when is_number(lon2) <- contact.pos2["lon"] || contact.pos2["lng"] do + fetch_and_store_profile(contact_id, contact, lat1, lon1, lat2, lon2) + else + _ -> + Radio.set_enrichment_status!([contact_id], :terrain_status, :unavailable) + :ok + end + end + + defp fetch_and_store_profile(contact_id, contact, lat1, lon1, lat2, lon2) do + dist_km = Decimal.to_float(contact.distance_km || Decimal.new(0)) + freq_ghz = Decimal.to_float(contact.band) / 1000 + k = lookup_k_factor(contact) + + case ElevationClient.fetch_elevation_profile(lat1, lon1, lat2, lon2, 64, download: true) do + {:ok, profile} -> + store_terrain_result(contact_id, profile, dist_km, freq_ghz, k) + + {:error, reason} -> + {:error, reason} + end + end + + defp store_terrain_result(contact_id, profile, dist_km, freq_ghz, k) do + analysis = TerrainAnalysis.analyse(profile, dist_km, freq_ghz, k_factor: k) + + path_points = + Enum.map(profile, fn p -> + %{"lat" => p.lat, "lon" => p.lon, "d" => p.d, "elev" => p.elev, "dist_km" => p.dist_km} + end) + + Terrain.upsert_terrain_profile(%{ + contact_id: contact_id, + sample_count: length(profile), + path_points: path_points, + max_elevation_m: analysis.max_elevation_m, + min_clearance_m: analysis.min_clearance_m, + diffraction_db: analysis.diffraction_db, + fresnel_hit_count: analysis.fresnel_hit_count, + obstructed_count: analysis.obstructed_count, + verdict: analysis.verdict + }) + + Radio.set_enrichment_status!([contact_id], :terrain_status, :complete) + + Phoenix.PubSub.broadcast( + Microwaveprop.PubSub, + "contact_enrichment:#{contact_id}", + {:terrain_ready, contact_id} + ) + + :ok + end + defp lookup_k_factor(contact) do case Weather.hrrr_for_contact(contact) do %{min_refractivity_gradient: grad} when not is_nil(grad) -> diff --git a/lib/microwaveprop/workers/weather_fetch_worker.ex b/lib/microwaveprop/workers/weather_fetch_worker.ex index c773fbb7..6c2a1db3 100644 --- a/lib/microwaveprop/workers/weather_fetch_worker.ex +++ b/lib/microwaveprop/workers/weather_fetch_worker.ex @@ -37,44 +37,7 @@ defmodule Microwaveprop.Workers.WeatherFetchWorker do :ok station -> - if Weather.has_surface_observations?(station.id, start_dt, end_dt) do - Logger.debug("WeatherFetch ASOS: #{station_code} already has obs for #{start_dt_str}..#{end_dt_str}") - :ok - else - Logger.info("WeatherFetch ASOS: fetching #{station_code} for #{start_dt_str}..#{end_dt_str}") - - case IemClient.fetch_asos(station_code, start_dt, end_dt) do - {:ok, rows} -> - count = Enum.count(rows, & &1.observed_at) - - Enum.each(rows, fn row -> - if row.observed_at, do: Weather.upsert_surface_observation(station, row) - end) - - # Store stub so this station/window isn't retried on future backfills - if count == 0 do - midpoint = DateTime.add(start_dt, div(DateTime.diff(end_dt, start_dt), 2)) - - %SurfaceObservation{} - |> SurfaceObservation.changeset(%{station_id: station.id, observed_at: midpoint}) - |> Repo.insert(on_conflict: :nothing, conflict_target: [:station_id, :observed_at]) - end - - Logger.info("WeatherFetch ASOS: #{station_code} ingested #{count} observations") - - Phoenix.PubSub.broadcast( - Microwaveprop.PubSub, - "contact_enrichment:weather", - {:weather_ready, %{station_id: station_id}} - ) - - :ok - - {:error, reason} -> - Logger.warning("WeatherFetch ASOS: #{station_code} failed: #{inspect(reason)}") - {:error, reason} - end - end + fetch_asos(station, station_id, station_code, start_dt, end_dt, start_dt_str, end_dt_str) end end @@ -93,70 +56,125 @@ defmodule Microwaveprop.Workers.WeatherFetchWorker do :ok station -> - if Weather.has_sounding?(station.id, sounding_time) do - Logger.debug("WeatherFetch RAOB: #{station_code} already has sounding for #{sounding_time_str}") - :ok - else - Logger.info("WeatherFetch RAOB: fetching #{station_code} for #{sounding_time_str}") + fetch_raob(station, station_id, station_code, sounding_time, sounding_time_str) + end + end - case IemClient.fetch_raob(station_code, sounding_time) do - {:ok, [parsed | _]} -> - params = SoundingParams.derive(parsed.profile) + # -- ASOS helpers -- - sounding_attrs = - if params do - %{ - observed_at: parsed.observed_at, - profile: parsed.profile, - level_count: params.level_count, - surface_pressure_mb: params.surface_pressure_mb, - surface_temp_c: params.surface_temp_c, - surface_dewpoint_c: params.surface_dewpoint_c, - surface_refractivity: params.surface_refractivity, - min_refractivity_gradient: params.min_refractivity_gradient, - boundary_layer_depth_m: params.boundary_layer_depth_m, - precipitable_water_mm: params.precipitable_water_mm, - k_index: params.k_index, - lifted_index: params.lifted_index, - ducting_detected: params.ducting_detected, - duct_characteristics: params.duct_characteristics - } - else - %{ - observed_at: parsed.observed_at, - profile: parsed.profile, - level_count: length(parsed.profile) - } - end + defp fetch_asos(station, station_id, station_code, start_dt, end_dt, start_dt_str, end_dt_str) do + if Weather.has_surface_observations?(station.id, start_dt, end_dt) do + Logger.debug("WeatherFetch ASOS: #{station_code} already has obs for #{start_dt_str}..#{end_dt_str}") + :ok + else + Logger.info("WeatherFetch ASOS: fetching #{station_code} for #{start_dt_str}..#{end_dt_str}") + do_fetch_asos(station, station_id, station_code, start_dt, end_dt) + end + end - Weather.upsert_sounding(station, sounding_attrs) + defp do_fetch_asos(station, station_id, station_code, start_dt, end_dt) do + case IemClient.fetch_asos(station_code, start_dt, end_dt) do + {:ok, rows} -> + ingest_asos_rows(station, station_id, station_code, rows, start_dt, end_dt) - Logger.info("WeatherFetch RAOB: #{station_code} ingested sounding with #{length(parsed.profile)} levels") + {:error, reason} -> + Logger.warning("WeatherFetch ASOS: #{station_code} failed: #{inspect(reason)}") + {:error, reason} + end + end - Phoenix.PubSub.broadcast( - Microwaveprop.PubSub, - "contact_enrichment:weather", - {:weather_ready, %{station_id: station_id}} - ) + defp ingest_asos_rows(station, station_id, station_code, rows, start_dt, end_dt) do + count = Enum.count(rows, & &1.observed_at) - :ok + Enum.each(rows, fn row -> + if row.observed_at, do: Weather.upsert_surface_observation(station, row) + end) - {:ok, []} -> - # Store stub sounding so this station/time isn't retried on future backfills - Weather.upsert_sounding(station, %{ - observed_at: sounding_time, - profile: [], - level_count: 0 - }) + if count == 0, do: store_asos_stub(station, start_dt, end_dt) - Logger.info("WeatherFetch RAOB: #{station_code} no data for #{sounding_time_str}, stored stub") - :ok + Logger.info("WeatherFetch ASOS: #{station_code} ingested #{count} observations") - {:error, reason} -> - Logger.warning("WeatherFetch RAOB: #{station_code} failed: #{inspect(reason)}") - {:error, reason} - end - end + Phoenix.PubSub.broadcast( + Microwaveprop.PubSub, + "contact_enrichment:weather", + {:weather_ready, %{station_id: station_id}} + ) + + :ok + end + + defp store_asos_stub(station, start_dt, end_dt) do + midpoint = DateTime.add(start_dt, div(DateTime.diff(end_dt, start_dt), 2)) + + %SurfaceObservation{} + |> SurfaceObservation.changeset(%{station_id: station.id, observed_at: midpoint}) + |> Repo.insert(on_conflict: :nothing, conflict_target: [:station_id, :observed_at]) + end + + # -- RAOB helpers -- + + defp fetch_raob(station, station_id, station_code, sounding_time, sounding_time_str) do + if Weather.has_sounding?(station.id, sounding_time) do + Logger.debug("WeatherFetch RAOB: #{station_code} already has sounding for #{sounding_time_str}") + :ok + else + Logger.info("WeatherFetch RAOB: fetching #{station_code} for #{sounding_time_str}") + do_fetch_raob(station, station_id, station_code, sounding_time, sounding_time_str) + end + end + + defp do_fetch_raob(station, station_id, station_code, sounding_time, sounding_time_str) do + case IemClient.fetch_raob(station_code, sounding_time) do + {:ok, [parsed | _]} -> + ingest_raob(station, station_id, station_code, parsed) + + {:ok, []} -> + Weather.upsert_sounding(station, %{observed_at: sounding_time, profile: [], level_count: 0}) + Logger.info("WeatherFetch RAOB: #{station_code} no data for #{sounding_time_str}, stored stub") + :ok + + {:error, reason} -> + Logger.warning("WeatherFetch RAOB: #{station_code} failed: #{inspect(reason)}") + {:error, reason} + end + end + + defp ingest_raob(station, station_id, station_code, parsed) do + sounding_attrs = build_sounding_attrs(parsed) + Weather.upsert_sounding(station, sounding_attrs) + Logger.info("WeatherFetch RAOB: #{station_code} ingested sounding with #{length(parsed.profile)} levels") + + Phoenix.PubSub.broadcast( + Microwaveprop.PubSub, + "contact_enrichment:weather", + {:weather_ready, %{station_id: station_id}} + ) + + :ok + end + + defp build_sounding_attrs(parsed) do + case SoundingParams.derive(parsed.profile) do + nil -> + %{observed_at: parsed.observed_at, profile: parsed.profile, level_count: length(parsed.profile)} + + params -> + %{ + observed_at: parsed.observed_at, + profile: parsed.profile, + level_count: params.level_count, + surface_pressure_mb: params.surface_pressure_mb, + surface_temp_c: params.surface_temp_c, + surface_dewpoint_c: params.surface_dewpoint_c, + surface_refractivity: params.surface_refractivity, + min_refractivity_gradient: params.min_refractivity_gradient, + boundary_layer_depth_m: params.boundary_layer_depth_m, + precipitable_water_mm: params.precipitable_water_mm, + k_index: params.k_index, + lifted_index: params.lifted_index, + ducting_detected: params.ducting_detected, + duct_characteristics: params.duct_characteristics + } end end end diff --git a/lib/microwaveprop_web/components/core_components.ex b/lib/microwaveprop_web/components/core_components.ex index 91715709..24c69632 100644 --- a/lib/microwaveprop_web/components/core_components.ex +++ b/lib/microwaveprop_web/components/core_components.ex @@ -29,6 +29,7 @@ defmodule MicrowavepropWeb.CoreComponents do use Phoenix.Component use Gettext, backend: MicrowavepropWeb.Gettext + alias Phoenix.HTML.Form alias Phoenix.HTML.FormField alias Phoenix.LiveView.JS @@ -200,7 +201,7 @@ defmodule MicrowavepropWeb.CoreComponents do def input(%{type: "checkbox"} = assigns) do assigns = assign_new(assigns, :checked, fn -> - Phoenix.HTML.Form.normalize_value("checkbox", assigns[:value]) + Form.normalize_value("checkbox", assigns[:value]) end) ~H""" diff --git a/lib/microwaveprop_web/controllers/health_controller.ex b/lib/microwaveprop_web/controllers/health_controller.ex index f247b033..3bbd6dbf 100644 --- a/lib/microwaveprop_web/controllers/health_controller.ex +++ b/lib/microwaveprop_web/controllers/health_controller.ex @@ -1,9 +1,11 @@ defmodule MicrowavepropWeb.HealthController do use MicrowavepropWeb, :controller + alias Ecto.Adapters.SQL + def check(conn, _params) do # Verify DB is reachable - Ecto.Adapters.SQL.query!(Microwaveprop.Repo, "SELECT 1") + SQL.query!(Microwaveprop.Repo, "SELECT 1") conn |> put_resp_content_type("text/plain") diff --git a/lib/microwaveprop_web/live/backfill_live.ex b/lib/microwaveprop_web/live/backfill_live.ex index 2a71606c..09554e84 100644 --- a/lib/microwaveprop_web/live/backfill_live.ex +++ b/lib/microwaveprop_web/live/backfill_live.ex @@ -101,35 +101,15 @@ defmodule MicrowavepropWeb.BackfillLive do done = [:complete, :unavailable] total = Repo.one(from(c in Contact, where: not is_nil(c.pos1), select: count())) + terrain = count_incomplete(:terrain_status, incomplete) + hrrr = count_incomplete(:hrrr_status, incomplete) + weather = count_incomplete(:weather_status, incomplete) + iemre = count_incomplete(:iemre_status, incomplete) - terrain = - Repo.one(from(c in Contact, where: c.terrain_status in ^incomplete and not is_nil(c.pos1), select: count())) - - hrrr = - Repo.one(from(c in Contact, where: c.hrrr_status in ^incomplete and not is_nil(c.pos1), select: count())) - - weather = - Repo.one(from(c in Contact, where: c.weather_status in ^incomplete and not is_nil(c.pos1), select: count())) - - iemre = - Repo.one(from(c in Contact, where: c.iemre_status in ^incomplete and not is_nil(c.pos1), select: count())) - - # ERA5 targets contacts where HRRR is unavailable era5 = Repo.one(from(c in Contact, where: c.hrrr_status == :unavailable and not is_nil(c.pos1), select: count())) - all_done = - Repo.one( - from(c in Contact, - where: - not is_nil(c.pos1) and - c.hrrr_status in ^done and - c.weather_status in ^done and - c.terrain_status in ^done and - c.iemre_status in ^done, - select: count() - ) - ) + all_done = count_all_done(done) %{ terrain: terrain, @@ -143,6 +123,29 @@ defmodule MicrowavepropWeb.BackfillLive do } end + defp count_incomplete(status_field, statuses) do + Repo.one( + from(c in Contact, + where: field(c, ^status_field) in ^statuses and not is_nil(c.pos1), + select: count() + ) + ) + end + + defp count_all_done(done) do + Repo.one( + from(c in Contact, + where: + not is_nil(c.pos1) and + c.hrrr_status in ^done and + c.weather_status in ^done and + c.terrain_status in ^done and + c.iemre_status in ^done, + select: count() + ) + ) + end + defp fetch_stats do jobs = Repo.all( @@ -156,21 +159,7 @@ defmodule MicrowavepropWeb.BackfillLive do by_worker = jobs |> Enum.group_by(&elem(&1, 0), fn {_, state, count} -> {state, count} end) - |> Enum.map(fn {worker, states} -> - short_name = worker |> String.split(".") |> List.last() - total = Enum.reduce(states, 0, fn {_, c}, acc -> acc + c end) - executing = Enum.find_value(states, 0, fn {s, c} -> if s == "executing", do: c end) - available = Enum.find_value(states, 0, fn {s, c} -> if s == "available", do: c end) - retryable = Enum.find_value(states, 0, fn {s, c} -> if s == "retryable", do: c end) - - %{ - worker: short_name, - total: total, - executing: executing, - available: available, - retryable: retryable - } - end) + |> Enum.map(&summarize_worker/1) |> Enum.sort_by(& &1.total, :desc) completed_1h = @@ -184,6 +173,22 @@ defmodule MicrowavepropWeb.BackfillLive do %{by_worker: by_worker, completed_1h: completed_1h} end + defp summarize_worker({worker, states}) do + short_name = worker |> String.split(".") |> List.last() + + %{ + worker: short_name, + total: Enum.reduce(states, 0, fn {_, c}, acc -> acc + c end), + executing: state_count(states, "executing"), + available: state_count(states, "available"), + retryable: state_count(states, "retryable") + } + end + + defp state_count(states, name) do + Enum.find_value(states, 0, fn {s, c} -> if s == name, do: c end) + end + defp fetch_db_stats do # Regular tables + partitioned tables (summing child partition stats) %{rows: rows} = diff --git a/lib/microwaveprop_web/live/contact_live/show.ex b/lib/microwaveprop_web/live/contact_live/show.ex index 7013c73e..3b9c2d49 100644 --- a/lib/microwaveprop_web/live/contact_live/show.ex +++ b/lib/microwaveprop_web/live/contact_live/show.ex @@ -182,41 +182,50 @@ defmodule MicrowavepropWeb.ContactLive.Show do if is_nil(user) do {:noreply, put_flash(socket, :error, "You must be logged in to edit.")} else - contact = socket.assigns.contact + {:noreply, submit_edit(socket, user, params)} + end + end - if admin?(socket.assigns) do - # Admin: apply changes directly - proposed = Radio.diff_against_contact(contact, Radio.normalize_proposed(params)) + defp submit_edit(socket, user, params) do + contact = socket.assigns.contact - if proposed == %{} do - {:noreply, put_flash(socket, :error, "No changes detected.")} - else - Radio.apply_edit_to_contact(contact, proposed) - updated = Radio.get_contact!(contact.id) + if admin?(socket.assigns) do + apply_admin_edit(socket, contact, params) + else + submit_user_edit(socket, contact, user, params) + end + end - {:noreply, - socket - |> assign(contact: updated, editing: false, edit_form: nil) - |> put_flash(:info, "Contact updated.")} - end - else - case Radio.create_contact_edit(contact, user, params) do - {:ok, _edit} -> - {:noreply, - socket - |> assign(editing: false, edit_form: nil, pending_edit: Radio.pending_edit_for_user(contact.id, user.id)) - |> put_flash(:info, "Edit submitted for review.")} + defp apply_admin_edit(socket, contact, params) do + proposed = Radio.diff_against_contact(contact, Radio.normalize_proposed(params)) - {:error, changeset} -> - messages = - changeset - |> Ecto.Changeset.traverse_errors(fn {msg, _opts} -> msg end) - |> Enum.flat_map(fn {_field, msgs} -> msgs end) - |> Enum.join(", ") + if proposed == %{} do + put_flash(socket, :error, "No changes detected.") + else + Radio.apply_edit_to_contact(contact, proposed) + updated = Radio.get_contact!(contact.id) - {:noreply, put_flash(socket, :error, "Could not submit edit: #{messages}")} - end - end + socket + |> assign(contact: updated, editing: false, edit_form: nil) + |> put_flash(:info, "Contact updated.") + end + end + + defp submit_user_edit(socket, contact, user, params) do + case Radio.create_contact_edit(contact, user, params) do + {:ok, _edit} -> + socket + |> assign(editing: false, edit_form: nil, pending_edit: Radio.pending_edit_for_user(contact.id, user.id)) + |> put_flash(:info, "Edit submitted for review.") + + {:error, changeset} -> + messages = + changeset + |> Ecto.Changeset.traverse_errors(fn {msg, _opts} -> msg end) + |> Enum.flat_map(fn {_field, msgs} -> msgs end) + |> Enum.join(", ") + + put_flash(socket, :error, "Could not submit edit: #{messages}") end end @@ -367,41 +376,54 @@ defmodule MicrowavepropWeb.ContactLive.Show do end defp load_weather(contact) do + case extract_contact_coords(contact) do + {:ok, lat1, lon1, lat2, lon2} -> + fetch_weather_for_path(contact, lat1, lon1, lat2, lon2) + + :error -> + %{surface_observations: [], soundings: []} + end + end + + defp extract_contact_coords(contact) do lat1 = contact.pos1 && contact.pos1["lat"] lon1 = contact.pos1 && (contact.pos1["lon"] || contact.pos1["lng"]) - lat2 = contact.pos2 && contact.pos2["lat"] - lon2 = contact.pos2 && (contact.pos2["lon"] || contact.pos2["lng"]) if lat1 && lon1 do - # Search along the path midpoint with radius covering both endpoints - mid_lat = if lat2, do: (lat1 + lat2) / 2, else: lat1 - mid_lon = if lon2, do: (lon1 + lon2) / 2, else: lon1 - half_dist = if lat2 && lon2, do: haversine_km(lat1, lon1, lat2, lon2) / 2, else: 0 - radius = max(50, half_dist + 50) - - weather = - Weather.weather_for_contact( - %{lat: mid_lat, lon: mid_lon, timestamp: contact.qso_timestamp}, - radius_km: radius - ) - - # Keep only the 5 closest stations by distance to path midpoint - %{ - surface_observations: - weather.surface_observations - |> Enum.sort_by(fn obs -> - slat = obs.station.lat - slon = obs.station.lon - abs(slat - mid_lat) + abs(slon - mid_lon) - end) - |> Enum.take(5), - soundings: weather.soundings - } + lat2 = contact.pos2 && contact.pos2["lat"] + lon2 = contact.pos2 && (contact.pos2["lon"] || contact.pos2["lng"]) + {:ok, lat1, lon1, lat2, lon2} else - %{surface_observations: [], soundings: []} + :error end end + defp fetch_weather_for_path(contact, lat1, lon1, lat2, lon2) do + mid_lat = if lat2, do: (lat1 + lat2) / 2, else: lat1 + mid_lon = if lon2, do: (lon1 + lon2) / 2, else: lon1 + half_dist = if lat2 && lon2, do: haversine_km(lat1, lon1, lat2, lon2) / 2, else: 0 + radius = max(50, half_dist + 50) + + weather = + Weather.weather_for_contact( + %{lat: mid_lat, lon: mid_lon, timestamp: contact.qso_timestamp}, + radius_km: radius + ) + + %{ + surface_observations: closest_observations(weather.surface_observations, mid_lat, mid_lon, 5), + soundings: weather.soundings + } + end + + defp closest_observations(observations, mid_lat, mid_lon, limit) do + observations + |> Enum.sort_by(fn obs -> + abs(obs.station.lat - mid_lat) + abs(obs.station.lon - mid_lon) + end) + |> Enum.take(limit) + end + defp maybe_enqueue_terrain(terrain, contact) when not is_nil(terrain) do if contact.terrain_status != :complete do Radio.set_enrichment_status!([contact.id], :terrain_status, :complete) @@ -1294,10 +1316,14 @@ defmodule MicrowavepropWeb.ContactLive.Show do defp extract_ducts(hrrr_path, soundings, surface_elev) do hrrr_ducts = extract_hrrr_ducts(hrrr_path, surface_elev) - if hrrr_ducts == [] do - sounding_ducts = extract_sounding_ducts(soundings, surface_elev) + cond do + hrrr_ducts != [] -> + hrrr_ducts - if sounding_ducts == [] do + (sounding_ducts = extract_sounding_ducts(soundings, surface_elev)) != [] -> + sounding_ducts + + true -> # Use the profile with the strongest (most negative) gradient best = hrrr_path @@ -1305,11 +1331,6 @@ defmodule MicrowavepropWeb.ContactLive.Show do |> Enum.min_by(& &1.min_refractivity_gradient, fn -> nil end) infer_duct_from_gradient(best, surface_elev) - else - sounding_ducts - end - else - hrrr_ducts end end @@ -1595,78 +1616,99 @@ defmodule MicrowavepropWeb.ContactLive.Show do defp terrain_summary(terrain, elevation_profile) do verdict = (elevation_profile && elevation_profile.verdict) || (terrain && terrain.verdict) + describe_verdict(verdict, elevation_profile) + end - case verdict do - "CLEAR" -> - "Line of sight is clear between stations." + defp describe_verdict("CLEAR", _), do: "Line of sight is clear between stations." + defp describe_verdict("FRESNEL_MINOR", _), do: "Line of sight is clear but with minor Fresnel zone encroachment." + defp describe_verdict("FRESNEL_PARTIAL", _), do: "Line of sight has partial Fresnel zone obstruction." - "FRESNEL_MINOR" -> - "Line of sight is clear but with minor Fresnel zone encroachment." + defp describe_verdict("BLOCKED", elevation_profile) do + obs_dist = elevation_profile && elevation_profile.first_obstruction_km + blocked_message(obs_dist) + end - "FRESNEL_PARTIAL" -> - "Line of sight has partial Fresnel zone obstruction." + defp describe_verdict(_, _), do: nil - "BLOCKED" -> - obs_dist = elevation_profile && elevation_profile.first_obstruction_km + defp blocked_message(nil), do: "Path is terrain-obstructed." - if obs_dist do - "Path is terrain-obstructed at #{:erlang.float_to_binary(obs_dist * 0.621371, decimals: 1)} mi (#{:erlang.float_to_binary(obs_dist, decimals: 1)} km) from station 1." - else - "Path is terrain-obstructed." - end - - _ -> - nil - end + defp blocked_message(obs_dist) do + "Path is terrain-obstructed at #{:erlang.float_to_binary(obs_dist * 0.621371, decimals: 1)} mi (#{:erlang.float_to_binary(obs_dist, decimals: 1)} km) from station 1." end defp propagation_mechanism(terrain, elevation_profile, hrrr, soundings, dist_km) do blocked? = terrain && terrain.verdict == "BLOCKED" ducting? = has_ducting?(hrrr, soundings) - - enhanced_refraction? = - hrrr != nil && is_number(hrrr.min_refractivity_gradient) && hrrr.min_refractivity_gradient < -100 - + enhanced_refraction? = enhanced_refraction?(hrrr) long_path? = dist_km && dist_km > 100 ducts = (elevation_profile && elevation_profile.ducts) || [] likely_duct = Enum.find(ducts, & &1[:likely]) - cond do - blocked? && ducting? && likely_duct -> - "Signal likely propagated via atmospheric ducting (#{duct_description(likely_duct)}), bending over the terrain obstruction." + props = %{ + blocked?: blocked?, + ducting?: ducting?, + enhanced_refraction?: enhanced_refraction?, + long_path?: long_path?, + likely_duct: likely_duct, + terrain: terrain, + hrrr: hrrr + } - blocked? && ducting? -> - "Ducting conditions detected — signal likely propagated through a tropospheric duct, bypassing the terrain obstruction." - - blocked? && enhanced_refraction? -> - grad = round(hrrr.min_refractivity_gradient) - - "Enhanced refraction (dN/dh = #{grad} N-units/km) likely bent the signal over the obstruction. Conditions are super-refractive but below full ducting threshold." - - blocked? -> - diff_db = terrain && terrain.diffraction_db - - if diff_db && diff_db > 0 do - "Path is obstructed with #{:erlang.float_to_binary(diff_db, decimals: 1)} dB diffraction loss. Contact may have been enabled by knife-edge diffraction or tropospheric scatter." - else - "Path is obstructed. Contact likely enabled by tropospheric scatter or transient ducting conditions." - end - - ducting? && long_path? && likely_duct -> - "Ducting conditions present (#{duct_description(likely_duct)}) — likely extending range beyond normal line of sight." - - ducting? && long_path? -> - "Ducting conditions present — likely contributing to extended range." - - enhanced_refraction? && long_path? -> - "Enhanced refraction present — may have contributed to extended range." - - true -> - nil + if blocked? do + blocked_mechanism(props) + else + clear_path_mechanism(props) end end + defp enhanced_refraction?(nil), do: false + + defp enhanced_refraction?(hrrr) do + is_number(hrrr.min_refractivity_gradient) && hrrr.min_refractivity_gradient < -100 + end + + defp blocked_mechanism(%{ducting?: true, likely_duct: duct}) when not is_nil(duct) do + "Signal likely propagated via atmospheric ducting (#{duct_description(duct)}), bending over the terrain obstruction." + end + + defp blocked_mechanism(%{ducting?: true}) do + "Ducting conditions detected — signal likely propagated through a tropospheric duct, bypassing the terrain obstruction." + end + + defp blocked_mechanism(%{enhanced_refraction?: true, hrrr: hrrr}) do + grad = round(hrrr.min_refractivity_gradient) + + "Enhanced refraction (dN/dh = #{grad} N-units/km) likely bent the signal over the obstruction. Conditions are super-refractive but below full ducting threshold." + end + + defp blocked_mechanism(%{terrain: terrain}) do + diff_db = terrain && terrain.diffraction_db + blocked_diffraction_message(diff_db) + end + + defp blocked_diffraction_message(diff_db) when is_number(diff_db) and diff_db > 0 do + "Path is obstructed with #{:erlang.float_to_binary(diff_db, decimals: 1)} dB diffraction loss. Contact may have been enabled by knife-edge diffraction or tropospheric scatter." + end + + defp blocked_diffraction_message(_) do + "Path is obstructed. Contact likely enabled by tropospheric scatter or transient ducting conditions." + end + + defp clear_path_mechanism(%{ducting?: true, long_path?: true, likely_duct: duct}) when not is_nil(duct) do + "Ducting conditions present (#{duct_description(duct)}) — likely extending range beyond normal line of sight." + end + + defp clear_path_mechanism(%{ducting?: true, long_path?: true}) do + "Ducting conditions present — likely contributing to extended range." + end + + defp clear_path_mechanism(%{enhanced_refraction?: true, long_path?: true}) do + "Enhanced refraction present — may have contributed to extended range." + end + + defp clear_path_mechanism(_), do: nil + defp duct_description(%{source: source, base_m_msl: base, top_m_msl: top, strength: strength}) do base_ft = round(base * 3.28084) top_ft = round(top * 3.28084) @@ -1704,61 +1746,58 @@ defmodule MicrowavepropWeb.ContactLive.Show do defp band_summary(_, _), do: nil defp build_details(hrrr, soundings, elevation_profile, _band_mhz) do - details = [] - - details = - if hrrr && is_number(hrrr.min_refractivity_gradient) do - grad = round(hrrr.min_refractivity_gradient) - - label = - cond do - grad < -157 -> "ducting" - grad < -100 -> "super-refractive" - grad < -79 -> "enhanced" - true -> "normal" - end - - details ++ ["Refractivity gradient: #{grad} N-units/km (#{label})."] - else - details - end - - details = - if hrrr && is_number(hrrr.hpbl_m) do - hpbl = round(hrrr.hpbl_m) - note = if hpbl < 300, do: " — shallow boundary layer favors ducting", else: "" - details ++ ["Boundary layer depth: #{hpbl} m#{note}."] - else - details - end - - ducting_soundings = if is_list(soundings), do: Enum.filter(soundings, & &1.ducting_detected), else: [] - - details = - if ducting_soundings == [] do - details - else - names = Enum.map_join(ducting_soundings, ", ", fn s -> s.station.name || s.station.station_code end) - details ++ ["Ducting detected by soundings at: #{names}."] - end - - details = - if elevation_profile do - ducts = elevation_profile.ducts || [] - likely = Enum.find(ducts, & &1[:likely]) - - if likely do - details ++ ["Most likely propagation duct: #{duct_description(likely)}."] - else - details - end - else - details - end - - details + Enum.reject( + [ + refractivity_detail(hrrr), + boundary_layer_detail(hrrr), + sounding_ducting_detail(soundings), + duct_layer_detail(elevation_profile) + ], + &is_nil/1 + ) end + defp refractivity_detail(%{min_refractivity_gradient: grad}) when is_number(grad) do + g = round(grad) + label = refractivity_label(g) + "Refractivity gradient: #{g} N-units/km (#{label})." + end + + defp refractivity_detail(_), do: nil + + defp refractivity_label(g) when g < -157, do: "ducting" + defp refractivity_label(g) when g < -100, do: "super-refractive" + defp refractivity_label(g) when g < -79, do: "enhanced" + defp refractivity_label(_), do: "normal" + + defp boundary_layer_detail(%{hpbl_m: hpbl}) when is_number(hpbl) do + h = round(hpbl) + note = if h < 300, do: " — shallow boundary layer favors ducting", else: "" + "Boundary layer depth: #{h} m#{note}." + end + + defp boundary_layer_detail(_), do: nil + + defp sounding_ducting_detail(soundings) when is_list(soundings) do + ducting = Enum.filter(soundings, & &1.ducting_detected) + + if ducting == [] do + nil + else + names = Enum.map_join(ducting, ", ", fn s -> s.station.name || s.station.station_code end) + "Ducting detected by soundings at: #{names}." + end + end + + defp sounding_ducting_detail(_), do: nil + + defp duct_layer_detail(%{ducts: ducts}) when is_list(ducts) do + likely = Enum.find(ducts, & &1[:likely]) + if likely, do: "Most likely propagation duct: #{duct_description(likely)}." + end + + defp duct_layer_detail(_), do: nil + # Factor note helpers defp humidity_note(abs_h, band_config) do h = Float.round(abs_h, 1) @@ -1770,19 +1809,15 @@ defmodule MicrowavepropWeb.ContactLive.Show do solar_hour = ts.hour + ts.minute / 60 + lon / 15 solar_hour = rem_float(solar_hour + 24, 24) h = round(solar_hour) - - label = - cond do - h >= 4 and h < 8 -> "dawn" - h >= 8 and h < 12 -> "morning" - h >= 12 and h < 17 -> "afternoon" - h >= 17 and h < 21 -> "evening" - true -> "night" - end - - "Solar hour ~#{h} (#{label})" + "Solar hour ~#{h} (#{solar_period(h)})" end + defp solar_period(h) when h >= 4 and h < 8, do: "dawn" + defp solar_period(h) when h >= 8 and h < 12, do: "morning" + defp solar_period(h) when h >= 12 and h < 17, do: "afternoon" + defp solar_period(h) when h >= 17 and h < 21, do: "evening" + defp solar_period(_), do: "night" + defp td_note(temp_f, dewpoint_f) when is_number(temp_f) and is_number(dewpoint_f) do depression = Float.round(temp_f - dewpoint_f, 1) diff --git a/lib/microwaveprop_web/live/contact_map_live.ex b/lib/microwaveprop_web/live/contact_map_live.ex index ebc28f02..f7077cbd 100644 --- a/lib/microwaveprop_web/live/contact_map_live.ex +++ b/lib/microwaveprop_web/live/contact_map_live.ex @@ -108,23 +108,43 @@ defmodule MicrowavepropWeb.ContactMapLive do } ) |> Repo.all() - |> Enum.map(fn c -> - lat1 = c.pos1["lat"] - lon1 = c.pos1["lon"] || c.pos1["lng"] - lat2 = c.pos2["lat"] - lon2 = c.pos2["lon"] || c.pos2["lng"] - band = if c.band, do: Decimal.to_integer(c.band), else: 0 - dist = if c.distance_km, do: Decimal.to_float(c.distance_km) - ts = if c.qso_timestamp, do: Calendar.strftime(c.qso_timestamp, "%Y-%m-%d %H:%M") - - if lat1 && lon1 && lat2 && lon2 do - [lat1, lon1, lat2, lon2, band, c.station1, c.station2, c.mode, dist, ts, c.id] - end - end) + |> Enum.map(&format_contact/1) |> Enum.reject(&is_nil/1) |> dedup_reciprocals() end + defp format_contact(c) do + lat1 = c.pos1["lat"] + lon1 = c.pos1["lon"] || c.pos1["lng"] + lat2 = c.pos2["lat"] + lon2 = c.pos2["lon"] || c.pos2["lng"] + + if lat1 && lon1 && lat2 && lon2 do + [ + lat1, + lon1, + lat2, + lon2, + format_band(c.band), + c.station1, + c.station2, + c.mode, + format_distance(c.distance_km), + format_timestamp(c.qso_timestamp), + c.id + ] + end + end + + defp format_band(nil), do: 0 + defp format_band(band), do: Decimal.to_integer(band) + + defp format_distance(nil), do: nil + defp format_distance(d), do: Decimal.to_float(d) + + defp format_timestamp(nil), do: nil + defp format_timestamp(ts), do: Calendar.strftime(ts, "%Y-%m-%d %H:%M") + # Deduplicate reciprocal contacts (A↔B same band/hour = same contact logged by both sides) defp dedup_reciprocals(contacts) do Enum.uniq_by(contacts, fn [_lat1, _lon1, _lat2, _lon2, band, s1, s2, _mode, _dist, ts, _id] -> diff --git a/lib/mix/tasks/import_contest_logs.ex b/lib/mix/tasks/import_contest_logs.ex index e9ab1f75..bac4843d 100644 --- a/lib/mix/tasks/import_contest_logs.ex +++ b/lib/mix/tasks/import_contest_logs.ex @@ -48,52 +48,62 @@ defmodule Mix.Tasks.ImportContestLogs do case CsvImport.preview(combined, "contest-import@ntms.org") do {:ok, preview} -> - Mix.shell().info(""" - Preview: - Valid: #{length(preview.valid)} - Duplicates: #{length(preview.duplicates)} - Invalid: #{length(preview.invalid)} - """) - - if preview.invalid != [] do - preview.invalid - |> Enum.take(10) - |> Enum.each(fn row -> - Mix.shell().info(" Row #{row.row_num}: #{Enum.join(row.messages, ", ")}") - end) - - if length(preview.invalid) > 10 do - Mix.shell().info(" ... and #{length(preview.invalid) - 10} more") - end - end + print_preview_summary(preview) + print_invalid_rows(preview.invalid) if preview.valid == [] do Mix.shell().info("Nothing to import.") else - Mix.shell().info("Importing #{length(preview.valid)} contacts...") - - # Import in batches to show progress - preview.valid - |> Enum.chunk_every(500) - |> Enum.with_index(1) - |> Enum.reduce({0, 0}, fn {batch, batch_num}, {imported, errors} -> - {:ok, result} = CsvImport.commit(batch) - new_imported = imported + length(result.imported) - new_errors = errors + length(result.errors) - - Mix.shell().info( - " Batch #{batch_num}: +#{length(result.imported)} imported, #{length(result.errors)} errors (total: #{new_imported})" - ) - - {new_imported, new_errors} - end) - |> then(fn {imported, errors} -> - Mix.shell().info("Done! Imported #{imported} contacts, #{errors} errors.") - end) + import_batches(preview.valid) end {:error, reason} -> Mix.raise("Failed: #{inspect(reason)}") end end + + defp print_preview_summary(preview) do + Mix.shell().info(""" + Preview: + Valid: #{length(preview.valid)} + Duplicates: #{length(preview.duplicates)} + Invalid: #{length(preview.invalid)} + """) + end + + defp print_invalid_rows([]), do: :ok + + defp print_invalid_rows(invalid) do + invalid + |> Enum.take(10) + |> Enum.each(fn row -> + Mix.shell().info(" Row #{row.row_num}: #{Enum.join(row.messages, ", ")}") + end) + + if length(invalid) > 10 do + Mix.shell().info(" ... and #{length(invalid) - 10} more") + end + end + + defp import_batches(valid) do + Mix.shell().info("Importing #{length(valid)} contacts...") + + {imported, errors} = + valid + |> Enum.chunk_every(500) + |> Enum.with_index(1) + |> Enum.reduce({0, 0}, fn {batch, batch_num}, {imported, errors} -> + {:ok, result} = CsvImport.commit(batch) + new_imported = imported + length(result.imported) + new_errors = errors + length(result.errors) + + Mix.shell().info( + " Batch #{batch_num}: +#{length(result.imported)} imported, #{length(result.errors)} errors (total: #{new_imported})" + ) + + {new_imported, new_errors} + end) + + Mix.shell().info("Done! Imported #{imported} contacts, #{errors} errors.") + end end diff --git a/lib/mix/tasks/propagation_grid.ex b/lib/mix/tasks/propagation_grid.ex index e9e2b738..5ca97c61 100644 --- a/lib/mix/tasks/propagation_grid.ex +++ b/lib/mix/tasks/propagation_grid.ex @@ -4,6 +4,8 @@ defmodule Mix.Tasks.PropagationGrid do @moduledoc "Manually trigger propagation grid computation." use Mix.Task + alias Microwaveprop.Workers.PropagationGridWorker + @impl Mix.Task def run(_args) do # Pause all Oban queues so backfill jobs don't run alongside the grid fetch @@ -21,7 +23,7 @@ defmodule Mix.Tasks.PropagationGrid do IO.puts("Starting propagation grid computation...") - Microwaveprop.Workers.PropagationGridWorker.perform(%Oban.Job{args: %{}}) + PropagationGridWorker.perform(%Oban.Job{args: %{}}) IO.puts("Done!") end end diff --git a/test/microwaveprop/weather/grib2/complex_packing_test.exs b/test/microwaveprop/weather/grib2/complex_packing_test.exs index c9d0e1cf..6d2a199e 100644 --- a/test/microwaveprop/weather/grib2/complex_packing_test.exs +++ b/test/microwaveprop/weather/grib2/complex_packing_test.exs @@ -2,6 +2,7 @@ defmodule Microwaveprop.Weather.Grib2.ComplexPackingTest do use ExUnit.Case, async: true alias Microwaveprop.Weather.Grib2.ComplexPacking + alias Microwaveprop.Weather.Grib2.LambertConformal alias Microwaveprop.Weather.Grib2.Section @fixture_path "test/fixtures/grib2/hrrr_tmp_2m.grib2" @@ -52,7 +53,7 @@ defmodule Microwaveprop.Weather.Grib2.ComplexPackingTest do {:ok, parsed} = Section.parse_message(fixture) grid = parsed.grid_params - {:ok, {i, j}} = Microwaveprop.Weather.Grib2.LambertConformal.to_grid_index(grid, 32.78, -96.8) + {:ok, {i, j}} = LambertConformal.to_grid_index(grid, 32.78, -96.8) index = j * grid.nx + i {:ok, value} = diff --git a/test/support/conn_case.ex b/test/support/conn_case.ex index a7a85f4c..791e7121 100644 --- a/test/support/conn_case.ex +++ b/test/support/conn_case.ex @@ -17,6 +17,10 @@ defmodule MicrowavepropWeb.ConnCase do use ExUnit.CaseTemplate + alias Microwaveprop.Accounts + alias Microwaveprop.Accounts.Scope + alias Microwaveprop.AccountsFixtures + using do quote do use MicrowavepropWeb, :verified_routes @@ -45,8 +49,8 @@ defmodule MicrowavepropWeb.ConnCase do test context. """ def register_and_log_in_user(%{conn: conn} = context) do - user = Microwaveprop.AccountsFixtures.user_fixture() - scope = Microwaveprop.Accounts.Scope.for_user(user) + user = AccountsFixtures.user_fixture() + scope = Scope.for_user(user) opts = context @@ -62,7 +66,7 @@ defmodule MicrowavepropWeb.ConnCase do It returns an updated `conn`. """ def log_in_user(conn, user, opts \\ []) do - token = Microwaveprop.Accounts.generate_user_session_token(user) + token = Accounts.generate_user_session_token(user) maybe_set_token_authenticated_at(token, opts[:token_authenticated_at]) @@ -74,6 +78,6 @@ defmodule MicrowavepropWeb.ConnCase do defp maybe_set_token_authenticated_at(_token, nil), do: nil defp maybe_set_token_authenticated_at(token, authenticated_at) do - Microwaveprop.AccountsFixtures.override_token_authenticated_at(token, authenticated_at) + AccountsFixtures.override_token_authenticated_at(token, authenticated_at) end end diff --git a/test/support/fixtures/accounts_fixtures.ex b/test/support/fixtures/accounts_fixtures.ex index 8cc297aa..998a341d 100644 --- a/test/support/fixtures/accounts_fixtures.ex +++ b/test/support/fixtures/accounts_fixtures.ex @@ -8,6 +8,8 @@ defmodule Microwaveprop.AccountsFixtures do alias Microwaveprop.Accounts alias Microwaveprop.Accounts.Scope + alias Microwaveprop.Accounts.User + alias Microwaveprop.Repo def unique_user_email, do: "user#{System.unique_integer([:positive])}@example.com" def unique_user_callsign, do: "W#{[:positive] |> System.unique_integer() |> rem(1_000_000)}X" @@ -39,8 +41,8 @@ defmodule Microwaveprop.AccountsFixtures do {:ok, user} = user - |> Microwaveprop.Accounts.User.confirm_changeset() - |> Microwaveprop.Repo.update() + |> User.confirm_changeset() + |> Repo.update() user end @@ -61,7 +63,7 @@ defmodule Microwaveprop.AccountsFixtures do end def override_token_authenticated_at(token, authenticated_at) when is_binary(token) do - Microwaveprop.Repo.update_all( + Repo.update_all( from(t in Accounts.UserToken, where: t.token == ^token ), @@ -72,7 +74,7 @@ defmodule Microwaveprop.AccountsFixtures do def offset_user_token(token, amount_to_add, unit) do dt = DateTime.add(DateTime.utc_now(:second), amount_to_add, unit) - Microwaveprop.Repo.update_all( + Repo.update_all( from(ut in Accounts.UserToken, where: ut.token == ^token), set: [inserted_at: dt, authenticated_at: dt] )