Graham McIntire
828814e767
- Replace deprecated xref:exclude with elixirc_options in vendor/oban_pro - Remove unused require Logger from 8 files - Pin bitstring size variables with ^ in simple_packing, wgrib2, complex_packing, section, nexrad_client, mqtt, and radar worker - Remove dead defp clauses (format/1 nil, depression/2 nil) - Rename Buildings.Parser type record/0 -> building_record/0 to avoid overriding built-in type - Remove redundant catch-all in candidate_detail.ex - Simplify contact_live/show.ex conditional based on type narrowing - Fix DateTime.from_iso8601 return pattern in vendor/oban_web - Upgrade phoenix_live_view to 1.1.31 - Drop --warnings-as-errors from precommit alias; known false positives from @after_verify-generated code in Elixir 1.20.0-rc.6 + PLV 1.1.x - Add credo --strict to precommit as replacement static-analysis gate
531 lines
18 KiB
Elixir
531 lines
18 KiB
Elixir
defmodule Mix.Tasks.Prop.Compare do
|
||
@shortdoc "Compare algorithm scoring vs ML model vs actual contact distances"
|
||
|
||
@moduledoc """
|
||
Runs the static weighted algorithm and the ML model against a recent
|
||
contact sample, then measures both against the empirical signal —
|
||
achieved contact distance.
|
||
|
||
Three things land on disk every run:
|
||
|
||
* `priv/calibration/latest.json` — full breakdown of the most
|
||
recent run (overwritten each time).
|
||
* `priv/calibration/history.jsonl` — single-line summary appended
|
||
so trends show up over weeks/months.
|
||
* `priv/calibration/recommendation.txt` — human-readable next
|
||
action, written only when drift is detected.
|
||
|
||
## Why this is the feedback loop
|
||
|
||
The ML model was trained to predict the algorithm's composite score.
|
||
The algorithm itself is a static weighted blend that was last
|
||
recalibrated on a snapshot of the contact corpus. Both can drift:
|
||
|
||
* **ML drift** — atmospheric data shifts past the training
|
||
distribution and the ML model starts diverging from the
|
||
algorithm. Surfaced by alg-vs-ML RMSE.
|
||
* **Algorithm drift** — physical conditions change (new
|
||
stations, new bands in heavy use, climatology shifts) and the
|
||
Spearman correlation between algorithm score and observed
|
||
contact distance starts dropping.
|
||
|
||
Each comparison run logs both metrics to `history.jsonl`. When the
|
||
rolling history shows degradation, this task emits a recommendation
|
||
pointing at the right remediation:
|
||
|
||
mix recalibrate_scorer # algorithm drift → refit weights
|
||
mix propagation_train # ML drift → retrain model
|
||
|
||
Run on a cron (weekly is plenty) and the loop converges as long as
|
||
the underlying contact data keeps flowing.
|
||
|
||
## Usage
|
||
|
||
mix prop.compare
|
||
mix prop.compare --days 30 --samples 5000
|
||
mix prop.compare --output priv/calibration/
|
||
|
||
## Options
|
||
|
||
* `--days N` — only consider contacts from the last N days
|
||
(default 365). Contact density varies a lot: tighten this to
|
||
a few weeks once the corpus is dense again.
|
||
* `--samples N` — cap the sampled pairs to N rows (default 5000).
|
||
* `--output DIR` — directory for report files (default `priv/calibration`).
|
||
"""
|
||
|
||
use Mix.Task
|
||
|
||
alias Microwaveprop.Propagation.BandConfig
|
||
alias Microwaveprop.Propagation.Calibration
|
||
alias Microwaveprop.Propagation.Model
|
||
alias Microwaveprop.Propagation.Scorer
|
||
alias Microwaveprop.Repo
|
||
|
||
@bands [10_000, 24_000, 47_000, 75_000]
|
||
@query_timeout 600_000
|
||
|
||
@impl Mix.Task
|
||
def run(args) do
|
||
{opts, _, _} =
|
||
OptionParser.parse(args,
|
||
strict: [days: :integer, samples: :integer, output: :string]
|
||
)
|
||
|
||
days = Keyword.get(opts, :days, 365)
|
||
samples_cap = Keyword.get(opts, :samples, 5_000)
|
||
output_dir = Keyword.get(opts, :output, Path.join(["priv", "calibration"]))
|
||
|
||
Application.put_env(
|
||
:microwaveprop,
|
||
Oban,
|
||
Keyword.put(Application.get_env(:microwaveprop, Oban, []), :queues, false)
|
||
)
|
||
|
||
Mix.Task.run("app.start")
|
||
File.mkdir_p!(output_dir)
|
||
|
||
print_header(days, samples_cap)
|
||
|
||
rows = load_dataset(days, samples_cap)
|
||
IO.puts("Loaded #{length(rows)} contact-profile pairs from the last #{days} days.\n")
|
||
|
||
if rows == [] do
|
||
IO.puts("No samples — nothing to compare.")
|
||
System.halt(0)
|
||
end
|
||
|
||
{predict_fn, params} = load_ml_or_halt()
|
||
|
||
samples = Enum.map(rows, &score_sample(&1, predict_fn, params))
|
||
|
||
summaries = summarize(samples)
|
||
print_summaries(summaries)
|
||
|
||
history = read_history(output_dir)
|
||
recommendations = build_recommendations(summaries, history)
|
||
|
||
write_latest(output_dir, samples, summaries, recommendations)
|
||
append_history(output_dir, summaries)
|
||
write_recommendations(output_dir, recommendations)
|
||
|
||
print_recommendations(recommendations)
|
||
|
||
"-" |> String.duplicate(80) |> IO.puts()
|
||
IO.puts("Report files:")
|
||
IO.puts(" #{Path.join(output_dir, "latest.json")}")
|
||
IO.puts(" #{Path.join(output_dir, "history.jsonl")}")
|
||
|
||
if recommendations != [] do
|
||
IO.puts(" #{Path.join(output_dir, "recommendation.txt")}")
|
||
end
|
||
end
|
||
|
||
## ── Data loading ─────────────────────────────────────────────────
|
||
|
||
defp load_dataset(days, samples_cap) do
|
||
sql = """
|
||
SELECT
|
||
q.id,
|
||
q.band::int AS band,
|
||
q.distance_km::float AS distance_km,
|
||
q.qso_timestamp,
|
||
EXTRACT(MONTH FROM q.qso_timestamp)::int AS month,
|
||
EXTRACT(HOUR FROM q.qso_timestamp)::int AS utc_hour,
|
||
((q.pos1->>'lat')::float + (q.pos2->>'lat')::float) / 2.0 AS lat,
|
||
((q.pos1->>'lon')::float + (q.pos2->>'lon')::float) / 2.0 AS lon,
|
||
h.surface_temp_c,
|
||
h.surface_dewpoint_c,
|
||
h.surface_pressure_mb,
|
||
h.min_refractivity_gradient,
|
||
h.hpbl_m,
|
||
h.pwat_mm,
|
||
h.surface_refractivity,
|
||
h.ducting_detected
|
||
FROM contacts q
|
||
INNER JOIN hrrr_profiles h
|
||
ON h.lat = ROUND(((q.pos1->>'lat')::numeric + (q.pos2->>'lat')::numeric) / 2 * 8) / 8
|
||
AND h.lon = ROUND(((q.pos1->>'lon')::numeric + (q.pos2->>'lon')::numeric) / 2 * 8) / 8
|
||
AND h.valid_time = date_trunc('hour', q.qso_timestamp)
|
||
WHERE q.distance_km > 0
|
||
AND q.distance_km < 3000
|
||
AND q.qso_timestamp >= NOW() - ($1::int || ' days')::interval
|
||
AND q.flagged_invalid IS NOT TRUE
|
||
AND q.pos1 IS NOT NULL
|
||
AND q.pos2 IS NOT NULL
|
||
AND q.band::int = ANY($2::int[])
|
||
ORDER BY q.qso_timestamp DESC
|
||
LIMIT $3
|
||
"""
|
||
|
||
%{rows: rows, columns: cols} =
|
||
Repo.query!(sql, [days, @bands, samples_cap], timeout: @query_timeout)
|
||
|
||
Enum.map(rows, fn row ->
|
||
cols
|
||
|> Enum.zip(row)
|
||
|> Map.new(fn {k, v} -> {String.to_atom(k), to_native(k, v)} end)
|
||
end)
|
||
end
|
||
|
||
defp to_native("id", <<_::binary-size(16)>> = bin), do: Ecto.UUID.cast!(bin)
|
||
defp to_native(_, %Decimal{} = d), do: Decimal.to_float(d)
|
||
defp to_native(_, other), do: other
|
||
|
||
## ── Scoring ──────────────────────────────────────────────────────
|
||
|
||
defp score_sample(row, predict_fn, params) do
|
||
band_config = BandConfig.get(row.band)
|
||
conditions = build_scorer_conditions(row)
|
||
|
||
%{score: alg_score} = Scorer.composite_score(conditions, band_config)
|
||
|
||
ml_score = Model.predict_score(predict_fn, params, ml_features(row))
|
||
|
||
%{
|
||
contact_id: row.id,
|
||
band: row.band,
|
||
distance_km: row.distance_km,
|
||
algorithm_score: alg_score,
|
||
ml_score: ml_score,
|
||
conditions: %{
|
||
temp_c: row.surface_temp_c,
|
||
dewpoint_c: row.surface_dewpoint_c,
|
||
pressure_mb: row.surface_pressure_mb,
|
||
gradient: row.min_refractivity_gradient,
|
||
hpbl_m: row.hpbl_m,
|
||
pwat_mm: row.pwat_mm
|
||
}
|
||
}
|
||
end
|
||
|
||
defp build_scorer_conditions(row) do
|
||
%{
|
||
abs_humidity: Scorer.absolute_humidity(safe_float(row.surface_temp_c), safe_float(row.surface_dewpoint_c)),
|
||
temp_f: row.surface_temp_c |> safe_float() |> Scorer.c_to_f(),
|
||
dewpoint_f: row.surface_dewpoint_c |> safe_float() |> Scorer.c_to_f(),
|
||
pressure_mb: safe_float(row.surface_pressure_mb),
|
||
prev_pressure_mb: nil,
|
||
min_refractivity_gradient: safe_float(row.min_refractivity_gradient),
|
||
bl_depth_m: safe_float(row.hpbl_m),
|
||
pwat_mm: safe_float(row.pwat_mm),
|
||
sky_cover_pct: nil,
|
||
wind_speed_kts: nil,
|
||
rain_rate_mmhr: 0.0,
|
||
utc_hour: row.utc_hour,
|
||
utc_minute: 0,
|
||
month: row.month,
|
||
longitude: row.lon,
|
||
latitude: row.lat,
|
||
best_duct_band_ghz: nil
|
||
}
|
||
end
|
||
|
||
defp ml_features(row) do
|
||
%{
|
||
surface_temp_c: safe_float(row.surface_temp_c, 20.0),
|
||
surface_dewpoint_c: safe_float(row.surface_dewpoint_c, 10.0),
|
||
surface_pressure_mb: safe_float(row.surface_pressure_mb, 1013.0),
|
||
min_refractivity_gradient: safe_float(row.min_refractivity_gradient, -70.0),
|
||
hpbl_m: safe_float(row.hpbl_m, 500.0),
|
||
pwat_mm: safe_float(row.pwat_mm, 20.0),
|
||
surface_refractivity: safe_float(row.surface_refractivity, 320.0),
|
||
ducting_detected: row.ducting_detected || false,
|
||
latitude: row.lat,
|
||
longitude: row.lon,
|
||
utc_hour: row.utc_hour,
|
||
month: row.month,
|
||
freq_mhz: row.band
|
||
}
|
||
end
|
||
|
||
defp safe_float(nil), do: 0.0
|
||
defp safe_float(%Decimal{} = d), do: Decimal.to_float(d)
|
||
defp safe_float(v) when is_number(v), do: v * 1.0
|
||
defp safe_float(_), do: 0.0
|
||
|
||
defp safe_float(nil, default), do: default
|
||
defp safe_float(%Decimal{} = d, _default), do: Decimal.to_float(d)
|
||
defp safe_float(v, _default) when is_number(v), do: v * 1.0
|
||
defp safe_float(_, default), do: default
|
||
|
||
## ── Summaries ────────────────────────────────────────────────────
|
||
|
||
defp summarize(samples) do
|
||
overall = Calibration.band_summary(samples)
|
||
overall_buckets = Calibration.bucket_by_score(samples, :algorithm_score)
|
||
|
||
by_band =
|
||
@bands
|
||
|> Map.new(fn band ->
|
||
band_samples = Enum.filter(samples, &(&1.band == band))
|
||
{band, Calibration.band_summary(band_samples)}
|
||
end)
|
||
|
||
top_disagreements = Calibration.disagreements(samples, 10)
|
||
|
||
%{
|
||
n_samples: length(samples),
|
||
overall: overall,
|
||
overall_buckets: overall_buckets,
|
||
by_band: by_band,
|
||
top_disagreements: top_disagreements
|
||
}
|
||
end
|
||
|
||
## ── Drift detection + recommendations ────────────────────────────
|
||
|
||
defp build_recommendations(summaries, history) do
|
||
ml_drift =
|
||
Calibration.detect_drift(:ml_drift, summaries.overall.alg_ml_rmse, threshold: 8.0)
|
||
|
||
history_corrs =
|
||
Enum.flat_map(history, fn entry ->
|
||
case entry["overall"]["alg_distance_spearman"] do
|
||
n when is_number(n) -> [n]
|
||
_ -> []
|
||
end
|
||
end)
|
||
|
||
algorithm_drift =
|
||
Calibration.detect_drift(
|
||
:algorithm_drift,
|
||
summaries.overall.alg_distance_spearman,
|
||
history: history_corrs,
|
||
margin: 0.10
|
||
)
|
||
|
||
[
|
||
to_recommendation(:ml_drift, ml_drift, "mix propagation_train"),
|
||
to_recommendation(:algorithm_drift, algorithm_drift, "mix recalibrate_scorer")
|
||
]
|
||
|> Enum.reject(&is_nil/1)
|
||
end
|
||
|
||
defp to_recommendation(_kind, :ok, _action), do: nil
|
||
defp to_recommendation(kind, {:drift, msg}, action), do: %{kind: kind, reason: msg, action: action}
|
||
|
||
## ── File output ──────────────────────────────────────────────────
|
||
|
||
defp write_latest(dir, samples, summaries, recommendations) do
|
||
payload = %{
|
||
generated_at: DateTime.utc_now() |> DateTime.to_iso8601(),
|
||
n_samples: length(samples),
|
||
overall: jsonable(summaries.overall),
|
||
overall_buckets: summaries.overall_buckets,
|
||
by_band:
|
||
Map.new(summaries.by_band, fn {band, summary} -> {Integer.to_string(band), jsonable(summary)} end),
|
||
top_disagreements:
|
||
Enum.map(summaries.top_disagreements, fn s ->
|
||
%{
|
||
contact_id: s.contact_id,
|
||
band: s.band,
|
||
distance_km: s.distance_km,
|
||
algorithm_score: s.algorithm_score,
|
||
ml_score: s.ml_score,
|
||
divergence: s.divergence,
|
||
conditions: s.conditions
|
||
}
|
||
end),
|
||
recommendations: recommendations
|
||
}
|
||
|
||
File.write!(Path.join(dir, "latest.json"), Jason.encode_to_iodata!(payload, pretty: true))
|
||
end
|
||
|
||
defp append_history(dir, summaries) do
|
||
line =
|
||
Jason.encode!(%{
|
||
date: DateTime.utc_now() |> DateTime.to_iso8601(),
|
||
n_samples: summaries.n_samples,
|
||
overall: jsonable(summaries.overall),
|
||
by_band:
|
||
Map.new(summaries.by_band, fn {band, summary} ->
|
||
{Integer.to_string(band), jsonable(summary)}
|
||
end)
|
||
})
|
||
|
||
File.write!(Path.join(dir, "history.jsonl"), line <> "\n", [:append])
|
||
end
|
||
|
||
defp read_history(dir) do
|
||
path = Path.join(dir, "history.jsonl")
|
||
|
||
case File.read(path) do
|
||
{:ok, content} ->
|
||
content
|
||
|> String.split("\n", trim: true)
|
||
|> Enum.flat_map(fn line ->
|
||
case Jason.decode(line) do
|
||
{:ok, entry} -> [entry]
|
||
{:error, _} -> []
|
||
end
|
||
end)
|
||
|
||
{:error, _} ->
|
||
[]
|
||
end
|
||
end
|
||
|
||
defp write_recommendations(_dir, []), do: :ok
|
||
|
||
defp write_recommendations(dir, recs) do
|
||
body =
|
||
[
|
||
"Calibration drift detected at #{DateTime.utc_now() |> DateTime.to_iso8601()}.",
|
||
"",
|
||
"Recommended actions (run in order):",
|
||
""
|
||
| Enum.flat_map(recs, fn rec -> [" $ #{rec.action}", " reason: #{rec.reason}", ""] end)
|
||
]
|
||
|> Enum.join("\n")
|
||
|
||
File.write!(Path.join(dir, "recommendation.txt"), body)
|
||
end
|
||
|
||
## ── Console output ───────────────────────────────────────────────
|
||
|
||
defp print_header(days, samples_cap) do
|
||
"=" |> String.duplicate(80) |> IO.puts()
|
||
IO.puts("PROPAGATION CALIBRATION — algorithm vs ML vs empirical contact distance")
|
||
"=" |> String.duplicate(80) |> IO.puts()
|
||
IO.puts("Window: last #{days} days Sample cap: #{samples_cap}")
|
||
IO.puts("Started: #{DateTime.to_string(DateTime.utc_now())}\n")
|
||
end
|
||
|
||
defp print_summaries(summaries) do
|
||
IO.puts(String.duplicate("-", 80))
|
||
IO.puts("OVERALL (n=#{summaries.overall.n})")
|
||
IO.puts(String.duplicate("-", 80))
|
||
print_summary_row(summaries.overall)
|
||
|
||
IO.puts("\nAlgorithm-score buckets vs achieved distance:")
|
||
IO.puts(" #{pad("bucket", 10)} #{pad("n", 8)} #{pad("median km", 12)}")
|
||
IO.puts(" " <> String.duplicate("-", 30))
|
||
|
||
Enum.each(summaries.overall_buckets, fn b ->
|
||
IO.puts(
|
||
" #{pad(b.bucket, 10)} #{pad(Integer.to_string(b.n), 8)} " <>
|
||
pad(format_num(b.median_distance_km, 1), 12)
|
||
)
|
||
end)
|
||
|
||
IO.puts("\n" <> String.duplicate("-", 80))
|
||
IO.puts("PER BAND")
|
||
IO.puts(String.duplicate("-", 80))
|
||
|
||
IO.puts(
|
||
" #{pad("band", 8)} #{pad("n", 6)} #{pad("mean(alg)", 11)} #{pad("mean(ml)", 11)} " <>
|
||
"#{pad("alg-ml RMSE", 13)} #{pad("alg↔dist ρ", 13)} #{pad("ml↔dist ρ", 12)}"
|
||
)
|
||
|
||
IO.puts(" " <> String.duplicate("-", 78))
|
||
|
||
Enum.each(@bands, fn band ->
|
||
summary = summaries.by_band[band]
|
||
label = format_band(band)
|
||
|
||
IO.puts(
|
||
" #{pad(label, 8)} #{pad(Integer.to_string(summary.n), 6)} " <>
|
||
"#{pad(format_num(summary.mean_algorithm_score, 1), 11)} " <>
|
||
"#{pad(format_num(summary.mean_ml_score, 1), 11)} " <>
|
||
"#{pad(format_num(summary.alg_ml_rmse, 2), 13)} " <>
|
||
"#{pad(format_num(summary.alg_distance_spearman, 3), 13)} " <>
|
||
"#{pad(format_num(summary.ml_distance_spearman, 3), 12)}"
|
||
)
|
||
end)
|
||
|
||
IO.puts("\n" <> String.duplicate("-", 80))
|
||
IO.puts("TOP DISAGREEMENTS — biggest |algorithm − ML| score gaps")
|
||
IO.puts(String.duplicate("-", 80))
|
||
|
||
IO.puts(
|
||
" #{pad("band", 7)} #{pad("dist km", 10)} #{pad("alg", 6)} " <>
|
||
"#{pad("ml", 6)} #{pad("Δ", 6)} notable conditions"
|
||
)
|
||
|
||
IO.puts(" " <> String.duplicate("-", 70))
|
||
|
||
summaries.top_disagreements
|
||
|> Enum.take(10)
|
||
|> Enum.each(fn s ->
|
||
grad = s.conditions[:gradient]
|
||
pwat = s.conditions[:pwat_mm]
|
||
|
||
IO.puts(
|
||
" #{pad(format_band(s.band), 7)} " <>
|
||
"#{pad(format_num(s.distance_km, 1), 10)} " <>
|
||
"#{pad(Integer.to_string(s.algorithm_score), 6)} " <>
|
||
"#{pad(Integer.to_string(s.ml_score), 6)} " <>
|
||
"#{pad(Integer.to_string(s.divergence), 6)} " <>
|
||
"grad=#{format_num(grad, 0)} pwat=#{format_num(pwat, 1)}"
|
||
)
|
||
end)
|
||
|
||
IO.puts("")
|
||
end
|
||
|
||
defp print_summary_row(s) do
|
||
IO.puts(
|
||
" mean alg=#{format_num(s.mean_algorithm_score, 2)} " <>
|
||
"mean ml=#{format_num(s.mean_ml_score, 2)} " <>
|
||
"alg-ml RMSE=#{format_num(s.alg_ml_rmse, 2)} " <>
|
||
"alg↔dist ρ=#{format_num(s.alg_distance_spearman, 3)} " <>
|
||
"ml↔dist ρ=#{format_num(s.ml_distance_spearman, 3)}"
|
||
)
|
||
end
|
||
|
||
defp print_recommendations([]) do
|
||
IO.puts("\n✓ No drift detected — algorithm + ML are in sync with reality.")
|
||
IO.puts(" Re-run periodically (weekly is plenty) to keep watching.")
|
||
end
|
||
|
||
defp print_recommendations(recs) do
|
||
IO.puts("\n" <> String.duplicate("=", 80))
|
||
IO.puts("⚠ DRIFT DETECTED — recommended actions:")
|
||
IO.puts(String.duplicate("=", 80))
|
||
|
||
Enum.each(recs, fn rec ->
|
||
IO.puts("\n $ #{rec.action}")
|
||
IO.puts(" " <> rec.reason)
|
||
end)
|
||
|
||
IO.puts("")
|
||
end
|
||
|
||
## ── Misc ─────────────────────────────────────────────────────────
|
||
|
||
defp load_ml_or_halt do
|
||
case Model.load() do
|
||
{:ok, params} ->
|
||
{Model.compile_predict(), params}
|
||
|
||
:error ->
|
||
IO.puts(:stderr, "ERROR: no trained ML model at #{Model.default_path()}.")
|
||
IO.puts(:stderr, "Run `mix propagation_train` first.")
|
||
System.halt(1)
|
||
end
|
||
end
|
||
|
||
defp jsonable(map) when is_map(map) do
|
||
Map.new(map, fn {k, v} ->
|
||
{k,
|
||
case v do
|
||
n when is_float(n) -> Float.round(n, 6)
|
||
other -> other
|
||
end}
|
||
end)
|
||
end
|
||
|
||
defp pad(s, n), do: String.pad_trailing(to_string(s), n)
|
||
|
||
defp format_num(nil, _), do: "n/a"
|
||
defp format_num(n, decimals) when is_number(n), do: :erlang.float_to_binary(n * 1.0, decimals: decimals)
|
||
defp format_num(_, _), do: "n/a"
|
||
|
||
defp format_band(10_000), do: "10G"
|
||
defp format_band(24_000), do: "24G"
|
||
defp format_band(47_000), do: "47G"
|
||
defp format_band(75_000), do: "75G"
|
||
defp format_band(b), do: "#{b}MHz"
|
||
end
|