Let a GPX override a video whose own GPS is unusable

mapillary_tools/geotag/factory.py

@@ -27,6 +27,10 @@
 
 LOG = logging.getLogger(__name__)
 
+# Sources that read a GPS track from outside the video file, as opposed to
+# re-reading the telemetry embedded in it
+EXTERNAL_GPS_SOURCES = frozenset({SourceType.GPX, SourceType.NMEA})
+
 
 def parse_source_option(source: str) -> list[SourceOption]:
     """
@@ -68,10 +72,13 @@ def process(
 
     final_metadatas: list[types.MetadataOrError] = []
 
+    # Indexable, so each step can see which sources are still to come
+    option_list = list(options)
+
     # Paths (image path or video path) that will be sent to the next geotag process
     reprocessable_paths = set(paths)
 
-    for idx, option in enumerate(options):
+    for idx, option in enumerate(option_list):
         if LOG.isEnabledFor(logging.DEBUG):
             LOG.info(
                 f"==> Processing {len(reprocessable_paths)} files with source {option}..."
@@ -101,10 +108,10 @@ def process(
         else:
             video_metadata_or_errors = []
 
-        more_option = idx < len(options) - 1
+        remaining_options = option_list[idx + 1 :]
 
         for metadata in image_metadata_or_errors + video_metadata_or_errors:
-            if more_option and _is_reprocessable(metadata):
+            if remaining_options and _is_reprocessable(metadata, remaining_options):
                 # Leave what it is for the next geotag process
                 pass
             else:
@@ -118,18 +125,40 @@ def process(
     return final_metadatas
 
 
-def _is_reprocessable(metadata: types.MetadataOrError) -> bool:
-    if isinstance(metadata, types.ErrorMetadata):
-        if isinstance(
-            metadata.error,
-            (
-                exceptions.MapillaryGeoTaggingError,
-                exceptions.MapillaryVideoGPSNotFoundError,
-                exceptions.MapillaryExiftoolNotFoundError,
-                exceptions.MapillaryExifToolXMLNotFoundError,
-            ),
-        ):
-            return True
+def _is_reprocessable(
+    metadata: types.MetadataOrError,
+    remaining_options: T.Sequence[SourceOption] = (),
+) -> bool:
+    if not isinstance(metadata, types.ErrorMetadata):
+        return False
+
+    if isinstance(
+        metadata.error,
+        (
+            exceptions.MapillaryGeoTaggingError,
+            exceptions.MapillaryVideoGPSNotFoundError,
+            exceptions.MapillaryExiftoolNotFoundError,
+            exceptions.MapillaryExifToolXMLNotFoundError,
+        ),
+    ):
+        return True
+
+    # Unusable GPS is a verdict on the data, not on the reader that happened to
+    # report it, so only a source that supplies GPS from *outside* the video can
+    # rescue the file. Handing it to another reader of the same embedded
+    # telemetry just asks a second opinion of the same bad data, and the readers
+    # do not agree: exiftool reports no DoP at all, so it silently accepts a
+    # track that the native parser rejects as noise.
+    if isinstance(
+        metadata.error,
+        (
+            exceptions.MapillaryGPXEmptyError,
+            exceptions.MapillaryGPSNoiseError,
+        ),
+    ):
+        return any(
+            option.source in EXTERNAL_GPS_SOURCES for option in remaining_options
+        )
 
     return False
 

mapillary_tools/geotag/video_extractors/gpx.py

@@ -59,11 +59,20 @@ def extract(self) -> types.VideoMetadata:
 
         gpx_points: T.Sequence[geo.Point] = sum(gpx_tracks, [])
 
-        native_extractor = NativeVideoExtractor(self.video_path)
+        # The GPX track replaces the video's own GPS, so the native extractor is
+        # only a source of make/model and of a clock to sync against. Keep noisy
+        # points: they are never published, and their timestamps still sync.
+        native_extractor = NativeVideoExtractor(
+            self.video_path, filter_noisy_points=False
+        )
 
         try:
             native_video_metadata = native_extractor.extract()
-        except exceptions.MapillaryVideoGPSNotFoundError as ex:
+        except (
+            exceptions.MapillaryVideoGPSNotFoundError,
+            exceptions.MapillaryGPXEmptyError,
+            exceptions.MapillaryGPSNoiseError,
+        ) as ex:
             if self.sync_mode is SyncMode.STRICT_SYNC:
                 raise ex
             self._rebase_times(gpx_points)

mapillary_tools/geotag/video_extractors/native.py

@@ -22,6 +22,15 @@
 
 
 class GoProVideoExtractor(BaseVideoExtractor):
+    def __init__(self, video_path: Path, filter_noisy_points: bool = True):
+        super().__init__(video_path)
+        # The noise filter is a quality gate on the track we are about to
+        # publish. Callers that only need the video's make/model and its GPS
+        # clock (e.g. geotagging from a GPX file) pass False: discarding noisy
+        # points there would throw away a usable sync anchor and, if every
+        # point is dropped, fail the whole video over GPS we are not using.
+        self.filter_noisy_points = filter_noisy_points
+
     @override
     def extract(self) -> types.VideoMetadata:
         with self.video_path.open("rb") as fp:
@@ -37,11 +46,13 @@ def extract(self) -> types.VideoMetadata:
         if not gps_points:
             raise exceptions.MapillaryGPXEmptyError("Empty GPS data found")
 
-        gps_points = T.cast(
-            T.List[telemetry.GPSPoint], gpmf_gps_filter.remove_noisy_points(gps_points)
-        )
-        if not gps_points:
-            raise exceptions.MapillaryGPSNoiseError("GPS is too noisy")
+        if self.filter_noisy_points:
+            gps_points = T.cast(
+                T.List[telemetry.GPSPoint],
+                gpmf_gps_filter.remove_noisy_points(gps_points),
+            )
+            if not gps_points:
+                raise exceptions.MapillaryGPSNoiseError("GPS is too noisy")
 
         video_metadata = types.VideoMetadata(
             filename=self.video_path,
@@ -106,17 +117,25 @@ def extract(self) -> types.VideoMetadata:
 
 
 class NativeVideoExtractor(BaseVideoExtractor):
-    def __init__(self, video_path: Path, filetypes: set[types.FileType] | None = None):
+    def __init__(
+        self,
+        video_path: Path,
+        filetypes: set[types.FileType] | None = None,
+        filter_noisy_points: bool = True,
+    ):
         super().__init__(video_path)
         self.filetypes = filetypes
+        self.filter_noisy_points = filter_noisy_points
 
     @override
     def extract(self) -> types.VideoMetadata:
         ft = self.filetypes
         extractor: BaseVideoExtractor
 
         if ft is None or types.FileType.VIDEO in ft or types.FileType.GOPRO in ft:
-            extractor = GoProVideoExtractor(self.video_path)
+            extractor = GoProVideoExtractor(
+                self.video_path, filter_noisy_points=self.filter_noisy_points
+            )
             try:
                 return extractor.extract()
             except simple_mp4_parser.BoxNotFoundError as ex: