diff --git a/lib/microwaveprop/ionosphere.ex b/lib/microwaveprop/ionosphere.ex index fb51bdb9..6bc8dedd 100644 --- a/lib/microwaveprop/ionosphere.ex +++ b/lib/microwaveprop/ionosphere.ex @@ -9,10 +9,13 @@ defmodule Microwaveprop.Ionosphere do alias Microwaveprop.Ionosphere.Observation alias Microwaveprop.Repo + @observation_update_fields ~w(confidence_score fo_f2_mhz fo_e_mhz fo_es_mhz hm_f2_km mufd_mhz updated_at)a + @doc """ Upsert a list of parsed `GiroClient` observations for a station. - Conflicts on `(station_code, valid_time)` replace every column except - the primary key and inserted_at. Returns `{:ok, count}`. + Conflicts on `(station_code, valid_time)` replace the scaled + characteristic columns plus `updated_at`; `id`, `inserted_at`, + `station_code`, and `valid_time` are preserved. Returns `{:ok, count}`. """ @spec upsert_observations(String.t(), [map()]) :: {:ok, non_neg_integer()} def upsert_observations(_station_code, []), do: {:ok, 0} @@ -30,7 +33,7 @@ defmodule Microwaveprop.Ionosphere do {count, _} = Repo.insert_all(Observation, rows, - on_conflict: {:replace_all_except, [:id, :inserted_at]}, + on_conflict: {:replace, @observation_update_fields}, conflict_target: [:station_code, :valid_time] ) diff --git a/lib/microwaveprop/space_weather.ex b/lib/microwaveprop/space_weather.ex index 14c38781..8c72e4cf 100644 --- a/lib/microwaveprop/space_weather.ex +++ b/lib/microwaveprop/space_weather.ex @@ -15,11 +15,16 @@ defmodule Microwaveprop.SpaceWeather do # ---------- Kp ---------- + @kp_update_fields ~w(kp_index estimated_kp updated_at)a + @spec upsert_kp([map()]) :: {:ok, non_neg_integer()} def upsert_kp([]), do: {:ok, 0} def upsert_kp(rows) when is_list(rows) do - stamp_upsert(GeomagneticObservation, rows, conflict_target: :valid_time) + stamp_upsert(GeomagneticObservation, rows, + conflict_target: :valid_time, + update_fields: @kp_update_fields + ) end @spec latest_kp() :: GeomagneticObservation.t() | nil @@ -29,11 +34,16 @@ defmodule Microwaveprop.SpaceWeather do # ---------- F10.7 ---------- + @solar_flux_update_fields ~w(frequency_mhz flux_sfu reporting_schedule ninety_day_mean updated_at)a + @spec upsert_solar_flux([map()]) :: {:ok, non_neg_integer()} def upsert_solar_flux([]), do: {:ok, 0} def upsert_solar_flux(rows) when is_list(rows) do - stamp_upsert(SolarFluxObservation, rows, conflict_target: :valid_time) + stamp_upsert(SolarFluxObservation, rows, + conflict_target: :valid_time, + update_fields: @solar_flux_update_fields + ) end @spec latest_f107() :: SolarFluxObservation.t() | nil @@ -43,11 +53,16 @@ defmodule Microwaveprop.SpaceWeather do # ---------- GOES X-ray ---------- + @xray_update_fields ~w(satellite flux_wm2 electron_contaminated updated_at)a + @spec upsert_xray([map()]) :: {:ok, non_neg_integer()} def upsert_xray([]), do: {:ok, 0} def upsert_xray(rows) when is_list(rows) do - stamp_upsert(SolarXrayObservation, rows, conflict_target: [:valid_time, :energy_band]) + stamp_upsert(SolarXrayObservation, rows, + conflict_target: [:valid_time, :energy_band], + update_fields: @xray_update_fields + ) end @spec latest_xray() :: SolarXrayObservation.t() | nil @@ -69,7 +84,7 @@ defmodule Microwaveprop.SpaceWeather do {count, _} = Repo.insert_all(schema, stamped, - on_conflict: {:replace_all_except, [:id, :inserted_at]}, + on_conflict: {:replace, Keyword.fetch!(opts, :update_fields)}, conflict_target: Keyword.fetch!(opts, :conflict_target) ) diff --git a/lib/microwaveprop/workers/common_volume_radar_worker.ex b/lib/microwaveprop/workers/common_volume_radar_worker.ex index 1dcf862c..b3c7d6bf 100644 --- a/lib/microwaveprop/workers/common_volume_radar_worker.ex +++ b/lib/microwaveprop/workers/common_volume_radar_worker.ex @@ -265,7 +265,8 @@ defmodule Microwaveprop.Workers.CommonVolumeRadarWorker do %ContactCommonVolumeRadar{} |> ContactCommonVolumeRadar.changeset(Map.merge(stats, %{contact_id: contact_id, observed_at: observed_at})) |> Repo.insert( - on_conflict: {:replace_all_except, [:id, :contact_id, :inserted_at]}, + on_conflict: {:replace, ~w(observed_at common_volume_km2 pixel_count rain_pixel_count heavy_rain_pixel_count + core_pixel_count max_dbz mean_dbz coverage_pct updated_at)a}, conflict_target: :contact_id ) end diff --git a/lib/microwaveprop/workers/hrrr_native_grid_worker.ex b/lib/microwaveprop/workers/hrrr_native_grid_worker.ex index b57afc0b..4dd35386 100644 --- a/lib/microwaveprop/workers/hrrr_native_grid_worker.ex +++ b/lib/microwaveprop/workers/hrrr_native_grid_worker.ex @@ -157,7 +157,11 @@ defmodule Microwaveprop.Workers.HrrrNativeGridWorker do Repo.insert_all( HrrrNativeProfile, rows, - on_conflict: {:replace_all_except, [:id, :inserted_at]}, + on_conflict: {:replace, ~w(run_time level_count heights_m temp_k spfh pressure_pa + u_wind_ms v_wind_ms tke_m2s2 + surface_temp_k surface_spfh surface_pressure_pa + inversion_top_m bulk_richardson theta_e_jump_k shear_at_top_ms + ducts best_duct_band_ghz updated_at)a}, conflict_target: [:lat, :lon, :valid_time] ) diff --git a/lib/microwaveprop/workers/nexrad_worker.ex b/lib/microwaveprop/workers/nexrad_worker.ex index 63eed505..8ce87aab 100644 --- a/lib/microwaveprop/workers/nexrad_worker.ex +++ b/lib/microwaveprop/workers/nexrad_worker.ex @@ -83,7 +83,8 @@ defmodule Microwaveprop.Workers.NexradWorker do Repo.insert_all( NexradObservation, rows, - on_conflict: {:replace_all_except, [:id, :inserted_at]}, + on_conflict: + {:replace, ~w(mean_reflectivity_dbz max_reflectivity_dbz texture_variance pixel_count updated_at)a}, conflict_target: [:lat, :lon, :observed_at] ) diff --git a/lib/microwaveprop/workers/radar_frame_worker.ex b/lib/microwaveprop/workers/radar_frame_worker.ex index f8c23a39..2e1a9fbb 100644 --- a/lib/microwaveprop/workers/radar_frame_worker.ex +++ b/lib/microwaveprop/workers/radar_frame_worker.ex @@ -139,7 +139,8 @@ defmodule Microwaveprop.Workers.RadarFrameWorker do %ContactCommonVolumeRadar{} |> ContactCommonVolumeRadar.changeset(Map.merge(stats, %{contact_id: contact_id, observed_at: observed_at})) |> Repo.insert( - on_conflict: {:replace_all_except, [:id, :contact_id, :inserted_at]}, + on_conflict: {:replace, ~w(observed_at common_volume_km2 pixel_count rain_pixel_count heavy_rain_pixel_count + core_pixel_count max_dbz mean_dbz coverage_pct updated_at)a}, conflict_target: :contact_id ) end diff --git a/test/microwaveprop/workers/radar_frame_worker_test.exs b/test/microwaveprop/workers/radar_frame_worker_test.exs index 4f31e12a..776c966e 100644 --- a/test/microwaveprop/workers/radar_frame_worker_test.exs +++ b/test/microwaveprop/workers/radar_frame_worker_test.exs @@ -60,6 +60,43 @@ defmodule Microwaveprop.Workers.RadarFrameWorkerTest do assert %Contact{radar_status: :complete} = Repo.get!(Contact, c1.id) assert %Contact{radar_status: :complete} = Repo.get!(Contact, c2.id) end + + test "re-running process_frame/4 upserts in place (one row per contact, observed_at updated)" do + # Pins the on_conflict: {:replace, [...]} clause on contact_id. If a + # column gets added to ContactCommonVolumeRadar and the explicit list + # isn't updated, or if the list names a non-existent column, this + # test fails loudly. + contact = insert_contact(%{station1: "RR1"}) + + pixels = :binary.copy(<<0>>, 100) + width = 10 + + :ok = + RadarFrameWorker.process_frame( + [Repo.reload!(contact)], + ~U[2024-09-15 18:30:00Z], + pixels, + width + ) + + first = Repo.get_by!(ContactCommonVolumeRadar, contact_id: contact.id) + + :ok = + RadarFrameWorker.process_frame( + [Repo.reload!(contact)], + ~U[2024-09-15 18:35:00Z], + pixels, + width + ) + + rows = + Repo.all(from(r in ContactCommonVolumeRadar, where: r.contact_id == ^contact.id)) + + assert length(rows) == 1 + [second] = rows + assert second.id == first.id + assert second.observed_at == ~U[2024-09-15 18:35:00Z] + end end describe "perform/1" do