refactor: deduplicate hrrr download, simplify radio/mqtt, fix credo/perf issues

- hrrr_client: extract shared fetch_grib_ranges/2 to eliminate ~80% duplicated code
- mqtt: encode_varint/1 uses IO list accumulator instead of O(n^2) binary concat
- commercial: single-pass reduce for aggregate_degradation, remove unused average/1
- scorer: single-pass reduce for safe_avg and build_path_conditions
- terrain_analysis: split 76-line do_analyse/4 into 3 focused functions
- radio: build_position_changes/3 simplified from then-chains to Enum.reduce
- recalibrator_test: fix credo warning (length > 0 -> != [])
This commit is contained in:
Graham McIntire 2026-05-07 11:50:29 -05:00
parent fe2ad2df15
commit a0065e2f44
No known key found for this signature in database
GPG key ID: F4ABF488E6029E59
7 changed files with 109 additions and 127 deletions

View file

@ -102,14 +102,19 @@ defmodule Microwaveprop.Commercial do
defp aggregate_degradation([]), do: nil
defp aggregate_degradation(list) do
baseline_avg = average(Enum.map(list, & &1.baseline_dbm))
current_avg = average(Enum.map(list, & &1.current_dbm))
{baseline_total, current_total, count} =
Enum.reduce(list, {0.0, 0.0, 0}, fn item, {bt, ct, c} ->
{bt + item.baseline_dbm, ct + item.current_dbm, c + 1}
end)
baseline_avg = baseline_total / count
current_avg = current_total / count
%{
degradation_db: Float.round(baseline_avg - current_avg, 2),
baseline_dbm: Float.round(baseline_avg, 2),
current_dbm: Float.round(current_avg, 2),
n_links: length(list)
n_links: count
}
end
@ -172,9 +177,6 @@ defmodule Microwaveprop.Commercial do
# which module asked.
defp haversine_km(lat1, lon1, lat2, lon2), do: Microwaveprop.Radio.haversine_km(lat1, lon1, lat2, lon2)
defp average([]), do: 0.0
defp average(list), do: Enum.sum(list) / length(list)
@spec create_link(map()) :: {:ok, Link.t()} | {:error, Ecto.Changeset.t()}
def create_link(attrs) do
%Link{}

View file

@ -638,8 +638,10 @@ defmodule Microwaveprop.Propagation.Scorer do
defp build_path_conditions(%{temps: temps, dewpoints: dewpoints} = buckets, contact) do
lon = contact.pos1["lon"] || -97.0
avg_temp_c = Enum.sum(temps) / length(temps)
avg_dewpoint_c = Enum.sum(dewpoints) / length(dewpoints)
{sum_t, count_t} = Enum.reduce(temps, {0, 0}, fn x, {s, c} -> {s + x, c + 1} end)
avg_temp_c = sum_t / count_t
{sum_d, count_d} = Enum.reduce(dewpoints, {0, 0}, fn x, {s, c} -> {s + x, c + 1} end)
avg_dewpoint_c = sum_d / count_d
%{
abs_humidity: absolute_humidity(avg_temp_c, avg_dewpoint_c),
@ -661,5 +663,9 @@ defmodule Microwaveprop.Propagation.Scorer do
end
defp safe_avg([]), do: nil
defp safe_avg(list), do: Enum.sum(list) / length(list)
defp safe_avg(list) do
{sum, count} = Enum.reduce(list, {0, 0}, fn x, {s, c} -> {s + x, c + 1} end)
sum / count
end
end

View file

@ -152,10 +152,16 @@ defmodule Microwaveprop.Pskr.Mqtt do
@doc false
@spec encode_varint(non_neg_integer()) :: binary()
def encode_varint(n) when n in 0..127, do: <<n>>
def encode_varint(n) do
n
|> encode_varint_io([])
|> IO.iodata_to_binary()
end
def encode_varint(n) when n <= 268_435_455 do
<<(n &&& 0x7F) ||| 0x80>> <> encode_varint(n >>> 7)
defp encode_varint_io(n, acc) when n <= 127, do: [acc, <<n>>]
defp encode_varint_io(n, acc) when n <= 268_435_455 do
encode_varint_io(n >>> 7, [acc, <<(n &&& 0x7F) ||| 0x80>>])
end
@doc false

View file

@ -559,11 +559,8 @@ defmodule Microwaveprop.Radio do
defp compute_distance(_pos1, _pos2, existing_distance), do: existing_distance
defp build_position_changes(contact, pos1, pos2, distance) do
%{}
|> then(fn c -> if pos1 && is_nil(contact.pos1), do: Map.put(c, :pos1, pos1), else: c end)
|> then(fn c -> if pos2 && is_nil(contact.pos2), do: Map.put(c, :pos2, pos2), else: c end)
|> then(fn c ->
if distance && is_nil(contact.distance_km), do: Map.put(c, :distance_km, distance), else: c
Enum.reduce([pos1: pos1, pos2: pos2, distance_km: distance], %{}, fn {key, val}, acc ->
if val && is_nil(Map.get(contact, key)), do: Map.put(acc, key, val), else: acc
end)
end

View file

@ -68,63 +68,18 @@ defmodule Microwaveprop.Terrain.TerrainAnalysis do
lambda_m = 0.3 / freq_ghz
n = length(profile) - 1
elev1 = hd(profile).elev + ant_ht_a
elev2 = List.last(profile).elev + ant_ht_b
first = hd(profile)
last = List.last(profile)
elev1 = first.elev + ant_ht_a
elev2 = last.elev + ant_ht_b
# Build effective profile with earth curvature correction
points =
profile
|> Enum.with_index()
|> Enum.map(fn {p, i} ->
frac = i / n
d1_m = frac * dist_km * 1000
d2_m = (1 - frac) * dist_km * 1000
beam = beam_height(frac, elev1, elev2)
bulge = earth_bulge(frac, dist_km, k)
eff_terrain = p.elev + bulge
r1 = fresnel_radius(d1_m, d2_m, lambda_m)
clearance = beam - eff_terrain
f1_clear = if r1 > 0, do: clearance - r1, else: clearance
is_endpoint = i == 0 or i == n
%{
lat: p.lat,
lon: p.lon,
d: p.d,
elev: p.elev,
dist_km: p.dist_km,
idx: i,
beam: beam,
eff_terrain: eff_terrain,
bulge: bulge,
r1: r1,
d1_m: d1_m,
d2_m: d2_m,
clearance: clearance,
f1_clear: f1_clear,
obstructed: not is_endpoint and clearance < 0,
fresnel_penetrated: not is_endpoint and r1 > 0 and f1_clear < 0 and clearance >= 0
}
end)
points = build_analysis_points(profile, n, dist_km, lambda_m, k, elev1, elev2)
interior = Enum.slice(points, 1..-2//1)
obstructed = Enum.filter(interior, & &1.obstructed)
fresnel_hit = Enum.filter(interior, & &1.fresnel_penetrated)
# Deygout diffraction: compute loss across all interior points
diffraction_db = deygout_diffraction(interior, lambda_m)
verdict = classify_verdict(diffraction_db, obstructed, fresnel_hit)
elev_vals = Enum.map(profile, & &1.elev)
clearance_vals = Enum.map(interior, & &1.clearance)
max_elevation_m = Enum.max(elev_vals, fn -> 0.0 end)
min_clearance_m = if clearance_vals == [], do: 999.0, else: Enum.min(clearance_vals)
{max_elevation_m, min_clearance_m} = compute_terrain_summary(profile, interior)
%{
points: points,
@ -138,6 +93,56 @@ defmodule Microwaveprop.Terrain.TerrainAnalysis do
}
end
defp build_analysis_points(profile, n, dist_km, lambda_m, k, elev1, elev2) do
profile
|> Enum.with_index()
|> Enum.map(fn {p, i} ->
frac = i / n
d1_m = frac * dist_km * 1000
d2_m = (1 - frac) * dist_km * 1000
beam = beam_height(frac, elev1, elev2)
bulge = earth_bulge(frac, dist_km, k)
r1 = fresnel_radius(d1_m, d2_m, lambda_m)
clearance = beam - (p.elev + bulge)
is_endpoint = i == 0 or i == n
f1_clear = if r1 > 0, do: clearance - r1, else: clearance
%{
lat: p.lat,
lon: p.lon,
d: p.d,
elev: p.elev,
dist_km: p.dist_km,
idx: i,
beam: beam,
eff_terrain: p.elev + bulge,
bulge: bulge,
r1: r1,
d1_m: d1_m,
d2_m: d2_m,
clearance: clearance,
f1_clear: f1_clear,
obstructed: not is_endpoint and clearance < 0,
fresnel_penetrated: not is_endpoint and r1 > 0 and f1_clear < 0 and clearance >= 0
}
end)
end
defp compute_terrain_summary(profile, interior) do
max_elevation_m = Enum.reduce(profile, 0.0, fn p, acc -> max(p.elev, acc) end)
min_clearance_m =
case interior do
[] -> 999.0
_ -> Enum.reduce(interior, :infinity, fn p, acc -> min(p.clearance, acc) end)
end
{max_elevation_m, min_clearance_m}
end
@doc "First Fresnel zone radius at a point between two antennas."
@spec fresnel_radius(number(), number(), number()) :: number()
def fresnel_radius(d1_m, d2_m, lambda_m) do

View file

@ -134,10 +134,9 @@ defmodule Microwaveprop.Weather.HrrrClient do
{:ok, profiles}
{:error, reason} = error ->
{:error, _reason} = error ->
# Drain the pressure task so it doesn't leak past this scope.
_ = Task.await(prs_task, 20 * 60 * 1000)
_ = reason
error
end
end
@ -540,46 +539,11 @@ 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
# Fetch ranges concurrently — Range requests against the HRRR
# mirror are independent and typically complete in ~200-500ms each
# when sequential. With 40+ ranges per native-duct fetch, serial
# download dominated the per-forecast-hour wall time. 8-way
# parallelism drops a 20s download to ~3s while still respecting
# the mirror's modest connection concurrency. Routed through the
# partitioned Task.Supervisor to avoid contention when the hourly
# PropagationGridWorker fetches f00-f18 alongside recalibrator/GEFS.
merged = merge_ranges(ranges)
supervisor = {:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor, self()}}
case fetch_grib_ranges(url, ranges) do
{:error, _} = error ->
error
results =
supervisor
|> Task.Supervisor.async_stream_nolink(
merged,
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: 8,
timeout: 120_000
)
|> Enum.map(fn
{:ok, result} ->
result
{:exit, reason} ->
Logger.error("HRRR grib range download crashed (to_file): url=#{url} reason=#{inspect(reason)}")
{:error, reason}
end)
case Enum.find(results, &match?({:error, _}, &1)) do
{:error, reason} ->
{:error, reason}
nil ->
{:ok, results} ->
file = File.open!(dest_path, [:write, :binary])
try do
@ -595,13 +559,25 @@ defmodule Microwaveprop.Weather.HrrrClient do
end
defp download_grib_ranges_remote(url, ranges) do
# Merge adjacent/overlapping ranges to reduce HTTP requests
case fetch_grib_ranges(url, ranges) do
{:error, _} = error ->
error
{:ok, results} ->
binary =
results
|> Enum.sort_by(fn {:ok, offset, _} -> offset end)
|> Enum.map(fn {:ok, _, body} -> body end)
|> IO.iodata_to_binary()
{:ok, binary}
end
end
defp fetch_grib_ranges(url, ranges) do
merged = merge_ranges(ranges)
supervisor = {:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor, self()}}
# Download ranges in parallel, routed through the partitioned
# Task.Supervisor so concurrent f00-f18 fetches do not serialize on
# a single supervisor PID.
results =
supervisor
|> Task.Supervisor.async_stream_nolink(
@ -622,23 +598,13 @@ defmodule Microwaveprop.Weather.HrrrClient do
{:exit, reason} ->
Logger.error("HRRR grib range download crashed: url=#{url} reason=#{inspect(reason)}")
{:error, reason}
end)
# Sort by offset and concatenate
case Enum.find(results, &match?({:error, _}, &1)) do
{:error, reason} ->
{:error, reason}
nil ->
binary =
results
|> Enum.sort_by(fn {:ok, offset, _} -> offset end)
|> Enum.map(fn {:ok, _, body} -> body end)
|> IO.iodata_to_binary()
{:ok, binary}
if error = Enum.find(results, &match?({:error, _}, &1)) do
error
else
{:ok, results}
end
end

View file

@ -74,7 +74,7 @@ defmodule Microwaveprop.Pskr.RecalibratorTest do
assert run.avg_spots_per_sample > 0
bins = Repo.all(from(fb in FeatureBin, where: fb.run_id == ^run.id))
assert length(bins) > 0
assert bins != []
# Every bin row should reference the run, target this band, and
# carry non-negative stats.