From f2efdd4ece91ab2bffe45b89f9d65d6ae2f893f7 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Fri, 10 Apr 2026 15:44:31 -0500 Subject: [PATCH] Stream HRRR native downloads to disk to prevent OOM Instead of holding ~530MB GRIB binary in BEAM memory, download ranges directly to a temp file and run wgrib2 on it. Peak memory drops from ~530MB to just HTTP chunk buffers. --- lib/microwaveprop/weather/grib2/wgrib2.ex | 50 +++++++++++++++++++ lib/microwaveprop/weather/hrrr_client.ex | 50 +++++++++++++++++++ .../weather/hrrr_native_client.ex | 29 +++++++++++ .../workers/hrrr_native_grid_worker.ex | 16 ++++-- 4 files changed, 142 insertions(+), 3 deletions(-) diff --git a/lib/microwaveprop/weather/grib2/wgrib2.ex b/lib/microwaveprop/weather/grib2/wgrib2.ex index cc56af07..44ba4697 100644 --- a/lib/microwaveprop/weather/grib2/wgrib2.ex +++ b/lib/microwaveprop/weather/grib2/wgrib2.ex @@ -27,11 +27,61 @@ defmodule Microwaveprop.Weather.Grib2.Wgrib2 do end end + @doc """ + Like `extract_grid/3`, but takes a path to a GRIB2 file on disk instead + of a binary. Avoids loading the entire file into memory — useful for + large native-level HRRR files (~530 MB). + """ + def extract_grid_from_file(grib_path, match_pattern, grid_spec) do + if available?() do + extract_file_with_wgrib2(grib_path, match_pattern, grid_spec) + else + {:error, :wgrib2_not_available} + end + end + @doc "Check if wgrib2 is available on the system." def available?, do: wgrib2_path() != nil defp wgrib2_path, do: System.find_executable("wgrib2") + defp extract_file_with_wgrib2(grib_path, match_pattern, grid_spec) do + %{ + lon_start: lon_start, + lon_count: lon_count, + lon_step: lon_step, + lat_start: lat_start, + lat_count: lat_count, + lat_step: lat_step + } = grid_spec + + wgrib2_lon_start = normalize_lon(lon_start) + lon_spec = "#{wgrib2_lon_start}:#{lon_count}:#{lon_step}" + lat_spec = "#{lat_start}:#{lat_count}:#{lat_step}" + + tmp_bin = grib_path <> ".lola.bin" + + try do + args = [grib_path, "-match", match_pattern, "-lola", lon_spec, lat_spec, tmp_bin, "bin"] + + case System.cmd(wgrib2_path(), args, stderr_to_stdout: true) do + {output, 0} -> + messages = parse_wgrib2_inventory(output) + + case File.read(tmp_bin) do + {:ok, bin_data} -> parse_lola_binary(bin_data, messages, grid_spec) + {:error, :enoent} -> {:ok, %{}} + {:error, reason} -> {:error, "Failed to read wgrib2 output: #{inspect(reason)}"} + end + + {output, exit_code} -> + {:error, "wgrib2 failed (exit #{exit_code}): #{String.slice(output, 0, 200)}"} + end + after + File.rm(tmp_bin) + end + end + defp extract_with_wgrib2(grib_binary, match_pattern, grid_spec) do %{ lon_start: lon_start, diff --git a/lib/microwaveprop/weather/hrrr_client.ex b/lib/microwaveprop/weather/hrrr_client.ex index e665cd53..5ef9922d 100644 --- a/lib/microwaveprop/weather/hrrr_client.ex +++ b/lib/microwaveprop/weather/hrrr_client.ex @@ -340,6 +340,56 @@ defmodule Microwaveprop.Weather.HrrrClient do end end + @doc """ + Like `download_grib_ranges/2`, but writes the result to `dest_path` + instead of returning a binary. Each chunk is written to disk as it + arrives, so peak memory stays low even for ~530 MB native-level files. + + Returns `:ok` or `{:error, reason}`. + """ + def download_grib_ranges_to_file(_url, [], _dest_path), do: {:ok, <<>>} + + def download_grib_ranges_to_file(url, ranges, dest_path) do + merged = merge_ranges(ranges) + + results = + merged + |> Task.async_stream( + fn {start, stop} -> + case Req.get(url, [{:headers, [{"Range", "bytes=#{start}-#{stop}"}]} | req_options()]) do + {:ok, %{status: 206, body: body}} -> {:ok, start, body} + {:ok, %{status: status}} -> {:error, "HRRR grib HTTP #{status}"} + {:error, reason} -> {:error, reason} + end + end, + max_concurrency: 4, + timeout: 120_000 + ) + |> Enum.map(fn + {:ok, result} -> result + {:exit, reason} -> {:error, reason} + end) + + case Enum.find(results, &match?({:error, _}, &1)) do + {:error, reason} -> + {:error, reason} + + nil -> + sorted = Enum.sort_by(results, fn {:ok, offset, _} -> offset end) + file = File.open!(dest_path, [:write, :binary]) + + try do + Enum.each(sorted, fn {:ok, _offset, body} -> + IO.binwrite(file, body) + end) + after + File.close(file) + end + + :ok + end + end + defp download_grib_ranges_remote(url, ranges) do # Merge adjacent/overlapping ranges to reduce HTTP requests merged = merge_ranges(ranges) diff --git a/lib/microwaveprop/weather/hrrr_native_client.ex b/lib/microwaveprop/weather/hrrr_native_client.ex index 5d2c7fda..279d4dae 100644 --- a/lib/microwaveprop/weather/hrrr_native_client.ex +++ b/lib/microwaveprop/weather/hrrr_native_client.ex @@ -125,6 +125,35 @@ defmodule Microwaveprop.Weather.HrrrNativeClient do HrrrClient.byte_ranges_for_messages(idx_entries, native_messages()) end + @doc """ + Extract native profiles for a list of `{lat, lon}` points from a + GRIB2 file on disk, using wgrib2. Avoids loading the entire file + into memory (~530 MB for native-level files). + + Returns `{:ok, %{{lat, lon} => native_profile_map}}` or `{:error, reason}`. + """ + def extract_native_profiles_from_file(grib_path, points) when is_list(points) do + alias Microwaveprop.Weather.Grib2.Wgrib2 + + match_pattern = ":(#{Enum.join(@native_variables, "|")}):.*hybrid level:" + grid_spec = bounding_grid(points) + + case Wgrib2.extract_grid_from_file(grib_path, match_pattern, grid_spec) do + {:ok, grid_data} -> + result = + Map.new(points, fn {lat, lon} -> + nearest = nearest_grid_cell(grid_data, lat, lon) + profile = if nearest, do: build_native_profile(nearest), else: %{level_count: 0} + {{lat, lon}, profile} + end) + + {:ok, result} + + error -> + error + end + end + @doc """ Extract native profiles for a list of `{lat, lon}` points from a GRIB2 binary, using wgrib2 for speed. diff --git a/lib/microwaveprop/workers/hrrr_native_grid_worker.ex b/lib/microwaveprop/workers/hrrr_native_grid_worker.ex index e010ddb0..0741493b 100644 --- a/lib/microwaveprop/workers/hrrr_native_grid_worker.ex +++ b/lib/microwaveprop/workers/hrrr_native_grid_worker.ex @@ -105,13 +105,23 @@ defmodule Microwaveprop.Workers.HrrrNativeGridWorker do url = HrrrNativeClient.hrrr_native_url(date, hour) idx_url = url <> ".idx" + tmp_grib = Path.join(System.tmp_dir!(), "hrrr_native_#{System.unique_integer([:positive])}.grib2") + + try do + fetch_and_upsert_with_file(date, hour, valid_time, points, url, idx_url, tmp_grib) + after + File.rm(tmp_grib) + end + end + + defp fetch_and_upsert_with_file(_date, _hour, valid_time, points, url, idx_url, tmp_grib) do with {:ok, idx_text} <- fetch_idx(idx_url), idx_entries = HrrrClient.parse_idx(idx_text), ranges = HrrrNativeClient.essential_byte_ranges(idx_entries), _ = Logger.info("HRRR native downloading #{length(ranges)} ranges for #{valid_time}"), - {:ok, grib_binary} <- HrrrClient.download_grib_ranges(url, ranges), - _ = Logger.info("HRRR native downloaded #{byte_size(grib_binary)} bytes, extracting #{length(points)} points..."), - {:ok, profiles} <- HrrrNativeClient.extract_native_profiles(grib_binary, points) do + :ok <- HrrrClient.download_grib_ranges_to_file(url, ranges, tmp_grib), + _ = Logger.info("HRRR native downloaded to #{tmp_grib}, extracting #{length(points)} points..."), + {:ok, profiles} <- HrrrNativeClient.extract_native_profiles_from_file(tmp_grib, points) do rows = Enum.flat_map(profiles, fn {{lat, lon}, profile} -> if profile[:level_count] && profile.level_count > 0 do