full-stack-fastapi/backend/app/mqtt/parsers/nmea.py

from __future__ import annotations

from typing import Any

SUPPORTED_NMEA_TYPES = {"GGA", "RMC", "GSV", "GSA", "GST"}
TALKER_TO_CONSTELLATION = {
    "GP": "GPS",
    "GL": "GLONASS",
    "GA": "Galileo",
    "GB": "BeiDou",
    "BD": "BeiDou",
    "GQ": "QZSS",
    "GI": "NavIC",
    "GN": "Mixed",
}


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_meta(sentence: str) -> tuple[str, str] | None:
    if not sentence.startswith("$"):
        return None
    if len(sentence) < 6:
        return None
    talker = sentence[1:3].upper()
    nmea_type = sentence[3:6].upper()
    if nmea_type not in SUPPORTED_NMEA_TYPES:
        return None
    return talker, nmea_type


def _parse_satellite_blocks(fields: list[str]) -> list[dict[str, Any]]:
    satellites: list[dict[str, Any]] = []
    for i in range(4, len(fields), 4):
        if i + 3 >= len(fields):
            continue
        satellite_id = fields[i]
        if not satellite_id:
            continue
        satellites.append(
            {
                "satellite_id": satellite_id,
                "elevation_deg": _to_int(fields[i + 1]),
                "azimuth_deg": _to_int(fields[i + 2]),
                "snr_dbhz": _to_int(fields[i + 3]),
            }
        )
    return satellites


def _parse_nmea_fields(
    *, talker: str, nmea_type: str, fields: list[str]
) -> dict[str, Any]:
    payload: dict[str, Any] = {
        "sentence_type": nmea_type,
        "talker": talker,
        "constellation": TALKER_TO_CONSTELLATION.get(talker, "Unknown"),
        "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":
        satellites = _parse_satellite_blocks(fields)
        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 ""),
                "satellites": satellites,
                "snr_values": [
                    satellite["snr_dbhz"]
                    for satellite in satellites
                    if satellite["snr_dbhz"] is not None
                ],
            }
        )
    elif nmea_type == "GSA":
        used_satellite_ids = [sat_id for sat_id in fields[3:15] if sat_id]
        payload.update(
            {
                "mode": fields[1] if len(fields) > 1 else None,
                "fix_type": _to_int(fields[2] if len(fields) > 2 else ""),
                "used_satellite_ids": used_satellite_ids,
                "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()
        sentence_meta = _extract_sentence_meta(sentence)
        if sentence_meta is None:
            continue
        talker, nmea_type = sentence_meta

        main_part = sentence.split("*", 1)[0]
        fields = main_part.split(",")
        parsed_payload = _parse_nmea_fields(
            talker=talker, nmea_type=nmea_type, fields=fields
        )
        events.append(
            {"event_type": f"nmea_{nmea_type.lower()}", "payload": parsed_payload}
        )

    return events