chore: delete dead RTMA + METAR5 code paths
Both RTMA and METAR5 schemas + clients + workers were defined but never had a consumer. `find_nearest_rtma/3` and `recent_surface_obs/3` existed in `Microwaveprop.Weather` with zero callers cluster-wide; RtmaFetchWorker had test coverage but was never enqueued anywhere; no IEM 5-min ASOS fetcher was ever written. The tables sat empty in prod and the recalibrate audit was permanently flagging them BROKEN even though the empty state was the steady state. Drop the lot — schemas, clients, workers, queue config, Req.Test stubs, audit checks, and the per-table unique tests in the observation-changesets suite. New migration drops the now-unreferenced `rtma_observations` and `metar_5min_observations` tables (both 0 rows in prod). Trim the `:rtma` slot out of `backfill_only_queues` in runtime.exs so Oban Pro's Smart engine stops trying to manage a queue with no producer. Audit thresholds reset on the surviving NARR check: drop the "WARN < 5000 absolute rows" rule that didn't account for NarrFetchWorker's 0.13° spatial dedup, and point remediation at the existing BackfillEnqueueWorker cron instead of the dev-only `mix narr.backfill` task. scripts/recalibrate_algo.py: ROW_COUNT_SOURCES + DATA_GAP_SQL + DATA_GAP_RULES all stop referencing the dropped tables.
This commit is contained in:
parent
2f0424a47b
commit
1086f52c85
13 changed files with 74 additions and 893 deletions
|
|
@ -193,7 +193,7 @@ if config_env() == :prod do
|
|||
]
|
||||
|
||||
# Queues that are only allowed to run on the dedicated backfill pod.
|
||||
# Historical backfill (weather/narr/rtma/nexrad/contact_import) spins
|
||||
# Historical backfill (weather/narr/nexrad/contact_import) spins
|
||||
# against rate-limited upstreams (IEM especially) and stacks Req
|
||||
# exponential-backoff sleeps under its workers; running those slots
|
||||
# on the hot pods was thrashing the DB pool and tripping the /health
|
||||
|
|
@ -212,7 +212,6 @@ if config_env() == :prod do
|
|||
# but its workers/client are gone. See
|
||||
# docs/plans/2026-04-15-merra2-historical-backfill.md.
|
||||
narr: 6,
|
||||
rtma: 2,
|
||||
nexrad: 2,
|
||||
contact_import: 4
|
||||
]
|
||||
|
|
|
|||
|
|
@ -110,7 +110,6 @@ config :microwaveprop, narr_req_options: [plug: {Req.Test, Microwaveprop.Weather
|
|||
|
||||
# Route HTTP requests through Req.Test stubs
|
||||
config :microwaveprop, nexrad_req_options: [plug: {Req.Test, Microwaveprop.Weather.NexradClient}, retry: false]
|
||||
config :microwaveprop, rtma_req_options: [plug: {Req.Test, Microwaveprop.Weather.RtmaClient}, retry: false]
|
||||
config :microwaveprop, solar_req_options: [plug: {Req.Test, Microwaveprop.Weather.SolarClient}]
|
||||
config :microwaveprop, srtm_req_options: [plug: {Req.Test, Microwaveprop.Terrain.Srtm}, retry: false]
|
||||
config :microwaveprop, start_freshness_monitor: false
|
||||
|
|
|
|||
|
|
@ -15,9 +15,7 @@ defmodule Microwaveprop.Weather do
|
|||
alias Microwaveprop.Weather.HrrrProfile
|
||||
alias Microwaveprop.Weather.IemClient
|
||||
alias Microwaveprop.Weather.IemreObservation
|
||||
alias Microwaveprop.Weather.Metar5minObservation
|
||||
alias Microwaveprop.Weather.NarrProfile
|
||||
alias Microwaveprop.Weather.RtmaObservation
|
||||
alias Microwaveprop.Weather.ScalarFile
|
||||
alias Microwaveprop.Weather.SolarIndex
|
||||
alias Microwaveprop.Weather.Sounding
|
||||
|
|
@ -1802,36 +1800,6 @@ defmodule Microwaveprop.Weather do
|
|||
|> Repo.one()
|
||||
end
|
||||
|
||||
@spec find_nearest_rtma(float(), float(), DateTime.t()) :: RtmaObservation.t() | nil
|
||||
def find_nearest_rtma(lat, lon, timestamp) do
|
||||
dlat = 0.05
|
||||
dlon = 0.05
|
||||
time_start = DateTime.add(timestamp, -900, :second)
|
||||
time_end = DateTime.add(timestamp, 900, :second)
|
||||
|
||||
RtmaObservation
|
||||
|> where(
|
||||
[o],
|
||||
o.lat >= ^(lat - dlat) and o.lat <= ^(lat + dlat) and
|
||||
o.lon >= ^(lon - dlon) and o.lon <= ^(lon + dlon) and
|
||||
o.valid_time >= ^time_start and o.valid_time <= ^time_end
|
||||
)
|
||||
|> order_by([o],
|
||||
asc:
|
||||
fragment(
|
||||
"ABS(? - ?) + ABS(? - ?) + ABS(EXTRACT(EPOCH FROM ? - ?))",
|
||||
o.lat,
|
||||
^lat,
|
||||
o.lon,
|
||||
^lon,
|
||||
o.valid_time,
|
||||
^timestamp
|
||||
)
|
||||
)
|
||||
|> limit(1)
|
||||
|> Repo.one()
|
||||
end
|
||||
|
||||
@spec round_to_hrrr_grid(float(), float()) :: {float(), float()}
|
||||
def round_to_hrrr_grid(lat, lon) do
|
||||
{Float.round(lat / 1.0, 2), Float.round(lon / 1.0, 2)}
|
||||
|
|
@ -1941,51 +1909,4 @@ defmodule Microwaveprop.Weather do
|
|||
|> Enum.map(fn {lat, lon} -> find_nearest_iemre(lat, lon, contact.qso_timestamp) end)
|
||||
|> Enum.reject(&is_nil/1)
|
||||
end
|
||||
|
||||
@doc """
|
||||
Find the nearest surface observation to a given (lat, lon, time),
|
||||
preferring 5-minute METAR data when available, falling back to the
|
||||
hourly `surface_observations` table.
|
||||
|
||||
Returns a map with `:temp_f`, `:dewpoint_f`, `:wind_speed_kts`,
|
||||
`:observed_at`, etc. — the same shape regardless of which table the
|
||||
data came from. Returns `nil` if neither source has data.
|
||||
"""
|
||||
@spec recent_surface_obs(float(), float(), DateTime.t()) :: Metar5minObservation.t() | SurfaceObservation.t() | nil
|
||||
def recent_surface_obs(lat, lon, timestamp) do
|
||||
dlat = 0.5
|
||||
dlon = 0.5
|
||||
time_start = DateTime.add(timestamp, -1800, :second)
|
||||
time_end = DateTime.add(timestamp, 1800, :second)
|
||||
|
||||
station_ids =
|
||||
Station
|
||||
|> where(
|
||||
[s],
|
||||
s.lat >= ^(lat - dlat) and s.lat <= ^(lat + dlat) and
|
||||
s.lon >= ^(lon - dlon) and s.lon <= ^(lon + dlon)
|
||||
)
|
||||
|> select([s], s.id)
|
||||
|
||||
# Try 5-min first
|
||||
metar_5min =
|
||||
Metar5minObservation
|
||||
|> where([o], o.station_id in subquery(station_ids))
|
||||
|> where([o], o.observed_at >= ^time_start and o.observed_at <= ^time_end)
|
||||
|> order_by([o], asc: fragment("ABS(EXTRACT(EPOCH FROM ? - ?))", o.observed_at, ^timestamp))
|
||||
|> limit(1)
|
||||
|> Repo.one()
|
||||
|
||||
if metar_5min do
|
||||
metar_5min
|
||||
else
|
||||
# Fall back to hourly
|
||||
SurfaceObservation
|
||||
|> where([o], o.station_id in subquery(station_ids))
|
||||
|> where([o], o.observed_at >= ^time_start and o.observed_at <= ^time_end)
|
||||
|> order_by([o], asc: fragment("ABS(EXTRACT(EPOCH FROM ? - ?))", o.observed_at, ^timestamp))
|
||||
|> limit(1)
|
||||
|> Repo.one()
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -1,46 +0,0 @@
|
|||
defmodule Microwaveprop.Weather.Metar5minObservation do
|
||||
@moduledoc """
|
||||
5-minute ASOS/METAR observations from NCEI (C00418).
|
||||
|
||||
Schema-identical to `SurfaceObservation` but stored in a separate
|
||||
table to avoid mixing cadences and preserve backward compatibility
|
||||
for existing queries against the hourly feed.
|
||||
"""
|
||||
use Ecto.Schema
|
||||
|
||||
import Ecto.Changeset
|
||||
|
||||
@primary_key {:id, :binary_id, autogenerate: true}
|
||||
@foreign_key_type :binary_id
|
||||
|
||||
schema "metar_5min_observations" do
|
||||
belongs_to :station, Microwaveprop.Weather.Station
|
||||
field :observed_at, :utc_datetime
|
||||
field :temp_f, :float
|
||||
field :dewpoint_f, :float
|
||||
field :relative_humidity, :float
|
||||
field :wind_speed_kts, :float
|
||||
field :wind_direction_deg, :integer
|
||||
field :sea_level_pressure_mb, :float
|
||||
field :altimeter_setting, :float
|
||||
field :sky_condition, :string
|
||||
field :precip_1h_in, :float
|
||||
field :wx_codes, :string
|
||||
|
||||
timestamps(type: :utc_datetime)
|
||||
end
|
||||
|
||||
@type t :: %__MODULE__{}
|
||||
|
||||
@required_fields ~w(station_id observed_at)a
|
||||
@optional_fields ~w(temp_f dewpoint_f relative_humidity wind_speed_kts wind_direction_deg sea_level_pressure_mb altimeter_setting sky_condition precip_1h_in wx_codes)a
|
||||
|
||||
@spec changeset(t() | Ecto.Changeset.t(), map()) :: Ecto.Changeset.t()
|
||||
def changeset(observation, attrs) do
|
||||
observation
|
||||
|> cast(attrs, @required_fields ++ @optional_fields)
|
||||
|> validate_required(@required_fields)
|
||||
|> foreign_key_constraint(:station_id)
|
||||
|> unique_constraint([:station_id, :observed_at])
|
||||
end
|
||||
end
|
||||
|
|
@ -1,180 +0,0 @@
|
|||
defmodule Microwaveprop.Weather.RtmaClient do
|
||||
@moduledoc """
|
||||
Client for NOAA RTMA (Real-Time Mesoscale Analysis) data.
|
||||
2.5 km resolution, 15-minute analysis cycles, CONUS coverage.
|
||||
|
||||
Available on AWS S3 at s3://noaa-rtma-pds/. GRIB2 format with
|
||||
byte-range requests, same pattern as HRRR.
|
||||
|
||||
RTMA provides surface fields only (no pressure levels):
|
||||
- 2m temperature, 2m dewpoint
|
||||
- 10m wind U/V
|
||||
- Surface pressure
|
||||
- Visibility
|
||||
- Precipitation analysis
|
||||
"""
|
||||
|
||||
alias Microwaveprop.Weather.Grib2.Extractor
|
||||
|
||||
require Logger
|
||||
|
||||
@s3_base "https://noaa-rtma-pds.s3.amazonaws.com"
|
||||
|
||||
@wanted_fields [
|
||||
"TMP:2 m above ground",
|
||||
"DPT:2 m above ground",
|
||||
"PRES:surface",
|
||||
"UGRD:10 m above ground",
|
||||
"VGRD:10 m above ground",
|
||||
"VIS:surface"
|
||||
]
|
||||
|
||||
@doc """
|
||||
Fetch RTMA observation for a point at a specific time.
|
||||
Returns {:ok, attrs} or {:error, reason}.
|
||||
"""
|
||||
@spec fetch_observation(float(), float(), DateTime.t()) :: {:ok, map()} | {:error, term()}
|
||||
def fetch_observation(lat, lon, timestamp) do
|
||||
Microwaveprop.Instrument.span(
|
||||
[:rtma, :fetch_observation],
|
||||
%{},
|
||||
fn -> do_fetch_observation(lat, lon, timestamp) end
|
||||
)
|
||||
end
|
||||
|
||||
defp do_fetch_observation(lat, lon, timestamp) do
|
||||
# RTMA runs every hour with 15-min updates; round to nearest hour
|
||||
valid_time = %{DateTime.truncate(timestamp, :second) | minute: 0, second: 0}
|
||||
rlat = Float.round(lat * 40) / 40
|
||||
rlon = Float.round(lon * 40) / 40
|
||||
|
||||
url = rtma_url(valid_time)
|
||||
idx_url = "#{url}.idx"
|
||||
|
||||
with {:ok, idx_body} <- fetch_idx(idx_url),
|
||||
ranges = byte_ranges_for_fields(idx_body, @wanted_fields),
|
||||
{:ok, grib_data} <- download_ranges(url, ranges),
|
||||
{:ok, fields} <- extract_point(grib_data, rlat, rlon) do
|
||||
attrs = %{
|
||||
valid_time: valid_time,
|
||||
lat: rlat,
|
||||
lon: rlon,
|
||||
temp_c: kelvin_to_celsius(fields["TMP:2 m above ground"]),
|
||||
dewpoint_c: kelvin_to_celsius(fields["DPT:2 m above ground"]),
|
||||
pressure_mb: pa_to_mb(fields["PRES:surface"]),
|
||||
wind_u_ms: fields["UGRD:10 m above ground"],
|
||||
wind_v_ms: fields["VGRD:10 m above ground"],
|
||||
visibility_m: fields["VIS:surface"]
|
||||
}
|
||||
|
||||
{:ok, attrs}
|
||||
end
|
||||
end
|
||||
|
||||
@doc "Build S3 URL for an RTMA analysis file."
|
||||
@spec rtma_url(DateTime.t()) :: String.t()
|
||||
def rtma_url(valid_time) do
|
||||
date_str = Calendar.strftime(valid_time, "%Y%m%d")
|
||||
hour_str = valid_time.hour |> Integer.to_string() |> String.pad_leading(2, "0")
|
||||
"#{@s3_base}/rtma2p5.#{date_str}/rtma2p5.t#{hour_str}z.2dvaranl_ndfd.grb2_wexp"
|
||||
end
|
||||
|
||||
defp fetch_idx(url) do
|
||||
case Req.get(url, [receive_timeout: 15_000] ++ req_options()) do
|
||||
{:ok, %{status: 200, body: body}} -> {:ok, body}
|
||||
{:ok, %{status: status}} -> {:error, "RTMA idx HTTP #{status}"}
|
||||
{:error, reason} -> {:error, "RTMA idx failed: #{inspect(reason)}"}
|
||||
end
|
||||
end
|
||||
|
||||
defp byte_ranges_for_fields(idx_body, wanted_fields) do
|
||||
lines =
|
||||
idx_body
|
||||
|> String.split("\n", trim: true)
|
||||
|> Enum.map(fn line ->
|
||||
parts = String.split(line, ":")
|
||||
%{offset: String.to_integer(Enum.at(parts, 1, "0")), field: Enum.at(parts, 3, "") <> ":" <> Enum.at(parts, 4, "")}
|
||||
end)
|
||||
|
||||
offsets = Enum.map(lines, & &1.offset)
|
||||
|
||||
lines
|
||||
|> Enum.with_index()
|
||||
|> Enum.filter(fn {line, _i} ->
|
||||
Enum.any?(wanted_fields, &String.contains?(line.field, &1))
|
||||
end)
|
||||
|> Enum.map(fn {line, i} ->
|
||||
next_offset = Enum.at(offsets, i + 1, line.offset + 5_000_000)
|
||||
{line.offset, next_offset - 1}
|
||||
end)
|
||||
|> merge_ranges()
|
||||
end
|
||||
|
||||
defp merge_ranges(ranges) do
|
||||
ranges
|
||||
|> Enum.sort()
|
||||
|> Enum.reduce([], fn
|
||||
range, [] -> [range]
|
||||
{s2, e2}, [{s1, e1} | rest] when s2 <= e1 + 1 -> [{s1, max(e1, e2)} | rest]
|
||||
range, acc -> [range | acc]
|
||||
end)
|
||||
|> Enum.reverse()
|
||||
end
|
||||
|
||||
defp download_ranges(url, ranges) do
|
||||
data =
|
||||
ranges
|
||||
|> Task.async_stream(
|
||||
fn {range_start, range_end} ->
|
||||
Req.get(
|
||||
url,
|
||||
[
|
||||
headers: [{"Range", "bytes=#{range_start}-#{range_end}"}],
|
||||
receive_timeout: 30_000
|
||||
] ++ req_options()
|
||||
)
|
||||
end,
|
||||
max_concurrency: 4,
|
||||
timeout: 60_000
|
||||
)
|
||||
|> Enum.reduce({:ok, []}, fn
|
||||
{:ok, {:ok, %{status: status, body: chunk}}}, {:ok, acc} when status in [200, 206] ->
|
||||
{:ok, [chunk | acc]}
|
||||
|
||||
{:ok, {:ok, %{status: status}}}, _acc ->
|
||||
{:error, "RTMA download HTTP #{status}"}
|
||||
|
||||
{:ok, {:error, reason}}, _acc ->
|
||||
{:error, "RTMA download failed: #{inspect(reason)}"}
|
||||
|
||||
{:exit, reason}, _acc ->
|
||||
Logger.error("RtmaClient.download_ranges task crashed: url=#{url} reason=#{inspect(reason)}")
|
||||
{:error, "RTMA download crashed: #{inspect(reason)}"}
|
||||
|
||||
other, acc ->
|
||||
Logger.warning("RtmaClient.download_ranges unexpected stream entry: #{inspect(other)}")
|
||||
acc
|
||||
end)
|
||||
|
||||
case data do
|
||||
{:ok, chunks} -> {:ok, chunks |> Enum.reverse() |> IO.iodata_to_binary()}
|
||||
error -> error
|
||||
end
|
||||
end
|
||||
|
||||
defp extract_point(grib_data, lat, lon) do
|
||||
case Extractor.extract_points(grib_data, lat, lon) do
|
||||
{:ok, fields} when map_size(fields) > 0 -> {:ok, fields}
|
||||
{:ok, _} -> {:error, "RTMA: no data at #{lat},#{lon}"}
|
||||
{:error, reason} -> {:error, reason}
|
||||
end
|
||||
end
|
||||
|
||||
defp kelvin_to_celsius(nil), do: nil
|
||||
defp kelvin_to_celsius(k), do: k - 273.15
|
||||
|
||||
defp pa_to_mb(nil), do: nil
|
||||
defp pa_to_mb(pa), do: pa / 100.0
|
||||
|
||||
defp req_options, do: Application.get_env(:microwaveprop, :rtma_req_options, [])
|
||||
end
|
||||
|
|
@ -1,37 +0,0 @@
|
|||
defmodule Microwaveprop.Weather.RtmaObservation do
|
||||
@moduledoc false
|
||||
use Ecto.Schema
|
||||
|
||||
import Ecto.Changeset
|
||||
|
||||
@primary_key {:id, :binary_id, autogenerate: true}
|
||||
@foreign_key_type :binary_id
|
||||
|
||||
schema "rtma_observations" do
|
||||
field :valid_time, :utc_datetime
|
||||
field :lat, :float
|
||||
field :lon, :float
|
||||
field :temp_c, :float
|
||||
field :dewpoint_c, :float
|
||||
field :pressure_mb, :float
|
||||
field :wind_u_ms, :float
|
||||
field :wind_v_ms, :float
|
||||
field :visibility_m, :float
|
||||
field :precip_mm, :float
|
||||
|
||||
timestamps(type: :utc_datetime)
|
||||
end
|
||||
|
||||
@type t :: %__MODULE__{}
|
||||
|
||||
@required_fields ~w(valid_time lat lon)a
|
||||
@optional_fields ~w(temp_c dewpoint_c pressure_mb wind_u_ms wind_v_ms visibility_m precip_mm)a
|
||||
|
||||
@spec changeset(t() | Ecto.Changeset.t(), map()) :: Ecto.Changeset.t()
|
||||
def changeset(observation, attrs) do
|
||||
observation
|
||||
|> cast(attrs, @required_fields ++ @optional_fields)
|
||||
|> validate_required(@required_fields)
|
||||
|> unique_constraint([:lat, :lon, :valid_time])
|
||||
end
|
||||
end
|
||||
|
|
@ -1,71 +0,0 @@
|
|||
defmodule Microwaveprop.Workers.RtmaFetchWorker do
|
||||
@moduledoc """
|
||||
Fetches RTMA surface observations for real-time propagation scoring.
|
||||
15-minute resolution provides finer temporal detail than HRRR's hourly cycle.
|
||||
"""
|
||||
use Oban.Worker,
|
||||
queue: :rtma,
|
||||
max_attempts: 10,
|
||||
unique: [period: 300, fields: [:args], states: [:scheduled, :available]]
|
||||
|
||||
alias Microwaveprop.Repo
|
||||
alias Microwaveprop.Weather.RtmaClient
|
||||
alias Microwaveprop.Weather.RtmaObservation
|
||||
|
||||
require Logger
|
||||
|
||||
@impl Oban.Worker
|
||||
def backoff(%Oban.Job{attempt: attempt}) do
|
||||
min(60 * Integer.pow(2, attempt - 1), _six_hours = 21_600)
|
||||
end
|
||||
|
||||
@impl Oban.Worker
|
||||
def perform(%Oban.Job{args: %{"lat" => lat, "lon" => lon, "valid_time" => valid_time_str}}) do
|
||||
{:ok, valid_time, _} = DateTime.from_iso8601(valid_time_str)
|
||||
|
||||
rlat = Float.round(lat * 40) / 40
|
||||
rlon = Float.round(lon * 40) / 40
|
||||
|
||||
if has_rtma_observation?(rlat, rlon, valid_time) do
|
||||
Logger.debug("RTMA: observation exists for #{rlat},#{rlon} @ #{valid_time_str}")
|
||||
:ok
|
||||
else
|
||||
Logger.info("RTMA: fetching for #{rlat},#{rlon} @ #{valid_time_str}")
|
||||
|
||||
case RtmaClient.fetch_observation(lat, lon, valid_time) do
|
||||
{:ok, attrs} ->
|
||||
%RtmaObservation{}
|
||||
|> RtmaObservation.changeset(attrs)
|
||||
|> Repo.insert(
|
||||
on_conflict: :nothing,
|
||||
conflict_target: [:lat, :lon, :valid_time]
|
||||
)
|
||||
|
||||
Logger.info("RTMA: stored observation for #{rlat},#{rlon} @ #{valid_time_str}")
|
||||
:ok
|
||||
|
||||
{:error, reason} ->
|
||||
Logger.warning("RTMA: failed for #{rlat},#{rlon} @ #{valid_time_str}: #{inspect(reason)}")
|
||||
{:error, reason}
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
defp has_rtma_observation?(lat, lon, valid_time) do
|
||||
import Ecto.Query
|
||||
|
||||
dlat = 0.03
|
||||
dlon = 0.03
|
||||
time_start = DateTime.add(valid_time, -450, :second)
|
||||
time_end = DateTime.add(valid_time, 450, :second)
|
||||
|
||||
RtmaObservation
|
||||
|> where(
|
||||
[o],
|
||||
o.lat >= ^(lat - dlat) and o.lat <= ^(lat + dlat) and
|
||||
o.lon >= ^(lon - dlon) and o.lon <= ^(lon + dlon) and
|
||||
o.valid_time >= ^time_start and o.valid_time <= ^time_end
|
||||
)
|
||||
|> Repo.exists?()
|
||||
end
|
||||
end
|
||||
|
|
@ -0,0 +1,58 @@
|
|||
defmodule Microwaveprop.Repo.Migrations.DropUnusedRtmaMetar5Tables do
|
||||
use Ecto.Migration
|
||||
|
||||
# Both tables landed via earlier "planned multi-source atmospheric
|
||||
# data" work that never wired into scoring or enrichment. RtmaClient
|
||||
# / RtmaFetchWorker / Metar5minObservation existed but had zero
|
||||
# callers cluster-wide; recalibrate's audit kept flagging them as
|
||||
# BROKEN even though the empty state was the steady state. Drop the
|
||||
# tables along with the unused code so a future contributor can't
|
||||
# read them as "real but stalled" data sources.
|
||||
|
||||
def up do
|
||||
drop_if_exists table(:rtma_observations)
|
||||
drop_if_exists table(:metar_5min_observations)
|
||||
end
|
||||
|
||||
def down do
|
||||
create table(:rtma_observations, 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 :temp_c, :float
|
||||
add :dewpoint_c, :float
|
||||
add :pressure_mb, :float
|
||||
add :wind_u_ms, :float
|
||||
add :wind_v_ms, :float
|
||||
add :visibility_m, :float
|
||||
add :precip_mm, :float
|
||||
|
||||
timestamps(type: :utc_datetime)
|
||||
end
|
||||
|
||||
create unique_index(:rtma_observations, [:lat, :lon, :valid_time])
|
||||
create index(:rtma_observations, [:valid_time])
|
||||
|
||||
create table(:metar_5min_observations, primary_key: false) do
|
||||
add :id, :binary_id, primary_key: true, null: false
|
||||
add :station_id, references(:weather_stations, type: :binary_id, on_delete: :delete_all)
|
||||
add :observed_at, :utc_datetime, null: false
|
||||
add :temp_f, :float
|
||||
add :dewpoint_f, :float
|
||||
add :relative_humidity, :float
|
||||
add :wind_speed_kts, :float
|
||||
add :wind_direction_deg, :integer
|
||||
add :sea_level_pressure_mb, :float
|
||||
add :altimeter_setting, :float
|
||||
add :sky_condition, :string
|
||||
add :precip_1h_in, :float
|
||||
add :wx_codes, :string
|
||||
|
||||
timestamps(type: :utc_datetime)
|
||||
end
|
||||
|
||||
create unique_index(:metar_5min_observations, [:station_id, :observed_at])
|
||||
create index(:metar_5min_observations, [:observed_at])
|
||||
end
|
||||
end
|
||||
|
|
@ -82,7 +82,6 @@ ROW_COUNT_SOURCES = [
|
|||
("narr_profiles", "valid_time"),
|
||||
("iemre_observations", None),
|
||||
("nexrad_observations", "observed_at"),
|
||||
("rtma_observations", "valid_time"),
|
||||
("terrain_profiles", None),
|
||||
("propagation_scores", "valid_time"),
|
||||
]
|
||||
|
|
@ -278,8 +277,6 @@ SELECT
|
|||
(SELECT count(*) FROM narr_profiles) AS narr_rows,
|
||||
(SELECT count(*) FROM narr_profiles WHERE valid_time < '2014-10-02') AS narr_pre2014_rows,
|
||||
(SELECT count(*) FROM hrrr_climatology) AS hrrr_climatology_rows,
|
||||
(SELECT count(*) FROM rtma_observations) AS rtma_rows,
|
||||
(SELECT count(*) FROM metar_5min_observations) AS metar5_rows,
|
||||
(SELECT count(*) FROM contacts WHERE hrrr_status = 'complete') AS hrrr_complete_contacts,
|
||||
(SELECT count(*) FROM contacts WHERE hrrr_status != 'complete') AS hrrr_pending_contacts,
|
||||
(SELECT count(*) FROM contacts
|
||||
|
|
@ -324,19 +321,28 @@ GROUP BY 1 ORDER BY 1;
|
|||
DATA_GAP_RULES: list[dict] = [
|
||||
{
|
||||
"key": "narr_rows",
|
||||
# NARR feeds pre-2014 enrichment via BackfillEnqueueWorker's
|
||||
# 30-min cron. NarrFetchWorker dedupes by 0.13° grid, so the
|
||||
# equilibrium count is ~3 path points × pre2014_contacts after
|
||||
# spatial collapse — not the contact count itself. Threshold
|
||||
# follows: BROKEN at zero, WARN below half the expected
|
||||
# post-dedup ceiling.
|
||||
"broken_when": lambda v: v == 0,
|
||||
# 358 rows is the symptom that triggered this rewrite — anything
|
||||
# under ~5k means the NARR backfill hasn't meaningfully started.
|
||||
"warn_when": lambda v: v < 5_000,
|
||||
"remediation": "Run `mix narr.backfill` (see lib/microwaveprop/workers/narr_fetch_worker.ex).",
|
||||
"warn_when": lambda _v: False,
|
||||
"remediation": (
|
||||
"BackfillEnqueueWorker dispatches narr jobs every 30 min; "
|
||||
"if rows aren't climbing, check NarrFetchWorker for upstream "
|
||||
"NCEI errors in the Oban dashboard."
|
||||
),
|
||||
},
|
||||
{
|
||||
"key": "narr_pre2014_rows",
|
||||
"broken_when": lambda v: v == 0,
|
||||
"warn_when": lambda v: v < 5_000,
|
||||
"warn_when": lambda _v: False,
|
||||
"remediation": (
|
||||
"NARR's purpose is filling the pre-2014-10 gap; without rows here "
|
||||
"the historical-sanity correlation table is empty."
|
||||
"the historical-sanity correlation table is empty. Same self-heal "
|
||||
"as narr_rows."
|
||||
),
|
||||
},
|
||||
{
|
||||
|
|
@ -349,25 +355,6 @@ DATA_GAP_RULES: list[dict] = [
|
|||
"day, check Oban dashboard for failed admin jobs."
|
||||
),
|
||||
},
|
||||
{
|
||||
"key": "rtma_rows",
|
||||
"broken_when": lambda v: v == 0,
|
||||
"warn_when": lambda _v: False,
|
||||
"remediation": (
|
||||
"Enable the rtma queue and let RtmaFetchWorker run — needed for "
|
||||
"contacts whose HRRR enrichment misses the ±1 h window."
|
||||
),
|
||||
},
|
||||
{
|
||||
"key": "metar5_rows",
|
||||
"broken_when": lambda v: v == 0,
|
||||
"warn_when": lambda _v: False,
|
||||
"remediation": (
|
||||
"No 5-minute METAR ingestor exists yet; if this stays at 0 the "
|
||||
"schema metar_5min_observations is unused — consider dropping it "
|
||||
"or wiring an IEM 5-min ASOS fetcher."
|
||||
),
|
||||
},
|
||||
{
|
||||
"key": "hrrr_complete_contacts",
|
||||
# Healthy is the fraction complete, not the absolute number — handled
|
||||
|
|
|
|||
|
|
@ -13,7 +13,6 @@ defmodule Microwaveprop.SpaceWeather.ObservationChangesetsTest do
|
|||
alias Microwaveprop.SpaceWeather.SolarFluxObservation
|
||||
alias Microwaveprop.SpaceWeather.SolarXrayObservation
|
||||
alias Microwaveprop.Weather.HrrrClimatology
|
||||
alias Microwaveprop.Weather.Metar5minObservation
|
||||
|
||||
describe "GeomagneticObservation.changeset/2" do
|
||||
test "valid with only valid_time" do
|
||||
|
|
@ -175,52 +174,6 @@ defmodule Microwaveprop.SpaceWeather.ObservationChangesetsTest do
|
|||
end
|
||||
end
|
||||
|
||||
describe "Metar5minObservation.changeset/2" do
|
||||
@station_id Ecto.UUID.generate()
|
||||
|
||||
test "valid with station_id + observed_at" do
|
||||
cs =
|
||||
Metar5minObservation.changeset(%Metar5minObservation{}, %{
|
||||
station_id: @station_id,
|
||||
observed_at: ~U[2026-04-21 00:00:00Z]
|
||||
})
|
||||
|
||||
assert cs.valid?
|
||||
end
|
||||
|
||||
test "casts every observation field" do
|
||||
cs =
|
||||
Metar5minObservation.changeset(%Metar5minObservation{}, %{
|
||||
station_id: @station_id,
|
||||
observed_at: ~U[2026-04-21 00:05:00Z],
|
||||
temp_f: 72.5,
|
||||
dewpoint_f: 55.0,
|
||||
relative_humidity: 55.0,
|
||||
wind_speed_kts: 8.0,
|
||||
wind_direction_deg: 180,
|
||||
sea_level_pressure_mb: 1013.2,
|
||||
altimeter_setting: 29.92,
|
||||
sky_condition: "SCT030",
|
||||
precip_1h_in: 0.0,
|
||||
wx_codes: "VCTS"
|
||||
})
|
||||
|
||||
assert cs.valid?
|
||||
end
|
||||
|
||||
test "invalid without station_id" do
|
||||
cs = Metar5minObservation.changeset(%Metar5minObservation{}, %{observed_at: ~U[2026-04-21 00:00:00Z]})
|
||||
refute cs.valid?
|
||||
assert errors_on(cs)[:station_id] == ["can't be blank"]
|
||||
end
|
||||
|
||||
test "invalid without observed_at" do
|
||||
cs = Metar5minObservation.changeset(%Metar5minObservation{}, %{station_id: @station_id})
|
||||
refute cs.valid?
|
||||
assert errors_on(cs)[:observed_at] == ["can't be blank"]
|
||||
end
|
||||
end
|
||||
|
||||
# Local copy of the standard Phoenix.DataCase `errors_on/1` so this
|
||||
# file can stay on the lightweight `ExUnit.Case` (no DB sandbox
|
||||
# needed for changeset tests).
|
||||
|
|
|
|||
|
|
@ -1,265 +0,0 @@
|
|||
defmodule Microwaveprop.Weather.RtmaClientTest do
|
||||
use ExUnit.Case, async: true
|
||||
use ExUnitProperties
|
||||
|
||||
alias Microwaveprop.Weather.RtmaClient
|
||||
|
||||
describe "rtma_url/1" do
|
||||
test "builds the S3 URL for an RTMA analysis time" do
|
||||
vt = ~U[2026-04-15 18:00:00Z]
|
||||
|
||||
assert RtmaClient.rtma_url(vt) ==
|
||||
"https://noaa-rtma-pds.s3.amazonaws.com/rtma2p5.20260415/rtma2p5.t18z.2dvaranl_ndfd.grb2_wexp"
|
||||
end
|
||||
|
||||
test "zero-pads single-digit hours" do
|
||||
vt = ~U[2026-04-15 03:00:00Z]
|
||||
|
||||
assert RtmaClient.rtma_url(vt) ==
|
||||
"https://noaa-rtma-pds.s3.amazonaws.com/rtma2p5.20260415/rtma2p5.t03z.2dvaranl_ndfd.grb2_wexp"
|
||||
end
|
||||
|
||||
test "handles hour zero" do
|
||||
vt = ~U[2026-04-15 00:00:00Z]
|
||||
|
||||
assert RtmaClient.rtma_url(vt) ==
|
||||
"https://noaa-rtma-pds.s3.amazonaws.com/rtma2p5.20260415/rtma2p5.t00z.2dvaranl_ndfd.grb2_wexp"
|
||||
end
|
||||
end
|
||||
|
||||
describe "fetch_observation/3 — idx layer" do
|
||||
test "returns {:error, _} when the idx fetch returns 404 (not yet published)" do
|
||||
Req.Test.stub(RtmaClient, fn conn ->
|
||||
if String.ends_with?(conn.request_path, ".idx") do
|
||||
Plug.Conn.send_resp(conn, 404, "not found")
|
||||
else
|
||||
Plug.Conn.send_resp(conn, 500, "idx should have short-circuited")
|
||||
end
|
||||
end)
|
||||
|
||||
assert {:error, reason} =
|
||||
RtmaClient.fetch_observation(32.9, -97.0, ~U[2026-04-15 18:00:00Z])
|
||||
|
||||
assert reason =~ "RTMA idx HTTP 404"
|
||||
end
|
||||
|
||||
test "returns {:error, _} when the idx fetch returns a non-200 status" do
|
||||
Req.Test.stub(RtmaClient, fn conn ->
|
||||
if String.ends_with?(conn.request_path, ".idx") do
|
||||
Plug.Conn.send_resp(conn, 503, "unavailable")
|
||||
else
|
||||
Plug.Conn.send_resp(conn, 500, "unexpected")
|
||||
end
|
||||
end)
|
||||
|
||||
assert {:error, reason} =
|
||||
RtmaClient.fetch_observation(32.9, -97.0, ~U[2026-04-15 18:00:00Z])
|
||||
|
||||
assert reason =~ "RTMA idx HTTP 503"
|
||||
end
|
||||
end
|
||||
|
||||
describe "fetch_observation/3 — hour truncation" do
|
||||
test "truncates minutes and seconds to the analysis hour in the request URL" do
|
||||
test_pid = self()
|
||||
|
||||
Req.Test.stub(RtmaClient, fn conn ->
|
||||
send(test_pid, {:request_path, conn.request_path, conn.method})
|
||||
# Return an empty idx so downstream stages short-circuit on 0 ranges.
|
||||
Plug.Conn.send_resp(conn, 200, "")
|
||||
end)
|
||||
|
||||
# 18:47:33 must request the 18z (not 19z) analysis file — truncation
|
||||
# is load-bearing for matching the stored row's hour-aligned valid_time.
|
||||
assert {:error, _} =
|
||||
RtmaClient.fetch_observation(32.9, -97.0, ~U[2026-04-15 18:47:33Z])
|
||||
|
||||
assert_received {:request_path, path, "GET"}
|
||||
assert String.contains?(path, "rtma2p5.t18z")
|
||||
refute String.contains?(path, "rtma2p5.t19z")
|
||||
end
|
||||
|
||||
test "already-aligned hour passes through unchanged" do
|
||||
test_pid = self()
|
||||
|
||||
Req.Test.stub(RtmaClient, fn conn ->
|
||||
send(test_pid, {:request_path, conn.request_path})
|
||||
Plug.Conn.send_resp(conn, 200, "")
|
||||
end)
|
||||
|
||||
assert {:error, _} =
|
||||
RtmaClient.fetch_observation(32.9, -97.0, ~U[2026-04-15 12:00:00Z])
|
||||
|
||||
assert_received {:request_path, path}
|
||||
assert String.contains?(path, "rtma2p5.t12z")
|
||||
end
|
||||
end
|
||||
|
||||
describe "fetch_observation/3 — extract layer" do
|
||||
# With an empty (or non-matching) idx body, `byte_ranges_for_fields/2`
|
||||
# produces zero ranges, so `download_ranges/2` does no HTTP and returns
|
||||
# {:ok, ""}, which the GRIB2 Extractor splits into an empty message
|
||||
# list → {:ok, %{}} → {:error, "RTMA: no data..."} path.
|
||||
test "returns {:error, 'RTMA: no data ...'} when the extractor yields no fields" do
|
||||
Req.Test.stub(RtmaClient, fn conn ->
|
||||
if String.ends_with?(conn.request_path, ".idx") do
|
||||
# idx contains no wanted fields — produces zero byte-ranges.
|
||||
Plug.Conn.send_resp(conn, 200, "1:0:d=2026041518:FOOBAR:nothing:anl:\n")
|
||||
else
|
||||
# Should not be reached because no ranges means no GET.
|
||||
Plug.Conn.send_resp(conn, 500, "unexpected GRB request")
|
||||
end
|
||||
end)
|
||||
|
||||
assert {:error, reason} =
|
||||
RtmaClient.fetch_observation(32.9, -97.0, ~U[2026-04-15 18:00:00Z])
|
||||
|
||||
assert reason =~ "RTMA: no data"
|
||||
# rlat/rlon are reported to 2.5 km precision (1/40°).
|
||||
assert reason =~ "32.9"
|
||||
assert reason =~ "-97.0"
|
||||
end
|
||||
|
||||
test "returns {:error, 'RTMA: no data ...'} when grib bytes are non-GRIB garbage" do
|
||||
# Serve an idx that DOES match wanted fields so byte-ranges are non-empty,
|
||||
# then serve junk bytes on the ranged GET. The GRIB2 Extractor's
|
||||
# split_messages/1 skips unknown bytes and returns an empty message list,
|
||||
# so extract_point still yields the "no data" error.
|
||||
idx = """
|
||||
1:0:d=2026041518:TMP:2 m above ground:anl:
|
||||
2:100:d=2026041518:DPT:2 m above ground:anl:
|
||||
3:200:d=2026041518:PRES:surface:anl:
|
||||
"""
|
||||
|
||||
Req.Test.stub(RtmaClient, fn conn ->
|
||||
if String.ends_with?(conn.request_path, ".idx") do
|
||||
Plug.Conn.send_resp(conn, 200, idx)
|
||||
else
|
||||
# 206 Partial Content with non-GRIB junk.
|
||||
Plug.Conn.send_resp(conn, 206, :binary.copy(<<0>>, 1024))
|
||||
end
|
||||
end)
|
||||
|
||||
assert {:error, reason} =
|
||||
RtmaClient.fetch_observation(32.9, -97.0, ~U[2026-04-15 18:00:00Z])
|
||||
|
||||
assert reason =~ "RTMA: no data"
|
||||
end
|
||||
end
|
||||
|
||||
describe "fetch_observation/3 — malformed idx" do
|
||||
test "crashes on a malformed idx line with a non-integer offset" do
|
||||
# byte_ranges_for_fields/2 calls String.to_integer/1 on the offset
|
||||
# field without a guard — a non-numeric offset is a programmer/server
|
||||
# contract violation and the process raises. Characterize this so
|
||||
# any future change to forgiving-parsing is intentional.
|
||||
Req.Test.stub(RtmaClient, fn conn ->
|
||||
if String.ends_with?(conn.request_path, ".idx") do
|
||||
Plug.Conn.send_resp(conn, 200, "1:notanumber:d=2026041518:TMP:2 m above ground:anl:\n")
|
||||
else
|
||||
Plug.Conn.send_resp(conn, 500, "unexpected")
|
||||
end
|
||||
end)
|
||||
|
||||
assert_raise ArgumentError, fn ->
|
||||
RtmaClient.fetch_observation(32.9, -97.0, ~U[2026-04-15 18:00:00Z])
|
||||
end
|
||||
end
|
||||
|
||||
test "tolerates an idx line with missing trailing fields (uses empty field name)" do
|
||||
# A single-field line falls back to empty strings for var/level via
|
||||
# Enum.at/3 defaults, so it simply doesn't match any wanted field.
|
||||
Req.Test.stub(RtmaClient, fn conn ->
|
||||
if String.ends_with?(conn.request_path, ".idx") do
|
||||
Plug.Conn.send_resp(conn, 200, "1:0\n")
|
||||
else
|
||||
Plug.Conn.send_resp(conn, 500, "unexpected")
|
||||
end
|
||||
end)
|
||||
|
||||
assert {:error, reason} =
|
||||
RtmaClient.fetch_observation(32.9, -97.0, ~U[2026-04-15 18:00:00Z])
|
||||
|
||||
# Empty idx body → zero ranges → empty GRIB → "no data" at extraction.
|
||||
assert reason =~ "RTMA: no data"
|
||||
end
|
||||
end
|
||||
|
||||
describe "fetch_observation/3 — download layer" do
|
||||
# download_ranges/2's Task.async_stream reduces over results; any
|
||||
# non-200/206 downgrades the accumulator to {:error, _}. But a catch-all
|
||||
# `_, acc -> acc` clause in the reducer silently swallows :exit/:timeout
|
||||
# task results. If all tasks time out, the accumulator stays {:ok, []},
|
||||
# which produces an empty binary and ultimately "RTMA: no data ...".
|
||||
# Characterizing, not prescribing — tighten in a follow-up if desired.
|
||||
test "surfaces a non-200 status from the ranged GET as an error" do
|
||||
idx = """
|
||||
1:0:d=2026041518:TMP:2 m above ground:anl:
|
||||
2:100:d=2026041518:DPT:2 m above ground:anl:
|
||||
"""
|
||||
|
||||
Req.Test.stub(RtmaClient, fn conn ->
|
||||
if String.ends_with?(conn.request_path, ".idx") do
|
||||
Plug.Conn.send_resp(conn, 200, idx)
|
||||
else
|
||||
Plug.Conn.send_resp(conn, 500, "s3 blew up")
|
||||
end
|
||||
end)
|
||||
|
||||
assert {:error, reason} =
|
||||
RtmaClient.fetch_observation(32.9, -97.0, ~U[2026-04-15 18:00:00Z])
|
||||
|
||||
assert reason =~ "RTMA download HTTP 500"
|
||||
end
|
||||
|
||||
test "surfaces a transport timeout on the idx GET as an {:error, _}" do
|
||||
Req.Test.stub(RtmaClient, fn conn ->
|
||||
Req.Test.transport_error(conn, :timeout)
|
||||
end)
|
||||
|
||||
assert {:error, reason} =
|
||||
RtmaClient.fetch_observation(32.9, -97.0, ~U[2026-04-15 18:00:00Z])
|
||||
|
||||
# fetch_idx wraps transport errors into a human-readable string.
|
||||
assert reason =~ "RTMA idx failed"
|
||||
end
|
||||
end
|
||||
|
||||
describe "rtma_url/1 round-trip" do
|
||||
property "embeds YYYYMMDD and a zero-padded hour for any valid analysis time" do
|
||||
check all(
|
||||
year <- integer(2020..2030),
|
||||
month <- integer(1..12),
|
||||
day <- integer(1..28),
|
||||
hour <- integer(0..23)
|
||||
) do
|
||||
dt = %DateTime{
|
||||
year: year,
|
||||
month: month,
|
||||
day: day,
|
||||
hour: hour,
|
||||
minute: 0,
|
||||
second: 0,
|
||||
microsecond: {0, 0},
|
||||
std_offset: 0,
|
||||
utc_offset: 0,
|
||||
zone_abbr: "UTC",
|
||||
time_zone: "Etc/UTC"
|
||||
}
|
||||
|
||||
url = RtmaClient.rtma_url(dt)
|
||||
|
||||
date_str =
|
||||
Integer.to_string(year) <>
|
||||
String.pad_leading(Integer.to_string(month), 2, "0") <>
|
||||
String.pad_leading(Integer.to_string(day), 2, "0")
|
||||
|
||||
hour_str = String.pad_leading(Integer.to_string(hour), 2, "0")
|
||||
|
||||
assert String.contains?(url, "rtma2p5.#{date_str}/")
|
||||
assert String.contains?(url, "rtma2p5.t#{hour_str}z")
|
||||
assert String.ends_with?(url, ".grb2_wexp")
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
@ -278,36 +278,6 @@ defmodule Microwaveprop.WeatherExtraTest do
|
|||
end
|
||||
end
|
||||
|
||||
describe "find_nearest_rtma/3" do
|
||||
test "returns nil when no RTMA observation exists nearby" do
|
||||
assert Weather.find_nearest_rtma(40.0, -80.0, ~U[2026-04-20 18:00:00Z]) == nil
|
||||
end
|
||||
end
|
||||
|
||||
describe "recent_surface_obs/3" do
|
||||
test "returns nil when no nearby stations have data in the time window" do
|
||||
assert Weather.recent_surface_obs(40.0, -80.0, ~U[2026-04-20 18:00:00Z]) == nil
|
||||
end
|
||||
|
||||
test "returns a surface observation when a nearby station has one in the ±30min window" do
|
||||
station = station!(%{lat: 32.9, lon: -97.04})
|
||||
|
||||
{:ok, _} =
|
||||
%SurfaceObservation{}
|
||||
|> SurfaceObservation.changeset(%{
|
||||
station_id: station.id,
|
||||
observed_at: ~U[2026-04-20 18:05:00Z],
|
||||
temp_f: 78.0,
|
||||
dewpoint_f: 60.0
|
||||
})
|
||||
|> Repo.insert()
|
||||
|
||||
obs = Weather.recent_surface_obs(32.9, -97.04, ~U[2026-04-20 18:00:00Z])
|
||||
assert obs
|
||||
assert obs.temp_f == 78.0
|
||||
end
|
||||
end
|
||||
|
||||
describe "property: round_to_hrrr_grid/2" do
|
||||
property "rounded output stays within 0.005 of input for any CONUS lat/lon" do
|
||||
check all(
|
||||
|
|
|
|||
|
|
@ -1,107 +0,0 @@
|
|||
defmodule Microwaveprop.Workers.RtmaFetchWorkerTest do
|
||||
use Microwaveprop.DataCase, async: false
|
||||
use Oban.Testing, repo: Microwaveprop.Repo
|
||||
|
||||
alias Microwaveprop.Weather.RtmaClient
|
||||
alias Microwaveprop.Weather.RtmaObservation
|
||||
alias Microwaveprop.Workers.RtmaFetchWorker
|
||||
|
||||
describe "perform/1" do
|
||||
test "skips the fetch (no HTTP) when a matching row already exists in the snapped bounding box" do
|
||||
valid_time = ~U[2026-04-15 18:00:00Z]
|
||||
|
||||
# Pre-populate a row at the snapped coordinates (32.9 → 32.9 at 1/40°,
|
||||
# -97.0 → -97.0). The worker's bounding-box short-circuit must find it
|
||||
# and return :ok without contacting RtmaClient.
|
||||
%RtmaObservation{}
|
||||
|> RtmaObservation.changeset(%{
|
||||
lat: 32.9,
|
||||
lon: -97.0,
|
||||
valid_time: valid_time,
|
||||
temp_c: 20.0
|
||||
})
|
||||
|> Repo.insert!()
|
||||
|
||||
Req.Test.stub(RtmaClient, fn _conn ->
|
||||
raise "RtmaClient should not be contacted when an observation already exists"
|
||||
end)
|
||||
|
||||
args = %{
|
||||
"lat" => 32.9,
|
||||
"lon" => -97.0,
|
||||
"valid_time" => DateTime.to_iso8601(valid_time)
|
||||
}
|
||||
|
||||
assert :ok = RtmaFetchWorker.perform(%Oban.Job{args: args})
|
||||
assert Repo.aggregate(RtmaObservation, :count, :id) == 1
|
||||
end
|
||||
|
||||
# With garbage-bytes-on-the-wire (no real GRIB2 fixture), the extractor
|
||||
# yields no fields and RtmaClient returns {:error, "RTMA: no data ..."},
|
||||
# which the worker surfaces as {:error, _}. This characterizes that the
|
||||
# worker propagates the client error (and does NOT insert a blank row).
|
||||
test "returns {:error, _} and inserts nothing when the client fails to extract" do
|
||||
idx = """
|
||||
1:0:d=2026041518:TMP:2 m above ground:anl:
|
||||
2:100:d=2026041518:DPT:2 m above ground:anl:
|
||||
"""
|
||||
|
||||
Req.Test.stub(RtmaClient, fn conn ->
|
||||
if String.ends_with?(conn.request_path, ".idx") do
|
||||
Plug.Conn.send_resp(conn, 200, idx)
|
||||
else
|
||||
Plug.Conn.send_resp(conn, 206, :binary.copy(<<0>>, 1024))
|
||||
end
|
||||
end)
|
||||
|
||||
args = %{
|
||||
"lat" => 32.9,
|
||||
"lon" => -97.0,
|
||||
"valid_time" => "2026-04-15T18:00:00Z"
|
||||
}
|
||||
|
||||
assert {:error, reason} = RtmaFetchWorker.perform(%Oban.Job{args: args})
|
||||
assert reason =~ "RTMA: no data"
|
||||
assert Repo.aggregate(RtmaObservation, :count, :id) == 0
|
||||
end
|
||||
|
||||
test "returns {:error, _} when upstream idx returns 404" do
|
||||
Req.Test.stub(RtmaClient, fn conn ->
|
||||
if String.ends_with?(conn.request_path, ".idx") do
|
||||
Plug.Conn.send_resp(conn, 404, "not yet published")
|
||||
else
|
||||
Plug.Conn.send_resp(conn, 500, "unexpected")
|
||||
end
|
||||
end)
|
||||
|
||||
args = %{
|
||||
"lat" => 32.9,
|
||||
"lon" => -97.0,
|
||||
"valid_time" => "2026-04-15T18:00:00Z"
|
||||
}
|
||||
|
||||
assert {:error, reason} = RtmaFetchWorker.perform(%Oban.Job{args: args})
|
||||
assert reason =~ "RTMA idx HTTP 404"
|
||||
assert Repo.aggregate(RtmaObservation, :count, :id) == 0
|
||||
end
|
||||
|
||||
test "raises FunctionClauseError when args are missing required keys" do
|
||||
# The worker only matches the shape with lat/lon/valid_time — missing
|
||||
# keys must crash loudly rather than silently succeed. This pins the
|
||||
# current fail-fast contract.
|
||||
assert_raise FunctionClauseError, fn ->
|
||||
RtmaFetchWorker.perform(%Oban.Job{args: %{}})
|
||||
end
|
||||
end
|
||||
|
||||
test "raises when valid_time is not an ISO8601 string" do
|
||||
# DateTime.from_iso8601/1 returns {:error, _}, and the `{:ok, _, _} =`
|
||||
# match raises — another fail-fast contract worth pinning.
|
||||
assert_raise MatchError, fn ->
|
||||
RtmaFetchWorker.perform(%Oban.Job{
|
||||
args: %{"lat" => 32.9, "lon" => -97.0, "valid_time" => "not-a-timestamp"}
|
||||
})
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
Loading…
Add table
Reference in a new issue