Judge GPX gaps against the limit in GPS seconds

mapillary_tools/geotag/factory.py

@@ -257,6 +257,8 @@ def process(
     options: T.Collection[SourceOption],
     # --filetypes, which the filetypes of every option are limited to already
     filetypes: set[types.FileType] | None = None,
+    # If given, filled with the option that geotagged each file, of those geotagged
+    sources: dict[Path, SourceOption] | None = None,
 ) -> list[types.MetadataOrError]:
     if not options:
         raise ValueError("No geotag options provided")
@@ -378,10 +380,11 @@ def process(
                 reprocessable_paths.remove(metadata.filename)
                 pending_errors.pop(metadata.filename, None)
                 rejection = gpx_rejections.pop(metadata.filename, None)
-                if rejection is not None and not isinstance(
-                    metadata, types.ErrorMetadata
-                ):
-                    rescued.append((metadata, *rejection))
+                if not isinstance(metadata, types.ErrorMetadata):
+                    if sources is not None:
+                        sources[metadata.filename] = option
+                    if rejection is not None:
+                        rescued.append((metadata, *rejection))
 
         _warn_geotagged_despite_gpx(rescued, option)
 

mapillary_tools/geotag/geotag_images_from_gpx_file.py

@@ -48,25 +48,30 @@ def _read_gpx_points(source_path: Path) -> list[geo.Point]:
     geotagged from it, and if it is missing or has no timestamped point: the
     file was given to geotag the images from, so it fails them rather than
     leaving them to the next geotag source.
+
+    The file is parsed once in a process (see utils.read_gpx_tracks()), and
+    its points are shared with its other readers, which must not change them.
     """
     if not source_path.is_file():
         raise exceptions.MapillaryInvalidGPXError(f"GPX file not found: {source_path}")
 
     try:
-        tracks = utils.parse_gpx(source_path)
+        tracks = utils.read_gpx_tracks(source_path)
     except (OSError, UnicodeError, gpxpy.gpx.GPXException) as ex:
         raise exceptions.MapillaryInvalidGPXError(
             f"{source_path.name} is not a valid GPX file. "
             "Replace it with a valid GPX file."
         ) from ex
 
     if 1 < len(tracks):
-        LOG.warning(
+        utils.warn_once(
+            source_path,
+            ("tracks", len(tracks)),
             "Found %s tracks in the GPX file %s. Will merge points in all the tracks as a single track for interpolation",
             len(tracks),
             source_path,
         )
-    points: list[geo.Point] = sum(tracks, [])
+    points: list[geo.Point] = [point for track in tracks for point in track]
 
     if not points:
         raise exceptions.MapillaryInvalidGPXError(

mapillary_tools/geotag/geotag_images_from_nmea_file.py

@@ -37,9 +37,19 @@
 # What a byte that is not text decodes to
 _NOT_TEXT = "\ufffd"
 
-# The checksum that ends a sentence: *HH
+# 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
+_SNIFF_BYTES = 64 * 1024
+
+# 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 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
 # date: an RMC sentence cut inside its date "2212" as 2012-02-02
@@ -85,8 +95,7 @@ def _read_nmea_points(source_path: Path) -> list[geo.Point]:
     # 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. "
-            "Replace it with a valid NMEA file."
+            describe_invalid_nmea(source_path, ex)
         ) from ex
 
     if not points:
@@ -97,6 +106,29 @@ def _read_nmea_points(source_path: Path) -> list[geo.Point]:
     return points
 
 
+def describe_invalid_nmea(nmea_file: Path, ex: Exception) -> str:
+    """
+    The message of the error for the NMEA file **nmea_file** that reading it
+    failed with **ex**, which says why
+
+    >>> describe_invalid_nmea(Path("a/track.nmea"), ValueError("it is not text"))
+    'track.nmea is not a valid NMEA file: it is not text. Replace it with a valid NMEA file.'
+    >>> describe_invalid_nmea(Path("track.nmea"), PermissionError(13, "Permission denied"))
+    'track.nmea is not a valid NMEA file: Permission denied. Replace it with a valid NMEA file.'
+    """
+    if isinstance(ex, OSError):
+        reason = ex.strerror or str(ex)
+    elif isinstance(ex, ValueError):
+        reason = str(ex)
+    else:
+        reason = ""
+    because = f": {reason}" if reason else ""
+    return (
+        f"{nmea_file.name} is not a valid NMEA file{because}. "
+        "Replace it with a valid NMEA file."
+    )
+
+
 def get_lat_lon_time_from_nmea(nmea_file: Path) -> list[geo.Point]:
     """
     Read the fixes of an NMEA file: the time and position of each GGA
@@ -117,13 +149,21 @@ def get_lat_lon_time_from_nmea(nmea_file: Path) -> list[geo.Point]:
     time.
 
     A malformed sentence, such as one whose checksum does not match, one cut
-    short, or one with a byte that is not text, is skipped with a warning. A
-    file of which no fix can be read raises ValueError if it has malformed
-    sentences or bytes that are not text.
+    short, or one with a byte that is not text, is skipped with a warning.
+    Checksums are optional: a sentence without one is cut short only if its
+    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 is read once in a process, however many times it is asked for.
     """
-    data = nmea_file.read_bytes()
+    with nmea_file.open("rb") as f:
+        head = f.read(_SNIFF_BYTES)
+        if not _looks_like_text(head):
+            raise ValueError("it is not text")
+        data = head + f.read()
     key = (str(nmea_file.resolve()), hashlib.sha256(data).hexdigest())
     read = _READ.get(key)
     if read is None:
@@ -139,6 +179,27 @@ 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:
+    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
+
+    >>> _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")
+    False
+    >>> _looks_like_text(b"\xca\xfe\xba\xbe" * 100)
+    False
+    """
+    if b"\x00" in head:
+        return False
+    text = head.decode("utf-8", errors="replace")
+    return text.count(_NOT_TEXT) * 2 <= len(text)
+
+
 def _read_nmea_data(nmea_file: Path, data: bytes) -> list[geo.Point]:
     """
     Read the fixes of the content **data** of an NMEA file, as
@@ -158,29 +219,28 @@ def _read_nmea_data(nmea_file: Path, data: bytes) -> list[geo.Point]:
         for number, line in enumerate(lines, start=1)
         if _SENTENCE_RE.search(line)
     ]
-    # Receivers end each sentence with its checksum, or none of them. One
-    # without, where the others have theirs, was cut short, such as the last
-    # one of a log when the recording stopped
-    checksummed = any(_CHECKSUM_RE.search(line) for _, line in sentences)
 
     messages = []
     # (line number, line, error) of each malformed sentence
     malformed: list[tuple[int, str, ValueError]] = []
     for number, line in sentences:
         try:
-            messages.append(_parse_sentence(line, checksummed=checksummed))
+            messages.append(_parse_sentence(line))
         except ValueError as ex:
             malformed.append((number, line, ex))
 
     points = _read_fixes(messages, nmea_file)
 
     if not points and not_text and not malformed:
-        raise ValueError(f"{nmea_file} is not text")
+        raise ValueError("it is not text")
 
     if malformed:
         number, line, error = malformed[0]
         if not points:
-            raise error
+            raise ValueError(
+                f"{len(malformed)} of its {len(sentences)} RMC and GGA sentences are malformed, "
+                f"the first on line {number} ({_quote(str(error))}): {_quote(line)}"
+            ) from 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),
@@ -259,14 +319,14 @@ def _read_fixes(
     return points
 
 
-def _parse_sentence(line: str, checksummed: bool = False) -> pynmea2.NMEASentence:
+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 but should
-    (**checksummed**), 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
+    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
+    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
 
     >>> _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):
@@ -276,21 +336,24 @@ def _parse_sentence(line: str, checksummed: bool = False) -> pynmea2.NMEASentenc
     Traceback (most recent call last):
     ...
     ValueError: checksum does not match: 00 != 4F
-    >>> _parse_sentence("$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,2212", checksummed=True)
-    Traceback (most recent call last):
-    ...
-    ValueError: no checksum
     >>> _parse_sentence("$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,2212")
     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")
+    Traceback (most recent call last):
+    ...
+    ValueError: cut short: 9 of the 11 fields of an RMC sentence
+    >>> _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__
+    'GGA'
     """
     if _NOT_TEXT in line:
         raise ValueError("a byte that is not text")
 
-    if checksummed and not _CHECKSUM_RE.search(line):
-        raise ValueError("no checksum")
-
     try:
         message = pynmea2.parse(line)
     except pynmea2.ParseError as ex:
@@ -306,6 +369,15 @@ def _parse_sentence(line: str, checksummed: bool = False) -> pynmea2.NMEASentenc
         if value and not pattern.fullmatch(value):
             raise ValueError(f"malformed {name}: {value!r}")
 
+    # A sentence with its checksum ended where it ends: pynmea2 checks it
+    if not _CHECKSUM_RE.search(line):
+        kind = message.sentence_type
+        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"
+            )
+
     fields: dict[str, type] = {}
     if isinstance(message, pynmea2.RMC):
         fields = {"timestamp": datetime.time, "datestamp": datetime.date}

mapillary_tools/geotag/geotag_videos_from_gpx.py

@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+import collections
 import logging
 import sys
 import typing as T
@@ -16,7 +17,7 @@
     from typing_extensions import override
 
 from .. import exceptions, types
-from . import options
+from . import options, utils
 from .base import GeotagVideosFromGeneric
 from .video_extractors.gpx import GPXVideoExtractor
 
@@ -72,4 +73,28 @@ def _generate_video_extractors(
                         filetype=types.FileType.VIDEO,
                     )
                 )
+
+        # The videos are geotagged in a pool of processes, each of which would
+        # read the track file of its videos. Read here, once, and warn about
+        # once, a file that geotags several of them, or that this process read
+        # already (see _read_here())
+        extractors = [
+            result for result in results if isinstance(result, GPXVideoExtractor)
+        ]
+        num_videos = collections.Counter(
+            extractor.gpx_path.resolve() for extractor in extractors
+        )
+        for extractor in extractors:
+            if self._read_here(extractor, num_videos[extractor.gpx_path.resolve()]):
+                extractor.read_tracks()
+
         return results
+
+    def _read_here(self, extractor: GPXVideoExtractor, num_videos: int) -> bool:
+        """
+        Whether to read the track file of **extractor**, which geotags
+        **num_videos** videos, in this process rather than in the pool: if it
+        geotags several, or was parsed in this process already. A file of
+        each video is left to the pool, which parses them side by side
+        """
+        return 1 < num_videos or utils.was_gpx_read(extractor.gpx_path)

mapillary_tools/geotag/geotag_videos_from_nmea.py

@@ -6,15 +6,12 @@
 from __future__ import annotations
 
 import sys
-import typing as T
-from pathlib import Path
 
 if sys.version_info >= (3, 12):
     from typing import override
 else:
     from typing_extensions import override
 
-from .. import types
 from .geotag_videos_from_gpx import GeotagVideosFromGPX
 from .video_extractors.gpx import GPXVideoExtractor
 from .video_extractors.nmea import NMEAVideoExtractor
@@ -30,14 +27,7 @@ class GeotagVideosFromNMEA(GeotagVideosFromGPX):
     default_pattern = "%g.nmea"
 
     @override
-    def _generate_video_extractors(
-        self, video_paths: T.Sequence[Path]
-    ) -> T.Sequence[GPXVideoExtractor | types.ErrorMetadata]:
-        extractors = super()._generate_video_extractors(video_paths)
-        # Read each NMEA file here, once however many videos it geotags, and
-        # warn about it once: the videos are geotagged in a pool of processes,
-        # each of which would read it again
-        for extractor in extractors:
-            if isinstance(extractor, NMEAVideoExtractor):
-                extractor.read_tracks()
-        return extractors
+    def _read_here(self, extractor: GPXVideoExtractor, num_videos: int) -> bool:
+        # Every NMEA file, which is read once in a process however many videos
+        # it geotags, and warned about once
+        return True