Do not require the optional fields of NMEA sentences without checksums

mapillary_tools/geotag/geotag_images_from_nmea_file.py

@@ -40,18 +40,28 @@
 # The byte order mark, which some editors write at the start of a UTF-8 file
 _BOM = "\ufeff"
 
-# How much of a file tells whether it is text at all, before it is read whole:
-# a video given as an NMEA file is gigabytes
+# How much of a file tells whether it is an NMEA log at all, before it is read
+# whole: a video given as an NMEA file is gigabytes
 _SNIFF_BYTES = 64 * 1024
 
+# An RMC or a GGA sentence, as in _SENTENCE_RE, in the bytes of a file
+_SENTENCE_BYTES_RE = re.compile(rb"\$[A-Z]{2}(RMC|GGA),")
+
+# A NUL byte, which a logger leaves in a log after a power cut, such as on an
+# SD card: it ends a line, as no sentence holds one
+_NUL = "\x00"
+
 # The checksum that ends a sentence: *HH. It is optional
 _CHECKSUM_RE = re.compile(r"\*[0-9A-Fa-f]{2}\s*$")
 
-# The fields of a complete sentence: an RMC sentence has 11 (NMEA 2.0 to
-# 2.2), 12 with the mode of NMEA 2.3, and 13 with the navigational status of
-# NMEA 4.1; a GGA sentence has 14. One without its checksum that has fewer was
-# cut short, such as the last one of a log when the recording stopped
-_MIN_FIELDS = {"RMC": 11, "GGA": 14}
+# The fields that a fix is read from: those of an RMC sentence through the
+# date, the 9th, and those of a GGA sentence through the altitude and its
+# unit, the 10th. One without its checksum that has fewer was cut short, such
+# as the last one of a log when the recording stopped. The fields after them
+# are optional, and receivers leave them off: the magnetic variation, the mode
+# and the navigational status of an RMC sentence, the height of the geoid and
+# the DGPS fields of a GGA sentence
+_MIN_FIELDS = {"RMC": 9, "GGA": 10}
 
 # The time of a sentence (hhmmss, and a fraction of a second), and the date of
 # an RMC sentence (ddmmyy). pynmea2 reads a field cut short as another time or
@@ -157,15 +167,16 @@ def get_lat_lon_time_from_nmea(nmea_file: Path) -> list[geo.Point]:
     own fields show it. A file of which no fix can be read raises ValueError,
     which says why, if it has malformed sentences or bytes that are not text.
 
-    A file that is not text at all, which its first bytes show, raises
-    ValueError without being read further.
+    A file that is no NMEA log at all, such as a video, which its first bytes
+    show (see _looks_binary()), raises ValueError without being read further.
+    NUL bytes, which a logger leaves in a log after a power cut, end a line.
 
     A file is read once in a process, however many times it is asked for.
     """
     with nmea_file.open("rb") as f:
         head = f.read(_SNIFF_BYTES)
-        if not _looks_like_text(head):
-            raise ValueError("it is not text")
+        if _looks_binary(head):
+            raise ValueError(_describe_binary(head))
         data = head + f.read()
     key = (str(nmea_file.resolve()), hashlib.sha256(data).hexdigest())
     read = _READ.get(key)
@@ -182,25 +193,49 @@ def get_lat_lon_time_from_nmea(nmea_file: Path) -> list[geo.Point]:
     return [dataclasses.replace(p) for p in read]
 
 
-def _looks_like_text(head: bytes) -> bool:
+def _looks_binary(head: bytes) -> bool:
     r"""
-    Whether the first bytes **head** of a file can be those of a text file:
-    without NUL bytes, and mostly bytes that decode as UTF-8, where a log may
-    have a few bytes damaged
+    Whether the first bytes **head** of a file show it to be no NMEA log at
+    all, such as a video: they hold no RMC or GGA sentence, and NUL bytes or
+    mostly bytes that do not decode as UTF-8. A log padded with NUL bytes, as a
+    logger leaves it after a power cut, holds sentences, and so does one with
+    a few bytes damaged
 
-    >>> _looks_like_text(b"$GPRMC,024636.00,A,3735.30670,N\r\n")
-    True
-    >>> _looks_like_text(b"$GPRMC,024636.00,\xe9,3735.30670,N\r\n")
-    True
-    >>> _looks_like_text(b"\x00\x00\x00\x18ftypmp42")
+    >>> _looks_binary(b"$GPRMC,024636.00,A,3735.30670,N\r\n" + b"\x00" * 100)
     False
-    >>> _looks_like_text(b"\xca\xfe\xba\xbe" * 100)
+    >>> _looks_binary(b"\x00" * 100 + b"$GPGGA,024636.00,3735.30670,N\r\n")
     False
+    >>> _looks_binary(b"$GPRMC,024636.00,\xe9,3735.30670,N\r\n")
+    False
+    >>> _looks_binary(b"$GPGSV,3,1,11,03,03,111,00\r\n")
+    False
+    >>> _looks_binary(b"\x00\x00\x00\x18ftypmp42")
+    True
+    >>> _looks_binary(b"\xca\xfe\xba\xbe" * 100)
+    True
     """
-    if b"\x00" in head:
+    if _SENTENCE_BYTES_RE.search(head):
         return False
+    if b"\x00" in head:
+        return True
     text = head.decode("utf-8", errors="replace")
-    return text.count(_NOT_TEXT) * 2 <= len(text)
+    return len(text) < text.count(_NOT_TEXT) * 2
+
+
+def _describe_binary(head: bytes) -> str:
+    r"""
+    Why a file whose first bytes **head** show it to be no NMEA log is not
+    read: it is UTF-16 text, which some Windows tools write, or no text at all
+
+    >>> _describe_binary("$GPRMC,024636.00,A".encode("utf-16"))
+    'it is UTF-16 text, which is not read: save it as UTF-8 or ASCII'
+    >>> _describe_binary(b"\x00\x00\x00\x18ftypmp42")
+    'it is not text'
+    """
+    for encoding in ("utf-16", "utf-16-le", "utf-16-be"):
+        if _SENTENCE_RE.search(head.decode(encoding, errors="ignore")):
+            return "it is UTF-16 text, which is not read: save it as UTF-8 or ASCII"
+    return "it is not text"
 
 
 def _read_nmea_data(nmea_file: Path, data: bytes) -> list[geo.Point]:
@@ -217,14 +252,15 @@ def _read_nmea_data(nmea_file: Path, data: bytes) -> list[geo.Point]:
         io.BytesIO(data), encoding="utf-8-sig", errors="replace"
     ) as f:
         lines = f.readlines()
-        lines = [line.rstrip("\n\r").lstrip(_BOM) for line in lines]
     not_text = any(_NOT_TEXT in line for line in lines)
 
-    # (line number, line) of each RMC and GGA sentence
+    # (line number, line) of each RMC and GGA sentence, of the lines that NUL
+    # bytes end too (see _NUL)
     sentences = [
-        (number, line)
+        (number, piece.lstrip(_BOM))
         for number, line in enumerate(lines, start=1)
-        if _SENTENCE_RE.search(line)
+        for piece in line.rstrip("\n\r").split(_NUL)
+        if _SENTENCE_RE.search(piece)
     ]
 
     messages = []
@@ -330,7 +366,7 @@ 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, if it has no checksum and fewer
-    fields than a complete sentence (see _MIN_FIELDS), if its time or the date
+    fields than a fix is read from (see _MIN_FIELDS), if its time or the date
     of an RMC sentence is not hhmmss[.ss] or ddmmyy, or if a field that a fix
     is read from does not parse, which pynmea2 leaves a string, or if it has a
     byte that is not text, decoded as U+FFFD
@@ -347,15 +383,17 @@ def _parse_sentence(line: str) -> pynmea2.NMEASentence:
     Traceback (most recent call last):
     ...
     ValueError: malformed datestamp: '2212'
-    >>> _parse_sentence("$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,221222")
+    >>> _parse_sentence("$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0")
     Traceback (most recent call last):
     ...
-    ValueError: cut short: 9 of the 11 fields of an RMC sentence
+    ValueError: cut short: 8 of the 9 fields of an RMC sentence that a fix is read from
+    >>> type(_parse_sentence("$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,221222")).__name__
+    'RMC'
     >>> _parse_sentence("$GPGGA,024636.00,3735.30670,N,12220.25810,W,1,08,1.0,12")
     Traceback (most recent call last):
     ...
-    ValueError: cut short: 9 of the 14 fields of a GGA sentence
-    >>> type(_parse_sentence("$GPGGA,024636.00,3735.30670,N,12220.25810,W,1,08,1.0,12.5,M,0.0,M,,")).__name__
+    ValueError: cut short: 9 of the 10 fields of a GGA sentence that a fix is read from
+    >>> type(_parse_sentence("$GPGGA,024636.00,3735.30670,N,12220.25810,W,1,08,1.0,12.5,M")).__name__
     'GGA'
     """
     if _NOT_TEXT in line:
@@ -382,7 +420,7 @@ def _parse_sentence(line: str) -> pynmea2.NMEASentence:
         if kind in _MIN_FIELDS and len(message.data) < _MIN_FIELDS[kind]:
             article = "an" if kind == "RMC" else "a"
             raise ValueError(
-                f"cut short: {len(message.data)} of the {_MIN_FIELDS[kind]} fields of {article} {kind} sentence"
+                f"cut short: {len(message.data)} of the {_MIN_FIELDS[kind]} fields of {article} {kind} sentence that a fix is read from"
             )
 
     fields: dict[str, type] = {}

tests/unit/test_nmea_parser.py

@@ -269,7 +269,7 @@ def test_sentences_cut_short_are_skipped_with_one_warning(tmp_path, caplog):
         lines += [_rmc(f"02463{second}.00"), _gga(f"02463{second}.00")]
     # Every other fix cut short, at another field each time
     for idx in range(1, 20, 4):
-        lines[idx] = lines[idx][: 10 + 3 * idx]
+        lines[idx] = lines[idx][: 6 + 3 * idx]
 
     with caplog.at_level(logging.WARNING):
         points = _read(tmp_path, *lines)
@@ -388,17 +388,43 @@ def test_complete_sentences_without_checksums_are_read(tmp_path, caplog, rmc_tai
     assert not caplog.text
 
 
+@pytest.mark.parametrize(
+    "rmc, gga",
+    [
+        # Through the date, and without the height of the geoid and the DGPS
+        # fields
+        (
+            "$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,221222",
+            "$GPGGA,024636.00,3735.30670,N,12220.25810,W,1,08,1.0,12.5,M",
+        ),
+        # Through the magnetic variation, and without the DGPS fields
+        (
+            "$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,221222,",
+            "$GPGGA,024636.00,3735.30670,N,12220.25810,W,1,08,1.0,12.5,M,0.0,M",
+        ),
+    ],
+    ids=["9 and 10 fields", "10 and 12 fields"],
+)
+def test_sentences_without_their_optional_fields_are_read(tmp_path, caplog, rmc, gga):
+    """A log without checksums whose sentences left them off was failed whole"""
+    with caplog.at_level(logging.WARNING):
+        fixes = _read(tmp_path, rmc, gga)
+
+    assert fixes == [_fix("02:46:36")]
+    assert not caplog.text
+
+
 @pytest.mark.parametrize(
     "cut, reason",
     [
         # Whole fields, the last of them maybe cut short too
         (
-            "$GPRMC,024637.00,A,3735.30670,N,12220.25810,W,0.0,0.0,221222",
-            "cut short: 9 of the 11 fields of an RMC sentence",
+            "$GPRMC,024637.00,A,3735.30670,N,12220.25810,W,0.0,0.0",
+            "cut short: 8 of the 9 fields of an RMC sentence that a fix is read from",
         ),
         (
             "$GPGGA,024637.00,3735.30670,N,12220.25810,W,1,08,1.0,12",
-            "cut short: 9 of the 14 fields of a GGA sentence",
+            "cut short: 9 of the 10 fields of a GGA sentence that a fix is read from",
         ),
         # A time that does not parse
         ("$GPGGA,02463", "malformed timestamp: '02463'"),
@@ -503,6 +529,54 @@ def test_a_file_that_is_not_text_fails_without_being_read_whole(tmp_path, head):
         _read_bytes(tmp_path, head + padding + sentences)
 
 
+def test_a_log_padded_with_nul_bytes_is_read(tmp_path, caplog):
+    """
+    A logger leaves NUL bytes in a log after a power cut, such as on an SD card:
+    one NUL byte failed the whole log as not text
+    """
+    lines = [_rmc("024636.00"), _gga("024636.00"), _rmc("024637.00"), _gga("024637.00")]
+
+    with caplog.at_level(logging.WARNING):
+        times = _read_bytes(tmp_path, _crlf(lines) + b"\x00" * 4096)
+
+    assert len(times) == 2
+    assert not caplog.text
+
+
+def test_nul_bytes_before_the_first_sentence_are_skipped(tmp_path, caplog):
+    lines = [_rmc("024636.00"), _gga("024636.00")]
+
+    with caplog.at_level(logging.WARNING):
+        times = _read_bytes(tmp_path, b"\x00" * 512 + _crlf(lines))
+
+    assert len(times) == 1
+    assert not caplog.text
+
+
+def test_nul_bytes_end_a_sentence_as_a_line_end_does(tmp_path, caplog):
+    """A logger that resumed after a power cut wrote on after the NUL bytes"""
+    first = (_rmc("024636.00") + "\r\n" + _gga("024636.00")).encode("ascii")
+    second = _crlf([_rmc("024637.00"), _gga("024637.00")])
+
+    with caplog.at_level(logging.WARNING):
+        times = _read_bytes(tmp_path, first + b"\x00" * 64 + second)
+
+    assert len(times) == 2
+    assert not caplog.text
+
+
+def test_a_sentence_with_nul_bytes_inside_is_skipped(tmp_path, caplog):
+    damaged = _gga("024637.00").replace(",N,", ",N\0\0", 1)
+    lines = [_rmc("024636.00"), _gga("024636.00"), _rmc("024637.00"), damaged]
+    content = "".join(f"{line}\r\n" for line in lines).encode("ascii")
+
+    with caplog.at_level(logging.WARNING):
+        times = _read_bytes(tmp_path, content)
+
+    assert len(times) == 1
+    assert "Skipping 1 of the 4 RMC and GGA sentences" in caplog.text
+
+
 def test_a_log_longer_than_what_tells_text_is_read_whole(tmp_path):
     lines = []
     for minute in range(60):
@@ -543,13 +617,14 @@ def test_logs_with_byte_order_marks_appended_are_read(tmp_path, caplog):
     assert not caplog.text
 
 
-def test_a_utf16_log_is_not_read(tmp_path):
-    """As one that is not text: NMEA is ASCII"""
+@pytest.mark.parametrize("encoding", ["utf-16", "utf-16-le", "utf-16-be"])
+def test_a_utf16_log_is_not_read(tmp_path, encoding):
+    """NMEA is ASCII: UTF-16 is not read, and the error says it is UTF-16"""
     content = "".join(
         f"{line}\r\n" for line in [_rmc("024636.00"), _gga("024636.00")]
-    ).encode("utf-16")
+    ).encode(encoding)
 
-    with pytest.raises(ValueError, match="^it is not text$"):
+    with pytest.raises(ValueError, match="^it is UTF-16 text, which is not read"):
         _read_bytes(tmp_path, content)