From ef9aac60c066390aa4e7ede658c54d4e401c03e5 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sun, 29 Mar 2026 13:29:29 -0500 Subject: [PATCH] Move all weather fetching to Oban, import script only enqueues jobs Instead of fetching inline with Task.async_stream (which hammered IEM with concurrent requests), the import script now just checks what data is missing and enqueues WeatherFetchWorker jobs. Oban handles all HTTP with concurrency 3 and automatic backoff on failures. --- priv/repo/import_weather.exs | 215 +++++++++++++---------------------- 1 file changed, 77 insertions(+), 138 deletions(-) diff --git a/priv/repo/import_weather.exs b/priv/repo/import_weather.exs index a6f87cc2..9c91b9c9 100644 --- a/priv/repo/import_weather.exs +++ b/priv/repo/import_weather.exs @@ -4,7 +4,6 @@ alias Microwaveprop.Repo alias Microwaveprop.Weather alias Microwaveprop.Weather.IemClient alias Microwaveprop.Weather.SolarClient -alias Microwaveprop.Weather.SoundingParams alias Microwaveprop.Workers.WeatherFetchWorker # --- Helpers --- @@ -173,9 +172,9 @@ sounding_station_list = |> Enum.map(fn s -> Map.put(s, :station_type, "sounding") end) |> Enum.uniq_by(& &1.station_code) -# --- Phase 2: Fetch weather for each QSO --- +# --- Phase 2: Enqueue weather fetches for each QSO --- -IO.puts("\n=== Phase 2: Importing weather data ===") +IO.puts("\n=== Phase 2: Enqueueing weather fetch jobs ===") qsos = Microwaveprop.Radio.Qso @@ -185,154 +184,88 @@ qsos = IO.puts("Processing #{length(qsos)} QSOs with positions...") -# Track which station+time combos we've already fetched to avoid duplicates -asos_fetched = :ets.new(:asos_fetched, [:set, :public]) -raob_fetched = :ets.new(:raob_fetched, [:set, :public]) +# Track which station+time combos we've already enqueued to avoid duplicates +asos_seen = :ets.new(:asos_seen, [:set, :public]) +raob_seen = :ets.new(:raob_seen, [:set, :public]) total = length(qsos) -progress = :counters.new(1, [:atomics]) +asos_enqueued = :counters.new(1, [:atomics]) +raob_enqueued = :counters.new(1, [:atomics]) +skipped = :counters.new(1, [:atomics]) -qsos -|> Task.async_stream( - fn qso -> - lat = qso.pos1["lat"] || qso.pos1["latitude"] - lon = qso.pos1["lng"] || qso.pos1["lon"] || qso.pos1["longitude"] +Enum.with_index(qsos, 1) +|> Enum.each(fn {qso, idx} -> + if rem(idx, 500) == 0, do: IO.puts(" Scanned #{idx}/#{total} QSOs") - if lat && lon do - :counters.add(progress, 1, 1) - val = :counters.get(progress, 1) - if rem(val, 500) == 0, do: IO.puts(" Processed #{val}/#{total} QSOs") + lat = qso.pos1["lat"] || qso.pos1["latitude"] + lon = qso.pos1["lng"] || qso.pos1["lon"] || qso.pos1["longitude"] - # Find nearby ASOS stations (within 150km) - nearby_asos = ImportHelpers.nearest_stations(lat, lon, asos_station_list, 150) + if lat && lon do + # Find nearby ASOS stations (within 150km) + nearby_asos = ImportHelpers.nearest_stations(lat, lon, asos_station_list, 150) - Enum.each(nearby_asos, fn s -> - # Fetch a 4-hour window around QSO time - start_dt = DateTime.add(qso.qso_timestamp, -2 * 3600, :second) - end_dt = DateTime.add(qso.qso_timestamp, 2 * 3600, :second) - key = {s.station_code, DateTime.to_date(start_dt), div(start_dt.hour, 4)} + Enum.each(nearby_asos, fn s -> + start_dt = DateTime.add(qso.qso_timestamp, -2 * 3600, :second) + end_dt = DateTime.add(qso.qso_timestamp, 2 * 3600, :second) + key = {s.station_code, DateTime.to_date(start_dt), div(start_dt.hour, 4)} - if !:ets.member(asos_fetched, key) do - :ets.insert(asos_fetched, {key, true}) + if !:ets.member(asos_seen, key) do + :ets.insert(asos_seen, {key, true}) + station = Map.get(asos_station_map, s.station_code) - station = Map.get(asos_station_map, s.station_code) + if station && !Weather.has_surface_observations?(station.id, start_dt, end_dt) do + %{ + "fetch_type" => "asos", + "station_id" => station.id, + "station_code" => s.station_code, + "start_dt" => DateTime.to_iso8601(start_dt), + "end_dt" => DateTime.to_iso8601(end_dt) + } + |> WeatherFetchWorker.new() + |> Oban.insert() - if station do - # Skip HTTP fetch if we already have observations in this window - if !Weather.has_surface_observations?(station.id, start_dt, end_dt) do - case IemClient.fetch_asos(s.station_code, start_dt, end_dt) do - {:ok, rows} -> - Enum.each(rows, fn row -> - if row.observed_at do - Weather.upsert_surface_observation(station, row) - end - end) + :counters.add(asos_enqueued, 1, 1) + else + :counters.add(skipped, 1, 1) + end + end + end) - {:error, reason} -> - jitter = :rand.uniform(3600) + # Find nearby sounding stations (within 300km) + nearby_soundings = ImportHelpers.nearest_stations(lat, lon, sounding_station_list, 300) - IO.puts( - " ASOS error #{s.station_code}: #{inspect(reason)}, enqueueing retry" - ) + Enum.each(nearby_soundings, fn s -> + sounding_times = ImportHelpers.sounding_times_around(qso.qso_timestamp) - %{ - "fetch_type" => "asos", - "station_id" => station.id, - "station_code" => s.station_code, - "start_dt" => DateTime.to_iso8601(start_dt), - "end_dt" => DateTime.to_iso8601(end_dt) - } - |> WeatherFetchWorker.new(schedule_in: 7200 + jitter) - |> Oban.insert() - end - end + Enum.each(sounding_times, fn sounding_time -> + key = {s.station_code, sounding_time} + + if !:ets.member(raob_seen, key) do + :ets.insert(raob_seen, {key, true}) + station = Map.get(sounding_station_map, s.station_code) + + if station && !Weather.has_sounding?(station.id, sounding_time) do + %{ + "fetch_type" => "raob", + "station_id" => station.id, + "station_code" => s.station_code, + "sounding_time" => DateTime.to_iso8601(sounding_time) + } + |> WeatherFetchWorker.new() + |> Oban.insert() + + :counters.add(raob_enqueued, 1, 1) + else + :counters.add(skipped, 1, 1) end end end) + end) + end +end) - # Find nearby sounding stations (within 300km) - nearby_soundings = ImportHelpers.nearest_stations(lat, lon, sounding_station_list, 300) - - Enum.each(nearby_soundings, fn s -> - sounding_times = ImportHelpers.sounding_times_around(qso.qso_timestamp) - - Enum.each(sounding_times, fn sounding_time -> - key = {s.station_code, sounding_time} - - if !:ets.member(raob_fetched, key) do - :ets.insert(raob_fetched, {key, true}) - - station = Map.get(sounding_station_map, s.station_code) - - if station do - # Skip HTTP fetch if we already have this sounding - if !Weather.has_sounding?(station.id, sounding_time) do - case IemClient.fetch_raob(s.station_code, sounding_time) do - {:ok, [parsed | _]} -> - params = SoundingParams.derive(parsed.profile) - - sounding_attrs = - if params do - %{ - observed_at: parsed.observed_at, - profile: parsed.profile, - level_count: params.level_count, - surface_pressure_mb: params.surface_pressure_mb, - surface_temp_c: params.surface_temp_c, - surface_dewpoint_c: params.surface_dewpoint_c, - surface_refractivity: params.surface_refractivity, - min_refractivity_gradient: params.min_refractivity_gradient, - boundary_layer_depth_m: params.boundary_layer_depth_m, - precipitable_water_mm: params.precipitable_water_mm, - k_index: params.k_index, - lifted_index: params.lifted_index, - ducting_detected: params.ducting_detected, - duct_characteristics: params.duct_characteristics - } - else - %{ - observed_at: parsed.observed_at, - profile: parsed.profile, - level_count: length(parsed.profile) - } - end - - Weather.upsert_sounding(station, sounding_attrs) - - {:ok, []} -> - :ok - - {:error, reason} -> - jitter = :rand.uniform(3600) - - IO.puts( - " RAOB error #{s.station_code}: #{inspect(reason)}, enqueueing retry" - ) - - %{ - "fetch_type" => "raob", - "station_id" => station.id, - "station_code" => s.station_code, - "sounding_time" => DateTime.to_iso8601(sounding_time) - } - |> WeatherFetchWorker.new(schedule_in: 7200 + jitter) - |> Oban.insert() - end - end - end - end - end) - end) - end - end, - max_concurrency: 5, - timeout: 60_000, - ordered: false -) -|> Stream.run() - -:ets.delete(asos_fetched) -:ets.delete(raob_fetched) +:ets.delete(asos_seen) +:ets.delete(raob_seen) # --- Summary --- @@ -340,12 +273,18 @@ asos_count = Repo.aggregate(Microwaveprop.Weather.SurfaceObservation, :count) sounding_count = Repo.aggregate(Microwaveprop.Weather.Sounding, :count) station_count = Repo.aggregate(Microwaveprop.Weather.Station, :count) solar_count = Repo.aggregate(Microwaveprop.Weather.SolarIndex, :count) +job_count = Oban.Job |> Repo.aggregate(:count) IO.puts(""" -Import complete: +Scan complete: Solar index days: #{solar_count} Stations: #{station_count} - Surface observations: #{asos_count} - Soundings: #{sounding_count} + Surface observations: #{asos_count} (existing) + Soundings: #{sounding_count} (existing) + ASOS jobs enqueued: #{:counters.get(asos_enqueued, 1)} + RAOB jobs enqueued: #{:counters.get(raob_enqueued, 1)} + Total Oban jobs: #{job_count} + +Oban will now process the weather queue (concurrency: 3) with automatic retries. """)