diff --git a/lib/microwaveprop/propagation.ex b/lib/microwaveprop/propagation.ex index 4a3a52f2..dd2cbef7 100644 --- a/lib/microwaveprop/propagation.ex +++ b/lib/microwaveprop/propagation.ex @@ -106,26 +106,24 @@ defmodule Microwaveprop.Propagation do def upsert_scores(scores, opts \\ []) do now = DateTime.truncate(DateTime.utc_now(), :second) - entries = - Enum.map(scores, fn s -> - %{ - id: Ecto.UUID.generate(), - lat: s.lat, - lon: s.lon, - valid_time: s.valid_time, - band_mhz: s.band_mhz, - score: s.score, - factors: s.factors, - inserted_at: now, - updated_at: now - } - end) - result = Repo.transaction( fn -> - entries - |> Enum.chunk_every(500) + scores + |> Stream.map(fn s -> + %{ + id: Ecto.UUID.generate(), + lat: s.lat, + lon: s.lon, + valid_time: s.valid_time, + band_mhz: s.band_mhz, + score: s.score, + factors: s.factors, + inserted_at: now, + updated_at: now + } + end) + |> Stream.chunk_every(500) |> Enum.reduce(0, fn chunk, acc -> {count, _} = Repo.insert_all(GridScore, chunk, diff --git a/lib/microwaveprop/propagation/scorer.ex b/lib/microwaveprop/propagation/scorer.ex index 81e7d008..fa096ab3 100644 --- a/lib/microwaveprop/propagation/scorer.ex +++ b/lib/microwaveprop/propagation/scorer.ex @@ -388,8 +388,18 @@ defmodule Microwaveprop.Propagation.Scorer do def path_integrated_conditions(profiles, contact) do lon = Kernel.||(contact.pos1["lon"] || contact.pos1["lng"], -97.0) - temps = profiles |> Enum.map(& &1.surface_temp_c) |> Enum.reject(&is_nil/1) - dewpoints = profiles |> Enum.map(& &1.surface_dewpoint_c) |> Enum.reject(&is_nil/1) + # Single-pass extraction of all fields to avoid 6 separate Enum traversals + {temps, dewpoints, pressures, gradients, bl_depths, pwats} = + Enum.reduce(profiles, {[], [], [], [], [], []}, fn p, {t, d, pr, g, b, pw} -> + { + if(is_nil(p.surface_temp_c), do: t, else: [p.surface_temp_c | t]), + if(is_nil(p.surface_dewpoint_c), do: d, else: [p.surface_dewpoint_c | d]), + if(is_nil(p.surface_pressure_mb), do: pr, else: [p.surface_pressure_mb | pr]), + if(is_nil(p.min_refractivity_gradient), do: g, else: [p.min_refractivity_gradient | g]), + if(is_nil(p.hpbl_m), do: b, else: [p.hpbl_m | b]), + if(is_nil(p.pwat_mm), do: pw, else: [p.pwat_mm | pw]) + } + end) if temps == [] or dewpoints == [] do nil @@ -399,11 +409,6 @@ defmodule Microwaveprop.Propagation.Scorer do avg_temp_f = c_to_f(avg_temp_c) avg_dewpoint_f = c_to_f(avg_dewpoint_c) - pressures = profiles |> Enum.map(& &1.surface_pressure_mb) |> Enum.reject(&is_nil/1) - gradients = profiles |> Enum.map(& &1.min_refractivity_gradient) |> Enum.reject(&is_nil/1) - bl_depths = profiles |> Enum.map(& &1.hpbl_m) |> Enum.reject(&is_nil/1) - pwats = profiles |> Enum.map(& &1.pwat_mm) |> Enum.reject(&is_nil/1) - %{ abs_humidity: absolute_humidity(avg_temp_c, avg_dewpoint_c), temp_f: avg_temp_f, diff --git a/lib/microwaveprop/weather/hrrr_client.ex b/lib/microwaveprop/weather/hrrr_client.ex index 6962e184..8de3e9c0 100644 --- a/lib/microwaveprop/weather/hrrr_client.ex +++ b/lib/microwaveprop/weather/hrrr_client.ex @@ -63,10 +63,12 @@ defmodule Microwaveprop.Weather.HrrrClient do %{} end - merged = merge_grid_data(sfc_grid, prs_grid) - + # Merge and build profiles in one pass to avoid a full intermediate map copy. + # Use Map.merge to include any pressure-only points, then map directly to profiles. profiles = - Map.new(merged, fn {point, data} -> + sfc_grid + |> Map.merge(prs_grid, fn _point, sfc_data, prs_data -> Map.merge(sfc_data, prs_data) end) + |> Map.new(fn {point, data} -> profile = build_profile(data) {point, Map.put(profile, :run_time, hour_dt)} end) @@ -350,7 +352,7 @@ defmodule Microwaveprop.Weather.HrrrClient do 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) |> Enum.sort_by(&elem(&1, 0)) + merged = ranges |> merge_ranges() |> Enum.sort_by(&elem(&1, 0)) file = File.open!(dest_path, [:write, :binary]) diff --git a/lib/microwaveprop/workers/hrrr_native_grid_worker.ex b/lib/microwaveprop/workers/hrrr_native_grid_worker.ex index 0741493b..04a3b1a0 100644 --- a/lib/microwaveprop/workers/hrrr_native_grid_worker.ex +++ b/lib/microwaveprop/workers/hrrr_native_grid_worker.ex @@ -129,7 +129,7 @@ defmodule Microwaveprop.Workers.HrrrNativeGridWorker do [ Map.merge(profile, %{ - id: Ecto.UUID.bingenerate(), + id: Ecto.UUID.generate(), valid_time: valid_time, run_time: valid_time, lat: lat, diff --git a/lib/microwaveprop/workers/propagation_grid_worker.ex b/lib/microwaveprop/workers/propagation_grid_worker.ex index 60a3bdf9..8e3b5cec 100644 --- a/lib/microwaveprop/workers/propagation_grid_worker.ex +++ b/lib/microwaveprop/workers/propagation_grid_worker.ex @@ -40,6 +40,8 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do for fh <- 0..@max_forecast_hour do valid_time = DateTime.add(run_time, fh * 3600, :second) result = process_forecast_hour(points, run_time, fh, valid_time) + # Reclaim grid_data/profiles from this forecast hour before starting the next + :erlang.garbage_collect() case result do :ok -> valid_time @@ -74,6 +76,7 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do end) do {:ok, grid_data} -> store_hrrr_profiles(grid_data, valid_time) + :erlang.garbage_collect() Phoenix.PubSub.broadcast( Microwaveprop.PubSub, @@ -111,8 +114,9 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do defp format_duration(ms), do: "#{Float.round(ms / 1000, 1)}s" defp store_hrrr_profiles(grid_data, valid_time) do - stored = - Enum.flat_map(grid_data, fn {{lat, lon}, profile} -> + count = + grid_data + |> Stream.flat_map(fn {{lat, lon}, profile} -> temp = profile.surface_temp_c if temp && temp > -80 && temp < 60 do @@ -150,14 +154,13 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do [] end end) + |> Stream.chunk_every(500) + |> Enum.reduce(0, fn chunk, acc -> + Weather.upsert_hrrr_profiles_batch(chunk) + acc + length(chunk) + end) - stored - |> Enum.chunk_every(500) - |> Enum.each(fn chunk -> - Weather.upsert_hrrr_profiles_batch(chunk) - end) - - Logger.info("PropagationGrid: stored #{length(stored)} HRRR profiles") + Logger.info("PropagationGrid: stored #{count} HRRR profiles") end defp compute_scores(grid_data, valid_time) do @@ -178,7 +181,7 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do max_concurrency: System.schedulers_online() * 2, timeout: 30_000 ) - |> Enum.flat_map(fn + |> Stream.flat_map(fn {:ok, results} -> results {:exit, _reason} -> [] end)