Graham McIntire
ddf86ed806
Enqueues ERA5 fetch jobs for contacts where HRRR is unavailable. Deduplicates by rounded grid point and hour, skips existing profiles. Usage: mix era5_backfill [limit]
74 lines
2 KiB
Elixir
74 lines
2 KiB
Elixir
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
|