From d41ced58501d2e222be1b2554362c986f3b174ca Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sun, 19 Apr 2026 18:04:34 -0500 Subject: [PATCH] feat(profiles_file): read Rust-written MessagePack alongside legacy ETF MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../propagation/profiles_file.ex | 117 +++++++++++++++++- mix.exs | 1 + mix.lock | 1 + .../propagation/profiles_file_test.exs | 82 ++++++++++++ 4 files changed, 196 insertions(+), 5 deletions(-) diff --git a/lib/microwaveprop/propagation/profiles_file.ex b/lib/microwaveprop/propagation/profiles_file.ex index 4909b810..89334588 100644 --- a/lib/microwaveprop/propagation/profiles_file.ex +++ b/lib/microwaveprop/propagation/profiles_file.ex @@ -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 diff --git a/mix.exs b/mix.exs index 676004d6..bbae8aaa 100644 --- a/mix.exs +++ b/mix.exs @@ -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"}, diff --git a/mix.lock b/mix.lock index efd4b42f..1d6200c2 100644 --- a/mix.lock +++ b/mix.lock @@ -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"}, diff --git a/test/microwaveprop/propagation/profiles_file_test.exs b/test/microwaveprop/propagation/profiles_file_test.exs index 80bf45e4..35805792 100644 --- a/test/microwaveprop/propagation/profiles_file_test.exs +++ b/test/microwaveprop/propagation/profiles_file_test.exs @@ -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