Honour --geotag_source in distance-based sampling

mapillary_tools/exceptions.py

@@ -89,6 +89,13 @@ class MapillaryStationaryVideoError(MapillaryDescriptionError):
     pass
 
 
+class MapillaryGPXVideoStartUnknownError(MapillaryDescriptionError):
+    """
+    Nothing says where along a GPX track a video without GPS starts, and the
+    track is too long to assume that it starts with the video.
+    """
+
+
 class MapillaryOutsideGPXTrackError(MapillaryDescriptionError):
     def __init__(
         self, message: str, image_time: str, gpx_start_time: str, gpx_end_time: str

mapillary_tools/geotag/factory.py

@@ -22,11 +22,19 @@
     geotag_videos_from_gpx,
     geotag_videos_from_video,
 )
-from .options import InterpolationOption, SOURCE_TYPE_ALIAS, SourceOption, SourceType
+from .options import (
+    InterpolationOption,
+    SOURCE_TYPE_ALIAS,
+    SourceOption,
+    SourcePathOption,
+    SourceType,
+)
 
 
 LOG = logging.getLogger(__name__)
 
+_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)
@@ -63,6 +71,73 @@ def parse_source_option(source: str) -> list[SourceOption]:
     return [SourceOption(SourceType(SOURCE_TYPE_ALIAS.get(s, s))) for s in sources]
 
 
+def parse_source_options(
+    geotag_source: T.Sequence[str],
+    video_geotag_source: T.Sequence[str],
+    geotag_source_path: Path | None,
+) -> list[SourceOption]:
+    """
+    Turn the raw --geotag_source / --video_geotag_source / --geotag_source_path
+    arguments into the option list that process() consumes.
+    """
+    parsed_options: list[SourceOption] = []
+
+    if video_geotag_source and geotag_source:
+        LOG.warning(
+            "Video source options will be processed BEFORE the generic source options"
+        )
+
+    for s in video_geotag_source:
+        for video_option in parse_source_option(s):
+            video_option.filetypes = types.combine_filetype_filters(
+                video_option.filetypes, {types.FileType.VIDEO}
+            )
+            parsed_options.append(video_option)
+
+    for s in geotag_source:
+        parsed_options.extend(parse_source_option(s))
+
+    if geotag_source_path is not None:
+        for parsed_option in parsed_options:
+            if parsed_option.source_path is None:
+                parsed_option.source_path = SourcePathOption(
+                    source_path=Path(geotag_source_path)
+                )
+            else:
+                source_path_option = parsed_option.source_path
+                if source_path_option.source_path is None:
+                    source_path_option.source_path = Path(geotag_source_path)
+                else:
+                    LOG.warning(
+                        "The option --geotag_source_path is ignored for source %s",
+                        parsed_option,
+                    )
+
+    return parsed_options
+
+
+def can_geotag_videos(option: SourceOption) -> bool:
+    """
+    Whether process() can read positions for videos from the source: it has a
+    video extractor, and its filetypes let videos through. process() returns no
+    metadata at all for a video that no source can read.
+
+    Examples:
+        >>> can_geotag_videos(SourceOption(SourceType.GPX))
+        True
+        >>> can_geotag_videos(SourceOption(SourceType.EXIF))
+        False
+        >>> can_geotag_videos(
+        ...     SourceOption(SourceType.NATIVE, filetypes={types.FileType.IMAGE})
+        ... )
+        False
+    """
+    if option.filetypes is not None and not option.filetypes & _ALL_VIDEO_TYPES:
+        return False
+
+    return _build_video_geotag(option) is not None
+
+
 def process(
     # Collection: ABC for sized iterable container classes
     paths: T.Iterable[Path],
@@ -179,14 +254,12 @@ def _filter_images_and_videos(
     image_paths = []
     video_paths = []
 
-    ALL_VIDEO_TYPES = {types.FileType.VIDEO, *types.NATIVE_VIDEO_FILETYPES}
-
     if filetypes is None:
         include_images = True
         include_videos = True
     else:
         include_images = types.FileType.IMAGE in filetypes
-        include_videos = bool(filetypes & ALL_VIDEO_TYPES)
+        include_videos = bool(filetypes & _ALL_VIDEO_TYPES)
 
     for path in paths:
         if utils.is_image_file(path):
@@ -332,7 +405,9 @@ def _build_video_geotag(option: SourceOption) -> base.GeotagVideosFromGeneric |
 
     elif option.source is SourceType.GPX:
         return geotag_videos_from_gpx.GeotagVideosFromGPX(
-            source_path=option.source_path, num_processes=option.num_processes
+            source_path=option.source_path,
+            num_processes=option.num_processes,
+            video_sync=option.video_sync,
         )
 
     elif option.source is SourceType.NMEA:

mapillary_tools/geotag/geotag_videos_from_gpx.py

@@ -29,11 +29,13 @@ def __init__(
         self,
         source_path: options.SourcePathOption | None = None,
         num_processes: int | None = None,
+        video_sync: options.VideoSyncOption | None = None,
     ):
         super().__init__(num_processes=num_processes)
         if source_path is None:
             source_path = options.SourcePathOption(pattern="%g.gpx")
         self.source_path = source_path
+        self.video_sync = video_sync
 
     @override
     def _generate_video_extractors(
@@ -43,7 +45,11 @@ def _generate_video_extractors(
         for video_path in video_paths:
             source_path = self.source_path.resolve(video_path)
             if source_path.is_file():
-                results.append(GPXVideoExtractor(video_path, source_path))
+                results.append(
+                    GPXVideoExtractor(
+                        video_path, source_path, video_sync=self.video_sync
+                    )
+                )
             else:
                 results.append(
                     types.describe_error_metadata(

mapillary_tools/geotag/options.py

@@ -51,6 +51,9 @@ class SourceOption:
 
     interpolation: InterpolationOption | None = None
 
+    # Set by sample_video only, and not in the JSON schema
+    video_sync: VideoSyncOption | None = None
+
     @classmethod
     def from_dict(cls, data: dict[str, T.Any]) -> SourceOption:
         validate_option(data)
@@ -143,6 +146,27 @@ class InterpolationOption:
     use_gpx_start_time: bool = False
 
 
+@dataclasses.dataclass
+class VideoSyncOption:
+    """
+    How to line up a GPX track with a video that is being sampled, so that its
+    frames are picked where the geotagging stage of video_process will place
+    them. process geotags videos without it, and places a GPX as it always has.
+    """
+
+    # Unix time of the first frame of the video (--video_start_time), to place
+    # a GPX by when the video has no GPS of its own to sync it with
+    start_time: float | None = None
+
+    # --interpolation_offset_time: seconds the geotagging stage adds to the
+    # capture time of a frame to find its position along the GPX
+    offset_time: float = 0.0
+
+    # --interpolation_use_gpx_start_time: the geotagging stage puts the first
+    # frame at the start of the GPX
+    use_gpx_start_time: bool = False
+
+
 SourceOptionSchema = {
     "type": "object",
     "properties": {