Geotag videos from NMEA files

README.md

@@ -462,6 +462,8 @@ with a GPX file called `foo.gpx`.
 ```sh
 mapillary_tools process --video_geotag_source gpx VIDEO_DIR/
 ```
+NMEA files work the same way with `--video_geotag_source nmea`, which associates `foo.mp4` with `foo.nmea`. What
+follows about GPX tracks applies to NMEA tracks too.
 
 #### Insta360 stitched videos
 

mapillary_tools/geotag/factory.py

@@ -22,6 +22,7 @@
     geotag_images_from_video,
     geotag_videos_from_exiftool,
     geotag_videos_from_gpx,
+    geotag_videos_from_nmea,
     geotag_videos_from_video,
 )
 from .options import (
@@ -38,8 +39,7 @@
 _ALL_VIDEO_TYPES = {types.FileType.VIDEO, *types.NATIVE_VIDEO_FILETYPES}
 
 # Sources that read a GPS track from outside the video file, as opposed to
-# re-reading the telemetry embedded in it. (NMEA cannot geotag videos yet: it
-# skips them, and they keep the error they had)
+# re-reading the telemetry embedded in it
 EXTERNAL_GPS_SOURCES = frozenset({SourceType.GPX, SourceType.NMEA})
 
 
@@ -146,6 +146,8 @@ def can_geotag_videos(option: SourceOption) -> bool:
     Examples:
         >>> can_geotag_videos(SourceOption(SourceType.GPX))
         True
+        >>> can_geotag_videos(SourceOption(SourceType.NMEA))
+        True
         >>> can_geotag_videos(SourceOption(SourceType.EXIF))
         False
         >>> can_geotag_videos(
@@ -478,8 +480,9 @@ def _build_video_geotag(option: SourceOption) -> base.GeotagVideosFromGeneric |
         'GeotagVideosFromGPX'
 
         >>> opt = SourceOption(SourceType.NMEA, source_path=Path("/tmp/test.nmea"))
-        >>> _build_video_geotag(opt) is None
-        True
+        >>> geotagger = _build_video_geotag(opt)
+        >>> geotagger.__class__.__name__
+        'GeotagVideosFromNMEA'
 
         >>> opt = SourceOption(SourceType.EXIF)
         >>> _build_video_geotag(opt) is None
@@ -523,8 +526,12 @@ def _build_video_geotag(option: SourceOption) -> base.GeotagVideosFromGeneric |
         )
 
     elif option.source is SourceType.NMEA:
-        # TODO: geotag videos from NMEA
-        return None
+        return geotag_videos_from_nmea.GeotagVideosFromNMEA(
+            source_path=option.source_path,
+            num_processes=option.num_processes,
+            video_sync=option.video_sync,
+            interpolation=option.interpolation,
+        )
 
     elif option.source is SourceType.EXIF:
         # Legacy image-specific geotag types

mapillary_tools/geotag/geotag_images_from_nmea_file.py

@@ -66,7 +66,10 @@ def _read_nmea_points(source_path: Path) -> list[geo.Point]:
 
 
 def get_lat_lon_time_from_nmea(nmea_file: Path) -> list[geo.Point]:
-    with nmea_file.open("r") as f:
+    # 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
+    # default encoding
+    with nmea_file.open("r", encoding="utf-8") as f:
         lines = f.readlines()
         lines = [line.rstrip("\n\r") for line in lines]
 

mapillary_tools/geotag/geotag_videos_from_gpx.py

@@ -25,6 +25,10 @@
 
 
 class GeotagVideosFromGPX(GeotagVideosFromGeneric):
+    # Reads the track file of each video, and where to find it by default
+    extractor_class: type[GPXVideoExtractor] = GPXVideoExtractor
+    default_pattern = "%g.gpx"
+
     def __init__(
         self,
         source_path: options.SourcePathOption | None = None,
@@ -34,7 +38,7 @@ def __init__(
     ):
         super().__init__(num_processes=num_processes)
         if source_path is None:
-            source_path = options.SourcePathOption(pattern="%g.gpx")
+            source_path = options.SourcePathOption(pattern=self.default_pattern)
         self.source_path = source_path
         self.video_sync = video_sync
         self.interpolation = interpolation
@@ -48,7 +52,7 @@ def _generate_video_extractors(
             source_path = self.source_path.resolve(video_path)
             if source_path.is_file():
                 results.append(
-                    GPXVideoExtractor(
+                    self.extractor_class(
                         video_path,
                         source_path,
                         video_sync=self.video_sync,
@@ -62,7 +66,7 @@ def _generate_video_extractors(
                 results.append(
                     types.describe_error_metadata(
                         exceptions.MapillaryVideoGPSNotFoundError(
-                            "GPX file not found for video"
+                            f"{self.extractor_class.file_format} file not found for video"
                         ),
                         filename=video_path,
                         filetype=types.FileType.VIDEO,

mapillary_tools/geotag/geotag_videos_from_nmea.py

@@ -0,0 +1,19 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+#
+# This source code is licensed under the BSD license found in the
+# LICENSE file in the root directory of this source tree.
+
+from __future__ import annotations
+
+from .geotag_videos_from_gpx import GeotagVideosFromGPX
+from .video_extractors.nmea import NMEAVideoExtractor
+
+
+class GeotagVideosFromNMEA(GeotagVideosFromGPX):
+    """
+    Geotag videos from NMEA files, which are placed on the videos as GPX
+    files are
+    """
+
+    extractor_class = NMEAVideoExtractor
+    default_pattern = "%g.nmea"

mapillary_tools/geotag/utils.py

@@ -29,28 +29,42 @@ def parse_gpx(gpx_file: Path) -> list[Track]:
             tracks.append([])
             for point in segment.points:
                 if point.time is not None:
-                    unix_time = geo.as_unix_time(point.time)
                     tracks[-1].append(
-                        telemetry.CAMMGPSPoint(
-                            time=unix_time,
+                        track_point(
+                            geo.as_unix_time(point.time),
                             lat=point.latitude,
                             lon=point.longitude,
                             alt=point.elevation,
-                            angle=None,
-                            epoch_time=unix_time,
-                            gps_fix_type=3 if point.elevation is not None else 2,
-                            horizontal_accuracy=0.0,
-                            vertical_accuracy=0.0,
-                            velocity_east=0.0,
-                            velocity_north=0.0,
-                            velocity_up=0.0,
-                            speed_accuracy=0.0,
                         )
                     )
 
     return tracks
 
 
+def track_point(
+    unix_time: float, lat: float, lon: float, alt: float | None
+) -> telemetry.CAMMGPSPoint:
+    """
+    A point of a track read from a file beside a video (GPX or NMEA), at its
+    Unix time, as it is written to the CAMM track uploaded with the video
+    """
+    return telemetry.CAMMGPSPoint(
+        time=unix_time,
+        lat=lat,
+        lon=lon,
+        alt=alt,
+        angle=None,
+        epoch_time=unix_time,
+        gps_fix_type=3 if alt is not None else 2,
+        horizontal_accuracy=0.0,
+        vertical_accuracy=0.0,
+        velocity_east=0.0,
+        velocity_north=0.0,
+        velocity_up=0.0,
+        speed_accuracy=0.0,
+    )
+
+
 def index_rdf_description_by_path(xml_paths: T.Sequence[Path]) -> dict[str, ET.Element]:
     rdf_by_path: dict[str, ET.Element] = {}
 

mapillary_tools/geotag/video_extractors/gpx.py

@@ -99,6 +99,9 @@ def to_unix_time(self, video_time: float) -> float:
 
 
 class GPXVideoExtractor(BaseVideoExtractor):
+    # The format of the track file, as messages name it
+    file_format = "GPX"
+
     def __init__(
         self,
         video_path: Path,
@@ -174,10 +177,11 @@ def extract(self) -> types.VideoMetadata:
                 ):
                     LOG.warning(
                         "The GPS clock of %s, whose GPS was rejected as noise, is %.1f days away from %s. "
-                        "Syncing the GPX from the start of the video instead",
+                        "Syncing the %s from the start of the video instead",
                         self.video_path,
                         abs(offset) / 86400,
                         self.gpx_path,
+                        self.file_format,
                     )
                     clock = None
 
@@ -222,26 +226,21 @@ def _read_gpx_points(self) -> list[geo.Point]:
         """
         Read the points of the GPX file, merged into one track in time order
         """
-        try:
-            gpx_tracks = parse_gpx(self.gpx_path)
-        except (OSError, UnicodeError, gpxpy.gpx.GPXException) as ex:
-            raise exceptions.MapillaryInvalidGPXError(
-                f"{self.gpx_path.name} is not a valid GPX file. "
-                "Replace it with a valid GPX file."
-            ) from ex
+        gpx_tracks = self._parse_tracks()
 
         if 1 < len(gpx_tracks):
             LOG.warning(
-                f"Found {len(gpx_tracks)} tracks in the GPX file {self.gpx_path}. Will merge points in all the tracks as a single track for interpolation"
+                f"Found {len(gpx_tracks)} tracks in the {self.file_format} file {self.gpx_path}. Will merge points in all the tracks as a single track for interpolation"
             )
 
         all_points: list[geo.Point] = sum(gpx_tracks, [])
         gpx_points = [p for p in all_points if _GPS_EPOCH <= p.time]
         if len(gpx_points) < len(all_points):
             LOG.warning(
-                "Ignoring %d of the %d points in the GPX file %s, which are timestamped before 1980",
+                "Ignoring %d of the %d points in the %s file %s, which are timestamped before 1980",
                 len(all_points) - len(gpx_points),
                 len(all_points),
+                self.file_format,
                 self.gpx_path,
             )
 
@@ -250,6 +249,18 @@ def _read_gpx_points(self) -> list[geo.Point]:
         gpx_points.sort(key=lambda p: p.time)
         return gpx_points
 
+    def _parse_tracks(self) -> list[list[geo.Point]]:
+        """
+        Parse the tracks of the GPX file, each a list of points with Unix times
+        """
+        try:
+            return parse_gpx(self.gpx_path)
+        except (OSError, UnicodeError, gpxpy.gpx.GPXException) as ex:
+            raise exceptions.MapillaryInvalidGPXError(
+                f"{self.gpx_path.name} is not a valid GPX file. "
+                "Replace it with a valid GPX file."
+            ) from ex
+
     def _clip_to_video(self, gpx_points: list[geo.Point]) -> list[geo.Point]:
         """
         Clip a GPX track to the video when a clock gives its pace along the
@@ -389,7 +400,7 @@ def _check_coverage(
             gpx_unix_times[0] if clock is None else clock.to_unix_time(0)
         ) + offset_time
 
-        gpx_track = f"The GPX track in {self.gpx_path} ({_isoformat(gpx_unix_times[0])} to {_isoformat(gpx_unix_times[1])})"
+        gpx_track = f"The {self.file_format} track in {self.gpx_path} ({_isoformat(gpx_unix_times[0])} to {_isoformat(gpx_unix_times[1])})"
         if clock is None:
             video = f"the video {self.video_path}"
         else:
@@ -420,14 +431,14 @@ def _check_coverage(
             advice = "Check --interpolation_offset_time"
         elif offset_time:
             advice = (
-                "Check --interpolation_offset_time, that the GPX file belongs to "
-                "this video, then the camera clock and the time zone of the GPX "
+                f"Check --interpolation_offset_time, that the {self.file_format} file belongs to "
+                f"this video, then the camera clock and the time zone of the {self.file_format} "
                 "timestamps"
             )
         else:
             advice = (
-                "Check that the GPX file belongs to this video, then the camera "
-                "clock and the time zone of the GPX timestamps"
+                f"Check that the {self.file_format} file belongs to this video, then the camera "
+                f"clock and the time zone of the {self.file_format} timestamps"
             )
         message = (
             f"{gpx_track} misses {video} by {_format_duration(gap)}{offset_clause}. "
@@ -468,7 +479,7 @@ def _check_assumed_start(
 
         if video_span < excess:
             raise exceptions.MapillaryGPXVideoStartUnknownError(
-                f"The GPX track in {self.gpx_path} lasts {_format_duration(gpx_duration)} "
+                f"The {self.file_format} track in {self.gpx_path} lasts {_format_duration(gpx_duration)} "
                 f"and {self._describe_video_span(video_span, ratio)}, so the video "
                 "could start anywhere along the track. Specify --video_start_time, or "
                 "--interpolation_use_gpx_start_time if the track was started with the "
@@ -477,11 +488,12 @@ def _check_assumed_start(
 
         if _LOGGER_TOLERANCE_SECONDS < excess:
             LOG.warning(
-                "Assuming that the GPX track in %s starts with the video %s, although "
+                "Assuming that the %s track in %s starts with the video %s, although "
                 "the track lasts %s longer. If it was started first, the frames are "
                 "placed up to that much too early along it. Specify --video_start_time "
                 "to place them exactly, or --interpolation_use_gpx_start_time if the "
                 "track was started with the video",
+                self.file_format,
                 self.gpx_path,
                 self.video_path,
                 _format_duration(excess),
@@ -505,7 +517,7 @@ def _check_given_start(
             return
 
         message = (
-            f"The GPX track in {self.gpx_path} lasts {_format_duration(gpx_duration)} "
+            f"The {self.file_format} track in {self.gpx_path} lasts {_format_duration(gpx_duration)} "
             f"and {self._describe_video_span(video_span, ratio)}, so it runs on long "
             "after the video ends. "
         )
@@ -595,11 +607,11 @@ def _warn_of_a_long_track(self, gpx_points: T.Sequence[geo.Point]) -> None:
         gpx_duration = max(p.time for p in gpx_points) - min(p.time for p in gpx_points)
         if 2 * video_duration < gpx_duration:
             LOG.warning(
-                f"The GPX track in {self.gpx_path} lasts {_format_duration(gpx_duration)} "
+                f"The {self.file_format} track in {self.gpx_path} lasts {_format_duration(gpx_duration)} "
                 f"and the video {self.video_path} only {_format_duration(video_duration)}, "
                 "so the video could start anywhere along the track, and is assumed to "
                 "start with it. Unless the video is a timelapse of the whole track, "
-                "trim the GPX to the video, or sample the video with video_process "
+                f"trim the {self.file_format} to the video, or sample the video with video_process "
                 "and --video_start_time"
             )
 

mapillary_tools/geotag/video_extractors/nmea.py

@@ -0,0 +1,43 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+#
+# This source code is licensed under the BSD license found in the
+# LICENSE file in the root directory of this source tree.
+
+from __future__ import annotations
+
+import sys
+
+if sys.version_info >= (3, 12):
+    from typing import override
+else:
+    from typing_extensions import override
+
+from ... import exceptions, geo
+from ..geotag_images_from_nmea_file import get_lat_lon_time_from_nmea
+from ..utils import track_point
+from .gpx import GPXVideoExtractor
+
+
+class NMEAVideoExtractor(GPXVideoExtractor):
+    """
+    Geotag a video from the track in an NMEA file, placed on the video as a
+    GPX track is
+    """
+
+    file_format = "NMEA"
+
+    @override
+    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) or a
+        # malformed sentence (pynmea2.ParseError); TypeError: a sentence
+        # without its time or date; UnboundLocalError: GGA sentences, and no
+        # RMC sentence in the file to give their date
+        except (OSError, ValueError, TypeError, UnboundLocalError) as ex:
+            raise exceptions.MapillaryInvalidGPXError(
+                f"{self.gpx_path.name} is not a valid NMEA file. "
+                "Replace it with a valid NMEA file."
+            ) from ex
+
+        return [[track_point(p.time, p.lat, p.lon, p.alt) for p in points]]