graham/prop
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Move all weather fetching to Oban, import script only enqueues jobs

Browse source 
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.
This commit is contained in:
Graham McIntire 2026-03-29 13:29:29 -05:00
parent 5124cc09c1
commit ef9aac60c0
No known key found for this signature in database
GPG key ID: F4ABF488E6029E59
1 changed files with 77 additions and 138 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

215
priv/repo/import_weather.exs
View file

@ -4,7 +4,6 @@ alias Microwaveprop.Repo
alias Microwaveprop.Weather alias Microwaveprop.Weather
alias Microwaveprop.Weather.IemClient alias Microwaveprop.Weather.IemClient
alias Microwaveprop.Weather.SolarClient alias Microwaveprop.Weather.SolarClient
alias Microwaveprop.Weather.SoundingParams
alias Microwaveprop.Workers.WeatherFetchWorker alias Microwaveprop.Workers.WeatherFetchWorker
# --- Helpers --- # --- Helpers ---
@ -173,9 +172,9 @@ sounding_station_list =
|> Enum.map(fn s -> Map.put(s, :station_type, "sounding") end) |> Enum.map(fn s -> Map.put(s, :station_type, "sounding") end)
|> Enum.uniq_by(& &1.station_code) |> 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 = qsos =
Microwaveprop.Radio.Qso Microwaveprop.Radio.Qso
@ -185,154 +184,88 @@ qsos =
IO.puts("Processing #{length(qsos)} QSOs with positions...") IO.puts("Processing #{length(qsos)} QSOs with positions...")
# Track which station+time combos we've already fetched to avoid duplicates # Track which station+time combos we've already enqueued to avoid duplicates
asos_fetched = :ets.new(:asos_fetched, [:set, :public]) asos_seen = :ets.new(:asos_seen, [:set, :public])
raob_fetched = :ets.new(:raob_fetched, [:set, :public]) raob_seen = :ets.new(:raob_seen, [:set, :public])
total = length(qsos) 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 Enum.with_index(qsos, 1)
|> Task.async_stream( |> Enum.each(fn {qso, idx} ->
fn qso -> if rem(idx, 500) == 0, do: IO.puts(" Scanned #{idx}/#{total} QSOs")
lat = qso.pos1["lat"] || qso.pos1["latitude"]
lon = qso.pos1["lng"] || qso.pos1["lon"] || qso.pos1["longitude"]
if lat && lon do lat = qso.pos1["lat"] || qso.pos1["latitude"]
:counters.add(progress, 1, 1) lon = qso.pos1["lng"] || qso.pos1["lon"] || qso.pos1["longitude"]
val = :counters.get(progress, 1)
if rem(val, 500) == 0, do: IO.puts(" Processed #{val}/#{total} QSOs")
# Find nearby ASOS stations (within 150km) if lat && lon do
nearby_asos = ImportHelpers.nearest_stations(lat, lon, asos_station_list, 150) # Find nearby ASOS stations (within 150km)
nearby_asos = ImportHelpers.nearest_stations(lat, lon, asos_station_list, 150)
Enum.each(nearby_asos, fn s -> Enum.each(nearby_asos, fn s ->
# Fetch a 4-hour window around QSO time start_dt = DateTime.add(qso.qso_timestamp, -2 * 3600, :second)
start_dt = DateTime.add(qso.qso_timestamp, -2 * 3600, :second) end_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)}
key = {s.station_code, DateTime.to_date(start_dt), div(start_dt.hour, 4)}
if !:ets.member(asos_fetched, key) do if !:ets.member(asos_seen, key) do
:ets.insert(asos_fetched, {key, true}) :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 :counters.add(asos_enqueued, 1, 1)
# Skip HTTP fetch if we already have observations in this window else
if !Weather.has_surface_observations?(station.id, start_dt, end_dt) do :counters.add(skipped, 1, 1)
case IemClient.fetch_asos(s.station_code, start_dt, end_dt) do end
{:ok, rows} -> end
Enum.each(rows, fn row -> end)
if row.observed_at do
Weather.upsert_surface_observation(station, row)
end
end)
{:error, reason} -> # Find nearby sounding stations (within 300km)
jitter = :rand.uniform(3600) nearby_soundings = ImportHelpers.nearest_stations(lat, lon, sounding_station_list, 300)
IO.puts( Enum.each(nearby_soundings, fn s ->
" ASOS error #{s.station_code}: #{inspect(reason)}, enqueueing retry" sounding_times = ImportHelpers.sounding_times_around(qso.qso_timestamp)
)
%{ Enum.each(sounding_times, fn sounding_time ->
"fetch_type" => "asos", key = {s.station_code, sounding_time}
"station_id" => station.id,
"station_code" => s.station_code, if !:ets.member(raob_seen, key) do
"start_dt" => DateTime.to_iso8601(start_dt), :ets.insert(raob_seen, {key, true})
"end_dt" => DateTime.to_iso8601(end_dt) station = Map.get(sounding_station_map, s.station_code)
}
|> WeatherFetchWorker.new(schedule_in: 7200 + jitter) if station && !Weather.has_sounding?(station.id, sounding_time) do
|> Oban.insert() %{
end "fetch_type" => "raob",
end "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)
end)
end
end)
# Find nearby sounding stations (within 300km) :ets.delete(asos_seen)
nearby_soundings = ImportHelpers.nearest_stations(lat, lon, sounding_station_list, 300) :ets.delete(raob_seen)
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)
# --- Summary --- # --- Summary ---
@ -340,12 +273,18 @@ asos_count = Repo.aggregate(Microwaveprop.Weather.SurfaceObservation, :count)
sounding_count = Repo.aggregate(Microwaveprop.Weather.Sounding, :count) sounding_count = Repo.aggregate(Microwaveprop.Weather.Sounding, :count)
station_count = Repo.aggregate(Microwaveprop.Weather.Station, :count) station_count = Repo.aggregate(Microwaveprop.Weather.Station, :count)
solar_count = Repo.aggregate(Microwaveprop.Weather.SolarIndex, :count) solar_count = Repo.aggregate(Microwaveprop.Weather.SolarIndex, :count)
job_count = Oban.Job |> Repo.aggregate(:count)
IO.puts(""" IO.puts("""
Import complete: Scan complete:
Solar index days: #{solar_count} Solar index days: #{solar_count}
Stations: #{station_count} Stations: #{station_count}
Surface observations: #{asos_count} Surface observations: #{asos_count} (existing)
Soundings: #{sounding_count} 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.
""") """)