graham/prop
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(profiles_file): read Rust-written MessagePack alongside legacy ETF

Browse source 
Phase 3 Stream A: Rust's prop-grid-rs writes f00 profile files as
MessagePack (c4f309c). Elixir needs to read both formats during the
cutover window and indefinitely afterward — .mp.gz wins when both
exist.

Reader:
- path_for/1 stays on .etf.gz (Elixir legacy writer)
- mp_path_for/1 returns the .mp.gz sibling Rust lands
- read/1 prefers .mp.gz, falls back to .etf.gz, :enoent if neither
- decode_mp_body handles the top-level {v, valid_time, cells} shape
- normalize_profile atomizes a whitelist of known keys (surface_*,
  native_min_gradient, ducts[], profile[] levels, ...) and leaves
  everything else as strings so no String.to_atom bomb is possible
- Directory scan picks up both extensions so retain_window and
  prune_older_than also work uniformly; list_valid_times dedupes
  by timestamp

Adds msgpax ~> 2.4 dependency. 4 new tests cover format preference,
key atomization, :enoent path, and dedupe.
This commit is contained in:
Graham McIntire 2026-04-19 18:04:34 -05:00
parent 6613804bbf
commit d41ced5850
No known key found for this signature in database
GPG key ID: F4ABF488E6029E59
4 changed files with 196 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

117
lib/microwaveprop/propagation/profiles_file.ex
View file

@ -32,13 +32,63 @@ defmodule Microwaveprop.Propagation.ProfilesFile do
)
end
@doc "Absolute path for the profile file covering `valid_time`."
@doc "Absolute path for the Elixir-written ETF profile file covering `valid_time`."
@spec path_for(DateTime.t()) :: String.t()
def path_for(%DateTime{} = valid_time) do
iso = valid_time |> DateTime.truncate(:second) |> DateTime.to_iso8601()
Path.join(base_dir(), "#{iso}.etf.gz")
Path.join(base_dir(), "#{iso_key(valid_time)}.etf.gz")
end
@doc """
Absolute path for the Rust-written MessagePack profile file covering
`valid_time`. Rust's `prop_grid_rs::profiles_file::write_atomic`
lands here. Reader prefers `.mp.gz` when both formats coexist during
Phase 3 Stream A cutover.
"""
@spec mp_path_for(DateTime.t()) :: String.t()
def mp_path_for(%DateTime{} = valid_time) do
Path.join(base_dir(), "#{iso_key(valid_time)}.mp.gz")
end
defp iso_key(%DateTime{} = valid_time) do
valid_time |> DateTime.truncate(:second) |> DateTime.to_iso8601()
end
# Keys Rust writes as strings; Elixir callers read them as atoms.
# Anything outside this whitelist stays a string and is ignored by
# downstream consumers (weather.ex uses the atom form throughout).
@mp_atom_keys MapSet.new([
"native_min_gradient",
"best_duct_freq_ghz",
"max_duct_thickness_m",
"duct_count",
"ducts",
"base_m",
"top_m",
"thickness_m",
"m_deficit",
"min_freq_ghz",
"surface_temp_c",
"surface_dewpoint_c",
"surface_pressure_mb",
"surface_refractivity",
"hpbl_m",
"pwat_mm",
"min_refractivity_gradient",
"ducting_detected",
"duct_characteristics",
"nexrad_max_reflectivity_dbz",
"commercial_link_degradation",
"degradation_db",
"baseline_dbm",
"current_dbm",
"n_links",
"profile",
"pres_mb",
"hght_m",
"tmpc",
"dwpc"
])
@doc """
Persist the enriched grid_data (`%{{lat, lon} => profile}`) for
`valid_time`. Writes compressed ETF; lat/lon keys are rounded to the
@ -80,12 +130,62 @@ defmodule Microwaveprop.Propagation.ProfilesFile do
"""
@spec read(DateTime.t()) :: {:ok, %{{float(), float()} => map()}} | {:error, :enoent}
def read(%DateTime{} = valid_time) do
# Rust-written MessagePack wins over Elixir-written ETF if both
# exist. After Phase 3 Stream A cutover only `.mp.gz` is produced;
# the ETF branch remains so pods picking up an older in-flight run
# during the rollout window still render correctly.
cond do
File.exists?(mp_path_for(valid_time)) -> read_mp(valid_time)
File.exists?(path_for(valid_time)) -> read_etf(valid_time)
true -> {:error, :enoent}
end
end
defp read_etf(valid_time) do
case File.read(path_for(valid_time)) do
{:ok, binary} -> {:ok, :erlang.binary_to_term(binary)}
{:error, _} -> {:error, :enoent}
end
end
defp read_mp(valid_time) do
with {:ok, gz} <- File.read(mp_path_for(valid_time)),
binary = :zlib.gunzip(gz),
{:ok, body} <- Msgpax.unpack(binary) do
{:ok, decode_mp_body(body)}
else
{:error, _} -> {:error, :enoent}
end
end
defp decode_mp_body(%{"cells" => cells}) when is_list(cells) do
Map.new(cells, fn cell ->
lat = cell |> Map.get("lat") |> to_float()
lon = cell |> Map.get("lon") |> to_float()
profile = cell |> Map.get("profile", %{}) |> normalize_profile()
{{lat, lon}, profile}
end)
end
defp decode_mp_body(_), do: %{}
defp normalize_profile(value) when is_map(value) do
Map.new(value, fn {k, v} ->
key = if is_binary(k) and MapSet.member?(@mp_atom_keys, k), do: String.to_atom(k), else: k
{key, normalize_profile(v)}
end)
end
defp normalize_profile(value) when is_list(value) do
Enum.map(value, &normalize_profile/1)
end
defp normalize_profile(value), do: value
defp to_float(v) when is_float(v), do: v
defp to_float(v) when is_integer(v), do: v * 1.0
defp to_float(_), do: 0.0
@doc """
Read a single profile for `(valid_time, lat, lon)`. Returns `nil` if
the file is missing or the point has no profile. Input lat/lon are
@ -172,7 +272,9 @@ defmodule Microwaveprop.Propagation.ProfilesFile do
def list_valid_times do
base_dir()
|> list_profile_files()
|> Enum.map(fn {_path, unix} -> DateTime.from_unix!(unix) end)
|> Enum.map(fn {_path, unix} -> unix end)
|> Enum.uniq()
|> Enum.map(&DateTime.from_unix!/1)
|> Enum.sort(DateTime)
end
@ -198,7 +300,12 @@ defmodule Microwaveprop.Propagation.ProfilesFile do
end
defp parse_valid_time(filename) do
with [_, iso] <- Regex.run(~r/^(.+)\.etf\.gz$/, filename),
# Matches both `.etf.gz` (Elixir legacy) and `.mp.gz` (Rust Phase 3
# Stream A). If both exist for the same valid_time the directory
# listing yields two entries with the same timestamp; the pipeline
# downstream of list_valid_times/0 uniq-sorts, and read/1 prefers
# the mp.gz variant via mp_path_for/1.
with [_, iso, _fmt] <- Regex.run(~r/^(.+)\.(etf|mp)\.gz$/, filename),
{:ok, dt, _} <- DateTime.from_iso8601(iso) do
DateTime.to_unix(dt)
else

1
mix.exs
View file

@ -69,6 +69,7 @@ defmodule Microwaveprop.MixProject do
{:prom_ex, "~> 1.11"},
{:gettext, "~> 0.26"},
{:jason, "~> 1.2"},
{:msgpax, "~> 2.4"},
{:bandit, "~> 1.5"},
{:styler, "~> 1.11", only: [:dev, :test], runtime: false},
{:oban, "~> 2.21"},

1
mix.lock
View file

@ -37,6 +37,7 @@
"live_table": {:hex, :live_table, "0.4.1", "26055f78af6295b6d6deefb5ffd31f87b49c8f74cdf65b6dff0da595767ea470", [:mix], [{:ecto, "~> 3.13", [hex: :ecto, repo: "hexpm", optional: false]}, {:ecto_sql, "~> 3.13", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:igniter, "~> 0.7.0", [hex: :igniter, repo: "hexpm", optional: true]}, {:jason, "~> 1.4", [hex: :jason, repo: "hexpm", optional: false]}, {:nimble_csv, "~> 1.3", [hex: :nimble_csv, repo: "hexpm", optional: false]}, {:oban, "~> 2.20", [hex: :oban, repo: "hexpm", optional: false]}, {:oban_web, "~> 2.11", [hex: :oban_web, repo: "hexpm", optional: false]}, {:phoenix_html, "~> 4.3", [hex: :phoenix_html, repo: "hexpm", optional: false]}, {:phoenix_live_view, "~> 1.1", [hex: :phoenix_live_view, repo: "hexpm", optional: false]}, {:postgrex, ">= 0.0.0", [hex: :postgrex, repo: "hexpm", optional: false]}, {:sutra_ui, "~> 0.3.0", [hex: :sutra_ui, repo: "hexpm", optional: true]}], "hexpm", "2f20a173e6944526607eb5c91196ce0774fc7235bf715bb2ba2d43828965fd9d"},
"mime": {:hex, :mime, "2.0.7", "b8d739037be7cd402aee1ba0306edfdef982687ee7e9859bee6198c1e7e2f128", [:mix], [], "hexpm", "6171188e399ee16023ffc5b76ce445eb6d9672e2e241d2df6050f3c771e80ccd"},
"mint": {:hex, :mint, "1.7.1", "113fdb2b2f3b59e47c7955971854641c61f378549d73e829e1768de90fc1abf1", [:mix], [{:castore, "~> 0.1.0 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:hpax, "~> 0.1.1 or ~> 0.2.0 or ~> 1.0", [hex: :hpax, repo: "hexpm", optional: false]}], "hexpm", "fceba0a4d0f24301ddee3024ae116df1c3f4bb7a563a731f45fdfeb9d39a231b"},
"msgpax": {:hex, :msgpax, "2.4.0", "4647575c87cb0c43b93266438242c21f71f196cafa268f45f91498541148c15d", [:mix], [{:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "ca933891b0e7075701a17507c61642bf6e0407bb244040d5d0a58597a06369d2"},
"nimble_csv": {:hex, :nimble_csv, "1.3.0", "b7f998dc62b222bce9596e46f028c7a5af04cb5dde6df2ea197c583227c54971", [:mix], [], "hexpm", "41ccdc18f7c8f8bb06e84164fc51635321e80d5a3b450761c4997d620925d619"},
"nimble_options": {:hex, :nimble_options, "1.1.1", "e3a492d54d85fc3fd7c5baf411d9d2852922f66e69476317787a7b2bb000a61b", [:mix], [], "hexpm", "821b2470ca9442c4b6984882fe9bb0389371b8ddec4d45a9504f00a66f650b44"},
"nimble_pool": {:hex, :nimble_pool, "1.1.0", "bf9c29fbdcba3564a8b800d1eeb5a3c58f36e1e11d7b7fb2e084a643f645f06b", [:mix], [], "hexpm", "af2e4e6b34197db81f7aad230c1118eac993acc0dae6bc83bac0126d4ae0813a"},

82
test/microwaveprop/propagation/profiles_file_test.exs
View file

@ -189,4 +189,86 @@ defmodule Microwaveprop.Propagation.ProfilesFileTest do
assert ProfilesFile.read_point(before_window, 25.0, -125.0) == nil
end
end
describe "MessagePack dual-format reader" do
test "read/1 prefers .mp.gz over .etf.gz when both exist", %{dir: dir} do
vt = ~U[2026-04-19 14:00:00Z]
# Elixir writes the legacy ETF path.
ProfilesFile.write!(vt, %{{32.9, -97.0} => %{surface_temp_c: 1.0}})
# Rust would land a .mp.gz sibling; simulate it.
mp_body = %{
"v" => 1,
"valid_time" => "2026-04-19T14:00:00Z",
"cells" => [
%{"lat" => 32.9, "lon" => -97.0, "profile" => %{"surface_temp_c" => 2.0}}
]
}
mp_path = ProfilesFile.mp_path_for(vt)
File.mkdir_p!(Path.dirname(mp_path))
File.write!(mp_path, :zlib.gzip(Msgpax.pack!(mp_body, iodata: false)))
# mp.gz wins.
assert {:ok, grid} = ProfilesFile.read(vt)
assert grid[{32.9, -97.0}] == %{surface_temp_c: 2.0}
end
test "read/1 atomizes whitelisted keys, leaves others as strings", %{dir: _dir} do
vt = ~U[2026-04-19 14:00:00Z]
mp_body = %{
"v" => 1,
"valid_time" => "2026-04-19T14:00:00Z",
"cells" => [
%{
"lat" => 32.9,
"lon" => -97.0,
"profile" => %{
"surface_temp_c" => 22.5,
"duct_count" => 1,
"ducts" => [
%{"base_m" => 100, "top_m" => 200, "thickness_m" => 100, "m_deficit" => 12.3, "min_freq_ghz" => 15.2}
],
"unknown_field" => "keep-as-string"
}
}
]
}
mp_path = ProfilesFile.mp_path_for(vt)
File.mkdir_p!(Path.dirname(mp_path))
File.write!(mp_path, :zlib.gzip(Msgpax.pack!(mp_body, iodata: false)))
{:ok, grid} = ProfilesFile.read(vt)
profile = grid[{32.9, -97.0}]
assert profile[:surface_temp_c] == 22.5
assert profile[:duct_count] == 1
assert [duct] = profile[:ducts]
assert duct[:base_m] == 100
assert duct[:min_freq_ghz] == 15.2
# Keys outside the whitelist stay strings so they can't crash a
# later String.to_atom/1 bomb.
assert profile["unknown_field"] == "keep-as-string"
end
test "read/1 returns :enoent when neither format exists", %{dir: _dir} do
vt = ~U[2030-01-01 00:00:00Z]
assert ProfilesFile.read(vt) == {:error, :enoent}
end
test "list_valid_times/0 dedupes when both formats share a timestamp", %{dir: _dir} do
vt = ~U[2026-04-19 14:00:00Z]
ProfilesFile.write!(vt, %{{32.9, -97.0} => %{surface_temp_c: 1.0}})
mp_body = %{"v" => 1, "valid_time" => "2026-04-19T14:00:00Z", "cells" => []}
mp_path = ProfilesFile.mp_path_for(vt)
File.mkdir_p!(Path.dirname(mp_path))
File.write!(mp_path, :zlib.gzip(Msgpax.pack!(mp_body, iodata: false)))
assert ProfilesFile.list_valid_times() == [vt]
end
end
end