Phase 8: 5-minute METAR ingestion pipeline
NCEI ASOS 5-minute data client (Weather.NceiMetarClient): - fetch/3 pulls per-station monthly .dat files from NCEI C00418 - parse/1 decodes the fixed-width METAR format including precise T-group temperatures (T02110094 → 21.1/9.4°C) - metar_5min_observations table: schema-identical to surface_observations, separate table to avoid mixing cadences Weather.recent_surface_obs/3 prefers 5-min data when available, falls back to the hourly surface_observations table. Data URL: https://www.ncei.noaa.gov/data/automated-surface-observing-system-five-minute/access/YYYY/MM/asos-5min-KXXX-YYYYMM.dat Available back to 1996.
This commit is contained in:
parent
1e17b29c5c
commit
3ea2548114
5 changed files with 356 additions and 0 deletions
|
|
@ -8,6 +8,7 @@ defmodule Microwaveprop.Weather do
|
|||
alias Microwaveprop.Weather.HrrrProfile
|
||||
alias Microwaveprop.Weather.IemClient
|
||||
alias Microwaveprop.Weather.IemreObservation
|
||||
alias Microwaveprop.Weather.Metar5minObservation
|
||||
alias Microwaveprop.Weather.RtmaObservation
|
||||
alias Microwaveprop.Weather.SolarIndex
|
||||
alias Microwaveprop.Weather.Sounding
|
||||
|
|
@ -701,4 +702,50 @@ 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.
|
||||
"""
|
||||
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
|
||||
|
|
|
|||
43
lib/microwaveprop/weather/metar_5min_observation.ex
Normal file
43
lib/microwaveprop/weather/metar_5min_observation.ex
Normal file
|
|
@ -0,0 +1,43 @@
|
|||
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
|
||||
|
||||
@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
|
||||
|
||||
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
|
||||
191
lib/microwaveprop/weather/ncei_metar_client.ex
Normal file
191
lib/microwaveprop/weather/ncei_metar_client.ex
Normal file
|
|
@ -0,0 +1,191 @@
|
|||
defmodule Microwaveprop.Weather.NceiMetarClient do
|
||||
@moduledoc """
|
||||
Fetches 5-minute ASOS observations from NCEI's C00418 dataset.
|
||||
|
||||
Data URL pattern:
|
||||
```
|
||||
https://www.ncei.noaa.gov/data/automated-surface-observing-system-five-minute/access/YYYY/MM/asos-5min-KXXX-YYYYMM.dat
|
||||
```
|
||||
|
||||
Each file is one station-month of 5-minute observations in a
|
||||
fixed-width METAR-like format. See the meteorologist's recommendation
|
||||
in the April 2026 review.
|
||||
"""
|
||||
|
||||
require Logger
|
||||
|
||||
@base_url "https://www.ncei.noaa.gov/data/automated-surface-observing-system-five-minute/access"
|
||||
|
||||
@doc """
|
||||
Fetch and parse 5-minute observations for a given station, year, and month.
|
||||
|
||||
Returns `{:ok, [parsed_obs]}` or `{:error, reason}`.
|
||||
"""
|
||||
def fetch(icao, year, month) do
|
||||
month_str = month |> Integer.to_string() |> String.pad_leading(2, "0")
|
||||
url = "#{@base_url}/#{year}/#{month_str}/asos-5min-#{icao}-#{year}#{month_str}.dat"
|
||||
|
||||
case Req.get(url, receive_timeout: 60_000) do
|
||||
{:ok, %{status: 200, body: body}} ->
|
||||
{:ok, parse(body)}
|
||||
|
||||
{:ok, %{status: status}} ->
|
||||
{:error, "NCEI 5-min HTTP #{status}"}
|
||||
|
||||
{:error, reason} ->
|
||||
{:error, reason}
|
||||
end
|
||||
end
|
||||
|
||||
@doc """
|
||||
Parse the fixed-width NCEI 5-minute ASOS format into maps.
|
||||
|
||||
Each line has a METAR embedded in a fixed-width wrapper. We extract
|
||||
the key fields without a full METAR parser — this covers the data
|
||||
the scorer and backtest need (temp, dewpoint, wind, pressure, sky).
|
||||
"""
|
||||
def parse(text) when is_binary(text) do
|
||||
text
|
||||
|> String.split("\n")
|
||||
|> Enum.flat_map(&parse_line/1)
|
||||
end
|
||||
|
||||
defp parse_line(line) when byte_size(line) < 60, do: []
|
||||
|
||||
defp parse_line(line) do
|
||||
# Fixed-width header: WBAN(5) ICAO(4) FAA(4) DATE(8) HHMM(4) ...
|
||||
# Then METAR text after the "5-MIN" marker
|
||||
with <<_wban::binary-5, icao::binary-4, _faa::binary-4, date::binary-8, hhmm::binary-4, _rest::binary>> <- line,
|
||||
{:ok, observed_at} <- parse_datetime(date, hhmm),
|
||||
metar_text <- extract_metar(line) do
|
||||
obs = parse_metar_fields(metar_text)
|
||||
[Map.merge(obs, %{icao: String.trim(icao), observed_at: observed_at})]
|
||||
else
|
||||
_ -> []
|
||||
end
|
||||
end
|
||||
|
||||
defp parse_datetime(<<y::binary-4, m::binary-2, d::binary-2>>, <<hh::binary-2, mm::binary-2>>) do
|
||||
with {year, ""} <- Integer.parse(y),
|
||||
{month, ""} <- Integer.parse(m),
|
||||
{day, ""} <- Integer.parse(d),
|
||||
{hour, ""} <- Integer.parse(hh),
|
||||
{minute, ""} <- Integer.parse(mm),
|
||||
{:ok, date} <- Date.new(year, month, day),
|
||||
{:ok, time} <- Time.new(hour, minute, 0),
|
||||
{:ok, dt} <- DateTime.new(date, time, "Etc/UTC") do
|
||||
{:ok, DateTime.truncate(dt, :second)}
|
||||
else
|
||||
_ -> :error
|
||||
end
|
||||
end
|
||||
|
||||
defp extract_metar(line) do
|
||||
case String.split(line, "5-MIN ", parts: 2) do
|
||||
[_, metar] -> metar
|
||||
_ -> ""
|
||||
end
|
||||
end
|
||||
|
||||
defp parse_metar_fields(metar) do
|
||||
parts = String.split(metar)
|
||||
|
||||
%{
|
||||
temp_f: extract_temp_f(parts),
|
||||
dewpoint_f: extract_dewpoint_f(parts),
|
||||
wind_speed_kts: extract_wind_speed(parts),
|
||||
wind_direction_deg: extract_wind_dir(parts),
|
||||
altimeter_setting: extract_altimeter(parts),
|
||||
sky_condition: extract_sky(parts),
|
||||
relative_humidity: nil,
|
||||
sea_level_pressure_mb: nil,
|
||||
precip_1h_in: nil,
|
||||
wx_codes: nil
|
||||
}
|
||||
end
|
||||
|
||||
# Extract precise temp/dew from RMK T-group: T02110094 → 21.1°C / 9.4°C
|
||||
defp extract_temp_f(parts) do
|
||||
with t_group when is_binary(t_group) <- Enum.find(parts, &String.starts_with?(&1, "T0")) do
|
||||
parse_t_group_temp(t_group)
|
||||
else
|
||||
_ ->
|
||||
# Fall back to temp/dew field: "21/09"
|
||||
with td when is_binary(td) <- Enum.find(parts, &Regex.match?(~r/^M?\d+\/M?\d+$/, &1)) do
|
||||
[t, _d] = String.split(td, "/")
|
||||
parse_metar_temp(t) |> c_to_f()
|
||||
else
|
||||
_ -> nil
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
defp extract_dewpoint_f(parts) do
|
||||
with t_group when is_binary(t_group) <- Enum.find(parts, &String.starts_with?(&1, "T0")) do
|
||||
parse_t_group_dew(t_group)
|
||||
else
|
||||
_ ->
|
||||
with td when is_binary(td) <- Enum.find(parts, &Regex.match?(~r/^M?\d+\/M?\d+$/, &1)) do
|
||||
[_t, d] = String.split(td, "/")
|
||||
parse_metar_temp(d) |> c_to_f()
|
||||
else
|
||||
_ -> nil
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
# T02110094 → temp = +21.1°C, dew = +9.4°C (leading 0=+, 1=-)
|
||||
defp parse_t_group_temp(<<"T", sign::binary-1, d1::binary-2, d2::binary-1, _rest::binary>>) do
|
||||
val = String.to_integer(d1) + String.to_integer(d2) / 10.0
|
||||
(if sign == "1", do: -val, else: val) |> c_to_f()
|
||||
end
|
||||
|
||||
defp parse_t_group_temp(_), do: nil
|
||||
|
||||
defp parse_t_group_dew(<<"T", _::binary-4, sign::binary-1, d1::binary-2, d2::binary-1, _rest::binary>>) do
|
||||
val = String.to_integer(d1) + String.to_integer(d2) / 10.0
|
||||
(if sign == "1", do: -val, else: val) |> c_to_f()
|
||||
end
|
||||
|
||||
defp parse_t_group_dew(_), do: nil
|
||||
|
||||
defp parse_metar_temp("M" <> rest), do: -String.to_integer(rest) * 1.0
|
||||
defp parse_metar_temp(s), do: String.to_integer(s) * 1.0
|
||||
|
||||
defp c_to_f(c) when is_number(c), do: c * 9.0 / 5.0 + 32.0
|
||||
defp c_to_f(_), do: nil
|
||||
|
||||
defp extract_wind_speed(parts) do
|
||||
with wind when is_binary(wind) <- Enum.find(parts, &Regex.match?(~r/^\d{3}\d{2,3}KT/, &1)),
|
||||
{speed, _} <- Integer.parse(String.slice(wind, 3..4)) do
|
||||
speed * 1.0
|
||||
else
|
||||
_ -> nil
|
||||
end
|
||||
end
|
||||
|
||||
defp extract_wind_dir(parts) do
|
||||
with wind when is_binary(wind) <- Enum.find(parts, &Regex.match?(~r/^\d{3}\d{2,3}KT/, &1)),
|
||||
{dir, _} <- Integer.parse(String.slice(wind, 0..2)) do
|
||||
dir
|
||||
else
|
||||
_ -> nil
|
||||
end
|
||||
end
|
||||
|
||||
defp extract_altimeter(parts) do
|
||||
with alt when is_binary(alt) <- Enum.find(parts, &String.starts_with?(&1, "A2")) do
|
||||
case Float.parse(String.slice(alt, 1..-1//1)) do
|
||||
{val, _} -> val / 100.0
|
||||
:error -> nil
|
||||
end
|
||||
else
|
||||
_ -> nil
|
||||
end
|
||||
end
|
||||
|
||||
defp extract_sky(parts) do
|
||||
sky_tokens = Enum.filter(parts, &Regex.match?(~r/^(CLR|FEW|SCT|BKN|OVC|VV)\d*/, &1))
|
||||
if sky_tokens != [], do: Enum.join(sky_tokens, " "), else: nil
|
||||
end
|
||||
end
|
||||
|
|
@ -0,0 +1,26 @@
|
|||
defmodule Microwaveprop.Repo.Migrations.CreateMetar5minObservations do
|
||||
use Ecto.Migration
|
||||
|
||||
def change do
|
||||
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
|
||||
49
test/microwaveprop/weather/ncei_metar_client_test.exs
Normal file
49
test/microwaveprop/weather/ncei_metar_client_test.exs
Normal file
|
|
@ -0,0 +1,49 @@
|
|||
defmodule Microwaveprop.Weather.NceiMetarClientTest do
|
||||
use ExUnit.Case, async: true
|
||||
|
||||
alias Microwaveprop.Weather.NceiMetarClient
|
||||
|
||||
@sample_line "03927KDFW DFW20260301000010303/01/26 00:00:31 5-MIN KDFW 010600Z 17012KT 10SM CLR 21/09 A2997 560 47 1400 160/12 RMK AO2 T02110094"
|
||||
|
||||
describe "parse/1" do
|
||||
test "parses a single KDFW observation" do
|
||||
[obs] = NceiMetarClient.parse(@sample_line)
|
||||
|
||||
assert obs.icao == "KDFW"
|
||||
assert obs.observed_at == ~U[2026-03-01 00:00:00Z]
|
||||
# T02110094 → 21.1°C = 70.0°F, 9.4°C = 48.9°F
|
||||
assert_in_delta obs.temp_f, 70.0, 0.2
|
||||
assert_in_delta obs.dewpoint_f, 48.9, 0.2
|
||||
assert obs.wind_speed_kts == 12.0
|
||||
assert obs.wind_direction_deg == 170
|
||||
assert_in_delta obs.altimeter_setting, 29.97, 0.01
|
||||
assert obs.sky_condition == "CLR"
|
||||
end
|
||||
|
||||
test "parses multiple lines" do
|
||||
text = """
|
||||
03927KDFW DFW20260301000010303/01/26 00:00:31 5-MIN KDFW 010600Z 17012KT 10SM CLR 21/09 A2997 560 47 1400 160/12 RMK AO2 T02110094
|
||||
03927KDFW DFW20260301000510303/01/26 00:05:31 5-MIN KDFW 010605Z 17012KT 10SM CLR 21/09 A2997 560 47 1400 160/12 RMK AO2 T02110094
|
||||
"""
|
||||
|
||||
obs = NceiMetarClient.parse(text)
|
||||
assert length(obs) == 2
|
||||
assert Enum.at(obs, 0).observed_at == ~U[2026-03-01 00:00:00Z]
|
||||
assert Enum.at(obs, 1).observed_at == ~U[2026-03-01 00:05:00Z]
|
||||
end
|
||||
|
||||
test "skips blank or too-short lines" do
|
||||
assert NceiMetarClient.parse("") == []
|
||||
assert NceiMetarClient.parse("short\n\n") == []
|
||||
end
|
||||
|
||||
test "handles negative temperatures via M prefix" do
|
||||
line = "03927KDFW DFW20260115120010103/15/26 12:00:31 5-MIN KDFW 151200Z 36005KT 10SM CLR M02/M08 A3032 560 47 1400 360/05 RMK AO2"
|
||||
[obs] = NceiMetarClient.parse(line)
|
||||
|
||||
# M02 = -2°C = 28.4°F, M08 = -8°C = 17.6°F
|
||||
assert_in_delta obs.temp_f, 28.4, 0.2
|
||||
assert_in_delta obs.dewpoint_f, 17.6, 0.2
|
||||
end
|
||||
end
|
||||
end
|
||||
Loading…
Add table
Reference in a new issue