historical packets working again
This commit is contained in:
parent
8cb4302c70
commit
1354b7da07
5 changed files with 161 additions and 64 deletions
|
|
@ -155,12 +155,57 @@ defmodule Aprsme.PacketConsumer do
|
||||||
# Explicitly remove raw_weather_data to prevent insert_all errors
|
# Explicitly remove raw_weather_data to prevent insert_all errors
|
||||||
|> Map.delete(:raw_weather_data)
|
|> Map.delete(:raw_weather_data)
|
||||||
|> Map.delete("raw_weather_data")
|
|> Map.delete("raw_weather_data")
|
||||||
|
# Create PostGIS geometry for location field
|
||||||
|
|> create_location_geometry()
|
||||||
rescue
|
rescue
|
||||||
error ->
|
error ->
|
||||||
Logger.error("Failed to prepare packet for batch insert: #{inspect(error)}")
|
Logger.error("Failed to prepare packet for batch insert: #{inspect(error)}")
|
||||||
nil
|
nil
|
||||||
end
|
end
|
||||||
|
|
||||||
|
# Create PostGIS geometry from lat/lon coordinates
|
||||||
|
defp create_location_geometry(attrs) do
|
||||||
|
lat = attrs[:lat]
|
||||||
|
lon = attrs[:lon]
|
||||||
|
|
||||||
|
if valid_coordinates?(lat, lon) do
|
||||||
|
location = create_point(lat, lon)
|
||||||
|
|
||||||
|
if location do
|
||||||
|
Map.put(attrs, :location, location)
|
||||||
|
else
|
||||||
|
attrs
|
||||||
|
end
|
||||||
|
else
|
||||||
|
attrs
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
# Helper functions for coordinate validation and point creation
|
||||||
|
defp valid_coordinates?(lat, lon) do
|
||||||
|
lat = normalize_coordinate(lat)
|
||||||
|
lon = normalize_coordinate(lon)
|
||||||
|
|
||||||
|
is_number(lat) && is_number(lon) &&
|
||||||
|
lat >= -90 && lat <= 90 &&
|
||||||
|
lon >= -180 && lon <= 180
|
||||||
|
end
|
||||||
|
|
||||||
|
defp normalize_coordinate(%Decimal{} = decimal), do: Decimal.to_float(decimal)
|
||||||
|
defp normalize_coordinate(coord), do: coord
|
||||||
|
|
||||||
|
defp create_point(lat, lon)
|
||||||
|
when (is_number(lat) or is_struct(lat, Decimal)) and (is_number(lon) or is_struct(lon, Decimal)) do
|
||||||
|
lat = normalize_coordinate(lat)
|
||||||
|
lon = normalize_coordinate(lon)
|
||||||
|
|
||||||
|
if valid_coordinates?(lat, lon) do
|
||||||
|
%Geo.Point{coordinates: {lon, lat}, srid: 4326}
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
defp create_point(_, _), do: nil
|
||||||
|
|
||||||
defp valid_packet?(nil), do: false
|
defp valid_packet?(nil), do: false
|
||||||
defp valid_packet?(%{sender: sender}) when is_binary(sender) and byte_size(sender) > 0, do: true
|
defp valid_packet?(%{sender: sender}) when is_binary(sender) and byte_size(sender) > 0, do: true
|
||||||
defp valid_packet?(_), do: false
|
defp valid_packet?(_), do: false
|
||||||
|
|
|
||||||
|
|
@ -348,14 +348,8 @@ defmodule Aprsme.Packets do
|
||||||
# Query building helpers
|
# Query building helpers
|
||||||
# Handle both start_time and end_time
|
# Handle both start_time and end_time
|
||||||
defp filter_by_time(query, %{start_time: start_time, end_time: end_time}) do
|
defp filter_by_time(query, %{start_time: start_time, end_time: end_time}) do
|
||||||
# Ensure we prioritize packets from the last hour
|
|
||||||
one_hour_ago = DateTime.add(DateTime.utc_now(), -3600, :second)
|
|
||||||
|
|
||||||
effective_start_time =
|
|
||||||
if DateTime.before?(start_time, one_hour_ago), do: one_hour_ago, else: start_time
|
|
||||||
|
|
||||||
from p in query,
|
from p in query,
|
||||||
where: p.received_at >= ^effective_start_time and p.received_at <= ^end_time
|
where: p.received_at >= ^start_time and p.received_at <= ^end_time
|
||||||
end
|
end
|
||||||
|
|
||||||
# Handle only start_time
|
# Handle only start_time
|
||||||
|
|
@ -442,8 +436,11 @@ defmodule Aprsme.Packets do
|
||||||
|
|
||||||
from p in query,
|
from p in query,
|
||||||
where: p.has_position == true,
|
where: p.has_position == true,
|
||||||
where: not is_nil(p.location),
|
# Use PostGIS spatial query if location is available
|
||||||
where: fragment("ST_Within(?, ST_GeomFromText(?, 4326))", p.location, ^bbox_wkt)
|
# Fall back to lat/lon comparison if location is null
|
||||||
|
where:
|
||||||
|
(not is_nil(p.location) and fragment("ST_Within(?, ST_GeomFromText(?, 4326))", p.location, ^bbox_wkt)) or
|
||||||
|
(is_nil(p.location) and p.lat >= ^min_lat and p.lat <= ^max_lat and p.lon >= ^min_lon and p.lon <= ^max_lon)
|
||||||
end
|
end
|
||||||
|
|
||||||
defp filter_by_map_bounds(query, _), do: query
|
defp filter_by_map_bounds(query, _), do: query
|
||||||
|
|
|
||||||
|
|
@ -15,7 +15,6 @@ defmodule AprsmeWeb.MapLive.Index do
|
||||||
@default_center %{lat: 39.8283, lng: -98.5795}
|
@default_center %{lat: 39.8283, lng: -98.5795}
|
||||||
@default_zoom 5
|
@default_zoom 5
|
||||||
@finch_name Aprsme.Finch
|
@finch_name Aprsme.Finch
|
||||||
@initialize_replay_delay Application.compile_env(:aprsme, :initialize_replay_delay, 500)
|
|
||||||
|
|
||||||
@impl true
|
@impl true
|
||||||
def mount(_params, _session, socket) do
|
def mount(_params, _session, socket) do
|
||||||
|
|
@ -212,8 +211,8 @@ defmodule AprsmeWeb.MapLive.Index do
|
||||||
def handle_event("map_ready", _params, socket) do
|
def handle_event("map_ready", _params, socket) do
|
||||||
socket = assign(socket, map_ready: true)
|
socket = assign(socket, map_ready: true)
|
||||||
|
|
||||||
# Start historical replay
|
# Always trigger reload of historical packets with the current value
|
||||||
Process.send_after(self(), :initialize_replay, @initialize_replay_delay)
|
send(self(), :reload_historical_packets)
|
||||||
|
|
||||||
# If we have pending geolocation, zoom to it now
|
# If we have pending geolocation, zoom to it now
|
||||||
socket =
|
socket =
|
||||||
|
|
@ -1004,12 +1003,9 @@ defmodule AprsmeWeb.MapLive.Index do
|
||||||
defp handle_reload_historical_packets(socket) do
|
defp handle_reload_historical_packets(socket) do
|
||||||
if socket.assigns.map_ready and socket.assigns.map_bounds do
|
if socket.assigns.map_ready and socket.assigns.map_bounds do
|
||||||
# Clear existing historical packets
|
# Clear existing historical packets
|
||||||
socket = assign(socket, historical_packets: %{})
|
|
||||||
|
|
||||||
# Clear only historical markers on the client (preserve live markers)
|
|
||||||
socket = push_event(socket, "clear_historical_packets", %{})
|
socket = push_event(socket, "clear_historical_packets", %{})
|
||||||
|
|
||||||
# Load historical packets with new time range
|
# Load historical packets for the current map bounds and time range
|
||||||
socket = load_historical_packets_for_bounds(socket, socket.assigns.map_bounds)
|
socket = load_historical_packets_for_bounds(socket, socket.assigns.map_bounds)
|
||||||
|
|
||||||
{:noreply, socket}
|
{:noreply, socket}
|
||||||
|
|
@ -1078,61 +1074,28 @@ defmodule AprsmeWeb.MapLive.Index do
|
||||||
unique_position_packets
|
unique_position_packets
|
||||||
|> Enum.with_index()
|
|> Enum.with_index()
|
||||||
|> Enum.map(fn {packet, index} ->
|
|> Enum.map(fn {packet, index} ->
|
||||||
build_historical_packet_data(packet, index, callsign)
|
# The first packet (index 0) is the most recent for this callsign
|
||||||
|
# Only show as red dot if it's not the most recent position
|
||||||
|
is_most_recent = index == 0
|
||||||
|
|
||||||
|
packet_data = PacketUtils.build_packet_data(packet, is_most_recent)
|
||||||
|
|
||||||
|
if packet_data do
|
||||||
|
packet_data
|
||||||
|
|> Map.put(:callsign, callsign)
|
||||||
|
|> Map.put(:historical, true)
|
||||||
|
end
|
||||||
end)
|
end)
|
||||||
|> Enum.filter(& &1)
|
|> Enum.filter(& &1)
|
||||||
end)
|
end)
|
||||||
|
|
||||||
socket = push_event(socket, "add_historical_packets", %{packets: packet_data_list})
|
if Enum.any?(packet_data_list) do
|
||||||
|
push_event(socket, "add_historical_packets", %{packets: packet_data_list})
|
||||||
historical_packets_map =
|
else
|
||||||
packet_data_list
|
socket
|
||||||
|> Enum.zip(historical_packets)
|
|
||||||
|> Enum.reduce(%{}, fn {packet_data, packet}, acc ->
|
|
||||||
Map.put(acc, packet_data["id"], packet)
|
|
||||||
end)
|
|
||||||
|
|
||||||
assign(socket,
|
|
||||||
historical_packets: historical_packets_map,
|
|
||||||
historical_loaded: true
|
|
||||||
)
|
|
||||||
end
|
|
||||||
|
|
||||||
defp build_historical_packet_data(packet, index, callsign) do
|
|
||||||
case PacketUtils.build_packet_data(packet) do
|
|
||||||
nil ->
|
|
||||||
nil
|
|
||||||
|
|
||||||
packet_data ->
|
|
||||||
packet_id =
|
|
||||||
"hist_#{if Map.has_key?(packet, :id), do: packet.id, else: System.unique_integer([:positive])}_#{index}"
|
|
||||||
|
|
||||||
is_most_recent = index == 0
|
|
||||||
|
|
||||||
packet_data
|
|
||||||
|> Map.put("id", packet_id)
|
|
||||||
|> Map.put("is_historical", true)
|
|
||||||
|> Map.put("is_most_recent_for_callsign", is_most_recent)
|
|
||||||
|> Map.put("callsign_group", callsign)
|
|
||||||
|> Map.put("timestamp", packet_timestamp_ms(packet))
|
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
defp packet_timestamp_ms(packet) do
|
|
||||||
case packet.inserted_at do
|
|
||||||
%NaiveDateTime{} = naive_dt ->
|
|
||||||
DateTime.to_unix(DateTime.from_naive!(naive_dt, "Etc/UTC"), :millisecond)
|
|
||||||
|
|
||||||
%DateTime{} = dt ->
|
|
||||||
DateTime.to_unix(dt, :millisecond)
|
|
||||||
|
|
||||||
_other ->
|
|
||||||
DateTime.to_unix(DateTime.utc_now(), :millisecond)
|
|
||||||
end
|
|
||||||
end
|
|
||||||
|
|
||||||
# Filter packets to only include those with unique positions (lat/lon changed)
|
|
||||||
@spec filter_unique_positions([struct()]) :: [struct()]
|
|
||||||
defp filter_unique_positions(packets) do
|
defp filter_unique_positions(packets) do
|
||||||
packets
|
packets
|
||||||
|> Enum.reduce([], fn packet, acc ->
|
|> Enum.reduce([], fn packet, acc ->
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,24 @@
|
||||||
|
defmodule Aprsme.Repo.Migrations.BackfillLocationFieldForExistingPackets do
|
||||||
|
use Ecto.Migration
|
||||||
|
|
||||||
|
def up do
|
||||||
|
# Update packets that have lat/lon but no location field
|
||||||
|
execute """
|
||||||
|
UPDATE packets
|
||||||
|
SET location = ST_SetSRID(ST_MakePoint(lon, lat), 4326)
|
||||||
|
WHERE lat IS NOT NULL
|
||||||
|
AND lon IS NOT NULL
|
||||||
|
AND location IS NULL
|
||||||
|
AND has_position = true;
|
||||||
|
"""
|
||||||
|
end
|
||||||
|
|
||||||
|
def down do
|
||||||
|
# Clear location field (if needed for rollback)
|
||||||
|
execute """
|
||||||
|
UPDATE packets
|
||||||
|
SET location = NULL
|
||||||
|
WHERE location IS NOT NULL;
|
||||||
|
"""
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
@ -82,6 +82,74 @@ defmodule AprsmeWeb.MapLive.IndexTest do
|
||||||
assert render(view) =~ "aprs-map"
|
assert render(view) =~ "aprs-map"
|
||||||
end
|
end
|
||||||
|
|
||||||
|
test "loads historical packets with correct time range", %{conn: conn} do
|
||||||
|
# Create mock historical packets from the last 2 hours
|
||||||
|
now = DateTime.utc_now()
|
||||||
|
two_hours_ago = DateTime.add(now, -7200, :second)
|
||||||
|
one_hour_ago = DateTime.add(now, -3600, :second)
|
||||||
|
|
||||||
|
mock_packets = [
|
||||||
|
%{
|
||||||
|
id: 1,
|
||||||
|
base_callsign: "TEST1",
|
||||||
|
sender: "TEST1-1",
|
||||||
|
ssid: "1",
|
||||||
|
data_type: "position",
|
||||||
|
symbol_table_id: "/",
|
||||||
|
symbol_code: ">",
|
||||||
|
lat: 39.8283,
|
||||||
|
lon: -98.5795,
|
||||||
|
received_at: two_hours_ago,
|
||||||
|
inserted_at: two_hours_ago,
|
||||||
|
has_position: true,
|
||||||
|
comment: "Historical packet 1"
|
||||||
|
},
|
||||||
|
%{
|
||||||
|
id: 2,
|
||||||
|
base_callsign: "TEST2",
|
||||||
|
sender: "TEST2-1",
|
||||||
|
ssid: "1",
|
||||||
|
data_type: "position",
|
||||||
|
symbol_table_id: "/",
|
||||||
|
symbol_code: ">",
|
||||||
|
lat: 40.0,
|
||||||
|
lon: -99.0,
|
||||||
|
received_at: one_hour_ago,
|
||||||
|
inserted_at: one_hour_ago,
|
||||||
|
has_position: true,
|
||||||
|
comment: "Historical packet 2"
|
||||||
|
}
|
||||||
|
]
|
||||||
|
|
||||||
|
# Mock the Packets.get_packets_for_replay function to return our test packets
|
||||||
|
Mox.stub(Aprsme.PacketsMock, :get_packets_for_replay, fn opts ->
|
||||||
|
# Verify that the time range is being passed correctly (not overridden to 1 hour)
|
||||||
|
assert Map.has_key?(opts, :start_time)
|
||||||
|
assert Map.has_key?(opts, :end_time)
|
||||||
|
|
||||||
|
# The start_time should be 2 hours ago (historical_hours = "2")
|
||||||
|
expected_start = DateTime.add(now, -7200, :second)
|
||||||
|
assert_in_delta DateTime.to_unix(opts.start_time), DateTime.to_unix(expected_start), 1
|
||||||
|
|
||||||
|
mock_packets
|
||||||
|
end)
|
||||||
|
|
||||||
|
{:ok, view, _html} = live(conn, "/")
|
||||||
|
|
||||||
|
# Set historical_hours to 2 to test the time range
|
||||||
|
render_hook(view, "update_historical_hours", %{"historical_hours" => "2"})
|
||||||
|
Process.sleep(10)
|
||||||
|
|
||||||
|
# Simulate map initialization which should trigger historical packet loading
|
||||||
|
assert render_hook(view, "map_ready", %{})
|
||||||
|
|
||||||
|
# Wait for the initialize_replay message to be processed
|
||||||
|
Process.sleep(20)
|
||||||
|
|
||||||
|
# The view should still be rendering without errors after loading historical packets
|
||||||
|
assert render(view) =~ "aprs-map"
|
||||||
|
end
|
||||||
|
|
||||||
test "handles clear_and_reload_markers event", %{conn: conn} do
|
test "handles clear_and_reload_markers event", %{conn: conn} do
|
||||||
{:ok, view, _html} = live(conn, "/")
|
{:ok, view, _html} = live(conn, "/")
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue