Graham McIntire
65e3159a85
Mix tasks that call app.start were also booting Oban's cron scheduler, causing PropagationGridWorker and other cron jobs to fire during backfills. Add Oban.pause_all_queues(Oban) immediately after app.start in every mix task that only needs Repo access.
82 lines
2.4 KiB
Elixir
82 lines
2.4 KiB
Elixir
defmodule Mix.Tasks.Era5Backfill do
|
|
@shortdoc "Enqueue ERA5 month-batch jobs for contacts without HRRR data"
|
|
@moduledoc """
|
|
Finds contacts where HRRR status is `:unavailable` (pre-2014 or missing
|
|
from the archive) and enqueues the month-tile batch worker for every
|
|
tile-month their path points need.
|
|
|
|
Usage:
|
|
mix era5_backfill # scan up to 1000 contacts
|
|
mix era5_backfill 5000 # scan up to 5000 contacts
|
|
|
|
Enqueues `Era5MonthBatchWorker` jobs deduplicated by
|
|
`(year, month, tile_lat, tile_lon)`. The per-point `Era5FetchWorker`
|
|
path still works for one-off enrichment, but this task goes straight
|
|
to the batch worker so backfills don't pay the per-point enqueue cost.
|
|
"""
|
|
use Mix.Task
|
|
|
|
import Ecto.Query
|
|
|
|
alias Microwaveprop.Radio
|
|
alias Microwaveprop.Radio.Contact
|
|
alias Microwaveprop.Repo
|
|
alias Microwaveprop.Weather.Era5BatchClient
|
|
alias Microwaveprop.Workers.Era5MonthBatchWorker
|
|
|
|
@impl Mix.Task
|
|
def run(args) do
|
|
Mix.Task.run("app.start")
|
|
Oban.pause_all_queues(Oban)
|
|
|
|
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")
|
|
|
|
batches =
|
|
contacts
|
|
|> Enum.flat_map(fn contact ->
|
|
valid_time = %{DateTime.truncate(contact.qso_timestamp, :second) | minute: 0, second: 0}
|
|
|
|
contact
|
|
|> Radio.contact_path_points()
|
|
|> Enum.map(fn {lat, lon} ->
|
|
{tile_lat, tile_lon} = Era5BatchClient.tile_for_point(lat, lon)
|
|
|
|
%{
|
|
year: valid_time.year,
|
|
month: valid_time.month,
|
|
tile_lat: tile_lat,
|
|
tile_lon: tile_lon
|
|
}
|
|
end)
|
|
end)
|
|
|> Enum.uniq()
|
|
|
|
if batches == [] do
|
|
Mix.shell().info("No ERA5 batches to enqueue")
|
|
else
|
|
# Use Oban.insert/1 (not insert_all) so the worker's unique
|
|
# constraint collapses duplicate tile-month jobs against anything
|
|
# already queued.
|
|
Enum.each(batches, fn args ->
|
|
args
|
|
|> Era5MonthBatchWorker.new()
|
|
|> Oban.insert()
|
|
end)
|
|
|
|
Mix.shell().info("Enqueued #{length(batches)} ERA5 month-tile batches")
|
|
end
|
|
end
|
|
end
|