Start a GPX track moved before the video at the video's start

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/commands/video_process.py

@@ -6,8 +6,10 @@
 from __future__ import annotations
 
 import logging
+from pathlib import Path
 
-from .. import exceptions
+from .. import constants, exceptions
+from ..sample_video import find_sample_dirs
 from ..types import FileType
 from .process import Command as ProcessCommand
 from .sample_video import Command as SampleCommand
@@ -43,7 +45,39 @@ def run(self, args: dict):
             )
             args[option] = {FileType.IMAGE}
 
+        if args["skip_subfolders"]:
+            _add_sample_dirs(args)
+
         ProcessCommand().run(args)
 
         if sample_error is not None:
             raise sample_error
+
+
+def _add_sample_dirs(args: dict) -> None:
+    """
+    Add the folders that the videos were sampled into to the import paths.
+    --skip_subfolders chose the videos to sample, and hides the subfolders of
+    the import path from processing (and from uploading in
+    video_process_and_upload), which the samples of each video are in
+    """
+    import_path: Path = args["import_path"]
+    sample_dirs = [
+        sample_dir
+        for sample_dir in find_sample_dirs(
+            args["video_import_path"], import_path, skip_subfolders=True
+        )
+        # Not a video that failed to sample
+        if sample_dir.is_dir()
+    ]
+    if not sample_dirs:
+        return
+
+    if args.get("desc_path") is None:
+        # Where process writes it for the import path alone, rather than to
+        # STDOUT, as for several import paths
+        args["desc_path"] = str(
+            import_path.joinpath(constants.IMAGE_DESCRIPTION_FILENAME)
+        )
+
+    args["import_path"] = [import_path, *sample_dirs]

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

@@ -6,6 +6,8 @@
 from __future__ import annotations
 
 import datetime
+import logging
+import re
 from pathlib import Path
 
 import pynmea2
@@ -14,6 +16,13 @@
 from .geotag_images_from_gpx import GeotagImagesFromGPX
 
 
+LOG = logging.getLogger(__name__)
+
+# An RMC or a GGA sentence, from any talker: GP (GPS), GN (several systems),
+# GL (GLONASS), GA (Galileo), ...
+_SENTENCE_RE = re.compile(r"\$[A-Z]{2}(RMC|GGA),")
+
+
 class GeotagImagesFromNMEAFile(GeotagImagesFromGPX):
     def __init__(
         self,
@@ -66,33 +75,69 @@ 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:
+    """
+    Read the fixes of an NMEA file: the time and position of each GGA
+    sentence, on the date of the last RMC sentence before it, or of the first
+    RMC sentence for the GGA sentences before any.
+
+    Receivers log sentences without a fix, such as before they acquire one:
+    RMC sentences that are void or have no date or time, and GGA sentences
+    without a time or a fix, which give their position as 0, 0. They are
+    skipped. A file without the date of any fix gives no points.
+
+    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.
+    """
+    # 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]
 
+    messages = [pynmea2.parse(line) for line in lines if _SENTENCE_RE.search(line)]
+
     # Get initial date
-    for line in lines:
-        if "GPRMC" in line:
-            data = pynmea2.parse(line)
-            date = data.datetime.date()
-            break
+    dates = [_fix_date(m) for m in messages if isinstance(m, pynmea2.RMC)]
+    date = next((d for d in dates if d is not None), None)
+    if date is None:
+        fixes = [m for m in messages if isinstance(m, pynmea2.GGA) and _has_fix(m)]
+        if fixes:
+            LOG.warning(
+                "Ignoring the %d fixes in the NMEA file %s, as no RMC sentence gives their date",
+                len(fixes),
+                nmea_file,
+            )
+        return []
 
     # Parse GPS trace
     points = []
-    for line in lines:
-        if "GPRMC" in line:
-            data = pynmea2.parse(line)
-            date = data.datetime.date()
-
-        if "$GPGGA" in line:
-            data = pynmea2.parse(line)
-            dt = datetime.datetime.combine(date, data.timestamp)
-            lat, lon, alt = data.latitude, data.longitude, data.altitude
-            points.append(
-                geo.Point(
-                    time=geo.as_unix_time(dt), lat=lat, lon=lon, alt=alt, angle=None
-                )
-            )
+    times: set[float] = set()
+    for message in messages:
+        if isinstance(message, pynmea2.RMC):
+            date = _fix_date(message) or date
+
+        elif isinstance(message, pynmea2.GGA) and _has_fix(message):
+            dt = datetime.datetime.combine(date, message.timestamp)
+            time = geo.as_unix_time(dt)
+            if time in times:
+                continue
+            times.add(time)
+            lat, lon, alt = message.latitude, message.longitude, message.altitude
+            points.append(geo.Point(time=time, lat=lat, lon=lon, alt=alt, angle=None))
 
     points.sort()
     return points
+
+
+def _fix_date(message: pynmea2.RMC) -> datetime.date | None:
+    """The date of an RMC sentence, or None if it is void or has no date or time"""
+    if message.status != "A" or message.timestamp is None:
+        return None
+    return message.datestamp
+
+
+def _has_fix(message: pynmea2.GGA) -> bool:
+    """Whether a GGA sentence has a time and a fix (quality 0 is none)"""
+    return message.timestamp is not None and bool(message.gps_qual)

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] = {}