Skip malformed NMEA sentences with a warning

mapillary_tools/geotag/geotag_images_from_nmea_file.py

@@ -56,10 +56,9 @@ def _read_nmea_points(source_path: Path) -> list[geo.Point]:
 
     try:
         points = get_lat_lon_time_from_nmea(source_path)
-    # OSError: unreadable; ValueError: not text (UnicodeError) or a malformed
-    # sentence (pynmea2.ParseError); TypeError: a sentence without its time or
-    # date; UnboundLocalError: GGA sentences but no RMC sentence to give their
-    # date
+    # OSError: unreadable; ValueError: not text (UnicodeError), or malformed
+    # sentences and no fix; TypeError: a sentence without its time or date;
+    # UnboundLocalError: GGA sentences but no RMC sentence to give their date
     except (OSError, ValueError, TypeError, UnboundLocalError) as ex:
         raise exceptions.MapillaryInvalidGPXError(
             f"{source_path.name} is not a valid NMEA file. "
@@ -92,6 +91,10 @@ def get_lat_lon_time_from_nmea(nmea_file: Path) -> list[geo.Point]:
     A receiver may log a fix from several talkers, such as $GPGGA and $GNGGA,
     for the same time: only the first is read, so that no two points share a
     time.
+
+    A malformed sentence, such as one whose checksum does not match, is
+    skipped with a warning. A file of which no fix can be read and that has
+    malformed sentences raises ValueError, as one that is not text does.
     """
     # NMEA sentences are ASCII. Decode them as UTF-8 on every platform, so
     # that a file that is not text fails to read, not only where UTF-8 is the
@@ -100,8 +103,42 @@ def get_lat_lon_time_from_nmea(nmea_file: Path) -> list[geo.Point]:
         lines = f.readlines()
         lines = [line.rstrip("\n\r") for line in lines]
 
-    messages = [pynmea2.parse(line) for line in lines if _SENTENCE_RE.search(line)]
+    messages = []
+    # (line number, line, error) of each malformed sentence
+    malformed: list[tuple[int, str, ValueError]] = []
+    for number, line in enumerate(lines, start=1):
+        if _SENTENCE_RE.search(line):
+            try:
+                messages.append(_parse_sentence(line))
+            except ValueError as ex:
+                malformed.append((number, line, ex))
+
+    points = _read_fixes(messages, nmea_file)
+
+    if malformed:
+        number, line, error = malformed[0]
+        if not points:
+            raise error
+        LOG.warning(
+            "Skipping %d of the %d RMC and GGA sentences of the NMEA file %s, which are malformed, the first on line %d (%s): %s",
+            len(malformed),
+            len(malformed) + len(messages),
+            nmea_file,
+            number,
+            error,
+            line,
+        )
+
+    return points
+
 
+def _read_fixes(
+    messages: list[pynmea2.NMEASentence], nmea_file: Path
+) -> list[geo.Point]:
+    """
+    The points of the fixes of the NMEA sentences read from **nmea_file**, as
+    get_lat_lon_time_from_nmea() describes
+    """
     # The RMC sentences with a date, and the GGA sentences with a fix
     fixes = [
         m
@@ -148,6 +185,45 @@ def get_lat_lon_time_from_nmea(nmea_file: Path) -> list[geo.Point]:
     return points
 
 
+def _parse_sentence(line: str) -> pynmea2.NMEASentence:
+    """
+    Parse an NMEA sentence, raising ValueError if it is malformed: if it does
+    not parse or its checksum does not match, or if a field that a fix is
+    read from does not parse, which pynmea2 leaves a string
+
+    >>> _parse_sentence("$GPGGA,0246,3735.30670,N,12220.25810,W,1,08,1.0,0.0,M,0.0,M,,*64")
+    Traceback (most recent call last):
+    ...
+    ValueError: malformed timestamp: '0246'
+    >>> _parse_sentence("$GPGGA,024636.00,3735.30670,N,12220.25810,W,1,08,1.0,0.0,M,0.0,M,,*00")
+    Traceback (most recent call last):
+    ...
+    ValueError: checksum does not match: 00 != 4F
+    """
+    try:
+        message = pynmea2.parse(line)
+    except pynmea2.ParseError as ex:
+        # Its argument is (reason, sentence)
+        raise ValueError(ex.args[0][0]) from ex
+
+    fields: dict[str, type] = {}
+    if isinstance(message, pynmea2.RMC):
+        fields = {"timestamp": datetime.time, "datestamp": datetime.date}
+    elif isinstance(message, pynmea2.GGA):
+        fields = {"timestamp": datetime.time, "gps_qual": int, "altitude": float}
+    for name, expected_type in fields.items():
+        value = getattr(message, name)
+        if value is not None and not isinstance(value, expected_type):
+            raise ValueError(f"malformed {name}: {value!r}")
+
+    if isinstance(message, pynmea2.GGA) and _has_fix(message):
+        # pynmea2 computes the position from its fields when it is read,
+        # raising ValueError if they are malformed
+        _position = (message.latitude, message.longitude)
+
+    return message
+
+
 def _rmc_datetime(message: pynmea2.RMC) -> datetime.datetime:
     return datetime.datetime.combine(message.datestamp, message.timestamp)
 

mapillary_tools/geotag/video_extractors/nmea.py

@@ -30,10 +30,9 @@ class NMEAVideoExtractor(GPXVideoExtractor):
     def _parse_tracks(self) -> list[list[geo.Point]]:
         try:
             points = get_lat_lon_time_from_nmea(self.gpx_path)
-        # OSError: unreadable; ValueError: not text (UnicodeError), a
-        # malformed sentence (pynmea2.ParseError) or position; TypeError: a
-        # malformed time, which pynmea2 leaves a string
-        except (OSError, ValueError, TypeError) as ex:
+        # OSError: unreadable; ValueError: not text (UnicodeError), or
+        # malformed sentences and no fix
+        except (OSError, ValueError) as ex:
             raise exceptions.MapillaryInvalidGPXError(
                 f"{self.gpx_path.name} is not a valid NMEA file. "
                 "Replace it with a valid NMEA file."

tests/unit/test_geotag_images_from_track_file.py

@@ -114,6 +114,16 @@ def test_a_missing_file_fails_every_image(self, tmp_path):
             GeotagImagesFromNMEAFile(tmp_path / "missing.nmea"), "NMEA file not found"
         )
 
+    def test_a_malformed_sentence_is_skipped(self, tmp_path):
+        """It failed every image"""
+        nmea = tmp_path / "track.nmea"
+        nmea.write_text(NMEA_RMC + NMEA_GGA.replace("$GPGGA", "$GLGGA") + NMEA_GGA)
+
+        geotag = GeotagImagesFromNMEAFile(nmea)
+
+        assert geotag.track_error is None
+        assert len(geotag.points) == 1
+
     def test_a_valid_file_is_read(self, tmp_path):
         nmea = tmp_path / "track.nmea"
         nmea.write_text(NMEA_RMC + NMEA_GGA)

tests/unit/test_nmea_parser.py

@@ -39,7 +39,9 @@ def _gga(time: str, quality: int = 1, talker="GP", lat="3735.30670") -> str:
 
 def _read(tmp_path: Path, *lines: str):
     path = tmp_path / "track.nmea"
-    path.write_text("".join(f"{line}\r\n" for line in lines))
+    # Bytes, so that the line ends are CRLF on every platform: write_text() on
+    # Windows would write each "\r\n" as "\r\r\n", an empty line more
+    path.write_bytes("".join(f"{line}\r\n" for line in lines).encode("ascii"))
     return [
         (
             datetime.datetime.fromtimestamp(p.time, datetime.timezone.utc).isoformat(),
@@ -89,22 +91,26 @@ def test_one_point_per_time(tmp_path):
     ) == [_fix("02:46:36"), _fix("02:46:37", lat=37.588435)]
 
 
-def test_sentences_before_the_first_fix_are_skipped(tmp_path):
+def test_sentences_before_the_first_fix_are_skipped(tmp_path, caplog):
     """
     A log that started before its first fix was rejected whole, and gave
     positions at 0, 0
     """
-    assert _read(
-        tmp_path,
-        # void, with the date of a clock that is not set yet, and empty
-        _rmc("000001.00", date="060180", status="V"),
-        _sentence("GPRMC,,V,,,,,,,,,,N"),
-        # no fix, with and without a time
-        _sentence("GPGGA,024635.00,,,,,0,00,99.99,,,,,,"),
-        _sentence("GPGGA,,,,,,0,00,,,M,,M,,"),
-        _rmc("024636.00"),
-        _gga("024636.00"),
-    ) == [_fix("02:46:36")]
+    with caplog.at_level(logging.WARNING):
+        assert _read(
+            tmp_path,
+            # void, with the date of a clock that is not set yet, and empty
+            _rmc("000001.00", date="060180", status="V"),
+            _sentence("GPRMC,,V,,,,,,,,,,N"),
+            # no fix, with and without a time
+            _sentence("GPGGA,024635.00,,,,,0,00,99.99,,,,,,"),
+            _sentence("GPGGA,,,,,,0,00,,,M,,M,,"),
+            _rmc("024636.00"),
+            _gga("024636.00"),
+        ) == [_fix("02:46:36")]
+
+    # They are not malformed
+    assert not caplog.text
 
 
 def test_a_void_rmc_gives_no_date(tmp_path):
@@ -184,6 +190,65 @@ def test_a_log_of_another_day_appended(tmp_path):
     ]
 
 
+def test_a_malformed_sentence_is_skipped_with_a_warning(tmp_path, caplog):
+    """One failed the whole file, and every image geotagged from it"""
+    bad_checksum = _gga("024636.50", talker="GL")[:-2] + "00"
+    with caplog.at_level(logging.WARNING):
+        assert _read(
+            tmp_path,
+            _rmc("024636.00"),
+            _gga("024636.00"),
+            bad_checksum,
+            _rmc("024637.00"),
+            _gga("024637.00"),
+        ) == [_fix("02:46:36"), _fix("02:46:37")]
+
+    assert "Skipping 1 of the 5 RMC and GGA sentences of the NMEA file" in caplog.text
+    assert "on line 3 (checksum does not match: 00 != " in caplog.text
+    assert bad_checksum in caplog.text
+
+
+@pytest.mark.parametrize(
+    "malformed",
+    [
+        # pynmea2 leaves the fields it cannot parse strings
+        _gga("0246"),
+        _sentence("GPGGA,024636.50,3735.30670,N,12220.25810,W,x,08,1.0,12.5,M,0.0,M,,"),
+        _gga("024636.50").replace(",12.5,", ",12.x,"),
+        _rmc("024636.50", date="2212x2"),
+        # and fails to compute the position from them
+        _gga("024636.50", lat="37x5.30670"),
+        # junk after the sentence type
+        "$GPGGA,\x01\x02",
+    ],
+)
+def test_malformed_fields_are_skipped(tmp_path, caplog, malformed):
+    with caplog.at_level(logging.WARNING):
+        assert _read(
+            tmp_path,
+            _rmc("024636.00"),
+            _gga("024636.00"),
+            malformed,
+            _rmc("024637.00"),
+            _gga("024637.00"),
+        ) == [_fix("02:46:36"), _fix("02:46:37")]
+
+    assert "Skipping 1 of the 5 RMC and GGA sentences" in caplog.text
+
+
+@pytest.mark.parametrize(
+    "lines",
+    [
+        [_rmc("024636.00")[:-2] + "00", _gga("024636.00")[:-2] + "00"],
+        [_rmc("024636.00"), _gga("0246")],
+    ],
+)
+def test_malformed_sentences_and_no_fix_fail_the_file(tmp_path, lines):
+    """As a file that is not NMEA does, rather than as one without a fix"""
+    with pytest.raises(ValueError):
+        _read(tmp_path, *lines)
+
+
 def test_gga_without_any_date_give_no_points(tmp_path, caplog):
     """They raised UnboundLocalError"""
     with caplog.at_level(logging.WARNING):

tests/unit/test_nmea_video.py

@@ -162,6 +162,21 @@ def test_a_track_across_midnight_is_placed_as_from_a_gpx(self, tmp_path):
         for nmea_point, gpx_point in zip(from_nmea, from_gpx):
             assert nmea_point == pytest.approx(gpx_point, abs=1e-6)
 
+    def test_a_malformed_sentence_is_skipped(self, tmp_path):
+        """It failed the whole file"""
+        gpx = _write_gpx(tmp_path / "track.gpx", 6)
+        nmea = _write_nmea(tmp_path / "track.nmea", 6)
+        lines = nmea.read_text().splitlines()
+        lines.insert(5, lines[4][:-2] + "00")
+        nmea.write_text("\r\n".join(lines))
+
+        from_gpx = _points(GPXVideoExtractor(NO_GPS_VIDEO, gpx))
+        from_nmea = _points(NMEAVideoExtractor(NO_GPS_VIDEO, nmea))
+
+        assert len(from_nmea) == len(from_gpx) == 7
+        for nmea_point, gpx_point in zip(from_nmea, from_gpx):
+            assert nmea_point == pytest.approx(gpx_point, abs=1e-6)
+
     def test_messages_name_the_nmea_file(self, tmp_path):
         nmea = _write_nmea(tmp_path / "track.nmea", seconds=60)