diff --git a/lib/microwaveprop/release.ex b/lib/microwaveprop/release.ex index 7b2d9947..2f41dd5c 100644 --- a/lib/microwaveprop/release.ex +++ b/lib/microwaveprop/release.ex @@ -1,7 +1,15 @@ defmodule Microwaveprop.Release do @moduledoc """ - Used for executing DB release tasks when run in production without Mix - installed. + Release tasks that run in production without Mix installed. + + Usage via `bin/microwaveprop eval`: + + bin/microwaveprop eval "Microwaveprop.Release.migrate" + bin/microwaveprop eval "Microwaveprop.Release.backtest_all" + bin/microwaveprop eval "Microwaveprop.Release.backtest(\"naive_gradient\")" + bin/microwaveprop eval "Microwaveprop.Release.climatology" + bin/microwaveprop eval "Microwaveprop.Release.native_backfill(500)" + bin/microwaveprop eval "Microwaveprop.Release.native_derive" """ @app :microwaveprop @@ -18,13 +26,243 @@ defmodule Microwaveprop.Release do {:ok, _, _} = Ecto.Migrator.with_repo(repo, &Ecto.Migrator.run(&1, :down, to: version)) end + @doc """ + Run a single feature backtest. Prints markdown report to stdout. + """ + def backtest(feature_name) when is_binary(feature_name) do + start_app() + + alias Microwaveprop.Backtest + alias Microwaveprop.Backtest.Features + + fun_atom = String.to_atom(feature_name) + + unless function_exported?(Features, fun_atom, 3) do + available = + Features.__info__(:functions) + |> Enum.filter(fn {_, a} -> a == 3 end) + |> Enum.map_join(", ", fn {f, _} -> to_string(f) end) + + IO.puts("Unknown feature: #{feature_name}\nAvailable: #{available}") + System.halt(1) + end + + feature_fun = &apply(Features, fun_atom, [&1, &2, &3]) + name = "Microwaveprop.Backtest.Features.#{feature_name}" + + report = Backtest.evaluate(feature_fun, sample_size: 5000, feature_name: name) + distance_bins = Backtest.lift_by_distance(feature_fun, sample_size: 5000) + band_stats = Backtest.lift_by_band(feature_fun, sample_size: 5000) + + IO.puts(Backtest.to_markdown(report, distance_bins: distance_bins, band_stats: band_stats)) + end + + @doc """ + Run consolidated backtest across all features. Prints markdown to stdout. + """ + def backtest_all do + start_app() + + alias Microwaveprop.Backtest + alias Microwaveprop.Backtest.Features + + features = Features.all_features() + IO.puts("Running consolidated backtest for #{map_size(features)} features...") + + results = Backtest.consolidated_report(features, sample_size: 5000) + IO.puts(Backtest.to_consolidated_markdown(results)) + end + + @doc """ + Build surface temperature climatology from hrrr_profiles. + """ + def climatology(min_samples \\ 3) do + start_app() + + alias Microwaveprop.Repo + + %{rows: combos} = + Repo.query!( + """ + SELECT EXTRACT(MONTH FROM valid_time)::int AS month, + EXTRACT(HOUR FROM valid_time)::int AS hour + FROM hrrr_profiles + WHERE surface_temp_c IS NOT NULL AND is_grid_point = true + GROUP BY 1, 2 + ORDER BY 1, 2 + """, + [], + timeout: 120_000 + ) + + IO.puts("Building climatology (min_samples=#{min_samples}, #{length(combos)} batches)...") + + total = + combos + |> Enum.with_index(1) + |> Enum.reduce(0, fn {[month, hour], idx}, acc -> + %{num_rows: count} = + Repo.query!( + """ + INSERT INTO hrrr_climatology (id, lat, lon, month, hour, + mean_surface_temp_c, stddev_surface_temp_c, sample_count) + SELECT gen_random_uuid(), lat, lon, + $2 AS month, + $3 AS hour, + AVG(surface_temp_c), + STDDEV_SAMP(surface_temp_c), + COUNT(*) + FROM hrrr_profiles + WHERE surface_temp_c IS NOT NULL + AND is_grid_point = true + AND EXTRACT(MONTH FROM valid_time)::int = $2 + AND EXTRACT(HOUR FROM valid_time)::int = $3 + GROUP BY lat, lon + HAVING COUNT(*) >= $1 + ON CONFLICT (lat, lon, month, hour) + DO UPDATE SET + mean_surface_temp_c = EXCLUDED.mean_surface_temp_c, + stddev_surface_temp_c = EXCLUDED.stddev_surface_temp_c, + sample_count = EXCLUDED.sample_count + """, + [min_samples, month, hour], + timeout: 300_000 + ) + + IO.puts(" [#{idx}/#{length(combos)}] month=#{month} hour=#{hour}: #{count} rows") + acc + count + end) + + IO.puts("Upserted #{total} climatology records total.") + end + + @doc """ + Enqueue native HRRR backfill jobs for the top-N hours by contact count. + """ + def native_backfill(limit \\ 500) do + start_app() + + alias Microwaveprop.Repo + + %{rows: rows} = + Repo.query!( + """ + SELECT EXTRACT(YEAR FROM qso_timestamp)::int AS year, + EXTRACT(MONTH FROM qso_timestamp)::int AS month, + EXTRACT(DAY FROM qso_timestamp)::int AS day, + EXTRACT(HOUR FROM qso_timestamp)::int AS hour, + COUNT(*) AS cnt + FROM contacts + WHERE pos1 IS NOT NULL + AND qso_timestamp >= '2019-01-01' + GROUP BY 1, 2, 3, 4 + ORDER BY cnt DESC + LIMIT $1 + """, + [limit], + timeout: 60_000 + ) + + IO.puts("Enqueuing #{length(rows)} native backfill jobs...") + + 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 + {:ok, _} -> IO.puts(" #{year}-#{month}-#{day} #{hour}Z (#{cnt} contacts)") + {:error, _} -> IO.puts(" #{year}-#{month}-#{day} #{hour}Z (skipped, already enqueued)") + end + end + + IO.puts("Done.") + end + + @doc """ + Compute derived fields (inversion, Ri, ducts) on native profiles that lack them. + """ + def native_derive(limit \\ 10_000) do + start_app() + + import Ecto.Query + + alias Microwaveprop.Propagation.Duct + alias Microwaveprop.Propagation.Inversion + alias Microwaveprop.Repo + alias Microwaveprop.Weather.HrrrNativeProfile + alias Microwaveprop.Weather.ThetaE + + profiles = + HrrrNativeProfile + |> where([p], is_nil(p.bulk_richardson) and p.level_count > 2) + |> limit(^limit) + |> Repo.all() + + IO.puts("Deriving fields for #{length(profiles)} profiles...") + + count = + Enum.count(profiles, fn profile -> + p = %{ + heights_m: profile.heights_m, + temp_k: profile.temp_k, + spfh: profile.spfh, + pressure_pa: profile.pressure_pa, + u_wind_ms: profile.u_wind_ms, + v_wind_ms: profile.v_wind_ms, + tke_m2s2: profile.tke_m2s2, + level_count: profile.level_count + } + + inversion_fields = + case Inversion.find_inversion_top(p) do + {:ok, %{height_m: top_h, level_idx: top_idx, base_idx: base_idx}} -> + ri = Inversion.bulk_richardson(p, base_idx, top_idx) + shear = Inversion.shear_magnitude(p, base_idx, top_idx) + theta_e_jump = ThetaE.theta_e_jump(p, base_idx, top_idx) + + [ + inversion_top_m: top_h, + bulk_richardson: ri, + shear_at_top_ms: shear, + theta_e_jump_k: theta_e_jump + ] + + :none -> + [ + inversion_top_m: nil, + bulk_richardson: nil, + shear_at_top_ms: nil, + theta_e_jump_k: nil + ] + end + + duct_result = Duct.analyze(p) + duct_fields = [ducts: duct_result.ducts, best_duct_band_ghz: duct_result.best_duct_band_ghz] + fields = inversion_fields ++ duct_fields + + {1, _} = + HrrrNativeProfile + |> where([pr], pr.id == ^profile.id) + |> Repo.update_all(set: fields) + + true + end) + + IO.puts("Updated #{count} profiles with derived fields.") + end + defp repos do Application.fetch_env!(@app, :ecto_repos) end defp load_app do - # Many platforms require SSL when connecting to the database Application.ensure_all_started(:ssl) Application.ensure_loaded(@app) end + + defp start_app do + Application.ensure_all_started(@app) + Oban.pause_all_queues(Oban) + end end