Geotag interval samples by the video's own GPS, as distance samples are

mapillary_tools/exif_write.py

@@ -124,12 +124,18 @@ def add_lat_lon(self, lat: float, lon: float) -> None:
             )
 
     def add_altitude(self, altitude: float) -> None:
-        """Add altitude."""
+        """
+        Add altitude. The terms of an EXIF rational are 32-bit, which limits the
+        denominator of altitudes above 4293 m
+        """
 
         ref = 0 if altitude > 0 else 1
+        max_denominator = min(
+            1_000_000, max(1, (2**32 - 1) // (math.ceil(math.fabs(altitude)) + 1))
+        )
         self._ef["GPS"][piexif.GPSIFD.GPSAltitude] = (
             Fraction.from_float(math.fabs(altitude))
-            .limit_denominator()
+            .limit_denominator(max_denominator)
             .as_integer_ratio()
         )
         self._ef["GPS"][piexif.GPSIFD.GPSAltitudeRef] = ref

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."

mapillary_tools/sample_video.py

@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+import dataclasses
 import datetime
 import logging
 import os
@@ -808,6 +809,10 @@ def _sample_single_video_by_interval(
                 f"Unable to extract video start time from {video_path}"
             )
 
+    # Located along the video's own GPS track, as distance samples are
+    track = _read_track_for_interval_samples(video_path, geotag_options)
+    interpolator = geo.Interpolator([track.points]) if track is not None else None
+
     stream_specifier = _interval_stream_specifier(probe)
     # The stream sampled, whether it is the only one or not
     video_stream = probe.probe_video_with_max_resolution()
@@ -835,9 +840,80 @@ def _sample_single_video_by_interval(
             exif_edit = ExifEdit(sample_paths[0])
             exif_edit.add_date_time_original(timestamp)
             exif_edit.add_gps_datetime(timestamp)
+            if (
+                track is not None
+                and interpolator is not None
+                and _within_track_time_range_buffered(track.points, video_time)
+            ):
+                _add_position(exif_edit, track, interpolator.interpolate(video_time))
             exif_edit.write()
 
 
+def _read_track_for_interval_samples(
+    video_path: Path, geotag_options: T.Sequence[SourceOption]
+) -> types.VideoMetadata | None:
+    """
+    The video's own GPS track, which its interval samples are located along
+    as distance samples are, or None to leave their positions to the
+    geotagging stage. That places them by their capture times along a GPX or
+    NMEA track, which interval sampling, unlike distance sampling, does not
+    place along the video: it stamps a video without a GPS clock from its
+    creation time, rather than from the start of the track
+    """
+    if any(option.source in factory.EXTERNAL_GPS_SOURCES for option in geotag_options):
+        return None
+
+    # Through the factory with the same sources as distance sampling, which
+    # reads the video's own GPS by default (see _parse_geotag_options()). One
+    # video is read in this process rather than in a pool of its own
+    video_metadatas = factory.process(
+        [video_path],
+        [
+            dataclasses.replace(option, num_processes=0)
+            for option in geotag_options or [SourceOption(SourceType.NATIVE)]
+        ],
+    )
+    assert len(video_metadatas) == 1, "expect 1 video metadata"
+    video_metadata = video_metadatas[0]
+    if isinstance(video_metadata, types.ErrorMetadata):
+        LOG.warning(
+            "The samples of %s are not geotagged, as no GPS track was read from it: %s",
+            video_path.name,
+            video_metadata.error,
+        )
+        return None
+
+    assert isinstance(video_metadata, types.VideoMetadata), (
+        f"expect VideoMetadata but got {type(video_metadata).__name__}"
+    )
+    if not video_metadata.points:
+        LOG.warning(
+            "The samples of %s are not geotagged, as it has no GPS points",
+            video_path.name,
+        )
+        return None
+
+    return video_metadata
+
+
+def _add_position(
+    exif_edit: ExifEdit, video_metadata: types.VideoMetadata, point: geo.Point
+) -> None:
+    """
+    Write the position of a sample, and the camera of the video it was cut
+    from, as distance sampling writes them
+    """
+    exif_edit.add_lat_lon(point.lat, point.lon)
+    if point.alt is not None:
+        exif_edit.add_altitude(point.alt)
+    if point.angle is not None:
+        exif_edit.add_direction(point.angle)
+    if video_metadata.make:
+        exif_edit.add_make(video_metadata.make)
+    if video_metadata.model:
+        exif_edit.add_model(video_metadata.model)
+
+
 def _within_track_time_range_buffered(points, t: float) -> bool:
     # apply 1ms buffer, which is MAPCaptureTime's precision
     start_point_time = points[0].time - 0.001
@@ -1131,13 +1207,5 @@ def _sample_single_video_by_distance(
             exif_edit = ExifEdit(sample_paths[0])
             exif_edit.add_date_time_original(timestamp)
             exif_edit.add_gps_datetime(timestamp)
-            exif_edit.add_lat_lon(interp.lat, interp.lon)
-            if interp.alt is not None:
-                exif_edit.add_altitude(interp.alt)
-            if interp.angle is not None:
-                exif_edit.add_direction(interp.angle)
-            if video_metadata.make:
-                exif_edit.add_make(video_metadata.make)
-            if video_metadata.model:
-                exif_edit.add_model(video_metadata.model)
+            _add_position(exif_edit, video_metadata, interp)
             exif_edit.write()

tests/integration/test_video_process.py

@@ -18,6 +18,7 @@
 
 from .fixtures import (
     assert_contains_image_descs,
+    assert_same_image_descs,
     EXIFTOOL_EXECUTABLE,
     pytest_skip_if_not_exiftool_installed,
     pytest_skip_if_not_ffmpeg_installed,
@@ -184,6 +185,44 @@ def test_video_process_by_distance_with_a_video_geotag_source_only(
     assert all("error" not in desc for desc in descs), descs
 
 
+def test_video_process_by_interval_with_the_default_geotag_sources(
+    setup_data: py.path.local,
+):
+    """
+    Interval samples are located along the video's own GPS track, as distance
+    samples are, where the legacy source that geotags images from videos would
+    place them. They used to carry no position, and the default sources failed
+    every one of them
+    """
+    pytest_skip_if_not_ffmpeg_installed()
+
+    video_path = setup_data.join("gopro_data").join("max-360mode.mp4")
+    interval = ["--video_sample_interval=2", "--video_sample_distance=-1"]
+
+    with tempfile.TemporaryDirectory() as dir:
+        descs = run_video_process_for_descs([*interval, str(video_path), dir])
+        sample_dir = Path(dir, "max-360mode.mp4")
+        assert descs
+        assert all("error" not in desc for desc in descs), descs
+
+        # sample_video and process, as video_process runs them
+        process_descs = run_process_for_descs([str(sample_dir)])
+        assert_same_image_descs(process_descs, descs)
+
+    with tempfile.TemporaryDirectory() as dir:
+        legacy_descs = run_video_process_for_descs(
+            [*interval, "--geotag_source=gopro_videos", str(video_path), dir]
+        )
+
+    by_name = {Path(d["filename"]).name: d for d in descs}
+    assert len(by_name) == len(legacy_descs)
+    for legacy in legacy_descs:
+        desc = by_name[Path(legacy["filename"]).name]
+        assert desc["MAPCaptureTime"] == legacy["MAPCaptureTime"]
+        assert desc["MAPLatitude"] == pytest.approx(legacy["MAPLatitude"], abs=1e-7)
+        assert desc["MAPLongitude"] == pytest.approx(legacy["MAPLongitude"], abs=1e-7)
+
+
 def _with_creation_time(video: py.path.local, creation_time: str, dir: str) -> Path:
     """A copy of the video whose container records the creation time."""
     copy = Path(dir, video.basename)

tests/unit/test_exifedit.py

@@ -171,6 +171,20 @@ def test_add_lat_lon(self):
     def test_add_altitude(self):
         add_altitude_general(self, EMPTY_EXIF_FILE_TEST)
 
+    def test_add_altitude_above_4293_m(self):
+        # Written with the 1e-6 precision of lower altitudes, the numerator of
+        # the rational would not fit in its 32 bits
+        test_altitude = 7112.573717404068
+
+        empty_exifedit = ExifEdit(EMPTY_EXIF_FILE)
+        empty_exifedit.add_altitude(test_altitude)
+        empty_exifedit.write(EMPTY_EXIF_FILE_TEST)
+
+        exif_data = ExifRead(EMPTY_EXIF_FILE_TEST)
+        altitude = exif_data.extract_altitude()
+        assert altitude is not None
+        self.assertAlmostEqual(test_altitude, altitude, places=5)
+
     def test_add_direction(self):
         add_direction_general(self, EMPTY_EXIF_FILE_TEST)
 

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)