149 lines
4.9 KiB
Python
149 lines
4.9 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
SUPPORTED_NMEA_TYPES = {"GGA", "RMC", "GSV", "GSA", "GST"}
|
|
|
|
|
|
def _to_float(value: str) -> float | None:
|
|
if not value:
|
|
return None
|
|
try:
|
|
return float(value)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _to_int(value: str) -> int | None:
|
|
if not value:
|
|
return None
|
|
try:
|
|
return int(value)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _parse_lat_lon(value: str, hemisphere: str) -> float | None:
|
|
if not value or not hemisphere:
|
|
return None
|
|
|
|
try:
|
|
degree_len = 2 if hemisphere in {"N", "S"} else 3
|
|
degrees = float(value[:degree_len])
|
|
minutes = float(value[degree_len:])
|
|
except ValueError:
|
|
return None
|
|
|
|
decimal = degrees + (minutes / 60.0)
|
|
if hemisphere in {"S", "W"}:
|
|
decimal *= -1
|
|
return decimal
|
|
|
|
|
|
def _extract_sentence_type(sentence: str) -> str | None:
|
|
if not sentence.startswith("$"):
|
|
return None
|
|
if len(sentence) < 6:
|
|
return None
|
|
nmea_type = sentence[3:6].upper()
|
|
if nmea_type not in SUPPORTED_NMEA_TYPES:
|
|
return None
|
|
return nmea_type
|
|
|
|
|
|
def _parse_nmea_fields(nmea_type: str, fields: list[str]) -> dict[str, Any]:
|
|
payload: dict[str, Any] = {"sentence_type": nmea_type, "raw_fields": fields}
|
|
|
|
if nmea_type == "GGA":
|
|
payload.update(
|
|
{
|
|
"utc_time": fields[1] if len(fields) > 1 else None,
|
|
"lat": _parse_lat_lon(
|
|
fields[2] if len(fields) > 2 else "",
|
|
fields[3] if len(fields) > 3 else "",
|
|
),
|
|
"lon": _parse_lat_lon(
|
|
fields[4] if len(fields) > 4 else "",
|
|
fields[5] if len(fields) > 5 else "",
|
|
),
|
|
"fix_quality": _to_int(fields[6] if len(fields) > 6 else ""),
|
|
"satellites_used": _to_int(fields[7] if len(fields) > 7 else ""),
|
|
"hdop": _to_float(fields[8] if len(fields) > 8 else ""),
|
|
"altitude_m": _to_float(fields[9] if len(fields) > 9 else ""),
|
|
}
|
|
)
|
|
elif nmea_type == "RMC":
|
|
payload.update(
|
|
{
|
|
"utc_time": fields[1] if len(fields) > 1 else None,
|
|
"status": fields[2] if len(fields) > 2 else None,
|
|
"lat": _parse_lat_lon(
|
|
fields[3] if len(fields) > 3 else "",
|
|
fields[4] if len(fields) > 4 else "",
|
|
),
|
|
"lon": _parse_lat_lon(
|
|
fields[5] if len(fields) > 5 else "",
|
|
fields[6] if len(fields) > 6 else "",
|
|
),
|
|
"speed_knots": _to_float(fields[7] if len(fields) > 7 else ""),
|
|
"course_deg": _to_float(fields[8] if len(fields) > 8 else ""),
|
|
"date": fields[9] if len(fields) > 9 else None,
|
|
}
|
|
)
|
|
elif nmea_type == "GSV":
|
|
payload.update(
|
|
{
|
|
"total_messages": _to_int(fields[1] if len(fields) > 1 else ""),
|
|
"message_index": _to_int(fields[2] if len(fields) > 2 else ""),
|
|
"satellites_in_view": _to_int(fields[3] if len(fields) > 3 else ""),
|
|
"snr_values": [
|
|
_to_int(fields[i])
|
|
for i in range(7, len(fields), 4)
|
|
if i < len(fields)
|
|
],
|
|
}
|
|
)
|
|
elif nmea_type == "GSA":
|
|
payload.update(
|
|
{
|
|
"mode": fields[1] if len(fields) > 1 else None,
|
|
"fix_type": _to_int(fields[2] if len(fields) > 2 else ""),
|
|
"pdop": _to_float(fields[15] if len(fields) > 15 else ""),
|
|
"hdop": _to_float(fields[16] if len(fields) > 16 else ""),
|
|
"vdop": _to_float(fields[17] if len(fields) > 17 else ""),
|
|
}
|
|
)
|
|
elif nmea_type == "GST":
|
|
payload.update(
|
|
{
|
|
"utc_time": fields[1] if len(fields) > 1 else None,
|
|
"rms": _to_float(fields[2] if len(fields) > 2 else ""),
|
|
"std_major": _to_float(fields[3] if len(fields) > 3 else ""),
|
|
"std_minor": _to_float(fields[4] if len(fields) > 4 else ""),
|
|
}
|
|
)
|
|
|
|
return payload
|
|
|
|
|
|
def parse_nmea_sentences(payload: bytes) -> list[dict[str, Any]]:
|
|
events: list[dict[str, Any]] = []
|
|
|
|
for raw_line in payload.splitlines():
|
|
if not raw_line.startswith(b"$"):
|
|
continue
|
|
|
|
sentence = raw_line.decode("ascii", errors="ignore").strip()
|
|
nmea_type = _extract_sentence_type(sentence)
|
|
if nmea_type is None:
|
|
continue
|
|
|
|
main_part = sentence.split("*", 1)[0]
|
|
fields = main_part.split(",")
|
|
parsed_payload = _parse_nmea_fields(nmea_type, fields)
|
|
events.append(
|
|
{"event_type": f"nmea_{nmea_type.lower()}", "payload": parsed_payload}
|
|
)
|
|
|
|
return events
|