Add HRRR model profile fetching for QSOs
HRRR provides hourly 3km-resolution atmospheric profiles, filling the temporal gaps (12-hourly soundings) and spatial gaps (only 9 sounding stations) in our current weather data. - Add hrrr_profiles table and hrrr_queued flag on QSOs - HrrrClient fetches GRIB2 data via HTTP Range requests + wgrib2 - HrrrFetchWorker derives refractivity/ducting params via SoundingParams - QsoWeatherEnqueueWorker now also enqueues HRRR jobs - QSO show page displays HRRR section with collapsible profile - Dockerfile builds wgrib2 from source for production
This commit is contained in:
parent
586d94fceb
commit
8bd0989fbe
17 changed files with 1219 additions and 5 deletions
14
Dockerfile
14
Dockerfile
|
|
@ -20,11 +20,17 @@ ARG RUNNER_IMAGE="docker.io/debian:${DEBIAN_VERSION}"
|
|||
|
||||
FROM ${BUILDER_IMAGE} AS builder
|
||||
|
||||
# install build dependencies
|
||||
# install build dependencies and build wgrib2
|
||||
RUN apt-get update \
|
||||
&& apt-get install -y --no-install-recommends build-essential git \
|
||||
&& apt-get install -y --no-install-recommends build-essential git gfortran wget \
|
||||
&& rm -rf /var/lib/apt/lists/*
|
||||
|
||||
RUN wget -q https://www.ftp.cpc.ncep.noaa.gov/wd51we/wgrib2/wgrib2.tgz \
|
||||
&& tar xzf wgrib2.tgz && cd grib2 \
|
||||
&& CC=gcc FC=gfortran make \
|
||||
&& cp wgrib2/wgrib2 /usr/local/bin/ \
|
||||
&& cd .. && rm -rf grib2 wgrib2.tgz
|
||||
|
||||
# prepare build dir
|
||||
WORKDIR /app
|
||||
|
||||
|
|
@ -71,9 +77,11 @@ RUN mix release
|
|||
FROM ${RUNNER_IMAGE} AS final
|
||||
|
||||
RUN apt-get update \
|
||||
&& apt-get install -y --no-install-recommends libstdc++6 openssl libncurses6 locales ca-certificates \
|
||||
&& apt-get install -y --no-install-recommends libstdc++6 openssl libncurses6 locales ca-certificates libgomp1 \
|
||||
&& rm -rf /var/lib/apt/lists/*
|
||||
|
||||
COPY --from=builder /usr/local/bin/wgrib2 /usr/local/bin/wgrib2
|
||||
|
||||
# Set the locale
|
||||
RUN sed -i '/en_US.UTF-8/s/^# //g' /etc/locale.gen \
|
||||
&& locale-gen
|
||||
|
|
|
|||
|
|
@ -44,7 +44,7 @@ config :microwaveprop, MicrowavepropWeb.Endpoint,
|
|||
|
||||
config :microwaveprop, Oban,
|
||||
repo: Microwaveprop.Repo,
|
||||
queues: [solar: 1, weather: 3, enqueue: 1],
|
||||
queues: [solar: 1, weather: 3, enqueue: 1, hrrr: 2],
|
||||
plugins: [
|
||||
{Oban.Plugins.Pruner, max_age: 3600 * 24},
|
||||
{Oban.Plugins.Cron,
|
||||
|
|
|
|||
|
|
@ -57,6 +57,20 @@ defmodule Microwaveprop.Radio do
|
|||
|> Repo.update_all(set: [weather_queued: true])
|
||||
end
|
||||
|
||||
def unprocessed_hrrr_qsos(limit \\ 500) do
|
||||
Qso
|
||||
|> where([q], q.hrrr_queued == false and not is_nil(q.pos1))
|
||||
|> order_by([q], asc: q.qso_timestamp)
|
||||
|> limit(^limit)
|
||||
|> Repo.all()
|
||||
end
|
||||
|
||||
def mark_hrrr_queued!(qso_ids) do
|
||||
Qso
|
||||
|> where([q], q.id in ^qso_ids)
|
||||
|> Repo.update_all(set: [hrrr_queued: true])
|
||||
end
|
||||
|
||||
@earth_radius_km 6371.0
|
||||
|
||||
def haversine_km(lat1, lon1, lat2, lon2) do
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ defmodule Microwaveprop.Radio.Qso do
|
|||
field :band, :decimal
|
||||
field :distance_km, :decimal
|
||||
field :weather_queued, :boolean, default: false
|
||||
field :hrrr_queued, :boolean, default: false
|
||||
|
||||
timestamps(type: :utc_datetime)
|
||||
end
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ defmodule Microwaveprop.Weather do
|
|||
import Ecto.Query
|
||||
|
||||
alias Microwaveprop.Repo
|
||||
alias Microwaveprop.Weather.HrrrProfile
|
||||
alias Microwaveprop.Weather.SolarIndex
|
||||
alias Microwaveprop.Weather.Sounding
|
||||
alias Microwaveprop.Weather.Station
|
||||
|
|
@ -164,4 +165,63 @@ defmodule Microwaveprop.Weather do
|
|||
|
||||
%{surface_observations: surface_observations, soundings: soundings}
|
||||
end
|
||||
|
||||
def upsert_hrrr_profile(attrs) do
|
||||
%HrrrProfile{}
|
||||
|> HrrrProfile.changeset(attrs)
|
||||
|> Repo.insert(
|
||||
on_conflict: {:replace_all_except, [:id, :inserted_at]},
|
||||
conflict_target: [:lat, :lon, :valid_time],
|
||||
returning: true
|
||||
)
|
||||
end
|
||||
|
||||
def has_hrrr_profile?(lat, lon, valid_time) do
|
||||
{rlat, rlon} = round_to_hrrr_grid(lat, lon)
|
||||
|
||||
HrrrProfile
|
||||
|> where([h], h.lat == ^rlat and h.lon == ^rlon and h.valid_time == ^valid_time)
|
||||
|> Repo.exists?()
|
||||
end
|
||||
|
||||
def hrrr_for_qso(%{pos1: nil}), do: nil
|
||||
|
||||
def hrrr_for_qso(qso) do
|
||||
lat = qso.pos1["lat"]
|
||||
lon = qso.pos1["lon"] || qso.pos1["lng"]
|
||||
|
||||
if lat && lon do
|
||||
# Search within ~0.05 degrees (~5km) and 1 hour
|
||||
dlat = 0.05
|
||||
dlon = 0.05
|
||||
time_start = DateTime.add(qso.qso_timestamp, -3600, :second)
|
||||
time_end = DateTime.add(qso.qso_timestamp, 3600, :second)
|
||||
|
||||
HrrrProfile
|
||||
|> where(
|
||||
[h],
|
||||
h.lat >= ^(lat - dlat) and h.lat <= ^(lat + dlat) and
|
||||
h.lon >= ^(lon - dlon) and h.lon <= ^(lon + dlon) and
|
||||
h.valid_time >= ^time_start and h.valid_time <= ^time_end
|
||||
)
|
||||
|> order_by([h],
|
||||
asc:
|
||||
fragment(
|
||||
"ABS(? - ?) + ABS(? - ?) + ABS(EXTRACT(EPOCH FROM ? - ?))",
|
||||
h.lat,
|
||||
^lat,
|
||||
h.lon,
|
||||
^lon,
|
||||
h.valid_time,
|
||||
^qso.qso_timestamp
|
||||
)
|
||||
)
|
||||
|> limit(1)
|
||||
|> Repo.one()
|
||||
end
|
||||
end
|
||||
|
||||
def round_to_hrrr_grid(lat, lon) do
|
||||
{Float.round(lat / 1.0, 2), Float.round(lon / 1.0, 2)}
|
||||
end
|
||||
end
|
||||
|
|
|
|||
255
lib/microwaveprop/weather/hrrr_client.ex
Normal file
255
lib/microwaveprop/weather/hrrr_client.ex
Normal file
|
|
@ -0,0 +1,255 @@
|
|||
defmodule Microwaveprop.Weather.HrrrClient do
|
||||
@moduledoc false
|
||||
|
||||
@hrrr_base "https://noaa-hrrr-bdp-pds.s3.amazonaws.com"
|
||||
|
||||
@pressure_levels [1000, 975, 950, 925, 900, 850, 800, 700]
|
||||
|
||||
@surface_messages [
|
||||
%{var: "TMP", level: "2 m above ground"},
|
||||
%{var: "DPT", level: "2 m above ground"},
|
||||
%{var: "PRES", level: "surface"},
|
||||
%{var: "HPBL", level: "surface"},
|
||||
%{var: "PWAT", level: "entire atmosphere (considered as a single layer)"}
|
||||
]
|
||||
|
||||
# --- Public API ---
|
||||
|
||||
def fetch_profile(lat, lon, valid_time) do
|
||||
hour_dt = nearest_hrrr_hour(valid_time)
|
||||
date = DateTime.to_date(hour_dt)
|
||||
hour = hour_dt.hour
|
||||
|
||||
with {:ok, sfc_data} <- fetch_product(date, hour, :surface, lat, lon),
|
||||
{:ok, prs_data} <- fetch_product(date, hour, :pressure, lat, lon) do
|
||||
merged = Map.merge(sfc_data, prs_data)
|
||||
result = build_profile_from_wgrib2(merged)
|
||||
{:ok, Map.put(result, :run_time, hour_dt)}
|
||||
end
|
||||
end
|
||||
|
||||
def nearest_hrrr_hour(dt) do
|
||||
total_seconds = dt.minute * 60 + dt.second
|
||||
rounded_dt = DateTime.add(dt, -total_seconds, :second)
|
||||
|
||||
if dt.minute >= 30 do
|
||||
DateTime.add(rounded_dt, 3600, :second)
|
||||
else
|
||||
rounded_dt
|
||||
end
|
||||
end
|
||||
|
||||
def hrrr_url(date, hour, product) do
|
||||
date_str = Calendar.strftime(date, "%Y%m%d")
|
||||
hour_str = hour |> Integer.to_string() |> String.pad_leading(2, "0")
|
||||
|
||||
file =
|
||||
case product do
|
||||
:surface -> "hrrr.t#{hour_str}z.wrfsfcf00.grib2"
|
||||
:pressure -> "hrrr.t#{hour_str}z.wrfprsf00.grib2"
|
||||
end
|
||||
|
||||
"#{@hrrr_base}/hrrr.#{date_str}/conus/#{file}"
|
||||
end
|
||||
|
||||
def parse_idx(text) do
|
||||
text
|
||||
|> String.split("\n")
|
||||
|> Enum.reject(&(&1 == ""))
|
||||
|> Enum.map(fn line ->
|
||||
parts = String.split(line, ":", parts: 8)
|
||||
|
||||
%{
|
||||
msg: String.to_integer(Enum.at(parts, 0)),
|
||||
offset: String.to_integer(Enum.at(parts, 1)),
|
||||
var: Enum.at(parts, 3),
|
||||
level: Enum.at(parts, 4)
|
||||
}
|
||||
end)
|
||||
end
|
||||
|
||||
def byte_ranges_for_messages(idx_entries, wanted) do
|
||||
Enum.flat_map(wanted, fn w ->
|
||||
idx_entries
|
||||
|> Enum.with_index()
|
||||
|> Enum.flat_map(fn {entry, idx} ->
|
||||
if entry.var == w.var && entry.level == w.level do
|
||||
end_offset =
|
||||
case Enum.at(idx_entries, idx + 1) do
|
||||
nil -> entry.offset + 10_000_000
|
||||
next -> next.offset - 1
|
||||
end
|
||||
|
||||
[{entry.offset, end_offset}]
|
||||
else
|
||||
[]
|
||||
end
|
||||
end)
|
||||
end)
|
||||
end
|
||||
|
||||
def parse_wgrib2_output(text) do
|
||||
text
|
||||
|> String.split("\n")
|
||||
|> Enum.reject(&(&1 == ""))
|
||||
|> Enum.reduce(%{}, fn line, acc ->
|
||||
case parse_wgrib2_line(line) do
|
||||
{:ok, key, value} -> Map.put(acc, key, value)
|
||||
:skip -> acc
|
||||
end
|
||||
end)
|
||||
end
|
||||
|
||||
def build_profile_from_wgrib2(parsed) do
|
||||
sfc_temp_k = parsed["TMP:2 m above ground"]
|
||||
sfc_dpt_k = parsed["DPT:2 m above ground"]
|
||||
sfc_pres_pa = parsed["PRES:surface"]
|
||||
|
||||
profile =
|
||||
Enum.flat_map(@pressure_levels, fn level ->
|
||||
level_str = "#{level} mb"
|
||||
tmp = parsed["TMP:#{level_str}"]
|
||||
dpt = parsed["DPT:#{level_str}"]
|
||||
hgt = parsed["HGT:#{level_str}"]
|
||||
|
||||
if tmp && dpt && hgt do
|
||||
[
|
||||
%{
|
||||
"pres" => level * 1.0,
|
||||
"tmpc" => tmp - 273.15,
|
||||
"dwpc" => dpt - 273.15,
|
||||
"hght" => hgt
|
||||
}
|
||||
]
|
||||
else
|
||||
[]
|
||||
end
|
||||
end)
|
||||
|
||||
%{
|
||||
surface_temp_c: if(sfc_temp_k, do: sfc_temp_k - 273.15),
|
||||
surface_dewpoint_c: if(sfc_dpt_k, do: sfc_dpt_k - 273.15),
|
||||
surface_pressure_mb: if(sfc_pres_pa, do: sfc_pres_pa / 100.0),
|
||||
hpbl_m: parsed["HPBL:surface"],
|
||||
pwat_mm: parsed["PWAT:entire atmosphere (considered as a single layer)"],
|
||||
profile: profile
|
||||
}
|
||||
end
|
||||
|
||||
# --- Private ---
|
||||
|
||||
defp fetch_product(date, hour, product, lat, lon) do
|
||||
url = hrrr_url(date, hour, product)
|
||||
idx_url = url <> ".idx"
|
||||
|
||||
wanted =
|
||||
case product do
|
||||
:surface -> @surface_messages
|
||||
:pressure -> pressure_messages()
|
||||
end
|
||||
|
||||
with {:ok, idx_text} <- fetch_idx(idx_url),
|
||||
idx_entries = parse_idx(idx_text),
|
||||
ranges = byte_ranges_for_messages(idx_entries, wanted),
|
||||
{:ok, grib_binary} <- download_grib_ranges(url, ranges),
|
||||
{:ok, output} <- extract_point(grib_binary, lat, lon) do
|
||||
{:ok, parse_wgrib2_output(output)}
|
||||
end
|
||||
end
|
||||
|
||||
defp pressure_messages do
|
||||
for level <- @pressure_levels, var <- ["TMP", "DPT", "HGT"] do
|
||||
%{var: var, level: "#{level} mb"}
|
||||
end
|
||||
end
|
||||
|
||||
defp fetch_idx(url) do
|
||||
case Req.get(url, req_options()) do
|
||||
{:ok, %{status: 200, body: body}} ->
|
||||
{:ok, body}
|
||||
|
||||
{:ok, %{status: status}} ->
|
||||
{:error, "HRRR idx HTTP #{status}"}
|
||||
|
||||
{:error, reason} ->
|
||||
{:error, reason}
|
||||
end
|
||||
end
|
||||
|
||||
defp download_grib_ranges(_url, []), do: {:ok, <<>>}
|
||||
|
||||
defp download_grib_ranges(url, ranges) do
|
||||
range_header =
|
||||
Enum.map_join(ranges, ", ", fn {start, stop} -> "#{start}-#{stop}" end)
|
||||
|
||||
case Req.get(url, [{:headers, [{"Range", "bytes=#{range_header}"}]} | req_options()]) do
|
||||
{:ok, %{status: status, body: body}} when status in [200, 206] ->
|
||||
{:ok, body}
|
||||
|
||||
{:ok, %{status: status}} ->
|
||||
{:error, "HRRR grib HTTP #{status}"}
|
||||
|
||||
{:error, reason} ->
|
||||
{:error, reason}
|
||||
end
|
||||
end
|
||||
|
||||
defp extract_point(grib_binary, lat, lon) do
|
||||
case System.find_executable("wgrib2") do
|
||||
nil ->
|
||||
{:error, "wgrib2 not found in PATH. Install via: brew install wgrib2"}
|
||||
|
||||
wgrib2_path ->
|
||||
tmp_dir = System.tmp_dir!()
|
||||
tmp_file = Path.join(tmp_dir, "hrrr_#{:erlang.unique_integer([:positive])}.grib2")
|
||||
|
||||
try do
|
||||
File.write!(tmp_file, grib_binary)
|
||||
lon_360 = if lon < 0, do: lon + 360, else: lon
|
||||
|
||||
case System.cmd(wgrib2_path, [tmp_file, "-lon", "#{lon_360}", "#{lat}"], stderr_to_stdout: true) do
|
||||
{output, 0} -> {:ok, output}
|
||||
{output, _} -> {:error, "wgrib2 failed: #{output}"}
|
||||
end
|
||||
after
|
||||
File.rm(tmp_file)
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
defp parse_wgrib2_line(line) do
|
||||
case Regex.run(~r/val=([0-9eE.+-]+)\s*$/, line) do
|
||||
[_, val_str] ->
|
||||
case Float.parse(val_str) do
|
||||
{val, _} ->
|
||||
parts = String.split(line, ":")
|
||||
var = Enum.at(parts, 3)
|
||||
level = Enum.at(parts, 4)
|
||||
|
||||
if var && level do
|
||||
{:ok, "#{var}:#{level}", val}
|
||||
else
|
||||
:skip
|
||||
end
|
||||
|
||||
:error ->
|
||||
:skip
|
||||
end
|
||||
|
||||
_ ->
|
||||
:skip
|
||||
end
|
||||
end
|
||||
|
||||
defp req_options do
|
||||
defaults = [retry: :transient, max_retries: 3, retry_delay: &retry_delay/1]
|
||||
overrides = Application.get_env(:microwaveprop, :hrrr_req_options, [])
|
||||
Keyword.merge(defaults, overrides)
|
||||
end
|
||||
|
||||
defp retry_delay(n) do
|
||||
base = Integer.pow(2, n) * 1_000
|
||||
jitter = :rand.uniform(500)
|
||||
base + jitter
|
||||
end
|
||||
end
|
||||
38
lib/microwaveprop/weather/hrrr_profile.ex
Normal file
38
lib/microwaveprop/weather/hrrr_profile.ex
Normal file
|
|
@ -0,0 +1,38 @@
|
|||
defmodule Microwaveprop.Weather.HrrrProfile do
|
||||
@moduledoc false
|
||||
use Ecto.Schema
|
||||
|
||||
import Ecto.Changeset
|
||||
|
||||
@primary_key {:id, :binary_id, autogenerate: true}
|
||||
@foreign_key_type :binary_id
|
||||
|
||||
schema "hrrr_profiles" do
|
||||
field :valid_time, :utc_datetime
|
||||
field :lat, :float
|
||||
field :lon, :float
|
||||
field :run_time, :utc_datetime
|
||||
field :profile, {:array, :map}
|
||||
field :hpbl_m, :float
|
||||
field :pwat_mm, :float
|
||||
field :surface_temp_c, :float
|
||||
field :surface_dewpoint_c, :float
|
||||
field :surface_pressure_mb, :float
|
||||
field :surface_refractivity, :float
|
||||
field :min_refractivity_gradient, :float
|
||||
field :ducting_detected, :boolean, default: false
|
||||
field :duct_characteristics, {:array, :map}
|
||||
|
||||
timestamps(type: :utc_datetime)
|
||||
end
|
||||
|
||||
@required_fields ~w(valid_time lat lon)a
|
||||
@optional_fields ~w(run_time profile hpbl_m pwat_mm surface_temp_c surface_dewpoint_c surface_pressure_mb surface_refractivity min_refractivity_gradient ducting_detected duct_characteristics)a
|
||||
|
||||
def changeset(hrrr_profile, attrs) do
|
||||
hrrr_profile
|
||||
|> cast(attrs, @required_fields ++ @optional_fields)
|
||||
|> validate_required(@required_fields)
|
||||
|> unique_constraint([:lat, :lon, :valid_time])
|
||||
end
|
||||
end
|
||||
59
lib/microwaveprop/workers/hrrr_fetch_worker.ex
Normal file
59
lib/microwaveprop/workers/hrrr_fetch_worker.ex
Normal file
|
|
@ -0,0 +1,59 @@
|
|||
defmodule Microwaveprop.Workers.HrrrFetchWorker do
|
||||
@moduledoc false
|
||||
use Oban.Worker, queue: :hrrr, max_attempts: 3
|
||||
|
||||
alias Microwaveprop.Weather
|
||||
alias Microwaveprop.Weather.HrrrClient
|
||||
alias Microwaveprop.Weather.SoundingParams
|
||||
|
||||
@impl Oban.Worker
|
||||
def perform(%Oban.Job{args: args}) do
|
||||
%{"lat" => raw_lat, "lon" => raw_lon, "valid_time" => valid_time_str} = args
|
||||
|
||||
{:ok, valid_time, _} = DateTime.from_iso8601(valid_time_str)
|
||||
{lat, lon} = Weather.round_to_hrrr_grid(raw_lat, raw_lon)
|
||||
|
||||
if Weather.has_hrrr_profile?(lat, lon, valid_time) do
|
||||
:ok
|
||||
else
|
||||
case HrrrClient.fetch_profile(lat, lon, valid_time) do
|
||||
{:ok, data} ->
|
||||
params = SoundingParams.derive(data.profile)
|
||||
|
||||
attrs =
|
||||
maybe_add_derived_params(
|
||||
%{
|
||||
valid_time: valid_time,
|
||||
lat: lat,
|
||||
lon: lon,
|
||||
run_time: data.run_time,
|
||||
profile: data.profile,
|
||||
hpbl_m: data.hpbl_m,
|
||||
pwat_mm: data.pwat_mm,
|
||||
surface_temp_c: data.surface_temp_c,
|
||||
surface_dewpoint_c: data.surface_dewpoint_c,
|
||||
surface_pressure_mb: data.surface_pressure_mb
|
||||
},
|
||||
params
|
||||
)
|
||||
|
||||
Weather.upsert_hrrr_profile(attrs)
|
||||
:ok
|
||||
|
||||
{:error, reason} ->
|
||||
{:error, reason}
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
defp maybe_add_derived_params(attrs, nil), do: attrs
|
||||
|
||||
defp maybe_add_derived_params(attrs, params) do
|
||||
Map.merge(attrs, %{
|
||||
surface_refractivity: params.surface_refractivity,
|
||||
min_refractivity_gradient: params.min_refractivity_gradient,
|
||||
ducting_detected: params.ducting_detected,
|
||||
duct_characteristics: params.duct_characteristics
|
||||
})
|
||||
end
|
||||
end
|
||||
|
|
@ -4,6 +4,8 @@ defmodule Microwaveprop.Workers.QsoWeatherEnqueueWorker do
|
|||
|
||||
alias Microwaveprop.Radio
|
||||
alias Microwaveprop.Weather
|
||||
alias Microwaveprop.Weather.HrrrClient
|
||||
alias Microwaveprop.Workers.HrrrFetchWorker
|
||||
alias Microwaveprop.Workers.WeatherFetchWorker
|
||||
|
||||
@asos_radius_km 150
|
||||
|
|
@ -11,6 +13,13 @@ defmodule Microwaveprop.Workers.QsoWeatherEnqueueWorker do
|
|||
|
||||
@impl Oban.Worker
|
||||
def perform(%Oban.Job{}) do
|
||||
enqueue_weather_jobs()
|
||||
enqueue_hrrr_jobs()
|
||||
|
||||
:ok
|
||||
end
|
||||
|
||||
defp enqueue_weather_jobs do
|
||||
qsos = Radio.unprocessed_qsos()
|
||||
|
||||
if qsos != [] do
|
||||
|
|
@ -25,8 +34,21 @@ defmodule Microwaveprop.Workers.QsoWeatherEnqueueWorker do
|
|||
qso_ids = Enum.map(qsos, & &1.id)
|
||||
Radio.mark_weather_queued!(qso_ids)
|
||||
end
|
||||
end
|
||||
|
||||
:ok
|
||||
defp enqueue_hrrr_jobs do
|
||||
qsos = Radio.unprocessed_hrrr_qsos()
|
||||
|
||||
if qsos != [] do
|
||||
jobs = build_hrrr_jobs(qsos)
|
||||
|
||||
if jobs != [] do
|
||||
Oban.insert_all(jobs)
|
||||
end
|
||||
|
||||
qso_ids = Enum.map(qsos, & &1.id)
|
||||
Radio.mark_hrrr_queued!(qso_ids)
|
||||
end
|
||||
end
|
||||
|
||||
def build_weather_jobs(qsos) do
|
||||
|
|
@ -35,6 +57,12 @@ defmodule Microwaveprop.Workers.QsoWeatherEnqueueWorker do
|
|||
|> Enum.uniq_by(fn changeset -> changeset.changes.args end)
|
||||
end
|
||||
|
||||
def build_hrrr_jobs(qsos) do
|
||||
qsos
|
||||
|> Enum.flat_map(&hrrr_job_for_qso/1)
|
||||
|> Enum.uniq_by(fn changeset -> changeset.changes.args end)
|
||||
end
|
||||
|
||||
defp jobs_for_qso(qso) do
|
||||
lat = qso.pos1["lat"]
|
||||
lon = qso.pos1["lon"] || qso.pos1["lng"]
|
||||
|
|
@ -45,6 +73,28 @@ defmodule Microwaveprop.Workers.QsoWeatherEnqueueWorker do
|
|||
asos_jobs ++ raob_jobs
|
||||
end
|
||||
|
||||
defp hrrr_job_for_qso(%{pos1: nil}), do: []
|
||||
|
||||
defp hrrr_job_for_qso(qso) do
|
||||
lat = qso.pos1["lat"]
|
||||
lon = qso.pos1["lon"] || qso.pos1["lng"]
|
||||
|
||||
if lat && lon do
|
||||
{rlat, rlon} = Weather.round_to_hrrr_grid(lat, lon)
|
||||
rounded_time = HrrrClient.nearest_hrrr_hour(qso.qso_timestamp)
|
||||
|
||||
[
|
||||
HrrrFetchWorker.new(%{
|
||||
"lat" => rlat,
|
||||
"lon" => rlon,
|
||||
"valid_time" => DateTime.to_iso8601(rounded_time)
|
||||
})
|
||||
]
|
||||
else
|
||||
[]
|
||||
end
|
||||
end
|
||||
|
||||
defp build_asos_jobs(lat, lon, timestamp) do
|
||||
start_dt = DateTime.add(timestamp, -2 * 3600, :second)
|
||||
end_dt = DateTime.add(timestamp, 2 * 3600, :second)
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@ defmodule MicrowavepropWeb.QsoLive.Show do
|
|||
|
||||
weather = load_weather(qso)
|
||||
solar = load_solar(qso)
|
||||
hrrr = Weather.hrrr_for_qso(qso)
|
||||
|
||||
{:ok,
|
||||
assign(socket,
|
||||
|
|
@ -19,6 +20,8 @@ defmodule MicrowavepropWeb.QsoLive.Show do
|
|||
surface_observations: weather.surface_observations,
|
||||
soundings: weather.soundings,
|
||||
solar: solar,
|
||||
hrrr: hrrr,
|
||||
hrrr_profile_expanded: false,
|
||||
obs_sort_by: "station_name",
|
||||
obs_sort_order: "asc",
|
||||
sounding_sort_by: "station_name",
|
||||
|
|
@ -42,6 +45,10 @@ defmodule MicrowavepropWeb.QsoLive.Show do
|
|||
)}
|
||||
end
|
||||
|
||||
def handle_event("toggle_hrrr_profile", _params, socket) do
|
||||
{:noreply, assign(socket, hrrr_profile_expanded: !socket.assigns.hrrr_profile_expanded)}
|
||||
end
|
||||
|
||||
def handle_event("toggle_profile", %{"id" => id}, socket) do
|
||||
expanded =
|
||||
if MapSet.member?(socket.assigns.expanded_soundings, id) do
|
||||
|
|
@ -255,6 +262,77 @@ defmodule MicrowavepropWeb.QsoLive.Show do
|
|||
|
||||
<div class="divider" />
|
||||
|
||||
<h2 class="text-base font-semibold mb-2">HRRR Model Profile</h2>
|
||||
<%= if @hrrr do %>
|
||||
<div class="mb-6 border border-base-300 rounded-lg">
|
||||
<button
|
||||
phx-click="toggle_hrrr_profile"
|
||||
class="w-full flex items-center justify-between p-4 hover:bg-base-200 transition-colors"
|
||||
>
|
||||
<div class="flex items-center gap-4 text-sm">
|
||||
<span class="font-semibold">
|
||||
{format_number(@hrrr.lat)}°, {format_number(@hrrr.lon)}°
|
||||
</span>
|
||||
<span>{Calendar.strftime(@hrrr.valid_time, "%H:%M UTC")}</span>
|
||||
<span>HPBL: {format_number(@hrrr.hpbl_m)} m</span>
|
||||
<span>PWAT: {format_number(@hrrr.pwat_mm)} mm</span>
|
||||
<span>N: {format_number(@hrrr.surface_refractivity)}</span>
|
||||
<span>dN/dh: {format_number(@hrrr.min_refractivity_gradient)}</span>
|
||||
<%= if @hrrr.ducting_detected do %>
|
||||
<span class="badge badge-warning badge-sm">Ducting</span>
|
||||
<% end %>
|
||||
</div>
|
||||
<.icon
|
||||
name={if @hrrr_profile_expanded, do: "hero-chevron-up", else: "hero-chevron-down"}
|
||||
class="w-5 h-5"
|
||||
/>
|
||||
</button>
|
||||
|
||||
<%= if @hrrr_profile_expanded do %>
|
||||
<div class="px-4 pb-4">
|
||||
<div class="grid grid-cols-2 gap-x-8 gap-y-1 text-sm mb-4">
|
||||
<div>Surface Temp: {format_number(@hrrr.surface_temp_c)}°C</div>
|
||||
<div>Surface Dewpoint: {format_number(@hrrr.surface_dewpoint_c)}°C</div>
|
||||
<div>Surface Pressure: {format_number(@hrrr.surface_pressure_mb)} mb</div>
|
||||
<div>
|
||||
Run Time: {if @hrrr.run_time,
|
||||
do: Calendar.strftime(@hrrr.run_time, "%Y-%m-%d %H:%M UTC"),
|
||||
else: "—"}
|
||||
</div>
|
||||
</div>
|
||||
<%= if @hrrr.profile && @hrrr.profile != [] do %>
|
||||
<div class="overflow-x-auto">
|
||||
<table class="table table-xs table-zebra">
|
||||
<thead>
|
||||
<tr>
|
||||
<th>Pressure (mb)</th>
|
||||
<th>Height (m)</th>
|
||||
<th>Temp (°C)</th>
|
||||
<th>Dewpoint (°C)</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
<%= for level <- @hrrr.profile do %>
|
||||
<tr>
|
||||
<td>{format_number(level["pres"])}</td>
|
||||
<td>{format_number(level["hght"])}</td>
|
||||
<td>{format_number(level["tmpc"])}</td>
|
||||
<td>{format_number(level["dwpc"])}</td>
|
||||
</tr>
|
||||
<% end %>
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
<% end %>
|
||||
</div>
|
||||
<% end %>
|
||||
</div>
|
||||
<% else %>
|
||||
<p class="text-sm text-base-content/50 italic">No HRRR profile available.</p>
|
||||
<% end %>
|
||||
|
||||
<div class="divider" />
|
||||
|
||||
<h2 class="text-base font-semibold mb-2">Solar Conditions</h2>
|
||||
<%= if @solar do %>
|
||||
<.list>
|
||||
|
|
|
|||
|
|
@ -0,0 +1,38 @@
|
|||
defmodule Microwaveprop.Repo.Migrations.CreateHrrrProfilesAndAddHrrrQueued do
|
||||
use Ecto.Migration
|
||||
|
||||
def change do
|
||||
create table(:hrrr_profiles, primary_key: false) do
|
||||
add :id, :binary_id, primary_key: true
|
||||
add :valid_time, :utc_datetime, null: false
|
||||
add :lat, :float, null: false
|
||||
add :lon, :float, null: false
|
||||
add :run_time, :utc_datetime
|
||||
add :profile, {:array, :map}, default: []
|
||||
add :hpbl_m, :float
|
||||
add :pwat_mm, :float
|
||||
add :surface_temp_c, :float
|
||||
add :surface_dewpoint_c, :float
|
||||
add :surface_pressure_mb, :float
|
||||
add :surface_refractivity, :float
|
||||
add :min_refractivity_gradient, :float
|
||||
add :ducting_detected, :boolean, default: false
|
||||
add :duct_characteristics, {:array, :map}
|
||||
|
||||
timestamps(type: :utc_datetime)
|
||||
end
|
||||
|
||||
create unique_index(:hrrr_profiles, [:lat, :lon, :valid_time])
|
||||
create index(:hrrr_profiles, [:valid_time])
|
||||
|
||||
alter table(:qsos) do
|
||||
add :hrrr_queued, :boolean, default: false, null: false
|
||||
end
|
||||
|
||||
# Backfill existing rows to false (need HRRR fetched)
|
||||
execute "UPDATE qsos SET hrrr_queued = false", "SELECT 1"
|
||||
|
||||
# Partial index for fast lookup of unprocessed QSOs
|
||||
create index(:qsos, [:hrrr_queued], where: "hrrr_queued = false")
|
||||
end
|
||||
end
|
||||
|
|
@ -219,4 +219,66 @@ defmodule Microwaveprop.RadioTest do
|
|||
end
|
||||
end
|
||||
end
|
||||
|
||||
describe "unprocessed_hrrr_qsos/1" do
|
||||
test "returns QSOs where hrrr_queued is false and pos1 is not nil" do
|
||||
q1 = create_qso(%{station1: "W5XD", qso_timestamp: ~U[2026-03-28 12:00:00Z]})
|
||||
_q2 = create_qso(%{station1: "K5TR", pos1: nil, qso_timestamp: ~U[2026-03-28 13:00:00Z]})
|
||||
|
||||
results = Radio.unprocessed_hrrr_qsos()
|
||||
ids = Enum.map(results, & &1.id)
|
||||
assert q1.id in ids
|
||||
end
|
||||
|
||||
test "excludes QSOs already marked as hrrr_queued" do
|
||||
q = create_qso()
|
||||
Radio.mark_hrrr_queued!([q.id])
|
||||
|
||||
assert Radio.unprocessed_hrrr_qsos() == []
|
||||
end
|
||||
|
||||
test "orders by qso_timestamp ascending" do
|
||||
q_late = create_qso(%{station1: "LATE", qso_timestamp: ~U[2026-03-28 20:00:00Z]})
|
||||
q_early = create_qso(%{station1: "EARLY", qso_timestamp: ~U[2026-03-28 10:00:00Z]})
|
||||
|
||||
results = Radio.unprocessed_hrrr_qsos()
|
||||
ids = Enum.map(results, & &1.id)
|
||||
assert ids == [q_early.id, q_late.id]
|
||||
end
|
||||
|
||||
test "respects limit parameter" do
|
||||
for i <- 1..5 do
|
||||
ts = DateTime.add(~U[2026-01-01 00:00:00Z], i * 3600, :second)
|
||||
create_qso(%{qso_timestamp: ts})
|
||||
end
|
||||
|
||||
assert length(Radio.unprocessed_hrrr_qsos(3)) == 3
|
||||
end
|
||||
end
|
||||
|
||||
describe "mark_hrrr_queued!/1" do
|
||||
test "sets hrrr_queued to true for given IDs" do
|
||||
q1 = create_qso(%{station1: "A1A"})
|
||||
q2 = create_qso(%{station1: "B2B"})
|
||||
|
||||
Radio.mark_hrrr_queued!([q1.id, q2.id])
|
||||
|
||||
assert Repo.get!(Qso, q1.id).hrrr_queued == true
|
||||
assert Repo.get!(Qso, q2.id).hrrr_queued == true
|
||||
end
|
||||
|
||||
test "does not affect other QSOs" do
|
||||
q1 = create_qso(%{station1: "A1A"})
|
||||
q2 = create_qso(%{station1: "B2B"})
|
||||
|
||||
Radio.mark_hrrr_queued!([q1.id])
|
||||
|
||||
assert Repo.get!(Qso, q1.id).hrrr_queued == true
|
||||
assert Repo.get!(Qso, q2.id).hrrr_queued == false
|
||||
end
|
||||
|
||||
test "handles empty list" do
|
||||
assert Radio.mark_hrrr_queued!([]) == {0, nil}
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
211
test/microwaveprop/weather/hrrr_client_test.exs
Normal file
211
test/microwaveprop/weather/hrrr_client_test.exs
Normal file
|
|
@ -0,0 +1,211 @@
|
|||
defmodule Microwaveprop.Weather.HrrrClientTest do
|
||||
use ExUnit.Case, async: true
|
||||
|
||||
alias Microwaveprop.Weather.HrrrClient
|
||||
|
||||
describe "nearest_hrrr_hour/1" do
|
||||
test "rounds down when within first 30 minutes" do
|
||||
assert HrrrClient.nearest_hrrr_hour(~U[2026-03-28 18:15:00Z]) ==
|
||||
~U[2026-03-28 18:00:00Z]
|
||||
end
|
||||
|
||||
test "rounds up when past 30 minutes" do
|
||||
assert HrrrClient.nearest_hrrr_hour(~U[2026-03-28 18:45:00Z]) ==
|
||||
~U[2026-03-28 19:00:00Z]
|
||||
end
|
||||
|
||||
test "stays same when exactly on the hour" do
|
||||
assert HrrrClient.nearest_hrrr_hour(~U[2026-03-28 18:00:00Z]) ==
|
||||
~U[2026-03-28 18:00:00Z]
|
||||
end
|
||||
|
||||
test "rounds at exactly 30 minutes" do
|
||||
result = HrrrClient.nearest_hrrr_hour(~U[2026-03-28 18:30:00Z])
|
||||
assert result in [~U[2026-03-28 18:00:00Z], ~U[2026-03-28 19:00:00Z]]
|
||||
end
|
||||
end
|
||||
|
||||
describe "hrrr_url/3" do
|
||||
test "builds surface product URL" do
|
||||
url = HrrrClient.hrrr_url(~D[2026-03-28], 18, :surface)
|
||||
|
||||
assert url ==
|
||||
"https://noaa-hrrr-bdp-pds.s3.amazonaws.com/hrrr.20260328/conus/hrrr.t18z.wrfsfcf00.grib2"
|
||||
end
|
||||
|
||||
test "builds pressure product URL" do
|
||||
url = HrrrClient.hrrr_url(~D[2026-03-28], 6, :pressure)
|
||||
|
||||
assert url ==
|
||||
"https://noaa-hrrr-bdp-pds.s3.amazonaws.com/hrrr.20260328/conus/hrrr.t06z.wrfprsf00.grib2"
|
||||
end
|
||||
|
||||
test "pads single-digit hours" do
|
||||
url = HrrrClient.hrrr_url(~D[2026-03-28], 3, :surface)
|
||||
assert String.contains?(url, "hrrr.t03z")
|
||||
end
|
||||
end
|
||||
|
||||
describe "parse_idx/1" do
|
||||
@sample_idx """
|
||||
1:0:d=2026032818:TMP:2 m above ground:anl:
|
||||
2:1234567:d=2026032818:DPT:2 m above ground:anl:
|
||||
3:2345678:d=2026032818:PRES:surface:anl:
|
||||
4:3456789:d=2026032818:HPBL:surface:anl:
|
||||
5:4567890:d=2026032818:PWAT:entire atmosphere (considered as a single layer):anl:
|
||||
6:5678901:d=2026032818:TMP:1000 mb:anl:
|
||||
"""
|
||||
|
||||
test "parses idx into structured list" do
|
||||
entries = HrrrClient.parse_idx(@sample_idx)
|
||||
|
||||
assert length(entries) == 6
|
||||
|
||||
first = hd(entries)
|
||||
assert first.msg == 1
|
||||
assert first.offset == 0
|
||||
assert first.var == "TMP"
|
||||
assert first.level == "2 m above ground"
|
||||
end
|
||||
|
||||
test "correctly extracts offsets" do
|
||||
entries = HrrrClient.parse_idx(@sample_idx)
|
||||
offsets = Enum.map(entries, & &1.offset)
|
||||
assert offsets == [0, 1_234_567, 2_345_678, 3_456_789, 4_567_890, 5_678_901]
|
||||
end
|
||||
|
||||
test "handles empty input" do
|
||||
assert HrrrClient.parse_idx("") == []
|
||||
end
|
||||
end
|
||||
|
||||
describe "byte_ranges_for_messages/2" do
|
||||
test "computes byte ranges from idx entries" do
|
||||
entries = [
|
||||
%{msg: 1, offset: 0, var: "TMP", level: "2 m above ground"},
|
||||
%{msg: 2, offset: 1000, var: "DPT", level: "2 m above ground"},
|
||||
%{msg: 3, offset: 2000, var: "PRES", level: "surface"},
|
||||
%{msg: 4, offset: 3000, var: "HPBL", level: "surface"}
|
||||
]
|
||||
|
||||
wanted = [
|
||||
%{var: "TMP", level: "2 m above ground"},
|
||||
%{var: "PRES", level: "surface"}
|
||||
]
|
||||
|
||||
ranges = HrrrClient.byte_ranges_for_messages(entries, wanted)
|
||||
|
||||
assert length(ranges) == 2
|
||||
assert {0, 999} in ranges
|
||||
assert {2000, 2999} in ranges
|
||||
end
|
||||
|
||||
test "handles last message in file" do
|
||||
entries = [
|
||||
%{msg: 1, offset: 0, var: "TMP", level: "2 m above ground"},
|
||||
%{msg: 2, offset: 1000, var: "DPT", level: "2 m above ground"}
|
||||
]
|
||||
|
||||
wanted = [%{var: "DPT", level: "2 m above ground"}]
|
||||
|
||||
ranges = HrrrClient.byte_ranges_for_messages(entries, wanted)
|
||||
|
||||
# Last message - uses a large end offset
|
||||
assert length(ranges) == 1
|
||||
[{start, _end}] = ranges
|
||||
assert start == 1000
|
||||
end
|
||||
end
|
||||
|
||||
describe "parse_wgrib2_output/1" do
|
||||
@sample_output """
|
||||
1:0:d=2026032818:TMP:2 m above ground:anl::lon=262.960000,lat=32.900000,val=298.5
|
||||
2:1234567:d=2026032818:DPT:2 m above ground:anl::lon=262.960000,lat=32.900000,val=291.2
|
||||
3:2345678:d=2026032818:PRES:surface:anl::lon=262.960000,lat=32.900000,val=101350
|
||||
4:3456789:d=2026032818:HPBL:surface:anl::lon=262.960000,lat=32.900000,val=1500
|
||||
5:4567890:d=2026032818:PWAT:entire atmosphere (considered as a single layer):anl::lon=262.960000,lat=32.900000,val=25.3
|
||||
"""
|
||||
|
||||
test "parses surface fields" do
|
||||
result = HrrrClient.parse_wgrib2_output(@sample_output)
|
||||
|
||||
assert result["TMP:2 m above ground"] == 298.5
|
||||
assert result["DPT:2 m above ground"] == 291.2
|
||||
assert result["PRES:surface"] == 101_350.0
|
||||
assert result["HPBL:surface"] == 1500.0
|
||||
end
|
||||
|
||||
test "handles pressure level data" do
|
||||
output = """
|
||||
1:0:d=2026032818:TMP:1000 mb:anl::lon=262.960000,lat=32.900000,val=297.3
|
||||
2:100:d=2026032818:TMP:975 mb:anl::lon=262.960000,lat=32.900000,val=295.1
|
||||
3:200:d=2026032818:HGT:1000 mb:anl::lon=262.960000,lat=32.900000,val=110
|
||||
"""
|
||||
|
||||
result = HrrrClient.parse_wgrib2_output(output)
|
||||
|
||||
assert result["TMP:1000 mb"] == 297.3
|
||||
assert result["TMP:975 mb"] == 295.1
|
||||
assert result["HGT:1000 mb"] == 110.0
|
||||
end
|
||||
|
||||
test "handles empty output" do
|
||||
assert HrrrClient.parse_wgrib2_output("") == %{}
|
||||
end
|
||||
end
|
||||
|
||||
describe "build_profile_from_wgrib2/1" do
|
||||
test "assembles profile from parsed wgrib2 data" do
|
||||
parsed = %{
|
||||
"TMP:1000 mb" => 298.0,
|
||||
"DPT:1000 mb" => 291.0,
|
||||
"HGT:1000 mb" => 110.0,
|
||||
"TMP:975 mb" => 296.0,
|
||||
"DPT:975 mb" => 289.0,
|
||||
"HGT:975 mb" => 350.0,
|
||||
"TMP:950 mb" => 294.0,
|
||||
"DPT:950 mb" => 287.0,
|
||||
"HGT:950 mb" => 590.0,
|
||||
"TMP:2 m above ground" => 299.0,
|
||||
"DPT:2 m above ground" => 292.0,
|
||||
"PRES:surface" => 101_350.0,
|
||||
"HPBL:surface" => 1500.0,
|
||||
"PWAT:entire atmosphere (considered as a single layer)" => 25.0
|
||||
}
|
||||
|
||||
result = HrrrClient.build_profile_from_wgrib2(parsed)
|
||||
|
||||
assert result.surface_temp_c == 299.0 - 273.15
|
||||
assert result.surface_dewpoint_c == 292.0 - 273.15
|
||||
assert_in_delta result.surface_pressure_mb, 1013.5, 0.1
|
||||
assert result.hpbl_m == 1500.0
|
||||
assert result.pwat_mm == 25.0
|
||||
assert length(result.profile) == 3
|
||||
|
||||
first_level = hd(result.profile)
|
||||
assert first_level["pres"] == 1000.0
|
||||
assert first_level["tmpc"] == 298.0 - 273.15
|
||||
assert first_level["dwpc"] == 291.0 - 273.15
|
||||
assert first_level["hght"] == 110.0
|
||||
end
|
||||
|
||||
test "skips pressure levels with missing data" do
|
||||
parsed = %{
|
||||
"TMP:1000 mb" => 298.0,
|
||||
"DPT:1000 mb" => 291.0,
|
||||
"HGT:1000 mb" => 110.0,
|
||||
# 975 mb missing HGT
|
||||
"TMP:975 mb" => 296.0,
|
||||
"DPT:975 mb" => 289.0,
|
||||
"TMP:2 m above ground" => 299.0,
|
||||
"DPT:2 m above ground" => 292.0,
|
||||
"PRES:surface" => 101_350.0,
|
||||
"HPBL:surface" => 1500.0,
|
||||
"PWAT:entire atmosphere (considered as a single layer)" => 25.0
|
||||
}
|
||||
|
||||
result = HrrrClient.build_profile_from_wgrib2(parsed)
|
||||
assert length(result.profile) == 1
|
||||
end
|
||||
end
|
||||
end
|
||||
84
test/microwaveprop/weather/hrrr_profile_test.exs
Normal file
84
test/microwaveprop/weather/hrrr_profile_test.exs
Normal file
|
|
@ -0,0 +1,84 @@
|
|||
defmodule Microwaveprop.Weather.HrrrProfileTest do
|
||||
use Microwaveprop.DataCase, async: true
|
||||
|
||||
alias Microwaveprop.Weather.HrrrProfile
|
||||
|
||||
@sample_profile [
|
||||
%{"pres" => 1000.0, "hght" => 110, "tmpc" => 25.0, "dwpc" => 18.0},
|
||||
%{"pres" => 975.0, "hght" => 350, "tmpc" => 23.0, "dwpc" => 16.0},
|
||||
%{"pres" => 950.0, "hght" => 590, "tmpc" => 21.0, "dwpc" => 14.0}
|
||||
]
|
||||
|
||||
@valid_attrs %{
|
||||
valid_time: ~U[2026-03-28 18:00:00Z],
|
||||
lat: 32.90,
|
||||
lon: -97.04,
|
||||
run_time: ~U[2026-03-28 18:00:00Z],
|
||||
profile: @sample_profile,
|
||||
hpbl_m: 1500.0,
|
||||
pwat_mm: 25.0,
|
||||
surface_temp_c: 25.0,
|
||||
surface_dewpoint_c: 18.0,
|
||||
surface_pressure_mb: 1013.0,
|
||||
surface_refractivity: 320.5,
|
||||
min_refractivity_gradient: -45.0,
|
||||
ducting_detected: false,
|
||||
duct_characteristics: nil
|
||||
}
|
||||
|
||||
describe "changeset/2" do
|
||||
test "valid attributes" do
|
||||
changeset = HrrrProfile.changeset(%HrrrProfile{}, @valid_attrs)
|
||||
assert changeset.valid?
|
||||
end
|
||||
|
||||
test "requires valid_time, lat, lon" do
|
||||
changeset = HrrrProfile.changeset(%HrrrProfile{}, %{})
|
||||
|
||||
assert %{
|
||||
valid_time: ["can't be blank"],
|
||||
lat: ["can't be blank"],
|
||||
lon: ["can't be blank"]
|
||||
} = errors_on(changeset)
|
||||
end
|
||||
|
||||
test "derived params are optional" do
|
||||
attrs = %{valid_time: ~U[2026-03-28 18:00:00Z], lat: 32.90, lon: -97.04}
|
||||
changeset = HrrrProfile.changeset(%HrrrProfile{}, attrs)
|
||||
assert changeset.valid?
|
||||
end
|
||||
|
||||
test "persists to the database with JSONB profile" do
|
||||
changeset = HrrrProfile.changeset(%HrrrProfile{}, @valid_attrs)
|
||||
assert {:ok, profile} = Repo.insert(changeset)
|
||||
assert length(profile.profile) == 3
|
||||
assert hd(profile.profile)["pres"] == 1000.0
|
||||
assert profile.ducting_detected == false
|
||||
assert profile.hpbl_m == 1500.0
|
||||
end
|
||||
|
||||
test "persists duct_characteristics as JSONB" do
|
||||
duct_chars = [%{"base" => 200, "top" => 500, "strength" => 5.2}]
|
||||
|
||||
attrs =
|
||||
@valid_attrs
|
||||
|> Map.put(:ducting_detected, true)
|
||||
|> Map.put(:duct_characteristics, duct_chars)
|
||||
|
||||
changeset = HrrrProfile.changeset(%HrrrProfile{}, attrs)
|
||||
assert {:ok, profile} = Repo.insert(changeset)
|
||||
assert profile.ducting_detected == true
|
||||
assert [%{"base" => 200, "top" => 500}] = profile.duct_characteristics
|
||||
end
|
||||
|
||||
test "enforces unique lat + lon + valid_time" do
|
||||
assert {:ok, _} =
|
||||
%HrrrProfile{} |> HrrrProfile.changeset(@valid_attrs) |> Repo.insert()
|
||||
|
||||
assert {:error, changeset} =
|
||||
%HrrrProfile{} |> HrrrProfile.changeset(@valid_attrs) |> Repo.insert()
|
||||
|
||||
assert %{lat: ["has already been taken"]} = errors_on(changeset)
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
@ -428,4 +428,95 @@ defmodule Microwaveprop.WeatherTest do
|
|||
assert record.sunspot_number == 22
|
||||
end
|
||||
end
|
||||
|
||||
@hrrr_profile_attrs %{
|
||||
valid_time: ~U[2026-03-28 18:00:00Z],
|
||||
lat: 32.90,
|
||||
lon: -97.04,
|
||||
run_time: ~U[2026-03-28 18:00:00Z],
|
||||
profile: [
|
||||
%{"pres" => 1000.0, "hght" => 110, "tmpc" => 25.0, "dwpc" => 18.0},
|
||||
%{"pres" => 975.0, "hght" => 350, "tmpc" => 23.0, "dwpc" => 16.0}
|
||||
],
|
||||
hpbl_m: 1500.0,
|
||||
pwat_mm: 25.0,
|
||||
surface_temp_c: 25.0,
|
||||
surface_dewpoint_c: 18.0,
|
||||
surface_pressure_mb: 1013.0,
|
||||
surface_refractivity: 320.5,
|
||||
min_refractivity_gradient: -45.0,
|
||||
ducting_detected: false
|
||||
}
|
||||
|
||||
describe "upsert_hrrr_profile/1" do
|
||||
test "inserts a new HRRR profile" do
|
||||
assert {:ok, profile} = Weather.upsert_hrrr_profile(@hrrr_profile_attrs)
|
||||
assert profile.lat == 32.90
|
||||
assert profile.hpbl_m == 1500.0
|
||||
end
|
||||
|
||||
test "updates existing profile on conflict" do
|
||||
{:ok, first} = Weather.upsert_hrrr_profile(@hrrr_profile_attrs)
|
||||
{:ok, second} = Weather.upsert_hrrr_profile(%{@hrrr_profile_attrs | hpbl_m: 2000.0})
|
||||
|
||||
assert first.id == second.id
|
||||
assert second.hpbl_m == 2000.0
|
||||
end
|
||||
end
|
||||
|
||||
describe "has_hrrr_profile?/3" do
|
||||
test "returns true when a profile exists at the given location and time" do
|
||||
Weather.upsert_hrrr_profile(@hrrr_profile_attrs)
|
||||
|
||||
assert Weather.has_hrrr_profile?(32.90, -97.04, ~U[2026-03-28 18:00:00Z])
|
||||
end
|
||||
|
||||
test "returns false when no profile exists" do
|
||||
refute Weather.has_hrrr_profile?(32.90, -97.04, ~U[2026-03-28 18:00:00Z])
|
||||
end
|
||||
|
||||
test "returns false for different location" do
|
||||
Weather.upsert_hrrr_profile(@hrrr_profile_attrs)
|
||||
|
||||
refute Weather.has_hrrr_profile?(40.0, -80.0, ~U[2026-03-28 18:00:00Z])
|
||||
end
|
||||
end
|
||||
|
||||
describe "hrrr_for_qso/1" do
|
||||
test "returns nearest HRRR profile for a QSO" do
|
||||
Weather.upsert_hrrr_profile(@hrrr_profile_attrs)
|
||||
|
||||
qso = %{
|
||||
pos1: %{"lat" => 32.91, "lon" => -97.05},
|
||||
qso_timestamp: ~U[2026-03-28 18:30:00Z]
|
||||
}
|
||||
|
||||
profile = Weather.hrrr_for_qso(qso)
|
||||
assert profile
|
||||
assert profile.lat == 32.90
|
||||
end
|
||||
|
||||
test "returns nil when no HRRR profiles exist nearby" do
|
||||
qso = %{
|
||||
pos1: %{"lat" => 40.0, "lon" => -80.0},
|
||||
qso_timestamp: ~U[2026-03-28 18:00:00Z]
|
||||
}
|
||||
|
||||
assert Weather.hrrr_for_qso(qso) == nil
|
||||
end
|
||||
|
||||
test "returns nil when QSO has no pos1" do
|
||||
assert Weather.hrrr_for_qso(%{pos1: nil, qso_timestamp: ~U[2026-03-28 18:00:00Z]}) == nil
|
||||
end
|
||||
end
|
||||
|
||||
describe "round_to_hrrr_grid/2" do
|
||||
test "rounds coordinates to 2 decimal places" do
|
||||
assert Weather.round_to_hrrr_grid(32.9047, -97.0382) == {32.90, -97.04}
|
||||
end
|
||||
|
||||
test "handles negative values correctly" do
|
||||
assert Weather.round_to_hrrr_grid(-33.456, -97.999) == {-33.46, -98.0}
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
106
test/microwaveprop/workers/hrrr_fetch_worker_test.exs
Normal file
106
test/microwaveprop/workers/hrrr_fetch_worker_test.exs
Normal file
|
|
@ -0,0 +1,106 @@
|
|||
defmodule Microwaveprop.Workers.HrrrFetchWorkerTest do
|
||||
use Microwaveprop.DataCase, async: true
|
||||
|
||||
alias Microwaveprop.Weather
|
||||
alias Microwaveprop.Weather.HrrrProfile
|
||||
alias Microwaveprop.Workers.HrrrFetchWorker
|
||||
|
||||
@sample_profile [
|
||||
%{"pres" => 1000.0, "hght" => 110.0, "tmpc" => 25.0, "dwpc" => 18.0},
|
||||
%{"pres" => 975.0, "hght" => 350.0, "tmpc" => 23.0, "dwpc" => 16.0},
|
||||
%{"pres" => 950.0, "hght" => 590.0, "tmpc" => 21.0, "dwpc" => 14.0}
|
||||
]
|
||||
|
||||
@sample_client_result %{
|
||||
surface_temp_c: 25.0,
|
||||
surface_dewpoint_c: 18.0,
|
||||
surface_pressure_mb: 1013.0,
|
||||
hpbl_m: 1500.0,
|
||||
pwat_mm: 25.0,
|
||||
run_time: ~U[2026-03-28 18:00:00Z],
|
||||
profile: @sample_profile
|
||||
}
|
||||
|
||||
describe "perform/1" do
|
||||
test "skips if HRRR profile already exists" do
|
||||
Weather.upsert_hrrr_profile(%{
|
||||
valid_time: ~U[2026-03-28 18:00:00Z],
|
||||
lat: 32.90,
|
||||
lon: -97.04,
|
||||
profile: @sample_profile
|
||||
})
|
||||
|
||||
job =
|
||||
HrrrFetchWorker.new(%{
|
||||
"lat" => 32.90,
|
||||
"lon" => -97.04,
|
||||
"valid_time" => "2026-03-28T18:00:00Z"
|
||||
})
|
||||
|
||||
assert :ok = HrrrFetchWorker.perform(%Oban.Job{args: job.changes.args})
|
||||
end
|
||||
|
||||
test "stores profile when client returns success" do
|
||||
# We test the perform logic by mocking at the module level
|
||||
# The actual HTTP calls are tested in HrrrClientTest
|
||||
# Here we verify the worker's coordination logic
|
||||
|
||||
lat = 32.90
|
||||
lon = -97.04
|
||||
valid_time = ~U[2026-03-28 18:00:00Z]
|
||||
|
||||
# Verify no profile exists initially
|
||||
refute Weather.has_hrrr_profile?(lat, lon, valid_time)
|
||||
|
||||
# Simulate what perform does after a successful fetch
|
||||
attrs = build_profile_attrs(lat, lon, valid_time, @sample_client_result)
|
||||
{:ok, profile} = Weather.upsert_hrrr_profile(attrs)
|
||||
|
||||
assert profile.lat == lat
|
||||
assert profile.lon == lon
|
||||
assert profile.hpbl_m == 1500.0
|
||||
assert profile.surface_temp_c == 25.0
|
||||
assert Weather.has_hrrr_profile?(lat, lon, valid_time)
|
||||
end
|
||||
|
||||
test "rounds lat/lon before checking existence" do
|
||||
# Profile at rounded coordinates
|
||||
Weather.upsert_hrrr_profile(%{
|
||||
valid_time: ~U[2026-03-28 18:00:00Z],
|
||||
lat: 32.90,
|
||||
lon: -97.04,
|
||||
profile: []
|
||||
})
|
||||
|
||||
# Worker called with slightly different coords that round to same
|
||||
job =
|
||||
HrrrFetchWorker.new(%{
|
||||
"lat" => 32.904,
|
||||
"lon" => -97.039,
|
||||
"valid_time" => "2026-03-28T18:00:00Z"
|
||||
})
|
||||
|
||||
# Should skip because rounded coords match
|
||||
assert :ok = HrrrFetchWorker.perform(%Oban.Job{args: job.changes.args})
|
||||
|
||||
# Only one profile should exist
|
||||
count = Repo.aggregate(HrrrProfile, :count)
|
||||
assert count == 1
|
||||
end
|
||||
end
|
||||
|
||||
defp build_profile_attrs(lat, lon, valid_time, client_result) do
|
||||
%{
|
||||
valid_time: valid_time,
|
||||
lat: lat,
|
||||
lon: lon,
|
||||
run_time: client_result.run_time,
|
||||
profile: client_result.profile,
|
||||
hpbl_m: client_result.hpbl_m,
|
||||
pwat_mm: client_result.pwat_mm,
|
||||
surface_temp_c: client_result.surface_temp_c,
|
||||
surface_dewpoint_c: client_result.surface_dewpoint_c,
|
||||
surface_pressure_mb: client_result.surface_pressure_mb
|
||||
}
|
||||
end
|
||||
end
|
||||
|
|
@ -168,5 +168,64 @@ defmodule Microwaveprop.Workers.QsoWeatherEnqueueWorkerTest do
|
|||
all_qsos = Repo.all(Qso)
|
||||
refute Enum.any?(all_qsos, & &1.weather_queued)
|
||||
end
|
||||
|
||||
test "enqueues HRRR jobs and marks QSOs as hrrr_queued" do
|
||||
qso = create_qso()
|
||||
assert qso.hrrr_queued == false
|
||||
|
||||
assert :ok = QsoWeatherEnqueueWorker.perform(%Oban.Job{args: %{}})
|
||||
|
||||
updated = Repo.get!(Qso, qso.id)
|
||||
assert updated.hrrr_queued == true
|
||||
end
|
||||
end
|
||||
|
||||
describe "build_hrrr_jobs/1" do
|
||||
test "builds one HRRR job per QSO with rounded lat/lon" do
|
||||
qso = create_qso(%{pos1: %{"lat" => 32.907, "lon" => -97.038}})
|
||||
|
||||
jobs = QsoWeatherEnqueueWorker.build_hrrr_jobs([qso])
|
||||
|
||||
assert length(jobs) == 1
|
||||
job = hd(jobs)
|
||||
assert job.changes.args["lat"] == 32.91
|
||||
assert job.changes.args["lon"] == -97.04
|
||||
end
|
||||
|
||||
test "rounds valid_time to nearest hour" do
|
||||
qso = create_qso(%{qso_timestamp: ~U[2026-03-28 18:45:00Z]})
|
||||
|
||||
jobs = QsoWeatherEnqueueWorker.build_hrrr_jobs([qso])
|
||||
|
||||
job = hd(jobs)
|
||||
assert job.changes.args["valid_time"] == "2026-03-28T19:00:00Z"
|
||||
end
|
||||
|
||||
test "deduplicates jobs with identical args" do
|
||||
q1 =
|
||||
create_qso(%{
|
||||
station1: "A1",
|
||||
qso_timestamp: ~U[2026-03-28 18:10:00Z],
|
||||
pos1: %{"lat" => 32.90, "lon" => -97.04}
|
||||
})
|
||||
|
||||
q2 =
|
||||
create_qso(%{
|
||||
station1: "A2",
|
||||
qso_timestamp: ~U[2026-03-28 18:20:00Z],
|
||||
pos1: %{"lat" => 32.90, "lon" => -97.04}
|
||||
})
|
||||
|
||||
jobs = QsoWeatherEnqueueWorker.build_hrrr_jobs([q1, q2])
|
||||
|
||||
# Both round to same hour and same location → 1 job
|
||||
assert length(jobs) == 1
|
||||
end
|
||||
|
||||
test "skips QSOs without pos1" do
|
||||
qso = create_qso(%{pos1: nil})
|
||||
|
||||
assert QsoWeatherEnqueueWorker.build_hrrr_jobs([qso]) == []
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue