Graham McIntire
1400c38f44
Rework of d61fbd3's dialyzer suppression: the 4 warnings from
LiveTable.LiveResource's macro expansion are now fixed at the
root rather than hidden via .dialyzer_ignore.exs.
Changes:
- Delete .dialyzer_ignore.exs and remove ignore_warnings from
mix.exs. No more hidden suppressions.
- New MicrowavepropWeb.LiveTableResource shim that wraps
use LiveTable.LiveResource and overrides:
* maybe_subscribe/1 — explicit :ok = on Phoenix.PubSub.subscribe/2
(previously discarded, tripping :unmatched_return).
* handle_info/2 — explicit _ = on File.rm, Process.send_after.
Switch all 4 LiveTable-using LiveViews (contact_live/index,
beacon_live/index, admin/contact_edit_live, user_management_live/index)
to use MicrowavepropWeb.LiveTableResource instead.
- New priv/dep_patches/live_table-0.4.1.patch adds _ = in front of
LiveTable.LiveSelectHelpers.restore_live_select_from_params/2 in
the dep's macro-expanded handle_params/3. The call's return was
silently discarded, tripping the fourth warning per LiveView.
- New lib/mix/tasks/deps.patch.ex applies every
priv/dep_patches/*.patch to its target dep idempotently after
mix deps.get. Wired into the :setup alias + overridden
mix deps.get so the patch survives cache regeneration in CI.
- Other modified files in this commit are the spec-coverage
additions from the parallel agent pass (beacons.ex,
commercial.ex, propagation.ex, weather.ex, narr_client.ex,
run_timing.ex, 3 workers) — tighter specs for bounds, tuple
shapes, schema-typed params. Dialyzer remains at 0 after
the changes.
Upstream TODO: send a PR for the live_table `_ =` fix. Once
merged and released, drop the patch + mix task + bump the dep.
mix dialyzer --format short | grep ^lib/ | wc -l -> 0
mix test: 2165 tests, 2 pre-existing flakes (MapLive timestamp +
PropagationPrune ScoresFile), 0 regressions.
98 lines
2.9 KiB
Elixir
98 lines
2.9 KiB
Elixir
defmodule Microwaveprop.Workers.NexradWorker do
|
|
@moduledoc """
|
|
Fetches one n0q NEXRAD composite frame and stores per-point
|
|
observations in `nexrad_observations`.
|
|
|
|
Jobs are unique on `{year, month, day, hour, minute}` so backfill
|
|
sweeps that enqueue duplicate timestamps collapse automatically.
|
|
"""
|
|
|
|
use Oban.Worker,
|
|
queue: :nexrad,
|
|
max_attempts: 3,
|
|
unique: [
|
|
period: :infinity,
|
|
states: [:available, :scheduled, :executing, :retryable],
|
|
keys: [:year, :month, :day, :hour, :minute]
|
|
]
|
|
|
|
import Ecto.Query
|
|
|
|
alias Microwaveprop.Radio.Contact
|
|
alias Microwaveprop.Repo
|
|
alias Microwaveprop.Weather.NexradClient
|
|
alias Microwaveprop.Weather.NexradObservation
|
|
|
|
require Logger
|
|
|
|
@impl Oban.Worker
|
|
def perform(%Oban.Job{args: args}) do
|
|
%{"year" => year, "month" => month, "day" => day, "hour" => hour} = args
|
|
minute = Map.get(args, "minute", 0)
|
|
{:ok, date} = Date.new(year, month, day)
|
|
{:ok, time} = Time.new(hour, minute, 0)
|
|
{:ok, timestamp} = DateTime.new(date, time, "Etc/UTC")
|
|
|
|
points = points_of_interest_for_time(timestamp)
|
|
|
|
if points == [] do
|
|
Logger.info("NexradWorker: no points for #{timestamp}, skipping")
|
|
:ok
|
|
else
|
|
fetch_and_upsert(timestamp, points)
|
|
end
|
|
end
|
|
|
|
@doc false
|
|
@spec points_of_interest_for_time(DateTime.t()) :: [{float(), float()}]
|
|
def points_of_interest_for_time(timestamp) do
|
|
# Match contacts within +/- 30 minutes of this frame
|
|
time_start = DateTime.add(timestamp, -1800, :second)
|
|
time_end = DateTime.add(timestamp, 1800, :second)
|
|
|
|
Contact
|
|
|> where([c], not is_nil(c.pos1))
|
|
|> where([c], c.qso_timestamp >= ^time_start and c.qso_timestamp <= ^time_end)
|
|
|> select([c], c.pos1)
|
|
|> Repo.all()
|
|
|> Enum.flat_map(fn pos ->
|
|
case {pos["lat"], pos["lon"]} do
|
|
{lat, lon} when is_number(lat) and is_number(lon) -> [{snap(lat), snap(lon)}]
|
|
_ -> []
|
|
end
|
|
end)
|
|
|> Enum.uniq()
|
|
end
|
|
|
|
defp snap(x), do: Float.round(x * 1.0, 3)
|
|
|
|
defp fetch_and_upsert(timestamp, points) do
|
|
case NexradClient.fetch_frame(timestamp, points) do
|
|
{:ok, observations} ->
|
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
|
|
|
rows =
|
|
Enum.map(observations, fn obs ->
|
|
obs
|
|
|> Map.put(:id, Ecto.UUID.generate())
|
|
|> Map.put(:inserted_at, now)
|
|
|> Map.put(:updated_at, now)
|
|
end)
|
|
|
|
{inserted, _} =
|
|
Repo.insert_all(
|
|
NexradObservation,
|
|
rows,
|
|
on_conflict: {:replace_all_except, [:id, :inserted_at]},
|
|
conflict_target: [:lat, :lon, :observed_at]
|
|
)
|
|
|
|
Logger.info("NexradWorker: upserted #{inserted} observations for #{timestamp}")
|
|
:ok
|
|
|
|
{:error, reason} ->
|
|
Logger.warning("NexradWorker failed for #{timestamp}: #{inspect(reason)}")
|
|
{:error, reason}
|
|
end
|
|
end
|
|
end
|