diff --git a/.gitignore b/.gitignore index 91caa6ca..20c58fac 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,9 @@ # The directory Mix will write compiled artifacts to. /_build/ +# Environment variables (secrets) +.envrc + # If you run "mix test --cover", coverage assets end up here. /cover/ diff --git a/lib/mix/tasks/era5_backfill.ex b/lib/mix/tasks/era5_backfill.ex new file mode 100644 index 00000000..1151f5f7 --- /dev/null +++ b/lib/mix/tasks/era5_backfill.ex @@ -0,0 +1,74 @@ +defmodule Mix.Tasks.Era5Backfill do + @shortdoc "Enqueue ERA5 fetch jobs for contacts without HRRR data" + @moduledoc """ + Finds contacts where HRRR status is 'unavailable' (pre-2014 or missing from archive) + and enqueues ERA5 fetch jobs for their path points. + + Usage: + mix era5_backfill # enqueue all (default limit 1000) + mix era5_backfill 5000 # enqueue up to 5000 + """ + use Mix.Task + + import Ecto.Query + + alias Microwaveprop.Radio + alias Microwaveprop.Radio.Contact + alias Microwaveprop.Repo + alias Microwaveprop.Weather + alias Microwaveprop.Workers.Era5FetchWorker + + @impl Mix.Task + def run(args) do + Mix.Task.run("app.start") + + limit = + case args do + [n | _] -> String.to_integer(n) + _ -> 1000 + end + + contacts = + Contact + |> where([c], c.hrrr_status == :unavailable and not is_nil(c.pos1)) + |> order_by([c], asc: c.qso_timestamp) + |> limit(^limit) + |> Repo.all() + + Mix.shell().info("Found #{length(contacts)} contacts needing ERA5 data") + + jobs = + contacts + |> Enum.flat_map(fn contact -> + contact + |> Radio.contact_path_points() + |> Enum.map(fn {lat, lon} -> + rlat = Float.round(lat * 4) / 4 + rlon = Float.round(lon * 4) / 4 + valid_time = %{DateTime.truncate(contact.qso_timestamp, :second) | minute: 0, second: 0} + + if Weather.find_nearest_era5(rlat, rlon, valid_time) do + nil + else + Era5FetchWorker.new(%{ + "lat" => rlat, + "lon" => rlon, + "valid_time" => DateTime.to_iso8601(valid_time) + }) + end + end) + |> Enum.reject(&is_nil/1) + end) + |> Enum.uniq_by(fn changeset -> changeset.changes.args end) + + if jobs == [] do + Mix.shell().info("No ERA5 jobs needed (all data already exists)") + else + jobs + |> Enum.chunk_every(400) + |> Enum.each(&Oban.insert_all/1) + + Mix.shell().info("Enqueued #{length(jobs)} ERA5 fetch jobs") + end + end +end