Graham McIntire
0d4314b880
Objects and items were being displayed at the origin station's location instead of their actual coordinates because the broadcast used the sender (origin callsign) as the identifier instead of object_name/item_name. For example, object CALGRY sent by VE6RWB-15 was appearing at VE6RWB-15's location instead of CALGRY's actual coordinates. Changes: - Modified broadcast_single_packet to use object_name for objects and item_name for items as the identifier instead of sender - Added tests verifying objects and items broadcast with correct identifier - Objects/items now display at their specified coordinates, not origin station Fixes the positioning bug where objects appeared at sender location.
749 lines
24 KiB
Elixir
749 lines
24 KiB
Elixir
defmodule Aprsme.PacketConsumer do
|
|
@moduledoc """
|
|
GenStage consumer that batches APRS packets and inserts them into the database
|
|
efficiently to reduce database load.
|
|
"""
|
|
use GenStage
|
|
|
|
alias Aprs.Types.ParseError
|
|
alias Aprsme.Cluster.PacketDistributor
|
|
alias Aprsme.LogSanitizer
|
|
alias Aprsme.Repo
|
|
alias AprsmeWeb.Live.Shared.CoordinateUtils
|
|
|
|
require Logger
|
|
|
|
@type state :: %{
|
|
batch: list(map()),
|
|
batch_length: non_neg_integer(),
|
|
batch_size: integer(),
|
|
batch_timeout: integer(),
|
|
max_batch_size: integer(),
|
|
timer: reference() | nil
|
|
}
|
|
|
|
@spec start_link(keyword()) :: GenServer.on_start()
|
|
def start_link(opts \\ []) do
|
|
# Allow unnamed consumers for pool usage
|
|
name = opts[:name]
|
|
|
|
if name do
|
|
GenStage.start_link(__MODULE__, opts, name: name)
|
|
else
|
|
GenStage.start_link(__MODULE__, opts)
|
|
end
|
|
end
|
|
|
|
@impl true
|
|
def init(opts) do
|
|
batch_size = opts[:batch_size] || 100
|
|
batch_timeout = opts[:batch_timeout] || 1000
|
|
# Maximum batch size to prevent unbounded memory growth
|
|
max_batch_size = opts[:max_batch_size] || 1000
|
|
|
|
# Start a timer for batch processing
|
|
timer = Process.send_after(self(), :process_batch, batch_timeout)
|
|
|
|
# Extract subscription options if provided
|
|
subscribe_to = opts[:subscribe_to] || [{Aprsme.PacketProducer, max_demand: opts[:max_demand] || 250}]
|
|
|
|
{:consumer,
|
|
%{
|
|
batch: [],
|
|
batch_length: 0,
|
|
batch_size: batch_size,
|
|
batch_timeout: batch_timeout,
|
|
max_batch_size: max_batch_size,
|
|
timer: timer
|
|
}, subscribe_to: subscribe_to}
|
|
end
|
|
|
|
@impl true
|
|
def handle_events(events, _from, state) do
|
|
handle_batch_update(events, state)
|
|
end
|
|
|
|
# Pattern matching for batch handling with optimized list operations
|
|
defp handle_batch_update(events, %{batch: batch, batch_length: batch_length} = state) do
|
|
# Optimize: Keep batch in reverse order for O(1) prepending
|
|
# Only reverse when processing
|
|
new_batch = Enum.reverse(events, batch)
|
|
new_batch_length = batch_length + length(events)
|
|
|
|
handle_batch_by_size(new_batch, new_batch_length, state)
|
|
end
|
|
|
|
# Pattern matching for different batch size scenarios
|
|
defp handle_batch_by_size(batch, length, %{max_batch_size: max} = state) when length >= max do
|
|
handle_oversized_batch(batch, max, state)
|
|
end
|
|
|
|
defp handle_batch_by_size(batch, length, %{batch_size: size} = state) when length >= size do
|
|
handle_full_batch(batch, state)
|
|
end
|
|
|
|
defp handle_batch_by_size(batch, new_batch_length, state) do
|
|
handle_partial_batch(batch, new_batch_length, state)
|
|
end
|
|
|
|
# Handle oversized batch with pattern matching
|
|
defp handle_oversized_batch(batch, max_size, state) do
|
|
# Reverse once for processing
|
|
reversed_batch = Enum.reverse(batch)
|
|
{process_batch, drop_batch} = Enum.split(reversed_batch, max_size)
|
|
|
|
process_batch(process_batch)
|
|
log_dropped_packets(drop_batch, process_batch)
|
|
|
|
new_state = reset_batch_timer(state)
|
|
{:noreply, [], new_state}
|
|
end
|
|
|
|
# Handle full batch
|
|
defp handle_full_batch(batch, state) do
|
|
# Reverse once for processing
|
|
process_batch(Enum.reverse(batch))
|
|
|
|
new_state = reset_batch_timer(state)
|
|
{:noreply, [], new_state}
|
|
end
|
|
|
|
# Handle partial batch
|
|
defp handle_partial_batch(batch, new_batch_length, state) do
|
|
# Keep batch in reverse order
|
|
{:noreply, [], %{state | batch: batch, batch_length: new_batch_length}}
|
|
end
|
|
|
|
# Helper functions with pattern matching
|
|
@spec reset_batch_timer(state()) :: state()
|
|
defp reset_batch_timer(%{timer: nil} = state) do
|
|
new_timer = Process.send_after(self(), :process_batch, state.batch_timeout)
|
|
%{state | batch: [], batch_length: 0, timer: new_timer}
|
|
end
|
|
|
|
defp reset_batch_timer(%{timer: timer} = state) do
|
|
Process.cancel_timer(timer)
|
|
new_timer = Process.send_after(self(), :process_batch, state.batch_timeout)
|
|
%{state | batch: [], batch_length: 0, timer: new_timer}
|
|
end
|
|
|
|
# Pattern matching for logging
|
|
@spec log_dropped_packets(list(), list()) :: :ok
|
|
defp log_dropped_packets([], _), do: :ok
|
|
|
|
defp log_dropped_packets(dropped, processed) do
|
|
Logger.warning("Dropped #{length(dropped)} packets due to batch size limit",
|
|
batch_info:
|
|
LogSanitizer.log_data(
|
|
dropped_count: length(dropped),
|
|
processed_count: length(processed),
|
|
reason: "batch_size_limit_exceeded"
|
|
)
|
|
)
|
|
end
|
|
|
|
@impl true
|
|
# Pattern matching for empty batch
|
|
def handle_info(:process_batch, %{batch: []} = state) do
|
|
# Just restart the timer for empty batch
|
|
new_state = start_batch_timer(state)
|
|
{:noreply, [], new_state}
|
|
end
|
|
|
|
# Pattern matching for non-empty batch
|
|
def handle_info(:process_batch, %{batch: batch, batch_length: batch_length} = state) when is_list(batch) do
|
|
check_batch_utilization(batch_length, state.max_batch_size)
|
|
# Remember to reverse the batch since we keep it in reverse order
|
|
process_batch(Enum.reverse(batch))
|
|
|
|
new_state = start_batch_timer(%{state | batch: [], batch_length: 0})
|
|
{:noreply, [], new_state}
|
|
end
|
|
|
|
# Helper to start batch timer
|
|
@spec start_batch_timer(state()) :: state()
|
|
defp start_batch_timer(%{batch_timeout: timeout} = state) do
|
|
timer = Process.send_after(self(), :process_batch, timeout)
|
|
%{state | timer: timer}
|
|
end
|
|
|
|
# Batch utilization check using pre-computed length to avoid O(n) traversals
|
|
@spec check_batch_utilization(non_neg_integer(), integer()) :: :ok
|
|
defp check_batch_utilization(batch_length, max_size) when batch_length > max_size * 0.8 do
|
|
Logger.warning("Batch size approaching limit",
|
|
batch_status:
|
|
LogSanitizer.log_data(
|
|
current_size: batch_length,
|
|
max_size: max_size,
|
|
utilization_percent: trunc(batch_length / max_size * 100)
|
|
)
|
|
)
|
|
end
|
|
|
|
defp check_batch_utilization(_batch_length, _max_size), do: :ok
|
|
|
|
@spec process_batch(list(map())) :: :ok
|
|
defp process_batch(packets) do
|
|
require Logger
|
|
|
|
# Monitor memory usage before processing (only this process)
|
|
{:memory, memory_before} = Process.info(self(), :memory)
|
|
start_time = System.monotonic_time(:millisecond)
|
|
|
|
# Chunk size optimized for PostgreSQL work_mem=16MB
|
|
# Using smaller batches to reduce memory pressure
|
|
chunk_size = Application.get_env(:aprsme, :packet_pipeline)[:batch_size] || 100
|
|
|
|
# Use Stream for memory-efficient processing
|
|
# Process and reduce in one pass to avoid materializing the entire list
|
|
{success_count, error_count} =
|
|
packets
|
|
|> Stream.chunk_every(chunk_size)
|
|
|> Stream.map(&process_chunk/1)
|
|
|> Enum.reduce({0, 0}, fn {success, error}, {total_success, total_error} ->
|
|
{total_success + success, total_error + error}
|
|
end)
|
|
|
|
end_time = System.monotonic_time(:millisecond)
|
|
duration = end_time - start_time
|
|
|
|
# Monitor memory usage after processing (only this process)
|
|
{:memory, memory_after} = Process.info(self(), :memory)
|
|
memory_diff = memory_after - memory_before
|
|
|
|
# Get current process memory info
|
|
process_info = Process.info(self(), [:memory, :heap_size, :total_heap_size])
|
|
|
|
# Force garbage collection if memory usage is high
|
|
# 50MB threshold for memory diff (per process)
|
|
# 100MB threshold for process memory
|
|
if memory_diff > 52_428_800 or process_info[:memory] > 104_857_600 do
|
|
:erlang.garbage_collect()
|
|
|
|
Logger.warning("High memory usage detected, forced garbage collection",
|
|
memory_info:
|
|
LogSanitizer.log_data(
|
|
memory_diff_bytes: memory_diff,
|
|
process_memory: process_info[:memory],
|
|
heap_size: process_info[:heap_size],
|
|
total_heap_size: process_info[:total_heap_size],
|
|
batch_size: length(packets),
|
|
gc_forced: true
|
|
)
|
|
)
|
|
end
|
|
|
|
:telemetry.execute(
|
|
[
|
|
:aprsme,
|
|
:packet_pipeline,
|
|
:batch
|
|
],
|
|
%{
|
|
count: length(packets),
|
|
success: success_count,
|
|
error: error_count,
|
|
duration_ms: duration,
|
|
memory_diff: memory_diff
|
|
},
|
|
%{}
|
|
)
|
|
end
|
|
|
|
@spec process_chunk(list(map())) :: {non_neg_integer(), non_neg_integer()}
|
|
defp process_chunk(packets) do
|
|
# Use Stream for memory-efficient packet preparation
|
|
# Note: truncate_datetimes_to_second is already called inside prepare_packet_for_insert
|
|
packet_stream =
|
|
packets
|
|
|> Stream.map(&prepare_packet_for_insert/1)
|
|
|> Stream.reject(&is_nil/1)
|
|
|
|
# Separate valid and invalid packets using Stream
|
|
{valid_packets, invalid_count} =
|
|
Enum.reduce(packet_stream, {[], 0}, fn packet, {valid_acc, invalid_acc} ->
|
|
if valid_packet?(packet) do
|
|
{[packet | valid_acc], invalid_acc}
|
|
else
|
|
{valid_acc, invalid_acc + 1}
|
|
end
|
|
end)
|
|
|
|
# Reverse to maintain order
|
|
valid_packets = Enum.reverse(valid_packets)
|
|
|
|
# Insert valid packets in batch
|
|
# Optimized for PostgreSQL with synchronous_commit=off
|
|
insert_opts = [
|
|
# Don't return IDs for better performance
|
|
returning: false,
|
|
# Skip conflicts to avoid blocking
|
|
on_conflict: :nothing,
|
|
# Increased timeout for large batches
|
|
timeout: 60_000
|
|
]
|
|
|
|
try do
|
|
{inserted_count, _} = Repo.insert_all(Aprsme.Packet, valid_packets, insert_opts)
|
|
|
|
# Broadcast successfully inserted packets to StreamingPacketsPubSub
|
|
broadcast_packets_async(valid_packets)
|
|
|
|
{inserted_count, invalid_count}
|
|
rescue
|
|
error ->
|
|
Logger.error("Batch insert failed: #{inspect(error)}, falling back to individual inserts")
|
|
|
|
# Fall back to individual inserts so partial success is possible
|
|
{fallback_inserted, fallback_packets} = insert_individually(valid_packets)
|
|
|
|
# Broadcast whatever was successfully inserted
|
|
if fallback_packets != [] do
|
|
broadcast_packets_async(fallback_packets)
|
|
end
|
|
|
|
{fallback_inserted, invalid_count + length(valid_packets) - fallback_inserted}
|
|
end
|
|
end
|
|
|
|
# Fall back to inserting packets one at a time when batch insert fails.
|
|
# Returns {inserted_count, list_of_successfully_inserted_packet_attrs}.
|
|
@spec insert_individually(list(map())) :: {non_neg_integer(), list(map())}
|
|
defp insert_individually(packets) do
|
|
Enum.reduce(packets, {0, []}, fn packet_attrs, {count, inserted} ->
|
|
try do
|
|
changeset = Aprsme.Packet.changeset(%Aprsme.Packet{}, packet_attrs)
|
|
|
|
case Repo.insert(changeset, on_conflict: :nothing) do
|
|
{:ok, _record} ->
|
|
{count + 1, [packet_attrs | inserted]}
|
|
|
|
{:error, _changeset} ->
|
|
{count, inserted}
|
|
end
|
|
rescue
|
|
error ->
|
|
Logger.error("Individual insert failed: #{inspect(error)}")
|
|
{count, inserted}
|
|
end
|
|
end)
|
|
end
|
|
|
|
# Broadcast packets asynchronously using supervised task pool
|
|
defp broadcast_packets_async(packets) do
|
|
Aprsme.BroadcastTaskSupervisor.async_execute(fn ->
|
|
cluster_enabled = Application.get_env(:aprsme, :cluster_enabled, false)
|
|
Enum.each(packets, &broadcast_single_packet(&1, cluster_enabled))
|
|
end)
|
|
end
|
|
|
|
defp broadcast_single_packet(packet_attrs, cluster_enabled) do
|
|
# For objects and items, use object_name/item_name as the identifier
|
|
# instead of sender (which is the origin station)
|
|
identifier =
|
|
cond do
|
|
packet_attrs[:object_name] -> packet_attrs[:object_name]
|
|
packet_attrs[:item_name] -> packet_attrs[:item_name]
|
|
true -> packet_attrs[:sender]
|
|
end
|
|
|
|
packet = %{
|
|
sender: identifier,
|
|
latitude: packet_attrs[:lat],
|
|
longitude: packet_attrs[:lon],
|
|
received_at: packet_attrs[:received_at],
|
|
data_type: packet_attrs[:data_type],
|
|
altitude: packet_attrs[:altitude],
|
|
speed: packet_attrs[:speed],
|
|
course: packet_attrs[:course],
|
|
comment: packet_attrs[:comment]
|
|
}
|
|
|
|
if cluster_enabled do
|
|
PacketDistributor.distribute_packet(packet)
|
|
else
|
|
Aprsme.StreamingPacketsPubSub.broadcast_packet(packet)
|
|
Aprsme.SpatialPubSub.broadcast_packet(packet)
|
|
end
|
|
end
|
|
|
|
defp prepare_packet_for_insert(packet_data) do
|
|
# Always set received_at timestamp to ensure consistency
|
|
current_time = DateTime.truncate(DateTime.utc_now(), :microsecond)
|
|
packet_data = Map.put(packet_data, :received_at, current_time)
|
|
|
|
# Convert to map before storing to avoid struct conversion issues
|
|
attrs = struct_to_map(packet_data)
|
|
|
|
# Extract additional data from the parsed packet including raw packet
|
|
attrs = Aprsme.Packet.extract_additional_data(attrs, attrs[:raw_packet] || "")
|
|
|
|
# Detect and set item/object fields
|
|
attrs = detect_item_or_object(attrs)
|
|
|
|
# Sanitize packet data to prevent database field overflow
|
|
attrs = Aprsme.PacketSanitizer.sanitize_packet(attrs)
|
|
|
|
# Normalize data_type to string if it's an atom
|
|
attrs = normalize_data_type(attrs)
|
|
|
|
# Apply the same processing as the original store_packet function
|
|
attrs
|
|
|> convert_coordinate_field_names()
|
|
|> convert_field_names()
|
|
|> normalize_packet_attrs()
|
|
|> set_received_at()
|
|
|> patch_lat_lon_from_data_extended()
|
|
|> then(fn attrs ->
|
|
{lat, lon} = extract_position(attrs)
|
|
set_lat_lon(attrs, lat, lon)
|
|
end)
|
|
|> normalize_ssid()
|
|
|> then(fn attrs ->
|
|
device_identifier = Aprsme.DeviceParser.extract_device_identifier(packet_data)
|
|
Map.put(attrs, :device_identifier, device_identifier)
|
|
end)
|
|
|> sanitize_packet_strings()
|
|
|> Map.put(:inserted_at, current_time)
|
|
|> Map.put(:updated_at, current_time)
|
|
|> Map.delete(:id)
|
|
|> Map.delete("id")
|
|
# Remove embedded field for batch insert
|
|
|> Map.delete(:data_extended)
|
|
|> normalize_numeric_types()
|
|
|> truncate_datetimes_to_second()
|
|
# Create PostGIS geometry for location field BEFORE filtering
|
|
|> create_location_geometry()
|
|
# Remove all non-schema fields - do this LAST to ensure all processing is done
|
|
|> remove_non_schema_fields()
|
|
rescue
|
|
error ->
|
|
Logger.error("Failed to prepare packet for batch insert: #{inspect(error)}")
|
|
nil
|
|
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 function to remove fields that exist in parser output but not in database schema
|
|
defp remove_non_schema_fields(attrs) do
|
|
# Use whitelist approach - only keep fields that are in our schema
|
|
Aprsme.PacketFieldWhitelist.filter_fields(attrs)
|
|
end
|
|
|
|
# Helper functions for coordinate validation and point creation
|
|
defp valid_coordinates?(lat, lon) do
|
|
CoordinateUtils.valid_coordinates_any_type?(lat, lon)
|
|
end
|
|
|
|
defp normalize_coordinate(coord) do
|
|
CoordinateUtils.normalize_coordinate(coord)
|
|
end
|
|
|
|
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?(%{sender: sender}) when is_binary(sender) and byte_size(sender) > 0, do: true
|
|
defp valid_packet?(_), do: false
|
|
|
|
# Detect if packet is an item or object and set appropriate fields
|
|
defp detect_item_or_object(attrs) do
|
|
cond do
|
|
object_packet?(attrs) -> apply_object_fields(attrs)
|
|
item_packet?(attrs) -> apply_item_fields(attrs)
|
|
true -> attrs
|
|
end
|
|
end
|
|
|
|
defp object_packet?(attrs) do
|
|
info_field = get_in(attrs, [:data, "information_field"])
|
|
|
|
attrs[:data_type] == "object" or attrs["data_type"] == "object" or
|
|
(is_binary(info_field) and String.starts_with?(info_field, ";"))
|
|
end
|
|
|
|
defp item_packet?(attrs) do
|
|
Map.has_key?(attrs, :itemname) or Map.has_key?(attrs, "itemname")
|
|
end
|
|
|
|
defp apply_object_fields(attrs) do
|
|
info_field = get_in(attrs, [:data, "information_field"])
|
|
|
|
object_name =
|
|
extract_object_name(attrs) ||
|
|
extract_object_name_from_info_field(info_field)
|
|
|
|
attrs
|
|
|> Map.put(:object_name, object_name)
|
|
|> Map.put(:is_object, true)
|
|
|> Map.delete(:itemname)
|
|
|> Map.delete("itemname")
|
|
end
|
|
|
|
defp apply_item_fields(attrs) do
|
|
item_name = Map.get(attrs, :itemname) || Map.get(attrs, "itemname")
|
|
|
|
attrs
|
|
|> Map.put(:item_name, item_name)
|
|
|> Map.put(:is_item, true)
|
|
|> Map.delete(:itemname)
|
|
|> Map.delete("itemname")
|
|
end
|
|
|
|
defp extract_object_name(attrs) do
|
|
case attrs[:data_extended] do
|
|
%{name: name} when is_binary(name) -> name
|
|
%{"name" => name} when is_binary(name) -> name
|
|
_ -> nil
|
|
end
|
|
end
|
|
|
|
defp extract_object_name_from_info_field(nil), do: nil
|
|
|
|
defp extract_object_name_from_info_field(info_field) do
|
|
# Object format: ;OBJECTNAM*DDHHMMz... or ;OBJECTNAM_DDHHMMz... (killed)
|
|
# Object names are 9 printable ASCII characters followed by * (alive) or _ (killed)
|
|
case Regex.run(~r/^;(.{9})[*_]/, info_field) do
|
|
[_, name] -> String.trim(name)
|
|
_ -> nil
|
|
end
|
|
end
|
|
|
|
# Convert field names that don't match our schema
|
|
defp convert_field_names(attrs) do
|
|
attrs
|
|
|> then(fn a ->
|
|
# Convert aprs_messaging? to aprs_messaging
|
|
case Map.get(a, :aprs_messaging?) || Map.get(a, "aprs_messaging?") do
|
|
nil ->
|
|
a
|
|
|
|
value ->
|
|
a
|
|
|> Map.put(:aprs_messaging, value)
|
|
|> Map.delete(:aprs_messaging?)
|
|
|> Map.delete("aprs_messaging?")
|
|
end
|
|
end)
|
|
|> then(fn a ->
|
|
# Convert timestamp from integer to string if needed
|
|
case Map.get(a, :timestamp) do
|
|
timestamp when is_integer(timestamp) ->
|
|
Map.put(a, :timestamp, Integer.to_string(timestamp))
|
|
|
|
_ ->
|
|
a
|
|
end
|
|
end)
|
|
end
|
|
|
|
# Convert latitude/longitude to lat/lon if present at top level
|
|
defp convert_coordinate_field_names(attrs) do
|
|
attrs
|
|
|> then(fn a ->
|
|
case Map.get(a, :latitude) do
|
|
nil -> a
|
|
lat -> a |> Map.put(:lat, lat) |> Map.delete(:latitude)
|
|
end
|
|
end)
|
|
|> then(fn a ->
|
|
case Map.get(a, :longitude) do
|
|
nil -> a
|
|
lon -> a |> Map.put(:lon, lon) |> Map.delete(:longitude)
|
|
end
|
|
end)
|
|
end
|
|
|
|
# Helper functions copied from Packets module for consistency
|
|
defp normalize_packet_attrs(attrs) do
|
|
attrs
|
|
|> Map.put_new(:base_callsign, attrs[:sender])
|
|
|> Map.put_new(:data_type, "unknown")
|
|
|> Map.put_new(:destination, "")
|
|
|> Map.put_new(:path, "")
|
|
|> Map.put_new(:ssid, "")
|
|
|> Map.put_new(:data_extended, %{})
|
|
end
|
|
|
|
defp set_received_at(attrs) do
|
|
received_at = attrs[:received_at] || DateTime.utc_now()
|
|
Map.put(attrs, :received_at, received_at)
|
|
end
|
|
|
|
defp patch_lat_lon_from_data_extended(attrs) do
|
|
case attrs[:data_extended] do
|
|
%{latitude: lat, longitude: lon} when not is_nil(lat) and not is_nil(lon) ->
|
|
attrs
|
|
|> Map.put(:lat, lat)
|
|
|> Map.put(:lon, lon)
|
|
|> Map.put(:has_position, true)
|
|
|
|
_ ->
|
|
attrs
|
|
end
|
|
end
|
|
|
|
defp extract_position(packet_data) do
|
|
if not is_nil(packet_data[:lat]) and not is_nil(packet_data[:lon]) do
|
|
{Aprsme.EncodingUtils.to_float(packet_data.lat), Aprsme.EncodingUtils.to_float(packet_data.lon)}
|
|
else
|
|
extract_position_from_data_extended(packet_data[:data_extended])
|
|
end
|
|
end
|
|
|
|
defp extract_position_from_data_extended(nil), do: {nil, nil}
|
|
|
|
defp extract_position_from_data_extended(data_extended) when is_map(data_extended) do
|
|
if has_standard_position?(data_extended) do
|
|
extract_standard_position(data_extended)
|
|
else
|
|
extract_position_from_data_extended_case(data_extended)
|
|
end
|
|
end
|
|
|
|
defp extract_position_from_data_extended(_), do: {nil, nil}
|
|
|
|
defp has_standard_position?(data_extended) when is_map(data_extended) and not is_struct(data_extended) do
|
|
not is_nil(data_extended[:latitude]) and not is_nil(data_extended[:longitude])
|
|
end
|
|
|
|
defp has_standard_position?(_), do: false
|
|
|
|
defp extract_standard_position(data_extended) when is_map(data_extended) and not is_struct(data_extended) do
|
|
{Aprsme.EncodingUtils.to_float(data_extended[:latitude]), Aprsme.EncodingUtils.to_float(data_extended[:longitude])}
|
|
end
|
|
|
|
defp extract_standard_position(_), do: {nil, nil}
|
|
|
|
defp extract_position_from_data_extended_case(data_extended) do
|
|
lat = extract_lat_from_ext_map(data_extended)
|
|
lon = extract_lon_from_ext_map(data_extended)
|
|
{Aprsme.EncodingUtils.to_float(lat), Aprsme.EncodingUtils.to_float(lon)}
|
|
end
|
|
|
|
defp extract_lat_from_ext_map(ext_map) do
|
|
ext_map[:latitude] || ext_map["latitude"] ||
|
|
(Map.has_key?(ext_map, :position) &&
|
|
(ext_map[:position][:latitude] || ext_map[:position]["latitude"])) ||
|
|
(Map.has_key?(ext_map, "position") &&
|
|
(ext_map["position"][:latitude] || ext_map["position"]["latitude"]))
|
|
end
|
|
|
|
defp extract_lon_from_ext_map(ext_map) do
|
|
ext_map[:longitude] || ext_map["longitude"] ||
|
|
(Map.has_key?(ext_map, :position) &&
|
|
(ext_map[:position][:longitude] || ext_map[:position]["longitude"])) ||
|
|
(Map.has_key?(ext_map, "position") &&
|
|
(ext_map["position"][:longitude] || ext_map["position"]["longitude"]))
|
|
end
|
|
|
|
defp set_lat_lon(attrs, lat, lon) do
|
|
round6 = fn
|
|
nil ->
|
|
nil
|
|
|
|
n when is_float(n) ->
|
|
Float.round(n, 6)
|
|
end
|
|
|
|
attrs
|
|
|> Map.put(:lat, round6.(lat))
|
|
|> Map.put(:lon, round6.(lon))
|
|
|> Map.put(:has_position, not is_nil(lat) and not is_nil(lon))
|
|
end
|
|
|
|
defp normalize_ssid(attrs) do
|
|
case Map.get(attrs, :ssid) do
|
|
nil -> attrs
|
|
ssid -> Map.put(attrs, :ssid, to_string(ssid))
|
|
end
|
|
end
|
|
|
|
defp sanitize_packet_strings(value), do: Aprsme.EncodingUtils.sanitize_packet_strings(value)
|
|
|
|
defp normalize_data_type(attrs), do: Aprsme.EncodingUtils.normalize_data_type(attrs)
|
|
|
|
defp struct_to_map(%{__struct__: ParseError} = error) do
|
|
# Handle ParseError specially to avoid Access behavior issues
|
|
%{
|
|
error_code: error.error_code,
|
|
message: error.error_message,
|
|
__original_struct__: ParseError
|
|
}
|
|
end
|
|
|
|
defp struct_to_map(%{__struct__: struct_type} = struct) do
|
|
converted_map =
|
|
struct
|
|
|> Map.from_struct()
|
|
|> Map.new(fn {k, v} -> {k, struct_to_map(v)} end)
|
|
|
|
Map.put(converted_map, :__original_struct__, struct_type)
|
|
end
|
|
|
|
defp struct_to_map(value) when is_list(value) do
|
|
Enum.map(value, &struct_to_map/1)
|
|
end
|
|
|
|
defp struct_to_map(value), do: value
|
|
|
|
defp truncate_datetimes_to_second(%DateTime{} = dt), do: DateTime.truncate(dt, :second)
|
|
defp truncate_datetimes_to_second({:ok, %DateTime{} = dt}), do: DateTime.truncate(dt, :second)
|
|
defp truncate_datetimes_to_second({:error, _reason}), do: nil
|
|
|
|
defp truncate_datetimes_to_second(term) when is_map(term) and not is_struct(term) do
|
|
Map.new(term, fn {k, v} -> {k, truncate_datetimes_to_second(v)} end)
|
|
end
|
|
|
|
defp truncate_datetimes_to_second(list) when is_list(list), do: Enum.map(list, &truncate_datetimes_to_second/1)
|
|
defp truncate_datetimes_to_second(other), do: other
|
|
|
|
defp normalize_numeric_types(attrs) do
|
|
# Convert integer values to floats for float fields
|
|
float_fields = [
|
|
:temperature,
|
|
:humidity,
|
|
:wind_speed,
|
|
:wind_gust,
|
|
:pressure,
|
|
:rain_1h,
|
|
:rain_24h,
|
|
:rain_since_midnight,
|
|
:snow,
|
|
:speed,
|
|
:altitude
|
|
]
|
|
|
|
Enum.reduce(float_fields, attrs, fn field, acc ->
|
|
case Map.get(acc, field) do
|
|
value when is_integer(value) -> Map.put(acc, field, value * 1.0)
|
|
_ -> acc
|
|
end
|
|
end)
|
|
end
|
|
end
|