Add mix era5_backfill task and gitignore .envrc
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]
This commit is contained in:
parent
7df636147a
commit
ddf86ed806
2 changed files with 77 additions and 0 deletions
3
.gitignore
vendored
3
.gitignore
vendored
|
|
@ -1,6 +1,9 @@
|
||||||
# The directory Mix will write compiled artifacts to.
|
# The directory Mix will write compiled artifacts to.
|
||||||
/_build/
|
/_build/
|
||||||
|
|
||||||
|
# Environment variables (secrets)
|
||||||
|
.envrc
|
||||||
|
|
||||||
# If you run "mix test --cover", coverage assets end up here.
|
# If you run "mix test --cover", coverage assets end up here.
|
||||||
/cover/
|
/cover/
|
||||||
|
|
||||||
|
|
|
||||||
74
lib/mix/tasks/era5_backfill.ex
Normal file
74
lib/mix/tasks/era5_backfill.ex
Normal file
|
|
@ -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
|
||||||
Loading…
Add table
Reference in a new issue