From 1086f52c85efd91fc3f67b01033f58a08538fc46 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Mon, 4 May 2026 15:09:19 -0500 Subject: [PATCH] chore: delete dead RTMA + METAR5 code paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- config/runtime.exs | 3 +- config/test.exs | 1 - lib/microwaveprop/weather.ex | 79 ------ .../weather/metar_5min_observation.ex | 46 --- lib/microwaveprop/weather/rtma_client.ex | 180 ------------ lib/microwaveprop/weather/rtma_observation.ex | 37 --- .../workers/rtma_fetch_worker.ex | 71 ----- ...4200705_drop_unused_rtma_metar5_tables.exs | 58 ++++ scripts/recalibrate_algo.py | 43 +-- .../observation_changesets_test.exs | 47 ---- .../weather/rtma_client_test.exs | 265 ------------------ test/microwaveprop/weather_extra_test.exs | 30 -- .../workers/rtma_fetch_worker_test.exs | 107 ------- 13 files changed, 74 insertions(+), 893 deletions(-) delete mode 100644 lib/microwaveprop/weather/metar_5min_observation.ex delete mode 100644 lib/microwaveprop/weather/rtma_client.ex delete mode 100644 lib/microwaveprop/weather/rtma_observation.ex delete mode 100644 lib/microwaveprop/workers/rtma_fetch_worker.ex create mode 100644 priv/repo/migrations/20260504200705_drop_unused_rtma_metar5_tables.exs delete mode 100644 test/microwaveprop/weather/rtma_client_test.exs delete mode 100644 test/microwaveprop/workers/rtma_fetch_worker_test.exs diff --git a/config/runtime.exs b/config/runtime.exs index c317dba6..8fd05af5 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -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 ] diff --git a/config/test.exs b/config/test.exs index a096844b..36cd48fc 100644 --- a/config/test.exs +++ b/config/test.exs @@ -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 diff --git a/lib/microwaveprop/weather.ex b/lib/microwaveprop/weather.ex index 0b304d5c..f5857872 100644 --- a/lib/microwaveprop/weather.ex +++ b/lib/microwaveprop/weather.ex @@ -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 diff --git a/lib/microwaveprop/weather/metar_5min_observation.ex b/lib/microwaveprop/weather/metar_5min_observation.ex deleted file mode 100644 index 5303ae88..00000000 --- a/lib/microwaveprop/weather/metar_5min_observation.ex +++ /dev/null @@ -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 diff --git a/lib/microwaveprop/weather/rtma_client.ex b/lib/microwaveprop/weather/rtma_client.ex deleted file mode 100644 index 6070c71b..00000000 --- a/lib/microwaveprop/weather/rtma_client.ex +++ /dev/null @@ -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 diff --git a/lib/microwaveprop/weather/rtma_observation.ex b/lib/microwaveprop/weather/rtma_observation.ex deleted file mode 100644 index 94e458e0..00000000 --- a/lib/microwaveprop/weather/rtma_observation.ex +++ /dev/null @@ -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 diff --git a/lib/microwaveprop/workers/rtma_fetch_worker.ex b/lib/microwaveprop/workers/rtma_fetch_worker.ex deleted file mode 100644 index 4191b80d..00000000 --- a/lib/microwaveprop/workers/rtma_fetch_worker.ex +++ /dev/null @@ -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 diff --git a/priv/repo/migrations/20260504200705_drop_unused_rtma_metar5_tables.exs b/priv/repo/migrations/20260504200705_drop_unused_rtma_metar5_tables.exs new file mode 100644 index 00000000..e9cf6569 --- /dev/null +++ b/priv/repo/migrations/20260504200705_drop_unused_rtma_metar5_tables.exs @@ -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 diff --git a/scripts/recalibrate_algo.py b/scripts/recalibrate_algo.py index 17f0bbd5..acbef642 100755 --- a/scripts/recalibrate_algo.py +++ b/scripts/recalibrate_algo.py @@ -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 diff --git a/test/microwaveprop/space_weather/observation_changesets_test.exs b/test/microwaveprop/space_weather/observation_changesets_test.exs index 30b53c6b..f77fa7dd 100644 --- a/test/microwaveprop/space_weather/observation_changesets_test.exs +++ b/test/microwaveprop/space_weather/observation_changesets_test.exs @@ -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). diff --git a/test/microwaveprop/weather/rtma_client_test.exs b/test/microwaveprop/weather/rtma_client_test.exs deleted file mode 100644 index 81b99d6a..00000000 --- a/test/microwaveprop/weather/rtma_client_test.exs +++ /dev/null @@ -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 diff --git a/test/microwaveprop/weather_extra_test.exs b/test/microwaveprop/weather_extra_test.exs index 5ff70952..c5e357c2 100644 --- a/test/microwaveprop/weather_extra_test.exs +++ b/test/microwaveprop/weather_extra_test.exs @@ -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( diff --git a/test/microwaveprop/workers/rtma_fetch_worker_test.exs b/test/microwaveprop/workers/rtma_fetch_worker_test.exs deleted file mode 100644 index eeb498f2..00000000 --- a/test/microwaveprop/workers/rtma_fetch_worker_test.exs +++ /dev/null @@ -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