From add6d99ff0b2a3b21efaf5ad628c029412fd7c91 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Tue, 20 Oct 2026 13:29:23 +0200 Subject: [PATCH] Apply the interpolation options to videos geotagged from a GPX process ignored --interpolation_offset_time and --interpolation_use_gpx_start_time for videos. They reached the image geotaggers through SourceOption.interpolation, but the GPX video geotagger read them only from VideoSyncOption, which only distance sampling sets. The track uploaded with a video stayed where the GPS clock of the video, or the start of the video, put it, whatever the options said. GPXVideoExtractor now reads them from SourceOption.interpolation, which process sets on every source. Sampling sets it too, unless a source has options of its own, as process does. VideoSyncOption keeps only the start time, and with it the start-time check, which stays out of process: process has no --video_start_time to offer. The offset moves the track along the video. The absolute times of its points stay those of the GPX, just as an image geotagged with the offset takes the time of its position along the GPX. Only sampling shifts them as well, for the geotagging stage to add the offset back. Without the options, process gives the same tracks as before, and video_process the same frames and descriptions, with or without them. Fixes #827 --- README.md | 17 ++- mapillary_tools/commands/process.py | 4 +- mapillary_tools/geotag/factory.py | 1 + .../geotag/geotag_videos_from_gpx.py | 7 +- mapillary_tools/geotag/options.py | 11 +- .../geotag/video_extractors/gpx.py | 30 +++-- mapillary_tools/sample_video.py | 21 +++- tests/integration/test_process.py | 50 ++++++++ tests/unit/test_gpx_video_coverage.py | 13 +- tests/unit/test_gpx_video_sync.py | 112 ++++++++++++++++-- tests/unit/test_sample_video.py | 21 +++- 11 files changed, 236 insertions(+), 51 deletions(-) diff --git a/README.md b/README.md index 655c1c91..11d765b2 100644 --- a/README.md +++ b/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 diff --git a/mapillary_tools/commands/process.py b/mapillary_tools/commands/process.py index fc205c5d..6ac77289 100644 --- a/mapillary_tools/commands/process.py +++ b/mapillary_tools/commands/process.py @@ -149,7 +149,7 @@ 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, ) @@ -157,7 +157,7 @@ def add_basic_arguments(self, parser: argparse.ArgumentParser): "--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( diff --git a/mapillary_tools/geotag/factory.py b/mapillary_tools/geotag/factory.py index 9cdd8479..dc9d21e2 100644 --- a/mapillary_tools/geotag/factory.py +++ b/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: diff --git a/mapillary_tools/geotag/geotag_videos_from_gpx.py b/mapillary_tools/geotag/geotag_videos_from_gpx.py index 05aeecc9..225f29ea 100644 --- a/mapillary_tools/geotag/geotag_videos_from_gpx.py +++ b/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: diff --git a/mapillary_tools/geotag/options.py b/mapillary_tools/geotag/options.py index 72f635f3..e60d4391 100644 --- a/mapillary_tools/geotag/options.py +++ b/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", diff --git a/mapillary_tools/geotag/video_extractors/gpx.py b/mapillary_tools/geotag/video_extractors/gpx.py index 75d4aa16..dd322eab 100644 --- a/mapillary_tools/geotag/video_extractors/gpx.py +++ b/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( diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py index e945d252..7a858413 100644 --- a/mapillary_tools/sample_video.py +++ b/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,8 +86,9 @@ 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 @@ -89,6 +96,8 @@ def _parse_geotag_options( 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: diff --git a/tests/integration/test_process.py b/tests/integration/test_process.py index 5d1a81bb..7a9ceff1 100644 --- a/tests/integration/test_process.py +++ b/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'' + f"" + for i in range(21) + ) + gpx_file = setup_data.join("track.gpx") + gpx_file.write( + '\n' + '' + f"{points}\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") diff --git a/tests/unit/test_gpx_video_coverage.py b/tests/unit/test_gpx_video_coverage.py index 73a6581a..1bf0c2f0 100644 --- a/tests/unit/test_gpx_video_coverage.py +++ b/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,7 +165,7 @@ 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", @@ -172,7 +173,9 @@ def _extract( 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 diff --git a/tests/unit/test_gpx_video_sync.py b/tests/unit/test_gpx_video_sync.py index 619e13e6..442d61c0 100644 --- a/tests/unit/test_gpx_video_sync.py +++ b/tests/unit/test_gpx_video_sync.py @@ -5,8 +5,8 @@ """ How GPXVideoExtractor lines up a GPX track with the video it geotags, when -distance sampling hands it a VideoSyncOption, and that process, which does -not, places a GPX as it always has. +distance sampling hands it a VideoSyncOption, and when process, which does not, +geotags the video for upload. """ from __future__ import annotations @@ -19,6 +19,7 @@ from mapillary_tools import exceptions, types from mapillary_tools.geotag import factory from mapillary_tools.geotag.options import ( + InterpolationOption, SourceOption, SourcePathOption, SourceType, @@ -57,8 +58,15 @@ def _write_gpx(path: Path, seconds: int, start=GPX_START) -> Path: return path -def _extract(video: Path, gpx: Path, video_sync: VideoSyncOption | None): - return GPXVideoExtractor(video, gpx, video_sync=video_sync).extract() +def _extract( + video: Path, + gpx: Path, + video_sync: VideoSyncOption | None, + interpolation: InterpolationOption | None = None, +): + return GPXVideoExtractor( + video, gpx, video_sync=video_sync, interpolation=interpolation + ).extract() class TestVideoWithoutGPS: @@ -181,7 +189,8 @@ def test_the_offset_shifts_the_track_and_its_times(self, tmp_path): metadata = _extract( NO_GPS_VIDEO, gpx, - VideoSyncOption(start_time=video_start.timestamp(), offset_time=2), + VideoSyncOption(start_time=video_start.timestamp()), + InterpolationOption(offset_time=2), ) # the point recorded at start + 2 is where the first frame is @@ -198,7 +207,9 @@ def test_the_offset_applies_to_a_track_synced_by_the_video_gps(self, tmp_path): ) synced = _extract(GPS_VIDEO, gpx, VideoSyncOption()) - shifted = _extract(GPS_VIDEO, gpx, VideoSyncOption(offset_time=2)) + shifted = _extract( + GPS_VIDEO, gpx, VideoSyncOption(), InterpolationOption(offset_time=2) + ) assert [p.time for p in shifted.points] == pytest.approx( [p.time - 2 for p in synced.points] @@ -215,22 +226,92 @@ def test_use_gpx_start_time_starts_the_track_with_the_video(self, tmp_path): NO_GPS_VIDEO, gpx, VideoSyncOption( - start_time=(GPX_START + datetime.timedelta(seconds=30)).timestamp(), - use_gpx_start_time=True, + start_time=(GPX_START + datetime.timedelta(seconds=30)).timestamp() ), + InterpolationOption(use_gpx_start_time=True), ) assert metadata.points[0].time == 0 - def test_process_applies_neither(self, tmp_path): - """process passes no VideoSyncOption, and ignores both for videos.""" + +class TestProcessInterpolationOptions: + """ + process applies --interpolation_offset_time and + --interpolation_use_gpx_start_time to the track it geotags a video with, as + it does when it geotags images from a GPX. It ignored them for videos. + """ + + # 2.59 seconds before the GPS clock of GPS_VIDEO starts + START = datetime.datetime(2019, 11, 18, 23, 44, 40, tzinfo=datetime.timezone.utc) + + def test_use_gpx_start_time_starts_the_track_with_a_video_with_gps(self, tmp_path): + """Instead of syncing it by the GPS clock of the video.""" + gpx = _write_gpx(tmp_path / "track.gpx", seconds=20, start=self.START) + + synced = _extract(GPS_VIDEO, gpx, None, InterpolationOption()) + started = _extract( + GPS_VIDEO, gpx, None, InterpolationOption(use_gpx_start_time=True) + ) + + assert synced.points[0].time == pytest.approx(-2.59) + assert started.points[0].time == 0 + + def test_the_offset_shifts_the_track_but_not_its_times(self, tmp_path): + """ + The frame at video time t is at the position the GPX recorded at + t + offset, and takes that time, as an image does. + """ + gpx = _write_gpx(tmp_path / "track.gpx", seconds=20, start=self.START) + + synced = _extract(GPS_VIDEO, gpx, None, InterpolationOption()) + shifted = _extract(GPS_VIDEO, gpx, None, InterpolationOption(offset_time=2)) + + assert [p.time for p in shifted.points] == pytest.approx( + [p.time - 2 for p in synced.points] + ) + assert [p.get_unix_time() for p in shifted.points] == [ + p.get_unix_time() for p in synced.points + ] + + def test_both_on_a_video_without_gps(self, tmp_path): + """As the README describes for a GPX started 2.5 seconds before the video.""" gpx = _write_gpx(tmp_path / "track.gpx", seconds=6) - metadata = _extract(NO_GPS_VIDEO, gpx, video_sync=None) + metadata = _extract( + NO_GPS_VIDEO, + gpx, + None, + InterpolationOption(use_gpx_start_time=True, offset_time=2.5), + ) + + assert metadata.points[0].time == pytest.approx(-2.5) + assert metadata.points[0].get_unix_time() == GPX_START.timestamp() + + def test_without_them_the_track_is_placed_as_before(self, tmp_path): + gpx = _write_gpx(tmp_path / "track.gpx", seconds=3600) + + metadata = _extract(NO_GPS_VIDEO, gpx, None, InterpolationOption()) assert metadata.points[0].time == 0 assert metadata.points[0].get_unix_time() == GPX_START.timestamp() + def test_the_factory_hands_them_to_the_gpx_source(self, tmp_path): + gpx = _write_gpx(tmp_path / "track.gpx", seconds=6) + + [metadata] = factory.process( + [NO_GPS_VIDEO], + [ + SourceOption( + SourceType.GPX, + source_path=SourcePathOption(source_path=gpx), + interpolation=InterpolationOption(offset_time=2), + ) + ], + ) + + assert isinstance(metadata, types.VideoMetadata) + assert metadata.points[0].time == pytest.approx(-2) + def _write_gpx_tracks(path: Path, tracks: list[list[str]]) -> Path: """A GPX file with one track per list of times, heading north at 10 m/s.""" @@ -289,7 +370,10 @@ def test_points_before_1980_are_dropped_and_the_rest_sorted( def test_use_gpx_start_time_starts_at_the_earliest_point(self, tmp_path): """Not at the first point in the file, nor at a junk one.""" metadata = _extract( - NO_GPS_VIDEO, self._gpx(tmp_path), VideoSyncOption(use_gpx_start_time=True) + NO_GPS_VIDEO, + self._gpx(tmp_path), + VideoSyncOption(), + InterpolationOption(use_gpx_start_time=True), ) assert metadata.points[0].time == 0 @@ -339,7 +423,9 @@ def test_the_offset_is_in_gps_seconds(self, tmp_path, monkeypatch): gpx = _write_gpx(tmp_path / "track.gpx", seconds=30, start=start) synced = _extract(GPS_VIDEO, gpx, VideoSyncOption()) - shifted = _extract(GPS_VIDEO, gpx, VideoSyncOption(offset_time=2)) + shifted = _extract( + GPS_VIDEO, gpx, VideoSyncOption(), InterpolationOption(offset_time=2) + ) assert [p.time for p in shifted.points] == pytest.approx( [p.time - 0.2 for p in synced.points] diff --git a/tests/unit/test_sample_video.py b/tests/unit/test_sample_video.py index cf4760b7..595d3086 100644 --- a/tests/unit/test_sample_video.py +++ b/tests/unit/test_sample_video.py @@ -25,7 +25,11 @@ sample_video, telemetry, ) -from mapillary_tools.geotag.options import SourceType, VideoSyncOption +from mapillary_tools.geotag.options import ( + InterpolationOption, + SourceType, + VideoSyncOption, +) from mapillary_tools.mp4 import construct_mp4_parser as cparser, mp4_sample_parser from mapillary_tools.serializer import description from mapillary_tools.types import describe_error_metadata, FileType, VideoMetadata @@ -1588,10 +1592,23 @@ def fake_process(paths, options): interpolation_use_gpx_start_time=True, ) - assert seen[0].video_sync == VideoSyncOption( + assert seen[0].interpolation == InterpolationOption( offset_time=2.5, use_gpx_start_time=True ) + def test_the_interpolation_options_of_a_source_win(self): + """As they do when the geotagging stage geotags the frames.""" + options = sample_video._parse_geotag_options( + ['{"source": "gpx", "interpolation_offset_time": 1}'], + Path("/data/track.gpx"), + None, + Path("/data/v.mp4"), + VideoSyncOption(), + InterpolationOption(offset_time=2.5), + ) + + assert options[0].interpolation == InterpolationOption(offset_time=1) + def test_without_it_the_start_time_is_left_unknown(self): """ A video without a GPS clock would get its start time from its creation