diff --git a/lib/mix/tasks/propagation_train.ex b/lib/mix/tasks/propagation_train.ex index f2d43503..4b398970 100644 --- a/lib/mix/tasks/propagation_train.ex +++ b/lib/mix/tasks/propagation_train.ex @@ -1,32 +1,45 @@ defmodule Mix.Tasks.PropagationTrain do - @shortdoc "Train ML model from propagation scores + HRRR data" + @shortdoc "Train ML model from QSO-HRRR matched data" @moduledoc """ - Builds training data from propagation_scores joined to hrrr_profiles, - trains the Axon model, evaluates on held-out test set, and saves weights. + Trains the propagation ML model on real QSO outcomes matched to HRRR + atmospheric conditions. The target is QSO distance normalized per-band + (0-1), so the model learns what atmospheric conditions actually produce + good propagation from historical contacts — not from the hand-tuned algorithm. - Uses stratified sampling across all 8 bands so the model learns - frequency-specific scoring behavior via the log_freq_mhz feature. + Features: HRRR conditions averaged across both QSO endpoints + solar time + frequency. + Target: distance_km / band_p99_distance (capped at 1.0). ## Usage mix propagation_train - mix propagation_train --epochs 100 --batch-size 512 --sample 500000 + mix propagation_train --epochs 100 --batch-size 512 """ use Mix.Task - alias Microwaveprop.Propagation.BandConfig alias Microwaveprop.Propagation.Model alias Microwaveprop.Repo @query_timeout 600_000 + # Per-band normalization: p99 distances from the QSO dataset. + # Distance is divided by these values and capped at 1.0. + @band_max_km %{ + 10_000 => 800.0, + 24_000 => 350.0, + 47_000 => 200.0, + 75_000 => 200.0, + 122_000 => 140.0, + 134_000 => 160.0, + 241_000 => 115.0 + } + @impl Mix.Task def run(args) do {opts, _, _} = OptionParser.parse(args, - strict: [epochs: :integer, batch_size: :integer, sample: :integer] + strict: [epochs: :integer, batch_size: :integer] ) Application.put_env( @@ -41,26 +54,27 @@ defmodule Mix.Tasks.PropagationTrain do Mix.Task.run("app.start") - epochs = Keyword.get(opts, :epochs, 50) + epochs = Keyword.get(opts, :epochs, 100) batch_size = Keyword.get(opts, :batch_size, 256) - sample_size = Keyword.get(opts, :sample, 500_000) "=" |> String.duplicate(70) |> IO.puts() - IO.puts("PROPAGATION MODEL TRAINING") + IO.puts("PROPAGATION MODEL TRAINING (QSO-HRRR)") "=" |> String.duplicate(70) |> IO.puts() - IO.puts("Config: epochs=#{epochs}, batch_size=#{batch_size}, sample=#{sample_size}") + IO.puts("Config: epochs=#{epochs}, batch_size=#{batch_size}") IO.puts("Started: #{DateTime.to_string(DateTime.utc_now())}\n") - IO.puts("Loading training data (stratified, sample=#{sample_size})...") - {features, targets, band_counts} = load_training_data(sample_size) + IO.puts("Loading QSO-HRRR training data...") + {features, targets, band_counts} = load_training_data() n = elem(Nx.shape(features), 0) - IO.puts("Loaded #{n} samples across #{map_size(band_counts)} bands:") + IO.puts("\nLoaded #{n} training samples from real QSOs:") band_counts |> Enum.sort_by(fn {band, _} -> band end) |> Enum.each(fn {band, count} -> - IO.puts(" #{div(band, 1000)} GHz: #{count} samples") + max_km = Map.get(@band_max_km, band) + suffix = if max_km, do: " (norm: #{trunc(max_km)} km)", else: " (excluded)" + IO.puts(" #{div(band, 1000)} GHz: #{count} QSOs#{suffix}") end) IO.puts("") @@ -120,45 +134,59 @@ defmodule Mix.Tasks.PropagationTrain do IO.puts("Finished: #{DateTime.to_string(DateTime.utc_now())}") end - defp load_training_data(sample_size) do - bands = BandConfig.all_freqs() - per_band = div(sample_size, length(bands)) + defp load_training_data do + # Join QSOs to HRRR at both endpoints, average conditions. + # Same query as the analysis task but returns raw values for encoding. + sql = """ + SELECT + q.band::integer AS band_mhz, + q.distance_km::float AS distance_km, + EXTRACT(HOUR FROM q.qso_timestamp)::int AS utc_hour, + EXTRACT(MONTH FROM q.qso_timestamp)::int AS month, + ((q.pos1->>'lng')::float + (q.pos2->>'lng')::float) / 2.0 AS avg_lon, + (COALESCE(h1.surface_temp_c, h2.surface_temp_c) + + COALESCE(h2.surface_temp_c, h1.surface_temp_c)) / 2.0 AS avg_temp, + (COALESCE(h1.surface_dewpoint_c, h2.surface_dewpoint_c) + + COALESCE(h2.surface_dewpoint_c, h1.surface_dewpoint_c)) / 2.0 AS avg_dewpoint, + (COALESCE(h1.surface_pressure_mb, h2.surface_pressure_mb) + + COALESCE(h2.surface_pressure_mb, h1.surface_pressure_mb)) / 2.0 AS avg_pressure, + (COALESCE(h1.min_refractivity_gradient, h2.min_refractivity_gradient) + + COALESCE(h2.min_refractivity_gradient, h1.min_refractivity_gradient)) / 2.0 AS avg_gradient, + (COALESCE(h1.hpbl_m, h2.hpbl_m) + + COALESCE(h2.hpbl_m, h1.hpbl_m)) / 2.0 AS avg_hpbl, + (COALESCE(h1.pwat_mm, h2.pwat_mm) + + COALESCE(h2.pwat_mm, h1.pwat_mm)) / 2.0 AS avg_pwat + FROM qsos q + INNER JOIN hrrr_profiles h1 + ON h1.lat = ROUND((q.pos1->>'lat')::numeric * 8) / 8 + AND h1.lon = ROUND((q.pos1->>'lng')::numeric * 8) / 8 + AND h1.valid_time = date_trunc('hour', q.qso_timestamp) + INNER JOIN hrrr_profiles h2 + ON h2.lat = ROUND((q.pos2->>'lat')::numeric * 8) / 8 + AND h2.lon = ROUND((q.pos2->>'lng')::numeric * 8) / 8 + AND h2.valid_time = date_trunc('hour', q.qso_timestamp) + WHERE q.distance_km > 0 + AND q.distance_km < 3000 + AND q.qso_timestamp >= '2016-06-30' + AND q.pos1->>'lng' IS NOT NULL + AND q.pos2->>'lng' IS NOT NULL + """ - band_queries = - Enum.map_join(bands, "\nUNION ALL\n", fn band_mhz -> - """ - (SELECT - ps.score, ps.band_mhz, - h.surface_temp_c, h.surface_dewpoint_c, h.surface_pressure_mb, - h.min_refractivity_gradient, h.hpbl_m, h.pwat_mm, - EXTRACT(HOUR FROM ps.valid_time) AS utc_hour, - EXTRACT(MONTH FROM ps.valid_time) AS month, - ps.lon - FROM propagation_scores ps - JOIN hrrr_profiles h - ON h.lat = ps.lat AND h.lon = ps.lon AND h.valid_time = ps.valid_time - WHERE ps.band_mhz = #{band_mhz} - AND h.surface_temp_c IS NOT NULL - AND h.surface_dewpoint_c IS NOT NULL - AND h.surface_pressure_mb IS NOT NULL - ORDER BY RANDOM() - LIMIT #{per_band}) - """ - end) + IO.puts(" Running QSO-HRRR join query...") + %{rows: rows} = Repo.query!(sql, [], timeout: @query_timeout) + IO.puts(" Matched #{length(rows)} QSOs to HRRR conditions") - %{rows: rows} = Repo.query!(band_queries, [], timeout: @query_timeout) - - band_counts = - Enum.frequencies_by(rows, fn row -> Enum.at(row, 1) end) + band_counts = Enum.frequencies_by(rows, fn row -> Enum.at(row, 0) end) {feature_rows, target_rows} = rows - |> Enum.map(fn [score, band_mhz, temp_c, dewpoint_c, pressure_mb, grad, hpbl, pwat, utc_hour, month, lon] -> + |> Enum.filter(fn [band_mhz | _] -> Map.has_key?(@band_max_km, band_mhz) end) + |> Enum.map(fn [band_mhz, distance_km, utc_hour, month, lon, temp, dewpoint, pressure, grad, hpbl, pwat] -> features = Model.encode_features(%{ - surface_temp_c: to_float(temp_c), - surface_dewpoint_c: to_float(dewpoint_c), - surface_pressure_mb: to_float(pressure_mb), + surface_temp_c: to_float(temp), + surface_dewpoint_c: to_float(dewpoint), + surface_pressure_mb: to_float(pressure), min_refractivity_gradient: to_float(grad), hpbl_m: to_float(hpbl), pwat_mm: to_float(pwat), @@ -168,12 +196,16 @@ defmodule Mix.Tasks.PropagationTrain do freq_mhz: band_mhz }) - target = [score / 100.0] + # Normalize distance by band's p99 range, cap at 1.0 + max_km = Map.fetch!(@band_max_km, band_mhz) + target = [min(distance_km / max_km, 1.0)] {features, target} end) |> Enum.unzip() + IO.puts(" After filtering: #{length(feature_rows)} training samples") + features_tensor = Nx.tensor(feature_rows, type: :f32) targets_tensor = Nx.tensor(target_rows, type: :f32)