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.
99 lines
2.9 KiB
Elixir
99 lines
2.9 KiB
Elixir
defmodule Microwaveprop.Workers.NarrFetchWorker do
|
|
@moduledoc """
|
|
Fetches one NCEP NARR profile from NCEI via `NarrClient.fetch_profile_at/2`
|
|
and inserts it as a `NarrProfile` row.
|
|
|
|
Covers pre-2014 contacts where the HRRR archive doesn't reach. Replaces
|
|
the retired ERA5/CDS pipeline. NARR fetches are per-point (one analysis
|
|
hour, one location), so this worker does the fetch + insert directly —
|
|
no downstream batch worker, no routing.
|
|
|
|
Jobs are unique on `{lat, lon, valid_time}` so duplicate enrichment
|
|
requests collapse. `valid_time` MUST already be snapped to a NARR
|
|
3-hourly analysis slot (00/03/06/09/12/15/18/21 UTC on the hour) by the
|
|
caller — the worker raises `ArgumentError` otherwise via
|
|
`NarrClient.url_for/1`.
|
|
|
|
See `docs/plans/2026-04-15-merra2-historical-backfill.md` for the full
|
|
architecture.
|
|
"""
|
|
use Oban.Worker,
|
|
queue: :narr,
|
|
max_attempts: 3,
|
|
unique: [
|
|
period: :infinity,
|
|
states: [:available, :scheduled, :executing, :retryable]
|
|
]
|
|
|
|
import Ecto.Query
|
|
|
|
alias Microwaveprop.Repo
|
|
alias Microwaveprop.Weather.NarrClient
|
|
alias Microwaveprop.Weather.NarrProfile
|
|
|
|
require Logger
|
|
|
|
@impl Oban.Worker
|
|
def perform(%Oban.Job{args: %{"lat" => lat, "lon" => lon, "valid_time" => valid_time_str}}) do
|
|
{:ok, valid_time, _} = DateTime.from_iso8601(valid_time_str)
|
|
|
|
rlat = Float.round(lat * 4) / 4
|
|
rlon = Float.round(lon * 4) / 4
|
|
|
|
if has_narr_profile?(rlat, rlon, valid_time) do
|
|
:ok
|
|
else
|
|
fetch_and_insert(rlat, rlon, valid_time)
|
|
end
|
|
end
|
|
|
|
defp fetch_and_insert(rlat, rlon, valid_time) do
|
|
case NarrClient.fetch_profile_at(valid_time, {rlat, rlon}) do
|
|
{:ok, attrs} ->
|
|
_ = insert_profile(attrs, rlat, rlon, valid_time)
|
|
:ok
|
|
|
|
{:error, reason} = error ->
|
|
Logger.warning(
|
|
"NARR: fetch_profile_at failed for #{rlat},#{rlon} @ #{DateTime.to_iso8601(valid_time)}: #{inspect(reason)}"
|
|
)
|
|
|
|
error
|
|
end
|
|
end
|
|
|
|
defp insert_profile(attrs, rlat, rlon, valid_time) do
|
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
|
|
|
row =
|
|
attrs
|
|
|> Map.put(:lat, rlat)
|
|
|> Map.put(:lon, rlon)
|
|
|> Map.put(:valid_time, valid_time)
|
|
|> Map.put(:inserted_at, now)
|
|
|> Map.put(:updated_at, now)
|
|
|
|
Repo.insert_all(NarrProfile, [row],
|
|
on_conflict: :nothing,
|
|
conflict_target: [:lat, :lon, :valid_time]
|
|
)
|
|
end
|
|
|
|
defp has_narr_profile?(lat, lon, valid_time) do
|
|
# NARR is exact 3-hourly, so the time window can be tight (±15 min).
|
|
# Lat/lon tolerance matches the 0.25° dedup granularity.
|
|
dlat = 0.13
|
|
dlon = 0.13
|
|
time_start = DateTime.add(valid_time, -900, :second)
|
|
time_end = DateTime.add(valid_time, 900, :second)
|
|
|
|
NarrProfile
|
|
|> where(
|
|
[p],
|
|
p.lat >= ^(lat - dlat) and p.lat <= ^(lat + dlat) and
|
|
p.lon >= ^(lon - dlon) and p.lon <= ^(lon + dlon) and
|
|
p.valid_time >= ^time_start and p.valid_time <= ^time_end
|
|
)
|
|
|> Repo.exists?()
|
|
end
|
|
end
|