defmodule Parser.Helpers do @moduledoc """ Public helper functions for APRS parsing (NMEA, PHG/DF, compressed position, etc). """ @type phg_power :: {integer() | nil, String.t()} @type phg_height :: {integer() | nil, String.t()} @type phg_gain :: {integer() | nil, String.t()} @type phg_directivity :: {integer() | nil, String.t()} @type df_strength :: {integer() | nil, String.t()} # NMEA helpers @spec parse_nmea_coordinate(String.t(), String.t()) :: {:ok, float()} | {:error, String.t()} def parse_nmea_coordinate(value, direction) when is_binary(value) and is_binary(direction) do case Float.parse(value) do {coord, _} -> coord = coord / 100.0 coord = case direction do "N" -> coord "S" -> -coord "E" -> coord "W" -> -coord _ -> {:error, "Invalid coordinate direction"} end if is_tuple(coord), do: coord, else: {:ok, coord} :error -> {:error, "Invalid coordinate value"} end end @spec parse_nmea_coordinate(any(), any()) :: {:error, String.t()} def parse_nmea_coordinate(_, _), do: {:error, "Invalid coordinate format"} # PHG/DF helpers @spec parse_phg_power(integer()) :: phg_power def parse_phg_power(?0), do: {1, "1 watt"} def parse_phg_power(?1), do: {4, "4 watts"} def parse_phg_power(?2), do: {9, "9 watts"} def parse_phg_power(?3), do: {16, "16 watts"} def parse_phg_power(?4), do: {25, "25 watts"} def parse_phg_power(?5), do: {36, "36 watts"} def parse_phg_power(?6), do: {49, "49 watts"} def parse_phg_power(?7), do: {64, "64 watts"} def parse_phg_power(?8), do: {81, "81 watts"} def parse_phg_power(?9), do: {81, "81 watts"} def parse_phg_power(p), do: {nil, "Unknown power: #{<
>}"}
@spec parse_phg_height(integer()) :: phg_height
def parse_phg_height(?0), do: {10, "10 feet"}
def parse_phg_height(?1), do: {20, "20 feet"}
def parse_phg_height(?2), do: {40, "40 feet"}
def parse_phg_height(?3), do: {80, "80 feet"}
def parse_phg_height(?4), do: {160, "160 feet"}
def parse_phg_height(?5), do: {320, "320 feet"}
def parse_phg_height(?6), do: {640, "640 feet"}
def parse_phg_height(?7), do: {1280, "1280 feet"}
def parse_phg_height(?8), do: {2560, "2560 feet"}
def parse_phg_height(?9), do: {5120, "5120 feet"}
def parse_phg_height(h), do: {nil, "Unknown height: #{<>}"}
# Compressed position helpers
@spec convert_compressed_lat(binary()) :: float()
def convert_compressed_lat(lat) do
[l1, l2, l3, l4] = to_charlist(lat)
90 - ((l1 - 33) * 91 ** 3 + (l2 - 33) * 91 ** 2 + (l3 - 33) * 91 + l4 - 33) / 380_926
end
@spec convert_compressed_lon(binary()) :: float()
def convert_compressed_lon(lon) do
[l1, l2, l3, l4] = to_charlist(lon)
-180 + ((l1 - 33) * 91 ** 3 + (l2 - 33) * 91 ** 2 + (l3 - 33) * 91 + l4 - 33) / 190_463
end
# PEET Logging parsing
@spec parse_peet_logging(binary()) :: map()
def parse_peet_logging(<<"*", peet_data::binary>>) do
%{
peet_data: peet_data,
data_type: :peet_logging
}
end
@spec parse_peet_logging(any()) :: map()
def parse_peet_logging(data) do
%{
peet_data: data,
data_type: :peet_logging
}
end
# Invalid/Test Data parsing
@spec parse_invalid_test_data(binary()) :: map()
def parse_invalid_test_data(<<",", test_data::binary>>) do
%{
test_data: test_data,
data_type: :invalid_test_data
}
end
@spec parse_invalid_test_data(any()) :: map()
def parse_invalid_test_data(data) do
%{
test_data: data,
data_type: :invalid_test_data
}
end
# NMEA sentence parsing (minimal public wrapper)
@spec parse_nmea_sentence(any()) :: {:error, String.t()}
def parse_nmea_sentence(_sentence) do
# This is a minimal wrapper; full NMEA parsing logic can be implemented as needed
{:error, "NMEA parsing not implemented"}
end
# KISS to TNC2 conversion
@spec kiss_to_tnc2(binary()) :: binary()
def kiss_to_tnc2(<<0xC0, 0x00, rest::binary>>) do
# Remove KISS framing and control byte
tnc2 =
rest
|> String.trim_trailing(<<0xC0>>)
|> String.replace(<<0xDB, 0xDC>>, <<0xC0>>)
|> String.replace(<<0xDB, 0xDD>>, <<0xDB>>)
tnc2
end
@spec kiss_to_tnc2(any()) :: map()
def kiss_to_tnc2(_), do: %{error_code: :packet_invalid, error_message: "Unknown error"}
# TNC2 to KISS conversion
@spec tnc2_to_kiss(binary()) :: binary()
def tnc2_to_kiss(tnc2) do
# Add KISS framing and escape special bytes
escaped =
tnc2
|> String.replace(<<0xDB>>, <<0xDB, 0xDD>>)
|> String.replace(<<0xC0>>, <<0xDB, 0xDC>>)
<<0xC0, 0x00>> <> escaped <> <<0xC0>>
end
# Telemetry helpers
@spec parse_telemetry_sequence(String.t()) :: integer() | nil
def parse_telemetry_sequence(seq) do
case Integer.parse(seq) do
{num, _} -> num
:error -> nil
end
end
def parse_analog_values(values) do
Enum.map(values, fn value ->
case value do
"" ->
nil
value ->
case Float.parse(value) do
{float_val, _} -> float_val
:error -> nil
end
end
end)
end
def parse_digital_values(values) do
values
|> Enum.map(fn value ->
cond do
value == "1" ->
true
is_binary(value) ->
value
|> String.graphemes()
|> Enum.map(fn
"1" -> true
"0" -> false
_ -> nil
end)
end
end)
|> List.flatten()
end
def parse_coefficient(coeff) do
case Float.parse(coeff) do
{float_val, _} ->
float_val
:error ->
case Integer.parse(coeff) do
{int_val, _} -> int_val
:error -> coeff
end
end
end
# Weather helpers
@spec extract_timestamp(String.t()) :: String.t() | nil
def extract_timestamp(weather_data) do
case Regex.run(~r/^(\d{6}[hz\/])/, weather_data) do
[_, timestamp] -> timestamp
nil -> nil
end
end
@spec remove_timestamp(String.t()) :: String.t()
def remove_timestamp(weather_data) do
case Regex.run(~r/^\d{6}[hz\/]/, weather_data) do
[timestamp] -> String.replace(weather_data, timestamp, "")
nil -> weather_data
end
end
@spec parse_wind_direction(String.t()) :: integer() | nil
def parse_wind_direction(weather_data) do
case Regex.run(~r/(\d{3})\//, weather_data) do
[_, direction] -> String.to_integer(direction)
nil -> nil
end
end
@spec parse_wind_speed(String.t()) :: integer() | nil
def parse_wind_speed(weather_data) do
case Regex.run(~r/\/(\d{3})/, weather_data) do
[_, speed] -> String.to_integer(speed)
nil -> nil
end
end
@spec parse_wind_gust(String.t()) :: integer() | nil
def parse_wind_gust(weather_data) do
case Regex.run(~r/g(\d{3})/, weather_data) do
[_, gust] -> String.to_integer(gust)
nil -> nil
end
end
@spec parse_temperature(String.t()) :: integer() | nil
def parse_temperature(weather_data) do
case Regex.run(~r/t(-?\d{3})/, weather_data) do
[_, temp] -> String.to_integer(temp)
nil -> nil
end
end
@spec parse_rainfall_1h(String.t()) :: integer() | nil
def parse_rainfall_1h(weather_data) do
case Regex.run(~r/r(\d{3})/, weather_data) do
[_, rain] -> String.to_integer(rain)
nil -> nil
end
end
@spec parse_rainfall_24h(String.t()) :: integer() | nil
def parse_rainfall_24h(weather_data) do
case Regex.run(~r/p(\d{3})/, weather_data) do
[_, rain] -> String.to_integer(rain)
nil -> nil
end
end
@spec parse_rainfall_since_midnight(String.t()) :: integer() | nil
def parse_rainfall_since_midnight(weather_data) do
case Regex.run(~r/P(\d{3})/, weather_data) do
[_, rain] -> String.to_integer(rain)
nil -> nil
end
end
@spec parse_humidity(String.t()) :: integer() | nil
def parse_humidity(weather_data) do
case Regex.run(~r/h(\d{2})/, weather_data) do
[_, humidity] ->
val = String.to_integer(humidity)
if val == 0, do: 100, else: val
nil ->
nil
end
end
@spec parse_pressure(String.t()) :: float() | nil
def parse_pressure(weather_data) do
case Regex.run(~r/b(\d{5})/, weather_data) do
[_, pressure] -> String.to_integer(pressure) / 10.0
nil -> nil
end
end
@spec parse_luminosity(String.t()) :: integer() | nil
def parse_luminosity(weather_data) do
case Regex.run(~r/[lL](\d{3})/, weather_data) do
[_, luminosity] -> String.to_integer(luminosity)
nil -> nil
end
end
@spec parse_snow(String.t()) :: integer() | nil
def parse_snow(weather_data) do
case Regex.run(~r/s(\d{3})/, weather_data) do
[_, snow] -> String.to_integer(snow)
nil -> nil
end
end
# Ambiguity and utility helpers
def count_spaces(str) do
str
|> String.graphemes()
|> Enum.count(fn c -> c == " " end)
end
def count_leading_braces(packet), do: count_leading_braces(packet, 0)
def count_leading_braces(<<"}", rest::binary>>, count), do: count_leading_braces(rest, count + 1)
def count_leading_braces(_packet, count), do: count
def calculate_position_ambiguity(latitude, longitude) do
lat_spaces = count_spaces(latitude)
lon_spaces = count_spaces(longitude)
Map.get(
%{{0, 0} => 0, {1, 1} => 1, {2, 2} => 2, {3, 3} => 3, {4, 4} => 4},
{lat_spaces, lon_spaces},
0
)
end
def calculate_compressed_ambiguity(compression_type) do
case compression_type do
" " -> 0
"!" -> 1
"\"" -> 2
"#" -> 3
"$" -> 4
_ -> 0
end
end
def find_matches(regex, text) do
case Regex.names(regex) do
[] ->
matches = Regex.run(regex, text)
Enum.reduce(Enum.with_index(matches), %{}, fn {match, index}, acc ->
Map.put(acc, index, match)
end)
_ ->
Regex.named_captures(regex, text)
end
end
def parse_manufacturer(symbols) do
Aprs.DeviceIdentification.identify_device(symbols)
end
def convert_to_base91(<