diff --git a/lib/microwaveprop/weather.ex b/lib/microwaveprop/weather.ex index f4249777..e00ea761 100644 --- a/lib/microwaveprop/weather.ex +++ b/lib/microwaveprop/weather.ex @@ -260,6 +260,45 @@ defmodule Microwaveprop.Weather do |> Repo.exists?() end + @doc """ + True if the station already has at least one surface observation + anywhere within the given UTC date. Used by the `asos_day` worker + to short-circuit the IEM fetch when the day has already been + ingested by a prior run. + """ + @spec station_day_covered?(Ecto.UUID.t(), Date.t()) :: boolean() + def station_day_covered?(station_id, date) do + {:ok, start_dt} = DateTime.new(date, ~T[00:00:00], "Etc/UTC") + {:ok, end_dt} = DateTime.new(date, ~T[23:59:59], "Etc/UTC") + has_surface_observations?(station_id, start_dt, end_dt) + end + + @doc """ + Returns the MapSet of `{station_id, date}` tuples already covered + (at least one obs in that UTC day). Used by the enqueuer to avoid + emitting jobs for combinations we've already fetched. + """ + @spec station_day_pairs_covered([{Ecto.UUID.t(), Date.t()}]) :: + MapSet.t({Ecto.UUID.t(), Date.t()}) + def station_day_pairs_covered([]), do: MapSet.new() + + def station_day_pairs_covered(pairs) do + station_ids = pairs |> Enum.map(&elem(&1, 0)) |> Enum.uniq() + dates = pairs |> Enum.map(&elem(&1, 1)) |> Enum.uniq() + {:ok, start_dt} = DateTime.new(Enum.min(dates, Date), ~T[00:00:00], "Etc/UTC") + {:ok, end_dt} = DateTime.new(Enum.max(dates, Date), ~T[23:59:59], "Etc/UTC") + + rows = + SurfaceObservation + |> where([o], o.station_id in ^station_ids) + |> where([o], o.observed_at >= ^start_dt and o.observed_at <= ^end_dt) + |> select([o], {o.station_id, fragment("(?::date)", o.observed_at)}) + |> distinct(true) + |> Repo.all() + + MapSet.new(rows) + end + @doc "Returns a MapSet of station_ids that have surface observations in the time window." @spec station_ids_with_surface_observations([Ecto.UUID.t()], DateTime.t(), DateTime.t()) :: MapSet.t(Ecto.UUID.t()) def station_ids_with_surface_observations(station_ids, start_dt, end_dt) do diff --git a/lib/microwaveprop/weather/rebatch_asos.ex b/lib/microwaveprop/weather/rebatch_asos.ex index 2793587f..714c974c 100644 --- a/lib/microwaveprop/weather/rebatch_asos.ex +++ b/lib/microwaveprop/weather/rebatch_asos.ex @@ -42,6 +42,162 @@ defmodule Microwaveprop.Weather.RebatchAsos do end end + @doc """ + Converts pending `asos_batch` and `asos` jobs into the newer + `asos_day` shape: one job per unique `(station_id, station_code, date)` + tuple, where `date` enumerates every UTC date the source window + touched. Cross-contact dedup is automatic via Oban's unique-args. + + Expected to shrink the pending queue substantially because many + QSOs reference the same (station, date) and currently have their + own 4h-window job. Idempotent; dry-run supported. + """ + @spec to_day_jobs(keyword()) :: %{ + source_jobs: non_neg_integer(), + unique_day_keys: non_neg_integer(), + inserted: non_neg_integer(), + cancelled: non_neg_integer() + } + def to_day_jobs(opts \\ []) do + dry_run? = Keyword.get(opts, :dry_run, false) + jobs = load_pending_window_jobs() + + if jobs == [] do + IO.puts("No pending `asos`/`asos_batch` jobs. Nothing to do.") + %{source_jobs: 0, unique_day_keys: 0, inserted: 0, cancelled: 0} + else + do_to_day(jobs, dry_run?) + end + end + + defp load_pending_window_jobs do + query = + from(j in Oban.Job, + where: + j.worker == "Microwaveprop.Workers.WeatherFetchWorker" and + j.state in ["available", "scheduled", "retryable"] and + fragment("?->>'fetch_type' IN ('asos', 'asos_batch')", j.args), + select: %{id: j.id, args: j.args} + ) + + Repo.all(query) + end + + defp do_to_day(jobs, dry_run?) do + # Expand each source job into a list of (station_id, station_code, date) + # tuples. asos_batch → one tuple per (station_code, date-in-window); + # asos → one tuple per (single station_code, date-in-window). + expanded = + Enum.flat_map(jobs, fn %{id: id, args: a} -> + a + |> expand_tuples() + |> Enum.map(fn {sid, code, date} -> {sid, code, date, id} end) + end) + + # Group by the unique dedup key (station_id, date) and keep one + # station_code per group (always the same since station_id is the + # key). Track source_ids to cancel afterwards. + by_key = + Enum.group_by(expanded, fn {sid, _code, date, _id} -> {sid, date} end) + + IO.puts("Expanded #{length(jobs)} source jobs into #{map_size(by_key)} unique (station, date) keys.") + + inserts = + Enum.map(by_key, fn {{sid, date}, members} -> + {_, code, _, _} = hd(members) + source_ids = members |> Enum.map(fn {_, _, _, id} -> id end) |> Enum.uniq() + + %{ + "fetch_type" => "asos_day", + "station_id" => sid, + "station_code" => code, + "date" => Date.to_iso8601(date), + "_source_ids" => source_ids + } + end) + + if dry_run? do + Enum.each(Enum.take(inserts, 20), fn a -> + IO.puts("[dry-run] #{a["station_code"]} #{a["date"]} (collapses #{length(a["_source_ids"])} sources)") + end) + + %{source_jobs: length(jobs), unique_day_keys: length(inserts), inserted: 0, cancelled: 0} + else + apply_day_conversion(inserts, length(jobs)) + end + end + + defp expand_tuples(%{"fetch_type" => "asos_batch"} = a) do + case window_dates(a["start_dt"], a["end_dt"]) do + [] -> + [] + + dates -> + ids = a["station_ids"] || [] + codes = a["station_codes"] || [] + + for {sid, code} <- Enum.zip(ids, codes), date <- dates do + {sid, code, date} + end + end + end + + defp expand_tuples(%{"fetch_type" => "asos"} = a) do + case window_dates(a["start_dt"], a["end_dt"]) do + [] -> [] + dates -> Enum.map(dates, fn date -> {a["station_id"], a["station_code"], date} end) + end + end + + defp window_dates(nil, _), do: [] + defp window_dates(_, nil), do: [] + + defp window_dates(start_str, end_str) do + with {:ok, start_dt, _} <- DateTime.from_iso8601(start_str), + {:ok, end_dt, _} <- DateTime.from_iso8601(end_str) do + start_date = DateTime.to_date(start_dt) + end_date = DateTime.to_date(end_dt) + + if Date.compare(start_date, end_date) == :eq do + [start_date] + else + [start_date, end_date] + end + else + _ -> [] + end + end + + defp apply_day_conversion(inserts, source_jobs) do + {inserted, cancelled} = + Enum.reduce(inserts, {0, 0}, fn a, {ins, can} -> + source_ids = a["_source_ids"] + job_args = Map.delete(a, "_source_ids") + + Repo.transaction(fn -> + # Unique-args dedup means duplicate inserts are silently no-ops. + _ = job_args |> WeatherFetchWorker.new() |> Oban.insert!() + end) + + {ins + 1, can + length(source_ids)} + end) + + # Cancel the source jobs in bulk after inserts succeed. + {:ok, cancelled_n} = + Oban.cancel_all_jobs( + from(j in Oban.Job, + where: + j.worker == "Microwaveprop.Workers.WeatherFetchWorker" and + j.state in ["available", "scheduled", "retryable"] and + fragment("?->>'fetch_type' IN ('asos', 'asos_batch')", j.args) + ) + ) + + IO.puts("Converted: inserted #{inserted} asos_day jobs, cancelled #{cancelled_n} source jobs.") + + %{source_jobs: source_jobs, unique_day_keys: length(inserts), inserted: inserted, cancelled: cancelled} + end + defp load_pending_asos_jobs do query = from(j in Oban.Job, diff --git a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex index 5b6f5e1d..dec54d3d 100644 --- a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex +++ b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex @@ -389,34 +389,45 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do end_dt = DateTime.add(timestamp, 2 * 3600, :second) stations = Weather.nearby_stations(lat, lon, "asos", @asos_radius_km) - station_ids = Enum.map(stations, & &1.id) - covered = Weather.station_ids_with_surface_observations(station_ids, start_dt, end_dt) - # Group every uncovered station for this (lat, lon, window) into one - # batch job. IEM's ASOS CSV endpoint accepts multiple station= params - # in a single request and we group the result rows by station_code in - # the worker. Collapsing ~N→1 requests per endpoint was the largest - # lever against the IemRateLimiter gap + 429 retry tail. - uncovered = - stations - |> Enum.reject(fn s -> MapSet.member?(covered, s.id) end) - # Sort for deterministic args → Oban unique dedup works across enqueues. - |> Enum.sort_by(& &1.station_code) + # A ±2h window around the QSO can straddle UTC midnight, so enumerate + # every date the window touches. Each (station, date) becomes one + # asos_day job — Oban's unique-args dedup collapses cross-QSO + # repeats (contact A's KDFW/2021-08-21 and contact B's KDFW/2021-08-21 + # share args → one job). A full-day fetch returns ~24 hourly rows, + # amortizing the IemRateLimiter gap over 24× the previous payload. + dates = window_dates(start_dt, end_dt) - case uncovered do - [] -> - [] + pairs = + for station <- stations, date <- dates do + {station, date} + end - stations -> - [ - WeatherFetchWorker.new(%{ - "fetch_type" => "asos_batch", - "station_ids" => Enum.map(stations, & &1.id), - "station_codes" => Enum.map(stations, & &1.station_code), - "start_dt" => DateTime.to_iso8601(start_dt), - "end_dt" => DateTime.to_iso8601(end_dt) - }) - ] + covered = + pairs + |> Enum.map(fn {s, d} -> {s.id, d} end) + |> Weather.station_day_pairs_covered() + + pairs + |> Enum.reject(fn {s, d} -> MapSet.member?(covered, {s.id, d}) end) + |> Enum.map(fn {station, date} -> + WeatherFetchWorker.new(%{ + "fetch_type" => "asos_day", + "station_id" => station.id, + "station_code" => station.station_code, + "date" => Date.to_iso8601(date) + }) + end) + end + + defp window_dates(start_dt, end_dt) do + start_date = DateTime.to_date(start_dt) + end_date = DateTime.to_date(end_dt) + + if Date.compare(start_date, end_date) == :eq do + [start_date] + else + [start_date, end_date] end end diff --git a/lib/microwaveprop/workers/weather_fetch_worker.ex b/lib/microwaveprop/workers/weather_fetch_worker.ex index 28bd1b14..af42f1d7 100644 --- a/lib/microwaveprop/workers/weather_fetch_worker.ex +++ b/lib/microwaveprop/workers/weather_fetch_worker.ex @@ -41,6 +41,30 @@ defmodule Microwaveprop.Workers.WeatherFetchWorker do end end + def perform(%Oban.Job{args: %{"fetch_type" => "asos_day"} = args}) do + %{ + "station_id" => station_id, + "station_code" => station_code, + "date" => date_str + } = args + + {:ok, date} = Date.from_iso8601(date_str) + + cond do + is_nil(Repo.get(Station, station_id)) -> + Logger.warning("WeatherFetch ASOS day: station #{station_id} (#{station_code}) missing, skipping") + :ok + + Weather.station_day_covered?(station_id, date) -> + Logger.debug("WeatherFetch ASOS day: #{station_code} #{date_str} already covered, skipping") + :ok + + true -> + station = Repo.get!(Station, station_id) + fetch_asos_day(station, station_id, station_code, date) + end + end + def perform(%Oban.Job{args: %{"fetch_type" => "asos_batch"} = args}) do %{ "station_ids" => station_ids, @@ -155,6 +179,26 @@ defmodule Microwaveprop.Workers.WeatherFetchWorker do :ok end + # Fetch a full UTC day of ASOS obs for one station and upsert. + # Amortizes the per-request IemRateLimiter gap + 429 retry tail + # across ~24 hourly observations, vs ~1-2 for the previous 4h + # windows. Oban unique-args dedup means cross-QSO collisions on + # the same (station, date) never re-enqueue. + defp fetch_asos_day(station, station_id, station_code, date) do + {:ok, start_dt} = DateTime.new(date, ~T[00:00:00], "Etc/UTC") + {:ok, end_dt} = DateTime.new(date, ~T[23:59:59], "Etc/UTC") + + Logger.info("WeatherFetch ASOS day: fetching #{station_code} for #{date}") + + case IemClient.fetch_asos(station_code, start_dt, end_dt) do + {:ok, rows} -> + ingest_asos_rows(station, station_id, station_code, rows, start_dt, end_dt) + + {:error, reason} -> + handle_iem_error("ASOS day", station_code, reason) + end + end + defp store_asos_stub(station, start_dt, end_dt) do midpoint = DateTime.add(start_dt, div(DateTime.diff(end_dt, start_dt), 2)) diff --git a/test/microwaveprop/workers/contact_weather_enqueue_worker_test.exs b/test/microwaveprop/workers/contact_weather_enqueue_worker_test.exs index e86db057..08079cf4 100644 --- a/test/microwaveprop/workers/contact_weather_enqueue_worker_test.exs +++ b/test/microwaveprop/workers/contact_weather_enqueue_worker_test.exs @@ -61,51 +61,67 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorkerTest do station end - describe "build_weather_jobs/1" do - test "builds ASOS batch jobs grouping nearby stations per time window" do - station = create_asos_station("KDFW", 32.90, -97.04) + describe "build_weather_jobs/1 — asos_day" do + test "emits one asos_day job per (station_id, date) pair" do + s = create_asos_station("KDFW", 32.90, -97.04) contact = create_contact() jobs = ContactWeatherEnqueueWorker.build_weather_jobs([contact]) - asos_jobs = + day_jobs = Enum.filter(jobs, fn j -> - j.changes.args["fetch_type"] == "asos_batch" + j.changes.args["fetch_type"] == "asos_day" end) - assert asos_jobs != [] - - # Every uncovered station near this endpoint rides in one batch job, - # not one job per station. - job = Enum.find(asos_jobs, &(station.id in &1.changes.args["station_ids"])) - assert job - assert "KDFW" in job.changes.args["station_codes"] - assert job.changes.args["start_dt"] - assert job.changes.args["end_dt"] + assert day_jobs != [] + job = Enum.find(day_jobs, &(&1.changes.args["station_id"] == s.id)) + assert job.changes.args["station_code"] == "KDFW" + assert job.changes.args["date"] == "2026-03-28" end - test "builds several batch jobs for nearby stations — one per endpoint window" do + test "dedups identical (station, date) across multiple contacts" do create_asos_station("KDFW", 32.90, -97.04) - create_asos_station("KFTW", 32.82, -97.36) - create_asos_station("KAFW", 32.98, -97.32) - contact = create_contact() + + # Different seconds, same UTC day, same nearby station → one job. + q1 = create_contact(%{station1: "A1", qso_timestamp: ~U[2026-03-28 14:00:00Z]}) + q2 = create_contact(%{station1: "A2", qso_timestamp: ~U[2026-03-28 18:00:00Z]}) + + jobs = ContactWeatherEnqueueWorker.build_weather_jobs([q1, q2]) + + day_jobs = + Enum.filter(jobs, fn j -> j.changes.args["fetch_type"] == "asos_day" end) + + # pos1 near KDFW → 1 day job; pos2 and midpoint have no KDFW-nearby station, + # so each endpoint contributes at most 1 day job; shared (KDFW, 2026-03-28) + # across both QSOs should collapse. + kdfw_jobs = + Enum.filter(day_jobs, fn j -> j.changes.args["station_code"] == "KDFW" end) + + assert length(kdfw_jobs) == 1 + end + + test "emits both UTC days when the window crosses midnight" do + create_asos_station("KDFW", 32.90, -97.04) + + # QSO at 23:30 UTC → ±2h window spans 21:30..01:30 next day. + contact = create_contact(%{qso_timestamp: ~U[2026-03-28 23:30:00Z]}) jobs = ContactWeatherEnqueueWorker.build_weather_jobs([contact]) - asos_jobs = - Enum.filter(jobs, fn j -> - j.changes.args["fetch_type"] == "asos_batch" + kdfw_dates = + jobs + |> Enum.filter(fn j -> + j.changes.args["fetch_type"] == "asos_day" and + j.changes.args["station_code"] == "KDFW" end) + |> Enum.map(& &1.changes.args["date"]) - # At least one batch covers pos1's cluster; all three nearby stations - # should land in the same batch (shared 4-hour window, same endpoint). - pos1_batch = Enum.find(asos_jobs, &(length(&1.changes.args["station_codes"]) == 3)) - assert pos1_batch - - assert MapSet.new(pos1_batch.changes.args["station_codes"]) == - MapSet.new(["KDFW", "KFTW", "KAFW"]) + assert "2026-03-28" in kdfw_dates + assert "2026-03-29" in kdfw_dates end + end + describe "build_weather_jobs/1" do test "builds RAOB jobs for nearby sounding stations" do station = create_sounding_station("FWD", 32.83, -97.30) contact = create_contact() @@ -125,23 +141,6 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorkerTest do assert job.changes.args["sounding_time"] end - test "deduplicates jobs with identical args" do - _station = create_asos_station("KDFW", 32.90, -97.04) - - # Two QSOs at the exact same time produce identical batch args → deduped - q1 = create_contact(%{station1: "A1", qso_timestamp: ~U[2026-03-28 18:00:00Z]}) - q2 = create_contact(%{station1: "A2", qso_timestamp: ~U[2026-03-28 18:00:00Z]}) - - jobs = ContactWeatherEnqueueWorker.build_weather_jobs([q1, q2]) - - asos_jobs = - Enum.filter(jobs, fn j -> - j.changes.args["fetch_type"] == "asos_batch" - end) - - assert length(asos_jobs) == 1 - end - test "builds ASOS jobs for stations near pos2 and midpoint" do # Station near pos2 (30.3, -97.7) but NOT near pos1 (32.9, -97.0) station_near_pos2 = create_asos_station("KAUS", 30.20, -97.68) @@ -149,10 +148,12 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorkerTest do jobs = ContactWeatherEnqueueWorker.build_weather_jobs([contact]) - all_station_ids = - Enum.flat_map(jobs, fn j -> j.changes.args["station_ids"] || [] end) + station_ids = + jobs + |> Enum.filter(&(&1.changes.args["fetch_type"] == "asos_day")) + |> Enum.map(& &1.changes.args["station_id"]) - assert station_near_pos2.id in all_station_ids + assert station_near_pos2.id in station_ids end test "returns empty list when no stations nearby" do @@ -175,7 +176,7 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorkerTest do asos_jobs = Enum.filter(jobs, fn j -> - j.changes.args["fetch_type"] == "asos_batch" + j.changes.args["fetch_type"] == "asos_day" end) assert asos_jobs == [] diff --git a/test/microwaveprop/workers/weather_fetch_worker_test.exs b/test/microwaveprop/workers/weather_fetch_worker_test.exs index 1ce653f3..e9c94416 100644 --- a/test/microwaveprop/workers/weather_fetch_worker_test.exs +++ b/test/microwaveprop/workers/weather_fetch_worker_test.exs @@ -59,6 +59,80 @@ defmodule Microwaveprop.Workers.WeatherFetchWorkerTest do } end + describe "ASOS day perform/1" do + test "fetches full UTC day for one station and upserts returned rows" do + {:ok, station} = + Weather.find_or_create_station(%{ + station_code: "KDFW", + station_type: "asos", + name: "DFW", + lat: 32.9, + lon: -97.0 + }) + + Req.Test.stub(IemClient, fn conn -> + # Assert the worker requested the whole UTC day 2021-08-21. + qs = URI.decode_query(conn.query_string) + assert qs["year1"] == "2021" + assert qs["month1"] == "8" + assert qs["day1"] == "21" + assert qs["year2"] == "2021" + assert qs["day2"] == "21" + + body = """ + station,valid,tmpf,dwpf,relh,sknt,drct,mslp,alti,skyc1,p01i,wxcodes + KDFW,2021-08-21 00:53,80.0,70.0,71.0,8,180,1012.0,29.90,CLR,null,null + KDFW,2021-08-21 12:53,85.0,68.0,58.0,10,180,1013.0,29.92,FEW,null,null + KDFW,2021-08-21 23:53,78.0,69.0,73.0,6,160,1011.0,29.89,SCT,null,null + """ + + Req.Test.text(conn, body) + end) + + args = %{ + "fetch_type" => "asos_day", + "station_id" => station.id, + "station_code" => "KDFW", + "date" => "2021-08-21" + } + + assert :ok = WeatherFetchWorker.perform(%Oban.Job{args: args}) + assert Repo.aggregate(SurfaceObservation, :count) == 3 + end + + test "skips fetch when day is already covered" do + {:ok, station} = + Weather.find_or_create_station(%{ + station_code: "KDFW", + station_type: "asos", + name: "DFW", + lat: 32.9, + lon: -97.0 + }) + + # Seed a prior observation inside the target day. + Weather.upsert_surface_observation(station, %{ + observed_at: ~U[2021-08-21 12:00:00Z], + temp_f: 85.0 + }) + + Req.Test.stub(IemClient, fn conn -> + Plug.Conn.send_resp(conn, 500, "should not be called") + end) + + args = %{ + "fetch_type" => "asos_day", + "station_id" => station.id, + "station_code" => "KDFW", + "date" => "2021-08-21" + } + + assert :ok = WeatherFetchWorker.perform(%Oban.Job{args: args}) + # Still only the seeded row — no fetch happened. + assert Repo.aggregate(SurfaceObservation, :count) == 1 + end + end + describe "ASOS batch perform/1" do test "fetches one URL covering all stations and ingests per-station rows" do {:ok, s1} =