Fix APRS parser for direwolf bracket-prefixed frames

This commit is contained in:
Smittix
2026-02-24 22:52:34 +00:00
parent f6b0edaf5a
commit 4bf452d462
3 changed files with 87 additions and 30 deletions
+11
View File
@@ -2862,6 +2862,17 @@ class ModeManager:
def _parse_aprs_packet(self, line: str) -> dict | None: def _parse_aprs_packet(self, line: str) -> dict | None:
"""Parse APRS packet from direwolf or multimon-ng.""" """Parse APRS packet from direwolf or multimon-ng."""
if not line:
return None
# Normalize common decoder prefixes before parsing.
# multimon-ng: "AFSK1200: ..."
# direwolf: "[0.4] ...", "[0L] ..."
line = line.strip()
if line.startswith('AFSK1200:'):
line = line[9:].strip()
line = re.sub(r'^(?:\[[^\]]+\]\s*)+', '', line)
match = re.match(r'([A-Z0-9-]+)>([^:]+):(.+)', line) match = re.match(r'([A-Z0-9-]+)>([^:]+):(.+)', line)
if not match: if not match:
return None return None
+26 -7
View File
@@ -109,6 +109,26 @@ MODEM 1200
return DIREWOLF_CONFIG_PATH return DIREWOLF_CONFIG_PATH
def normalize_aprs_output_line(line: str) -> str:
"""Normalize a decoder output line to raw APRS packet format.
Handles common decoder prefixes:
- multimon-ng: ``AFSK1200: ...``
- direwolf tags: ``[0.4] ...``, ``[0L] ...``, etc.
"""
if not line:
return ''
normalized = line.strip()
if normalized.startswith('AFSK1200:'):
normalized = normalized[9:].strip()
# Strip one or more leading bracket tags emitted by decoders.
# Examples: [0.4], [0L], [NONE]
normalized = re.sub(r'^(?:\[[^\]]+\]\s*)+', '', normalized)
return normalized
def parse_aprs_packet(raw_packet: str) -> Optional[dict]: def parse_aprs_packet(raw_packet: str) -> Optional[dict]:
"""Parse APRS packet into structured data. """Parse APRS packet into structured data.
@@ -125,6 +145,10 @@ def parse_aprs_packet(raw_packet: str) -> Optional[dict]:
- User-defined formats - User-defined formats
""" """
try: try:
raw_packet = normalize_aprs_output_line(raw_packet)
if not raw_packet:
return None
# Basic APRS packet format: CALLSIGN>PATH:DATA # Basic APRS packet format: CALLSIGN>PATH:DATA
# Example: N0CALL-9>APRS,TCPIP*:@092345z4903.50N/07201.75W_090/000g005t077 # Example: N0CALL-9>APRS,TCPIP*:@092345z4903.50N/07201.75W_090/000g005t077
@@ -1348,13 +1372,8 @@ def stream_aprs_output(rtl_process: subprocess.Popen, decoder_process: subproces
app_module.aprs_queue.put(meter_msg) app_module.aprs_queue.put(meter_msg)
continue # Audio level lines are not packets continue # Audio level lines are not packets
# multimon-ng prefixes decoded packets with "AFSK1200: " # Normalize decoder prefixes (multimon/direwolf) before parsing.
if line.startswith('AFSK1200:'): line = normalize_aprs_output_line(line)
line = line[9:].strip()
# direwolf often prefixes packets with "[0.4] " or similar audio level indicator
# Strip any leading bracket prefix like "[0.4] " before parsing
line = re.sub(r'^\[\d+\.\d+\]\s*', '', line)
# Skip non-packet lines (APRS format: CALL>PATH:DATA) # Skip non-packet lines (APRS format: CALL>PATH:DATA)
if '>' not in line or ':' not in line: if '>' not in line or ':' not in line:
+27
View File
@@ -0,0 +1,27 @@
"""APRS packet parser regression tests."""
from __future__ import annotations
import pytest
from routes.aprs import parse_aprs_packet
_BASE_PACKET = "N0CALL-9>APRS,TCPIP*:@092345z4903.50N/07201.75W_090/000g005t077"
@pytest.mark.parametrize(
"line",
[
_BASE_PACKET,
f"[0.4] {_BASE_PACKET}",
f"[0L] {_BASE_PACKET}",
f"AFSK1200: {_BASE_PACKET}",
f"AFSK1200: [0L] {_BASE_PACKET}",
],
)
def test_parse_aprs_packet_accepts_decoder_prefix_variants(line: str) -> None:
packet = parse_aprs_packet(line)
assert packet is not None
assert packet["callsign"] == "N0CALL-9"
assert packet["type"] == "aprs"