defmodule Aprs.Packets do @moduledoc """ The Packets context. """ import Ecto.Query, warn: false alias Aprs.Packet alias Aprs.Repo @doc """ Stores a packet in the database. ## Parameters * `packet_data` - Map containing packet data to be stored """ def store_packet(packet_data) do require Logger # Convert to map if it's a struct, or use as is if already a map packet_attrs = case packet_data do %Packet{} = packet -> packet |> Map.from_struct() |> Map.delete(:__meta__) %{} -> packet_data end # Convert data_type to string if it's an atom packet_attrs = case packet_attrs do %{data_type: data_type} when is_atom(data_type) -> Map.put(packet_attrs, :data_type, to_string(data_type)) _ -> packet_attrs end # Make sure received_at is set with explicit UTC DateTime current_time = DateTime.truncate(DateTime.utc_now(), :microsecond) packet_attrs = Map.put(packet_attrs, :received_at, current_time) # Extract position data {lat, lon} = extract_position(packet_attrs) # Set position fields if found packet_attrs = if lat && lon do packet_attrs |> Map.put(:lat, lat) |> Map.put(:lon, lon) |> Map.put(:has_position, true) |> Map.put(:region, "#{Float.round(lat, 1)},#{Float.round(lon, 1)}") else # Set region based on callsign if no position sender_region = if packet_attrs[:sender], do: String.slice(packet_attrs.sender || "", 0, 3), else: "unknown" Map.put(packet_attrs, :region, "call:#{sender_region}") end # Insert the packet %Packet{} |> Packet.changeset(packet_attrs) |> Repo.insert() end # Extracts position data from packet, checking various possible locations defp extract_position(packet_data) do cond do # Check for lat/lon at top level not is_nil(packet_data[:lat]) and not is_nil(packet_data[:lon]) -> {to_float(packet_data.lat), to_float(packet_data.lon)} # Check data_extended struct or map not is_nil(packet_data[:data_extended]) -> data_extended = packet_data.data_extended cond do # Standard position format is_map(data_extended) and not is_nil(data_extended[:latitude]) and not is_nil(data_extended[:longitude]) -> {to_float(data_extended.latitude), to_float(data_extended.longitude)} # MicE packet format with components is_map(data_extended) and data_extended.__struct__ == Parser.Types.MicE -> mic_e = data_extended if is_number(mic_e.lat_degrees) and is_number(mic_e.lat_minutes) and is_number(mic_e.lon_degrees) and is_number(mic_e.lon_minutes) do # Convert MicE components to decimal degrees lat = mic_e.lat_degrees + mic_e.lat_minutes / 60.0 lat = if mic_e.lat_direction == :south, do: -lat, else: lat lon = mic_e.lon_degrees + mic_e.lon_minutes / 60.0 lon = if mic_e.lon_direction == :west, do: -lon, else: lon {lat, lon} else {nil, nil} end true -> {nil, nil} end true -> {nil, nil} end end @doc """ Gets packets for replay. ## Parameters * `opts` - Map of options for filtering and pagination: * `:lat` - Latitude for center point filtering * `:lon` - Longitude for center point filtering * `:radius` - Radius in kilometers for filtering * `:callsign` - Filter by callsign * `:region` - Filter by region * `:start_time` - Start time for replay (DateTime) * `:end_time` - End time for replay (DateTime) * `:limit` - Maximum number of packets to return * `:page` - Page number for pagination """ def get_packets_for_replay(opts \\ %{}) do base_query = from(p in Packet, order_by: [asc: p.received_at], where: p.has_position == true) query = base_query |> filter_by_time(opts) |> filter_by_region(opts) |> filter_by_callsign(opts) |> filter_by_map_bounds(opts) |> limit_results(opts) Repo.all(query) end @doc """ Gets historical packet count for a map area. """ def get_historical_packet_count(opts \\ %{}) do base_query = from(p in Packet, select: count(p.id), where: p.has_position == true) query = base_query |> filter_by_time(opts) |> filter_by_region(opts) |> filter_by_callsign(opts) |> filter_by_map_bounds(opts) Repo.one(query) end # Query building helpers # Handle both start_time and end_time 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, where: p.received_at >= ^effective_start_time and p.received_at <= ^end_time end # Handle only start_time defp filter_by_time(query, %{start_time: start_time}) do from p in query, where: p.received_at >= ^start_time end # Handle only end_time defp filter_by_time(query, %{end_time: end_time}) do from p in query, where: p.received_at <= ^end_time end # Default case defp filter_by_time(query, _), do: query @doc """ Get packets only from the last hour for current display. This is used for initial map loading to show only recent packets. """ def get_recent_packets(opts \\ %{}) do # Always limit to the last hour one_hour_ago = DateTime.add(DateTime.utc_now(), -3600, :second) # Merge the one-hour limit with any other filters opts_with_time = Map.put(opts, :start_time, one_hour_ago) get_packets_for_replay(opts_with_time) end @doc """ Retrieves a continuous stream of stored packets for replay in chronological order. This function returns a Stream that can be used to process packets in chronological order, preserving the timing between packets. ## Parameters * `opts` - The same options as `get_packets_for_replay/1` * `:playback_speed` - Speed multiplier (1.0 = real-time, 2.0 = 2x speed, etc.) ## Returns * Stream of packets with timing information """ def stream_packets_for_replay(opts \\ %{}) do packets = get_packets_for_replay(opts) playback_speed = Map.get(opts, :playback_speed, 1.0) # Return a stream that emits packets with their original timing Stream.unfold({packets, nil}, fn {[], _} -> nil {[packet | rest], nil} -> {{0, packet}, {rest, packet}} {[next | rest], prev} -> # Calculate delay between packets in milliseconds, then convert to seconds delay_ms = DateTime.diff(next.received_at, prev.received_at, :millisecond) adjusted_delay = delay_ms / (playback_speed * 1000) {{adjusted_delay, next}, {rest, next}} end) end defp filter_by_region(query, %{region: region}) do from p in query, where: p.region == ^region end defp filter_by_region(query, _), do: query defp filter_by_callsign(query, %{callsign: callsign}) do pattern = "%#{callsign}%" from p in query, where: ilike(p.sender, ^pattern) or ilike(p.base_callsign, ^pattern) end defp filter_by_callsign(query, _), do: query defp filter_by_map_bounds(query, %{bounds: [min_lon, min_lat, max_lon, max_lat]}) when not is_nil(min_lon) and not is_nil(min_lat) and not is_nil(max_lon) and not is_nil(max_lat) do from p in query, where: p.has_position == true, where: p.lat >= ^min_lat and p.lat <= ^max_lat, where: p.lon >= ^min_lon and p.lon <= ^max_lon end defp filter_by_map_bounds(query, _), do: query defp limit_results(query, %{limit: limit, page: page}) when not is_nil(limit) and not is_nil(page) do offset = (page - 1) * limit from p in query, limit: ^limit, offset: ^offset end defp limit_results(query, %{limit: limit}) when not is_nil(limit) do from p in query, limit: ^limit end defp limit_results(query, _), do: query @doc """ Configure packet retention policy. Packets are retained based on these rules: - Default retention is 30 days (configurable via :packet_retention_days) - Returns the number of packets deleted """ def clean_old_packets do retention_days = Application.get_env(:aprs, :packet_retention_days, 30) cutoff_time = DateTime.add(DateTime.utc_now(), -retention_days * 86_400, :second) {deleted_count, _} = Repo.delete_all(from(p in Packet, where: p.received_at < ^cutoff_time)) deleted_count end # Helper to convert various types to float defp to_float(value) when is_float(value), do: value defp to_float(value) when is_integer(value), do: value * 1.0 defp to_float(%Decimal{} = value), do: Decimal.to_float(value) defp to_float(value) when is_binary(value) do case Float.parse(value) do {float, _} -> float :error -> nil end end defp to_float(_), do: nil # Get packets from last hour only - used to initialize the map def get_last_hour_packets do one_hour_ago = DateTime.add(DateTime.utc_now(), -3600, :second) Packet |> where([p], p.has_position == true) |> where([p], p.received_at >= ^one_hour_ago) |> order_by([p], asc: p.received_at) |> limit(500) |> Repo.all() end end