Database performance fixes and async backfill enqueue
- Fix score_pressure crash on nil pressure_mb (coastal HRRR points) - Set 10-min timeout on grid score upsert transaction (was :infinity) - Single DELETE for prune_old_scores instead of N queries in a loop - Remove dead load_hrrr_refractivity that loaded 95k rows into nil map - Pass selected_time to point_detail to skip latest_valid_time sub-query - Batch station existence checks (1 query per path point, not per station) - Batch solar index upserts via insert_all in chunks of 500 - Batch backfill_distances via single UPDATE FROM VALUES statement - Add is_grid_point boolean + partial index to hrrr_profiles (replaces non-sargable modular arithmetic filter on every weather map query) - Add partial index on contacts(qso_timestamp) WHERE pos1 IS NOT NULL - Move backfill enqueue to Oban worker so UI returns immediately
This commit is contained in:
parent
a209fc65ae
commit
16883591b4
16 changed files with 235 additions and 115 deletions
|
|
@ -71,7 +71,17 @@ config :microwaveprop, MicrowavepropWeb.Endpoint,
|
|||
]
|
||||
|
||||
config :microwaveprop, Oban,
|
||||
queues: [propagation: 1, solar: 1, weather: 20, enqueue: 1, hrrr: 5, terrain: 4, commercial: 2, iemre: 10],
|
||||
queues: [
|
||||
propagation: 1,
|
||||
solar: 1,
|
||||
weather: 20,
|
||||
enqueue: 1,
|
||||
hrrr: 5,
|
||||
terrain: 4,
|
||||
commercial: 2,
|
||||
iemre: 10,
|
||||
backfill_enqueue: 1
|
||||
],
|
||||
plugins: [
|
||||
{Oban.Plugins.Pruner, max_age: 3600 * 24},
|
||||
{Oban.Plugins.Lifeline, rescue_after: to_timeout(minute: 30)},
|
||||
|
|
|
|||
|
|
@ -117,7 +117,7 @@ if config_env() == :prod do
|
|||
|
||||
# Production Oban: live scoring, polling, and on-demand QSO enrichment (no cron backfill)
|
||||
config :microwaveprop, Oban,
|
||||
queues: [propagation: 1, commercial: 2, solar: 1, weather: 10, hrrr: 10, terrain: 5, iemre: 10],
|
||||
queues: [propagation: 1, commercial: 2, solar: 1, weather: 10, hrrr: 10, terrain: 5, iemre: 10, backfill_enqueue: 1],
|
||||
plugins: [
|
||||
{Oban.Plugins.Pruner, max_age: 3600 * 24},
|
||||
{Oban.Plugins.Lifeline, rescue_after: 30 * 60 * 1000},
|
||||
|
|
|
|||
|
|
@ -146,7 +146,7 @@ defmodule Microwaveprop.Propagation do
|
|||
acc + count
|
||||
end)
|
||||
end,
|
||||
timeout: :infinity
|
||||
timeout: 600_000
|
||||
)
|
||||
|
||||
if Keyword.get(opts, :prune, true) do
|
||||
|
|
@ -159,28 +159,14 @@ defmodule Microwaveprop.Propagation do
|
|||
result
|
||||
end
|
||||
|
||||
@doc "Remove scores with valid_times older than 1 hour before the earliest current forecast."
|
||||
@doc "Remove scores with valid_times older than 2 hours."
|
||||
def prune_old_scores do
|
||||
cutoff = DateTime.add(DateTime.utc_now(), -2, :hour)
|
||||
|
||||
# Get distinct old valid_times and delete one at a time to avoid massive single DELETEs
|
||||
old_times =
|
||||
Repo.all(
|
||||
from(gs in GridScore,
|
||||
where: gs.valid_time < ^cutoff,
|
||||
select: gs.valid_time,
|
||||
distinct: gs.valid_time
|
||||
)
|
||||
)
|
||||
{deleted, _} = Repo.delete_all(from(gs in GridScore, where: gs.valid_time < ^cutoff), timeout: 120_000)
|
||||
|
||||
total =
|
||||
Enum.reduce(old_times, 0, fn vt, acc ->
|
||||
{deleted, _} = Repo.delete_all(from(gs in GridScore, where: gs.valid_time == ^vt))
|
||||
acc + deleted
|
||||
end)
|
||||
|
||||
if total > 0 do
|
||||
Logger.info("PropagationScores: pruned #{total} old scores across #{length(old_times)} valid_times")
|
||||
if deleted > 0 do
|
||||
Logger.info("PropagationScores: pruned #{deleted} old scores (before #{cutoff})")
|
||||
end
|
||||
end
|
||||
|
||||
|
|
@ -265,21 +251,23 @@ defmodule Microwaveprop.Propagation do
|
|||
end
|
||||
|
||||
@doc "Get the full score and factors for a specific grid point, snapped to nearest grid."
|
||||
def point_detail(band_mhz, lat, lon) do
|
||||
def point_detail(band_mhz, lat, lon, valid_time \\ nil) do
|
||||
step = Grid.step()
|
||||
snapped_lat = Float.round(Float.round(lat / step) * step, 3)
|
||||
snapped_lon = Float.round(Float.round(lon / step) * step, 3)
|
||||
|
||||
case latest_valid_time(band_mhz) do
|
||||
time = valid_time || latest_valid_time(band_mhz)
|
||||
|
||||
case time do
|
||||
nil ->
|
||||
nil
|
||||
|
||||
latest_time ->
|
||||
_ ->
|
||||
Repo.one(
|
||||
from(gs in GridScore,
|
||||
where:
|
||||
gs.band_mhz == ^band_mhz and
|
||||
gs.valid_time == ^latest_time and
|
||||
gs.valid_time == ^time and
|
||||
gs.lat == ^snapped_lat and
|
||||
gs.lon == ^snapped_lon,
|
||||
select: %{
|
||||
|
|
|
|||
|
|
@ -303,7 +303,9 @@ defmodule Microwaveprop.Propagation.Scorer do
|
|||
Without previous reading, scores based on absolute pressure.
|
||||
With previous reading, scores based on pressure change (delta).
|
||||
"""
|
||||
@spec score_pressure(number(), number() | nil) :: integer()
|
||||
@spec score_pressure(number() | nil, number() | nil) :: integer()
|
||||
def score_pressure(nil, _previous_mb), do: 50
|
||||
|
||||
def score_pressure(current_mb, nil) do
|
||||
cond do
|
||||
current_mb < 1005 -> 80
|
||||
|
|
|
|||
|
|
@ -195,22 +195,35 @@ defmodule Microwaveprop.Radio do
|
|||
end
|
||||
|
||||
def backfill_distances(contacts) do
|
||||
Enum.each(contacts, fn contact ->
|
||||
if is_nil(contact.distance_km) && contact.pos1 && contact.pos2 do
|
||||
lat1 = contact.pos1["lat"]
|
||||
lon1 = contact.pos1["lon"] || contact.pos1["lng"]
|
||||
lat2 = contact.pos2["lat"]
|
||||
lon2 = contact.pos2["lon"] || contact.pos2["lng"]
|
||||
|
||||
if lat1 && lon1 && lat2 && lon2 do
|
||||
dist = lat1 |> haversine_km(lon1, lat2, lon2) |> round() |> Decimal.new()
|
||||
|
||||
Contact
|
||||
|> where([q], q.id == ^contact.id)
|
||||
|> Repo.update_all(set: [distance_km: dist])
|
||||
end
|
||||
updates =
|
||||
for contact <- contacts,
|
||||
is_nil(contact.distance_km) && contact.pos1 && contact.pos2,
|
||||
lat1 = contact.pos1["lat"],
|
||||
lon1 = contact.pos1["lon"] || contact.pos1["lng"],
|
||||
lat2 = contact.pos2["lat"],
|
||||
lon2 = contact.pos2["lon"] || contact.pos2["lng"],
|
||||
lat1 && lon1 && lat2 && lon2 do
|
||||
dist = lat1 |> haversine_km(lon1, lat2, lon2) |> round()
|
||||
{contact.id, dist}
|
||||
end
|
||||
end)
|
||||
|
||||
if updates != [] do
|
||||
# Build a single UPDATE ... FROM (VALUES ...) statement
|
||||
# Ecto.UUID.dump!/1 converts string UUIDs to the 16-byte binary Postgrex expects
|
||||
{values_sql, params} =
|
||||
updates
|
||||
|> Enum.with_index()
|
||||
|> Enum.reduce({"", []}, fn {{id, dist}, i}, {sql, params} ->
|
||||
frag = "($#{i * 2 + 1}::uuid, $#{i * 2 + 2}::numeric)"
|
||||
sep = if sql == "", do: "", else: ", "
|
||||
{sql <> sep <> frag, params ++ [Ecto.UUID.dump!(id), dist]}
|
||||
end)
|
||||
|
||||
Repo.query!(
|
||||
"UPDATE contacts AS c SET distance_km = v.dist FROM (VALUES #{values_sql}) AS v(id, dist) WHERE c.id = v.id",
|
||||
params
|
||||
)
|
||||
end
|
||||
end
|
||||
|
||||
defp deg_to_rad(deg), do: deg * :math.pi() / 180
|
||||
|
|
|
|||
|
|
@ -143,12 +143,80 @@ defmodule Microwaveprop.Weather do
|
|||
|> Repo.exists?()
|
||||
end
|
||||
|
||||
@doc "Returns a MapSet of station_ids that have surface observations in the time window."
|
||||
def station_ids_with_surface_observations(station_ids, start_dt, end_dt) do
|
||||
SurfaceObservation
|
||||
|> where([o], o.station_id in ^station_ids)
|
||||
|> where([o], o.observed_at >= ^start_dt and o.observed_at <= ^end_dt)
|
||||
|> select([o], o.station_id)
|
||||
|> distinct(true)
|
||||
|> Repo.all()
|
||||
|> MapSet.new()
|
||||
end
|
||||
|
||||
def has_sounding?(station_id, observed_at) do
|
||||
Sounding
|
||||
|> where([s], s.station_id == ^station_id and s.observed_at == ^observed_at)
|
||||
|> Repo.exists?()
|
||||
end
|
||||
|
||||
@doc "Returns a MapSet of {station_id, observed_at} tuples that have soundings."
|
||||
def station_ids_with_soundings(station_ids, sounding_times) do
|
||||
Sounding
|
||||
|> where([s], s.station_id in ^station_ids and s.observed_at in ^sounding_times)
|
||||
|> select([s], {s.station_id, s.observed_at})
|
||||
|> distinct(true)
|
||||
|> Repo.all()
|
||||
|> MapSet.new()
|
||||
end
|
||||
|
||||
@doc "Batch upsert solar indices using insert_all in chunks of 500."
|
||||
def upsert_solar_indices_batch(records) do
|
||||
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
|
||||
records
|
||||
|> Enum.chunk_every(500)
|
||||
|> Enum.reduce(0, fn chunk, acc ->
|
||||
entries =
|
||||
Enum.map(chunk, fn attrs ->
|
||||
%{
|
||||
id: Ecto.UUID.generate(),
|
||||
date: attrs[:date] || attrs.date,
|
||||
sfi: attrs[:sfi],
|
||||
sfi_adjusted: attrs[:sfi_adjusted],
|
||||
sunspot_number: attrs[:sunspot_number],
|
||||
ap_index: attrs[:ap_index],
|
||||
kp_values: attrs[:kp_values],
|
||||
inserted_at: now,
|
||||
updated_at: now
|
||||
}
|
||||
end)
|
||||
|
||||
{count, _} =
|
||||
Repo.insert_all(SolarIndex, entries,
|
||||
on_conflict:
|
||||
from(s in SolarIndex,
|
||||
update: [
|
||||
set: [
|
||||
sfi: fragment("EXCLUDED.sfi"),
|
||||
sfi_adjusted: fragment("EXCLUDED.sfi_adjusted"),
|
||||
sunspot_number: fragment("EXCLUDED.sunspot_number"),
|
||||
ap_index: fragment("EXCLUDED.ap_index"),
|
||||
kp_values: fragment("EXCLUDED.kp_values"),
|
||||
updated_at: fragment("EXCLUDED.updated_at")
|
||||
]
|
||||
],
|
||||
where:
|
||||
s.sfi != fragment("EXCLUDED.sfi") or
|
||||
s.ap_index != fragment("EXCLUDED.ap_index")
|
||||
),
|
||||
conflict_target: [:date]
|
||||
)
|
||||
|
||||
acc + count
|
||||
end)
|
||||
end
|
||||
|
||||
def get_solar_index(date) do
|
||||
Repo.get_by(SolarIndex, date: date)
|
||||
end
|
||||
|
|
@ -268,9 +336,7 @@ defmodule Microwaveprop.Weather do
|
|||
def latest_grid_valid_time do
|
||||
Repo.one(
|
||||
from(h in HrrrProfile,
|
||||
where:
|
||||
fragment("(? * 1000)::int % 125 = 0", h.lat) and
|
||||
fragment("(? * 1000)::int % 125 = 0", h.lon),
|
||||
where: h.is_grid_point == true,
|
||||
select: max(h.valid_time)
|
||||
)
|
||||
)
|
||||
|
|
@ -285,10 +351,9 @@ defmodule Microwaveprop.Weather do
|
|||
from(h in HrrrProfile,
|
||||
where:
|
||||
h.valid_time == ^latest_vt and
|
||||
h.is_grid_point == true and
|
||||
h.lat >= ^bounds["south"] and h.lat <= ^bounds["north"] and
|
||||
h.lon >= ^bounds["west"] and h.lon <= ^bounds["east"] and
|
||||
fragment("(? * 1000)::int % 125 = 0", h.lat) and
|
||||
fragment("(? * 1000)::int % 125 = 0", h.lon),
|
||||
h.lon >= ^bounds["west"] and h.lon <= ^bounds["east"],
|
||||
select: %{
|
||||
lat: h.lat,
|
||||
lon: h.lon,
|
||||
|
|
@ -369,8 +434,13 @@ defmodule Microwaveprop.Weather do
|
|||
|> Enum.reduce({0, nil}, fn chunk, {total_count, _} ->
|
||||
entries =
|
||||
Enum.map(chunk, fn attrs ->
|
||||
is_gp =
|
||||
rem(round(attrs.lat * 1000), 125) == 0 and
|
||||
rem(round(attrs.lon * 1000), 125) == 0
|
||||
|
||||
Map.merge(attrs, %{
|
||||
id: Ecto.UUID.generate(),
|
||||
is_grid_point: is_gp,
|
||||
inserted_at: now,
|
||||
updated_at: now
|
||||
})
|
||||
|
|
@ -465,8 +535,7 @@ defmodule Microwaveprop.Weather do
|
|||
"""
|
||||
DELETE FROM hrrr_profiles
|
||||
WHERE valid_time >= $2 AND valid_time < $1
|
||||
AND (lat * 1000)::int % 125 = 0
|
||||
AND (lon * 1000)::int % 125 = 0
|
||||
AND is_grid_point = true
|
||||
""",
|
||||
[cutoff, floor],
|
||||
timeout: 300_000
|
||||
|
|
|
|||
|
|
@ -22,12 +22,13 @@ defmodule Microwaveprop.Weather.HrrrProfile do
|
|||
field :min_refractivity_gradient, :float
|
||||
field :ducting_detected, :boolean, default: false
|
||||
field :duct_characteristics, {:array, :map}
|
||||
field :is_grid_point, :boolean, default: false
|
||||
|
||||
timestamps(type: :utc_datetime)
|
||||
end
|
||||
|
||||
@required_fields ~w(valid_time lat lon)a
|
||||
@optional_fields ~w(run_time profile hpbl_m pwat_mm surface_temp_c surface_dewpoint_c surface_pressure_mb surface_refractivity min_refractivity_gradient ducting_detected duct_characteristics)a
|
||||
@optional_fields ~w(run_time profile hpbl_m pwat_mm surface_temp_c surface_dewpoint_c surface_pressure_mb surface_refractivity min_refractivity_gradient ducting_detected duct_characteristics is_grid_point)a
|
||||
|
||||
def changeset(hrrr_profile, attrs) do
|
||||
hrrr_profile
|
||||
|
|
|
|||
|
|
@ -96,9 +96,6 @@ defmodule Microwaveprop.Workers.AsosAdjustmentWorker do
|
|||
}
|
||||
end)
|
||||
|
||||
# Get the last HRRR refractivity data for blending
|
||||
hrrr_refractivity = load_hrrr_refractivity()
|
||||
|
||||
# For each grid point, find nearest station and score
|
||||
Grid.conus_points()
|
||||
|> Task.async_stream(
|
||||
|
|
@ -106,7 +103,7 @@ defmodule Microwaveprop.Workers.AsosAdjustmentWorker do
|
|||
nearest = find_nearest_station(obs_list, lat, lon)
|
||||
|
||||
if nearest do
|
||||
score_point_from_obs(point, nearest, valid_time, hrrr_refractivity)
|
||||
score_point_from_obs(point, nearest, valid_time)
|
||||
end
|
||||
end,
|
||||
max_concurrency: System.schedulers_online() * 2,
|
||||
|
|
@ -133,16 +130,13 @@ defmodule Microwaveprop.Workers.AsosAdjustmentWorker do
|
|||
end
|
||||
end
|
||||
|
||||
defp score_point_from_obs({lat, lon}, obs, valid_time, hrrr_refractivity) do
|
||||
defp score_point_from_obs({lat, lon}, obs, valid_time) do
|
||||
temp_c = Scorer.f_to_c(obs.temp_f)
|
||||
dewpoint_c = Scorer.f_to_c(obs.dewpoint_f)
|
||||
|
||||
if is_nil(temp_c) or is_nil(dewpoint_c) do
|
||||
[]
|
||||
else
|
||||
# Get HRRR refractivity for this grid point (from last hourly computation)
|
||||
refractivity = Map.get(hrrr_refractivity, {lat, lon})
|
||||
|
||||
sky_pct = sky_condition_to_pct(obs.sky_condition)
|
||||
rain_rate = precip_to_rate(obs.precip_1h_in)
|
||||
|
||||
|
|
@ -159,7 +153,7 @@ defmodule Microwaveprop.Workers.AsosAdjustmentWorker do
|
|||
pressure_mb: obs.sea_level_pressure_mb,
|
||||
prev_pressure_mb: nil,
|
||||
rain_rate_mmhr: rain_rate,
|
||||
min_refractivity_gradient: refractivity,
|
||||
min_refractivity_gradient: nil,
|
||||
bl_depth_m: nil
|
||||
}
|
||||
|
||||
|
|
@ -178,23 +172,6 @@ defmodule Microwaveprop.Workers.AsosAdjustmentWorker do
|
|||
end
|
||||
end
|
||||
|
||||
# Load the refractivity gradient values from the last HRRR grid computation
|
||||
defp load_hrrr_refractivity do
|
||||
case Propagation.latest_valid_time() do
|
||||
nil ->
|
||||
%{}
|
||||
|
||||
_time ->
|
||||
# Use the 10 GHz band (arbitrary — refractivity factor is the same for all bands)
|
||||
10_000
|
||||
|> Propagation.latest_scores()
|
||||
|> Enum.reduce(%{}, fn %{lat: lat, lon: lon} = _score, acc ->
|
||||
# We don't store refractivity separately, so use nil (neutral score)
|
||||
Map.put(acc, {lat, lon}, nil)
|
||||
end)
|
||||
end
|
||||
end
|
||||
|
||||
defp sky_condition_to_pct(nil), do: nil
|
||||
defp sky_condition_to_pct("CLR"), do: 0.0
|
||||
defp sky_condition_to_pct("SKC"), do: 0.0
|
||||
|
|
|
|||
45
lib/microwaveprop/workers/backfill_enqueue_worker.ex
Normal file
45
lib/microwaveprop/workers/backfill_enqueue_worker.ex
Normal file
|
|
@ -0,0 +1,45 @@
|
|||
defmodule Microwaveprop.Workers.BackfillEnqueueWorker do
|
||||
@moduledoc """
|
||||
Runs backfill enrichment enqueue as an Oban job so the backfill dashboard
|
||||
returns immediately instead of blocking on the enqueue loop.
|
||||
"""
|
||||
use Oban.Worker, queue: :backfill_enqueue, max_attempts: 1
|
||||
|
||||
import Ecto.Query
|
||||
|
||||
alias Microwaveprop.Radio.Contact
|
||||
alias Microwaveprop.Repo
|
||||
alias Microwaveprop.Workers.ContactWeatherEnqueueWorker
|
||||
|
||||
require Logger
|
||||
|
||||
@enrichable [:pending, :failed]
|
||||
|
||||
@impl Oban.Worker
|
||||
def perform(%Oban.Job{args: %{"limit" => limit}}) do
|
||||
contacts =
|
||||
Repo.all(
|
||||
from(c in Contact,
|
||||
where:
|
||||
(not is_nil(c.pos1) or not is_nil(c.grid1)) and
|
||||
(c.hrrr_status in ^@enrichable or c.weather_status in ^@enrichable or
|
||||
c.terrain_status in ^@enrichable or c.iemre_status in ^@enrichable),
|
||||
order_by: [desc: c.qso_timestamp],
|
||||
limit: ^limit
|
||||
)
|
||||
)
|
||||
|
||||
count = length(contacts)
|
||||
Logger.info("BackfillEnqueue: processing #{count} contacts")
|
||||
|
||||
Enum.each(contacts, &ContactWeatherEnqueueWorker.enqueue_for_contact/1)
|
||||
|
||||
Phoenix.PubSub.broadcast(
|
||||
Microwaveprop.PubSub,
|
||||
"backfill:enqueue_complete",
|
||||
{:enqueue_complete, count}
|
||||
)
|
||||
|
||||
:ok
|
||||
end
|
||||
end
|
||||
|
|
@ -198,11 +198,12 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do
|
|||
start_dt = DateTime.add(timestamp, -2 * 3600, :second)
|
||||
end_dt = DateTime.add(timestamp, 2 * 3600, :second)
|
||||
|
||||
lat
|
||||
|> Weather.nearby_stations(lon, "asos", @asos_radius_km)
|
||||
|> Enum.reject(fn station ->
|
||||
Weather.has_surface_observations?(station.id, start_dt, end_dt)
|
||||
end)
|
||||
stations = Weather.nearby_stations(lat, lon, "asos", @asos_radius_km)
|
||||
station_ids = Enum.map(stations, & &1.id)
|
||||
covered = Weather.station_ids_with_surface_observations(station_ids, start_dt, end_dt)
|
||||
|
||||
stations
|
||||
|> Enum.reject(fn station -> MapSet.member?(covered, station.id) end)
|
||||
|> Enum.map(fn station ->
|
||||
WeatherFetchWorker.new(%{
|
||||
"fetch_type" => "asos",
|
||||
|
|
@ -240,12 +241,13 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do
|
|||
|
||||
defp build_raob_jobs(lat, lon, timestamp) do
|
||||
sounding_times = Weather.sounding_times_around(timestamp)
|
||||
|
||||
stations = Weather.nearby_stations(lat, lon, "sounding", @sounding_radius_km)
|
||||
station_ids = Enum.map(stations, & &1.id)
|
||||
covered = Weather.station_ids_with_soundings(station_ids, sounding_times)
|
||||
|
||||
for station <- stations,
|
||||
sounding_time <- sounding_times,
|
||||
not Weather.has_sounding?(station.id, sounding_time) do
|
||||
not MapSet.member?(covered, {station.id, sounding_time}) do
|
||||
WeatherFetchWorker.new(%{
|
||||
"fetch_type" => "raob",
|
||||
"station_id" => station.id,
|
||||
|
|
|
|||
|
|
@ -15,7 +15,7 @@ defmodule Microwaveprop.Workers.SolarIndexWorker do
|
|||
case SolarClient.fetch_solar_indices() do
|
||||
{:ok, all_records} ->
|
||||
matched = Enum.filter(all_records, fn r -> r.date == target_date end)
|
||||
Enum.each(matched, &Weather.upsert_solar_index/1)
|
||||
Weather.upsert_solar_indices_batch(matched)
|
||||
|
||||
if matched != [] do
|
||||
Phoenix.PubSub.broadcast(
|
||||
|
|
@ -39,7 +39,7 @@ defmodule Microwaveprop.Workers.SolarIndexWorker do
|
|||
{:ok, all_records} ->
|
||||
all_records
|
||||
|> SolarClient.filter_since(since_date)
|
||||
|> Enum.each(&Weather.upsert_solar_index/1)
|
||||
|> Weather.upsert_solar_indices_batch()
|
||||
|
||||
:ok
|
||||
|
||||
|
|
|
|||
|
|
@ -7,13 +7,14 @@ defmodule MicrowavepropWeb.BackfillLive do
|
|||
|
||||
alias Microwaveprop.Radio.Contact
|
||||
alias Microwaveprop.Repo
|
||||
alias Microwaveprop.Workers.ContactWeatherEnqueueWorker
|
||||
alias Microwaveprop.Workers.BackfillEnqueueWorker
|
||||
|
||||
@impl true
|
||||
def mount(_params, _session, socket) do
|
||||
if connected?(socket) do
|
||||
Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "db:contact_status")
|
||||
Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "db:oban_jobs")
|
||||
Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "backfill:enqueue_complete")
|
||||
end
|
||||
|
||||
stats = fetch_stats()
|
||||
|
|
@ -43,19 +44,15 @@ defmodule MicrowavepropWeb.BackfillLive do
|
|||
def handle_event("enqueue", %{"limit" => limit_str}, socket) do
|
||||
limit = String.to_integer(limit_str)
|
||||
|
||||
# Run in a task to not block the LiveView
|
||||
pid = self()
|
||||
|
||||
Task.start(fn ->
|
||||
count = enqueue_batch(limit)
|
||||
send(pid, {:enqueued, count})
|
||||
end)
|
||||
%{"limit" => limit}
|
||||
|> BackfillEnqueueWorker.new()
|
||||
|> Oban.insert()
|
||||
|
||||
{:noreply, socket |> assign(enqueuing: true, limit: limit) |> LiveStash.stash_assigns([:limit])}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_info({:enqueued, count}, socket) do
|
||||
def handle_info({:enqueue_complete, count}, socket) do
|
||||
{:noreply, assign(socket, last_enqueued: count, enqueuing: false)}
|
||||
end
|
||||
|
||||
|
|
@ -87,26 +84,6 @@ defmodule MicrowavepropWeb.BackfillLive do
|
|||
assign(socket, refresh_timer: ref)
|
||||
end
|
||||
|
||||
@enrichable [:pending, :failed]
|
||||
|
||||
defp enqueue_batch(limit) do
|
||||
contacts =
|
||||
Repo.all(
|
||||
from(c in Contact,
|
||||
where:
|
||||
(not is_nil(c.pos1) or not is_nil(c.grid1)) and
|
||||
(c.hrrr_status in ^@enrichable or c.weather_status in ^@enrichable or c.terrain_status in ^@enrichable or
|
||||
c.iemre_status in ^@enrichable),
|
||||
order_by: [desc: c.qso_timestamp],
|
||||
limit: ^limit
|
||||
)
|
||||
)
|
||||
|
||||
# ensure_positions! is called inside enqueue_for_contact
|
||||
Enum.each(contacts, &ContactWeatherEnqueueWorker.enqueue_for_contact/1)
|
||||
length(contacts)
|
||||
end
|
||||
|
||||
defp count_unprocessed do
|
||||
incomplete = [:pending, :queued, :processing, :failed]
|
||||
|
||||
|
|
|
|||
|
|
@ -106,7 +106,7 @@ defmodule MicrowavepropWeb.MapLive do
|
|||
|
||||
def handle_event("point_detail", %{"lat" => lat, "lon" => lon}, socket) do
|
||||
band = socket.assigns.selected_band
|
||||
detail = Propagation.point_detail(band, lat, lon)
|
||||
detail = Propagation.point_detail(band, lat, lon, socket.assigns.selected_time)
|
||||
forecast = Propagation.point_forecast(band, lat, lon)
|
||||
|
||||
forecast_data =
|
||||
|
|
|
|||
|
|
@ -0,0 +1,21 @@
|
|||
defmodule Microwaveprop.Repo.Migrations.AddIsGridPointToHrrrProfiles do
|
||||
use Ecto.Migration
|
||||
|
||||
def change do
|
||||
alter table(:hrrr_profiles) do
|
||||
add :is_grid_point, :boolean, default: false, null: false
|
||||
end
|
||||
|
||||
flush()
|
||||
|
||||
execute(
|
||||
"UPDATE hrrr_profiles SET is_grid_point = true WHERE (lat * 1000)::int % 125 = 0 AND (lon * 1000)::int % 125 = 0",
|
||||
"SELECT 1"
|
||||
)
|
||||
|
||||
create index(:hrrr_profiles, [:valid_time],
|
||||
where: "is_grid_point = true",
|
||||
name: :hrrr_profiles_grid_point_valid_time_idx
|
||||
)
|
||||
end
|
||||
end
|
||||
|
|
@ -0,0 +1,10 @@
|
|||
defmodule Microwaveprop.Repo.Migrations.AddContactsPos1PartialIndex do
|
||||
use Ecto.Migration
|
||||
|
||||
def change do
|
||||
create index(:contacts, [:qso_timestamp],
|
||||
where: "pos1 IS NOT NULL",
|
||||
name: :contacts_pos1_not_null_idx
|
||||
)
|
||||
end
|
||||
end
|
||||
|
|
@ -245,6 +245,10 @@ defmodule Microwaveprop.WeatherGridTest do
|
|||
defp insert_hrrr_profile(attrs) do
|
||||
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
|
||||
is_gp =
|
||||
rem(round(attrs.lat * 1000), 125) == 0 and
|
||||
rem(round(attrs.lon * 1000), 125) == 0
|
||||
|
||||
defaults = %{
|
||||
id: Ecto.UUID.bingenerate(),
|
||||
run_time: DateTime.add(attrs.valid_time, -3600, :second),
|
||||
|
|
@ -253,6 +257,7 @@ defmodule Microwaveprop.WeatherGridTest do
|
|||
min_refractivity_gradient: nil,
|
||||
ducting_detected: nil,
|
||||
duct_characteristics: nil,
|
||||
is_grid_point: is_gp,
|
||||
inserted_at: now,
|
||||
updated_at: now
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue