Graham McIntire
6f16395f44
- Radio context with QSO schema and CSV import script (58K contacts) - Solar index schema, GFZ client, and daily Oban cron worker - LiveView dashboard with QSO table and weather correlation views - Parallelize weather import script using Task.async_stream (10 concurrent workers) - Replace sequential rate_limit sleeps with concurrency-based backpressure - Atomic progress counter for interleave-safe reporting
323 lines
10 KiB
Elixir
323 lines
10 KiB
Elixir
import Ecto.Query
|
|
|
|
alias Microwaveprop.Repo
|
|
alias Microwaveprop.Weather
|
|
alias Microwaveprop.Weather.IemClient
|
|
alias Microwaveprop.Weather.SolarClient
|
|
alias Microwaveprop.Weather.SoundingParams
|
|
|
|
# --- Helpers ---
|
|
|
|
defmodule ImportHelpers do
|
|
@moduledoc false
|
|
|
|
@km_per_deg 111.0
|
|
|
|
@us_networks ~w(
|
|
AL_ASOS AK_ASOS AZ_ASOS AR_ASOS CA_ASOS CO_ASOS CT_ASOS DE_ASOS FL_ASOS
|
|
GA_ASOS HI_ASOS ID_ASOS IL_ASOS IN_ASOS IA_ASOS KS_ASOS KY_ASOS LA_ASOS
|
|
ME_ASOS MD_ASOS MA_ASOS MI_ASOS MN_ASOS MS_ASOS MO_ASOS MT_ASOS NE_ASOS
|
|
NV_ASOS NH_ASOS NJ_ASOS NM_ASOS NY_ASOS NC_ASOS ND_ASOS OH_ASOS OK_ASOS
|
|
OR_ASOS PA_ASOS RI_ASOS SC_ASOS SD_ASOS TN_ASOS TX_ASOS UT_ASOS VT_ASOS
|
|
VA_ASOS WA_ASOS WV_ASOS WI_ASOS WY_ASOS DC_ASOS PR_ASOS
|
|
)
|
|
|
|
def us_networks, do: @us_networks
|
|
|
|
def nearest_stations(lat, lon, stations, radius_km) do
|
|
Enum.filter(stations, fn s ->
|
|
dlat = (s.lat - lat) * @km_per_deg
|
|
dlon = (s.lon - lon) * @km_per_deg * :math.cos(lat * :math.pi() / 180)
|
|
:math.sqrt(dlat * dlat + dlon * dlon) <= radius_km
|
|
end)
|
|
end
|
|
|
|
def sounding_times_around(dt) do
|
|
# Return the 00Z and 12Z sounding times that bracket the QSO time
|
|
date = DateTime.to_date(dt)
|
|
hour = dt.hour
|
|
|
|
times =
|
|
if hour < 12 do
|
|
[
|
|
DateTime.new!(Date.add(date, -1), ~T[12:00:00], "Etc/UTC"),
|
|
DateTime.new!(date, ~T[00:00:00], "Etc/UTC")
|
|
]
|
|
else
|
|
[
|
|
DateTime.new!(date, ~T[00:00:00], "Etc/UTC"),
|
|
DateTime.new!(date, ~T[12:00:00], "Etc/UTC")
|
|
]
|
|
end
|
|
|
|
Enum.uniq(times)
|
|
end
|
|
end
|
|
|
|
# --- Phase 0: Import solar indices ---
|
|
|
|
IO.puts("=== Phase 0: Importing solar indices (since 2000-01-01) ===")
|
|
|
|
existing_solar = Weather.existing_solar_dates()
|
|
IO.puts("Found #{MapSet.size(existing_solar)} solar index dates already in DB")
|
|
|
|
IO.puts("Downloading GFZ solar index file...")
|
|
|
|
case SolarClient.fetch_solar_indices() do
|
|
{:ok, all_records} ->
|
|
IO.puts("Parsed #{length(all_records)} total daily records from GFZ file")
|
|
|
|
records =
|
|
all_records
|
|
|> SolarClient.filter_since(~D[2000-01-01])
|
|
|> Enum.reject(fn r -> MapSet.member?(existing_solar, r.date) end)
|
|
|
|
IO.puts("#{length(records)} new records to import (skipping dates already in DB)")
|
|
|
|
records
|
|
|> Enum.with_index(1)
|
|
|> Enum.each(fn {record, idx} ->
|
|
if rem(idx, 500) == 0, do: IO.puts(" Upserting solar index #{idx}/#{length(records)}...")
|
|
Weather.upsert_solar_index(record)
|
|
end)
|
|
|
|
IO.puts("Solar index import complete")
|
|
|
|
{:error, reason} ->
|
|
IO.puts("Warning: failed to fetch solar indices: #{inspect(reason)}")
|
|
end
|
|
|
|
# --- Phase 1: Discover and seed stations from IEM ---
|
|
|
|
IO.puts("=== Phase 1: Discovering stations from IEM ===")
|
|
|
|
# Fetch ASOS stations from all US state networks
|
|
IO.puts("Fetching ASOS stations from #{length(ImportHelpers.us_networks())} state networks...")
|
|
|
|
asos_stations =
|
|
ImportHelpers.us_networks()
|
|
|> Task.async_stream(
|
|
fn network ->
|
|
state = network |> String.split("_") |> hd()
|
|
|
|
case IemClient.fetch_network(network) do
|
|
{:ok, stations} ->
|
|
Enum.map(stations, &Map.put(&1, :state, state))
|
|
|
|
{:error, reason} ->
|
|
IO.puts(" Warning: failed to fetch #{network}: #{inspect(reason)}")
|
|
[]
|
|
end
|
|
end,
|
|
max_concurrency: 10,
|
|
timeout: 30_000,
|
|
ordered: false
|
|
)
|
|
|> Enum.flat_map(fn
|
|
{:ok, stations} -> stations
|
|
{:exit, _reason} -> []
|
|
end)
|
|
|
|
IO.puts("Discovered #{length(asos_stations)} ASOS stations across all US networks")
|
|
|
|
# Fetch RAOB (sounding) stations
|
|
IO.puts("Fetching RAOB stations...")
|
|
|
|
raob_stations =
|
|
case IemClient.fetch_network("RAOB") do
|
|
{:ok, stations} ->
|
|
stations
|
|
|
|
{:error, reason} ->
|
|
IO.puts(" Warning: failed to fetch RAOB network: #{inspect(reason)}")
|
|
[]
|
|
end
|
|
|
|
IO.puts("Discovered #{length(raob_stations)} RAOB stations")
|
|
|
|
# Seed ASOS stations
|
|
IO.puts("Seeding ASOS stations...")
|
|
|
|
asos_station_map =
|
|
Enum.reduce(asos_stations, %{}, fn attrs, acc ->
|
|
{:ok, station} =
|
|
Weather.find_or_create_station(Map.put(attrs, :station_type, "asos"))
|
|
|
|
Map.put(acc, station.station_code, station)
|
|
end)
|
|
|
|
IO.puts("Seeded #{map_size(asos_station_map)} unique ASOS stations")
|
|
|
|
# Seed sounding stations
|
|
IO.puts("Seeding sounding stations...")
|
|
|
|
sounding_station_map =
|
|
Enum.reduce(raob_stations, %{}, fn attrs, acc ->
|
|
{:ok, station} =
|
|
Weather.find_or_create_station(Map.put(attrs, :station_type, "sounding"))
|
|
|
|
Map.put(acc, station.station_code, station)
|
|
end)
|
|
|
|
IO.puts("Seeded #{map_size(sounding_station_map)} unique sounding stations")
|
|
|
|
# Build lookup lists with lat/lon for nearest-station searches
|
|
asos_station_list =
|
|
asos_stations
|
|
|> Enum.map(fn s -> Map.put(s, :station_type, "asos") end)
|
|
|> Enum.uniq_by(& &1.station_code)
|
|
|
|
sounding_station_list =
|
|
raob_stations
|
|
|> Enum.map(fn s -> Map.put(s, :station_type, "sounding") end)
|
|
|> Enum.uniq_by(& &1.station_code)
|
|
|
|
# --- Phase 2: Fetch weather for each QSO ---
|
|
|
|
IO.puts("\n=== Phase 2: Importing weather data ===")
|
|
|
|
qsos =
|
|
Microwaveprop.Radio.Qso
|
|
|> where([q], not is_nil(q.pos1))
|
|
|> order_by([q], q.qso_timestamp)
|
|
|> Repo.all()
|
|
|
|
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])
|
|
|
|
total = length(qsos)
|
|
progress = :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"]
|
|
|
|
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")
|
|
|
|
# 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)}
|
|
|
|
if !:ets.member(asos_fetched, key) do
|
|
:ets.insert(asos_fetched, {key, true})
|
|
|
|
station = Map.get(asos_station_map, s.station_code)
|
|
|
|
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)
|
|
|
|
{:error, reason} ->
|
|
IO.puts(" ASOS error #{s.station_code}: #{inspect(reason)}")
|
|
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} ->
|
|
IO.puts(" RAOB error #{s.station_code}: #{inspect(reason)}")
|
|
end
|
|
end
|
|
end
|
|
end
|
|
end)
|
|
end)
|
|
end
|
|
end,
|
|
max_concurrency: 10,
|
|
timeout: 60_000,
|
|
ordered: false
|
|
)
|
|
|> Stream.run()
|
|
|
|
:ets.delete(asos_fetched)
|
|
:ets.delete(raob_fetched)
|
|
|
|
# --- Summary ---
|
|
|
|
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)
|
|
|
|
IO.puts("""
|
|
|
|
Import complete:
|
|
Solar index days: #{solar_count}
|
|
Stations: #{station_count}
|
|
Surface observations: #{asos_count}
|
|
Soundings: #{sounding_count}
|
|
""")
|