Apply the interpolation options to videos geotagged from a GPX

README.md

@@ -456,9 +456,20 @@ VIDEO_DIR/
 
 ### Limitations of `--video_geotag_source`
 
-**External geolocation sources will be aligned with the start of video, there is
-currently no way of applying offsets or scaling the time.** This means, for instance, that GPX tracks must begin precisely
-at the instant of the video, and that timelapse videos are supported only for sources `camm, gopro, blackvue`.
+**A GPX track is synced by the GPS clock of the video, if the video has GPS of its own, and otherwise aligned with
+the start of the video.** A timelapse video is supported for a GPX only if it has GPS of its own.
+
+As with images, `--interpolation_offset_time N` adds N seconds to the time of the video to find its positions along
+the GPX track, and `--interpolation_use_gpx_start_time` aligns the start of the video with the start of the track, even
+if the video has GPS:
+
+```sh
+# Start recording the video 2.5 seconds after starting the GPS logger
+mapillary_tools process VIDEO_DIR/ \
+    --video_geotag_source gpx \
+    --interpolation_use_gpx_start_time \
+    --interpolation_offset_time 2.5
+```
 
 
 ## Authenticate

mapillary_tools/commands/process.py

@@ -149,15 +149,15 @@ def add_basic_arguments(self, parser: argparse.ArgumentParser):
         )
         group_geotagging.add_argument(
             "--interpolation_use_gpx_start_time",
-            help=f"If supplied, the first image will use the first GPX point time for interpolation, which means the image location will be interpolated to the first GPX point too. Only works for geotagging from {', '.join(geotag_gpx_based_sources)}.",
+            help=f"If supplied, the first image, or the start of a video, will use the first GPX point time for interpolation, which means the image location will be interpolated to the first GPX point too. Only works for geotagging from {', '.join(geotag_gpx_based_sources)}.",
             action="store_true",
             required=False,
         )
         group_geotagging.add_argument(
             "--interpolation_offset_time",
             default=0.0,
             type=float,
-            help=f"Time offset, in seconds, that will be added for GPX interpolation, which affects image locations. Note that it is applied after --interpolation_use_gpx_start_time. Only works for geotagging from {', '.join(geotag_gpx_based_sources)}. [default %(default)s]",
+            help=f"Time offset, in seconds, that will be added for GPX interpolation, which affects image and video locations. Note that it is applied after --interpolation_use_gpx_start_time. Only works for geotagging from {', '.join(geotag_gpx_based_sources)}. [default %(default)s]",
             required=False,
         )
         group_geotagging.add_argument(

mapillary_tools/geotag/factory.py

@@ -463,6 +463,7 @@ def _build_video_geotag(option: SourceOption) -> base.GeotagVideosFromGeneric |
             source_path=option.source_path,
             num_processes=option.num_processes,
             video_sync=option.video_sync,
+            interpolation=option.interpolation,
         )
 
     elif option.source is SourceType.NMEA:

mapillary_tools/geotag/geotag_videos_from_gpx.py

@@ -30,12 +30,14 @@ def __init__(
         source_path: options.SourcePathOption | None = None,
         num_processes: int | None = None,
         video_sync: options.VideoSyncOption | None = None,
+        interpolation: options.InterpolationOption | 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
+        self.interpolation = interpolation
 
     @override
     def _generate_video_extractors(
@@ -47,7 +49,10 @@ def _generate_video_extractors(
             if source_path.is_file():
                 results.append(
                     GPXVideoExtractor(
-                        video_path, source_path, video_sync=self.video_sync
+                        video_path,
+                        source_path,
+                        video_sync=self.video_sync,
+                        interpolation=self.interpolation,
                     )
                 )
             else:

mapillary_tools/geotag/options.py

@@ -151,21 +151,14 @@ 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.
+    them. The interpolation options, which process applies to videos as well,
+    are in SourceOption.interpolation.
     """
 
     # 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",

mapillary_tools/geotag/video_extractors/gpx.py

@@ -98,12 +98,14 @@ def __init__(
         gpx_path: Path,
         sync_mode: SyncMode = SyncMode.SYNC,
         video_sync: options.VideoSyncOption | None = None,
+        interpolation: options.InterpolationOption | None = None,
     ):
         self.video_path = video_path
         self.gpx_path = gpx_path
         self.sync_mode = sync_mode
         self.video_sync = video_sync
-        if video_sync is not None and video_sync.use_gpx_start_time:
+        self.interpolation = interpolation
+        if interpolation is not None and interpolation.use_gpx_start_time:
             self.sync_mode = SyncMode.REBASE
 
     @override
@@ -228,23 +230,29 @@ def _apply_offset_time(
         self, gpx_points: T.Sequence[geo.Point], rate: float = 1.0
     ) -> None:
         """
-        Shift the track back by --interpolation_offset_time, which the
-        geotagging stage adds to the capture time of a frame to find its
-        position along the GPX. The offset is in GPS seconds, of which a
-        timelapse packs **rate** into each second of video.
-
-        The absolute times shift with the track, so that a frame is stamped
-        with its capture time, and the geotagging stage, adding the offset back,
+        Shift the track back by --interpolation_offset_time, which is added to
+        the capture time of a frame to find its position along the GPX. The
+        offset is in GPS seconds, of which a timelapse packs **rate** into each
+        second of video.
+
+        The absolute times stay those of the GPX, just as an image geotagged
+        with the offset takes the time of its position along the GPX. When
+        sampling, they shift with the track, so that a frame is stamped with
+        its capture time, and the geotagging stage, adding the offset back,
         finds the position the frame was picked by.
         """
-        if self.video_sync is None or not self.video_sync.offset_time:
+        if self.interpolation is None or not self.interpolation.offset_time:
             return
 
-        offset_time = self.video_sync.offset_time
+        offset_time = self.interpolation.offset_time
         for p in gpx_points:
             p.time -= offset_time / rate
             # parse_gpx() returns CAMMGPSPoint, with epoch_time 0 for no time
-            if isinstance(p, telemetry.CAMMGPSPoint) and 0 < p.epoch_time:
+            if (
+                self.video_sync is not None
+                and isinstance(p, telemetry.CAMMGPSPoint)
+                and 0 < p.epoch_time
+            ):
                 p.epoch_time -= offset_time
 
     def _check_coverage(

mapillary_tools/sample_video.py

@@ -29,7 +29,12 @@
 )
 from .exif_write import ExifEdit
 from .geotag import factory
-from .geotag.options import SourceOption, SourceType, VideoSyncOption
+from .geotag.options import (
+    InterpolationOption,
+    SourceOption,
+    SourceType,
+    VideoSyncOption,
+)
 from .geotag.video_extractors.native import NativeVideoExtractor
 from .mp4 import mp4_sample_parser
 from .serializer.description import build_capture_time, parse_capture_time
@@ -70,6 +75,7 @@ def _parse_geotag_options(
     video_geotag_source: list[str] | None,
     video_import_path: Path,
     video_sync: VideoSyncOption | None = None,
+    interpolation: InterpolationOption | None = None,
 ) -> list[SourceOption]:
     """
     Resolve which sources distance sampling may take its positions from.
@@ -80,15 +86,18 @@ def _parse_geotag_options(
     cannot see every field the noise filter rejects on, so routing sampling
     through it would accept tracks the native parser refuses.
 
-    Every option carries video_sync, for a GPX source to place its track where
-    the geotagging stage will place the frames.
+    Every option carries video_sync and the interpolation options, for a GPX
+    source to place its track where the geotagging stage will place the frames.
+    As there, the interpolation options of a source override the global ones.
     """
     options = _parse_video_sources(
         geotag_source, geotag_source_path, video_geotag_source, video_import_path
     )
 
     for option in options:
         option.video_sync = video_sync
+        if option.interpolation is None:
+            option.interpolation = interpolation
 
     return options
 
@@ -173,8 +182,6 @@ def sample_video(
         start_time=(
             video_start_time_dt.timestamp() if video_start_time_dt is not None else None
         ),
-        offset_time=interpolation_offset_time,
-        use_gpx_start_time=interpolation_use_gpx_start_time,
     )
 
     geotag_options = _parse_geotag_options(
@@ -183,6 +190,10 @@ def sample_video(
         video_geotag_source,
         video_import_path,
         video_sync,
+        InterpolationOption(
+            offset_time=interpolation_offset_time,
+            use_gpx_start_time=interpolation_use_gpx_start_time,
+        ),
     )
 
     if rerun:

tests/integration/test_process.py

@@ -694,6 +694,56 @@ def test_process_video_geotag_source_with_gpx_outside_video(
     }
 
 
+def test_process_video_from_gpx_with_interpolation_options(
+    setup_data: py.path.local,
+):
+    """
+    The interpolation options line up a GPX with a video, as they do with
+    images. They were ignored for videos.
+    """
+    video_path = setup_data.join("gopro_data").join("max-360mode.mp4")
+    # 2.59 seconds before the GPS clock of the video starts
+    start = datetime.datetime(2019, 11, 18, 23, 44, 40, tzinfo=datetime.timezone.utc)
+    points = "".join(
+        f'<trkpt lat="{37.0 + i * 0.0001:.7f}" lon="-122.0">'
+        f"<time>{(start + datetime.timedelta(seconds=i)).isoformat()}</time></trkpt>"
+        for i in range(21)
+    )
+    gpx_file = setup_data.join("track.gpx")
+    gpx_file.write(
+        '<?xml version="1.0"?>\n'
+        '<gpx version="1.1" creator="test" xmlns="http://www.topografix.com/GPX/1/1">'
+        f"<trk><trkseg>{points}</trkseg></trk></gpx>\n"
+    )
+
+    def first_point(*options: str):
+        [desc] = run_process_for_descs(
+            [
+                *["--geotag_source", "gpx"],
+                *["--geotag_source_path", str(gpx_file)],
+                *options,
+                str(video_path),
+            ]
+        )
+        return desc["MAPGPSTrack"][0]
+
+    synced = first_point()
+    assert synced[0] == pytest.approx(-2590, abs=1)
+
+    assert first_point("--interpolation_use_gpx_start_time")[0] == 0
+    assert (
+        first_point(
+            "--interpolation_use_gpx_start_time", "--interpolation_offset_time=2.5"
+        )[0]
+        == -2500
+    )
+
+    shifted = first_point("--interpolation_offset_time=2")
+    assert shifted[0] == pytest.approx(synced[0] - 2000, abs=1)
+    # the video takes the times of its positions along the GPX, as images do
+    assert shifted[5] == synced[5] == start.timestamp()
+
+
 def test_process_geotag_with_gpx_pattern_not_found(setup_data: py.path.local):
     video_path = setup_data.join("gopro_data").join("max-360mode.mp4")
 

tests/unit/test_gpx_video_coverage.py

@@ -28,6 +28,7 @@
 from mapillary_tools.camm import camm_builder
 from mapillary_tools.geotag import factory
 from mapillary_tools.geotag.options import (
+    InterpolationOption,
     SourceOption,
     SourcePathOption,
     SourceType,
@@ -164,15 +165,17 @@ def _extract(
         gpx_seconds: float,
         video_track: T.Sequence[geo.Point] | None = None,
         duration: float = VIDEO_DURATION,
-        video_sync: VideoSyncOption | None = None,
+        interpolation: InterpolationOption | None = None,
     ) -> types.VideoMetadata:
         video_path = _write_camm_mp4(
             tmp_path / "video.mp4",
             _video_track() if video_track is None else video_track,
             duration,
         )
         gpx_path = _write_gpx(tmp_path / "track.gpx", gpx_start, gpx_seconds)
-        return GPXVideoExtractor(video_path, gpx_path, video_sync=video_sync).extract()
+        return GPXVideoExtractor(
+            video_path, gpx_path, interpolation=interpolation
+        ).extract()
 
     @pytest.mark.parametrize(
         "start, seconds",
@@ -291,15 +294,15 @@ def test_epoch_mixup_raises(self, tmp_path):
 
     def test_the_interpolation_offset_is_part_of_the_placement(self, tmp_path, caplog):
         """
-        The geotagging stage looks the frames up 5s later along the track, so
-        a track that covers the video exactly leaves its last 5s uncovered.
+        The frames are looked up 5s later along the track, so a track that
+        covers the video exactly leaves its last 5s uncovered.
         """
         with caplog.at_level(logging.WARNING):
             self._extract(
                 tmp_path,
                 A_UNIX_TIME,
                 VIDEO_DURATION,
-                video_sync=VideoSyncOption(offset_time=5),
+                interpolation=InterpolationOption(offset_time=5),
             )
         [warning] = _warnings(caplog)
         assert "5 seconds of the video fall outside the track" in warning