Graham McIntire
d61fbd346e
Enabled :error_handling, :unknown, :unmatched_returns, :extra_return, :missing_return in an earlier commit and landed a 129-warning baseline. Four parallel agents each fixed a directory slice: - Core contexts (29): Radio, Release, Weather, Beacons, Cache, Backtest.Features, Terrain.Srtm, Ionosphere.GiroClient, Propagation.RunTiming, Accounts.Scope, RepoListener. Fixes were (a) prefix side-effect calls (Task.start, Phoenix.PubSub, Logger, :ets.new) with _ = ; (b) tighten/widen specs that didn't match actual returns; (c) add missing @type t declarations; (d) drop dead parse_int(nil) clause. - Propagation + weather subdirs (15): FreshnessMonitor, NotifyListener, ScoreCache, ScoreCacheReconciler, Weather.FrontalAnalysis, Weather.Grib2.Extractor, Weather.Grib2.Wgrib2, GridCache, HrrrPointEnqueuer, NexradCache. Same patterns — mostly _ = on PubSub / :ets / Repo.insert_all; widened two specs (float -> number) where integer returns were reachable. - Workers (35): BackfillEnqueue, CanadianSoundingFetch, ContactImport, ContactWeatherEnqueue, GefsFetch, IemreFetch, NarrFetch, SolarIndex, TerrainProfile, WeatherFetch. Prefixed Repo.update_all / Radio.set_enrichment_status! / Weather.upsert_* side-effect calls. Fixed one :pattern_match in CanadianSoundingFetch.most_recent_sounding_time/1 where a tautological cond guard generated unreachable code. - Web + Mix tasks + lib_ml (46 of 50): controllers, LiveViews, UserAuth, and 11 mix tasks. Same prefix strategy. 4 remaining warnings originate in LiveTable.LiveResource dep macro expansion and can't be fixed without forking the dep — added .dialyzer_ignore.exs to suppress just those specific file:line pairs. Also wired ignore_warnings in mix.exs dialyzer config. mix dialyzer --format short | grep ^lib/ | wc -l -> 0 mix test: 2163 tests, 3 pre-existing flakes, 0 regressions.
101 lines
2.8 KiB
Elixir
101 lines
2.8 KiB
Elixir
defmodule Mix.Tasks.RadarBackfill do
|
|
@shortdoc "Enqueue CommonVolumeRadarWorker for contacts with radar_status :pending"
|
|
@moduledoc """
|
|
Bulk-enqueue per-contact common-volume radar enrichment for historical
|
|
contacts.
|
|
|
|
Eligible contacts: post-2014-10-02 (outside NARR coverage), both pos1
|
|
and pos2 present, `radar_status = :pending`. The worker's own unique
|
|
constraint on `contact_id` makes re-runs idempotent.
|
|
|
|
Options:
|
|
--limit N Maximum contacts to enqueue per run (default: all)
|
|
--year N Only enqueue contacts from this year (optional filter)
|
|
--dry-run Print counts; don't enqueue
|
|
|
|
Example:
|
|
|
|
mix radar_backfill --year 2024 --limit 5000
|
|
"""
|
|
use Mix.Task
|
|
|
|
import Ecto.Query
|
|
|
|
alias Microwaveprop.Radio
|
|
alias Microwaveprop.Radio.Contact
|
|
alias Microwaveprop.Repo
|
|
alias Microwaveprop.Workers.CommonVolumeRadarWorker
|
|
|
|
@narr_cutoff ~U[2014-10-02 00:00:00Z]
|
|
@chunk 400
|
|
|
|
@impl Mix.Task
|
|
def run(argv) do
|
|
Mix.Task.run("app.start")
|
|
|
|
{opts, _, _} =
|
|
OptionParser.parse(argv,
|
|
switches: [limit: :integer, year: :integer, dry_run: :boolean]
|
|
)
|
|
|
|
limit = Keyword.get(opts, :limit)
|
|
year = Keyword.get(opts, :year)
|
|
dry_run? = Keyword.get(opts, :dry_run, false)
|
|
|
|
total = count_eligible(year)
|
|
Mix.shell().info("Eligible contacts: #{total}#{if year, do: " (year #{year})", else: ""}")
|
|
|
|
if dry_run? or total == 0 do
|
|
:ok
|
|
else
|
|
ids = fetch_ids(year, limit)
|
|
|
|
Mix.shell().info("Enqueueing #{length(ids)} CommonVolumeRadarWorker jobs...")
|
|
|
|
ids
|
|
|> Enum.chunk_every(@chunk)
|
|
|> Enum.with_index(1)
|
|
|> Enum.each(fn {batch, i} ->
|
|
jobs = Enum.map(batch, &CommonVolumeRadarWorker.new(%{"contact_id" => &1}))
|
|
_ = Oban.insert_all(jobs)
|
|
_ = Radio.set_enrichment_status!(batch, :radar_status, :queued)
|
|
Mix.shell().info(" batch #{i}: #{length(batch)} enqueued")
|
|
end)
|
|
|
|
Mix.shell().info("Done.")
|
|
end
|
|
end
|
|
|
|
defp count_eligible(year) do
|
|
Contact
|
|
|> eligible_query(year)
|
|
|> Repo.aggregate(:count, :id)
|
|
end
|
|
|
|
defp fetch_ids(year, limit) do
|
|
query =
|
|
Contact
|
|
|> eligible_query(year)
|
|
|> order_by(asc: :qso_timestamp)
|
|
|> select([c], c.id)
|
|
|
|
query = if limit, do: limit(query, ^limit), else: query
|
|
Repo.all(query)
|
|
end
|
|
|
|
defp eligible_query(query, year) do
|
|
query
|
|
|> where([c], not is_nil(c.pos1) and not is_nil(c.pos2))
|
|
|> where([c], c.qso_timestamp >= ^@narr_cutoff)
|
|
|> where([c], c.radar_status == :pending)
|
|
|> maybe_filter_year(year)
|
|
end
|
|
|
|
defp maybe_filter_year(query, nil), do: query
|
|
|
|
defp maybe_filter_year(query, year) do
|
|
start_dt = DateTime.new!(Date.new!(year, 1, 1), ~T[00:00:00], "Etc/UTC")
|
|
end_dt = DateTime.new!(Date.new!(year, 12, 31), ~T[23:59:59], "Etc/UTC")
|
|
where(query, [c], c.qso_timestamp >= ^start_dt and c.qso_timestamp <= ^end_dt)
|
|
end
|
|
end
|