From 484d281c06a859fc5aea5acf169da0aabdc82e57 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Tue, 20 Oct 2026 14:01:22 +0200 Subject: [PATCH 1/5] Stamp interval samples of a timelapse with their capture times Interval sampling stamped each frame with the start time of the video plus its time in the video. A timelapse packs several seconds of capture into each second of video, so its frames were stamped that many times too close together. Geotagged from a GPX by those stamps, they all landed near the start of the track. Geotagged from the video's own track, they moved so far per second that the speed check could reject them: a 30x timelapse of a drive read as over 1,000 km/h. Interval sampling now reads the rate from the video's GPS clock, as GPX sync does, and stamps the frames at it. The start time is mapped back from the first GPS timestamps at the same rate. --video_duration_ratio, which used to apply to every video, now paces only the videos without a GPS clock, so that a ratio given for those leaves the other videos in the batch alone. Each video whose clock overrides a given ratio says so in a warning. The samples are then geotagged from the video's own track (the gopro, camm and blackvue sources) by their time since the first sample, while the track is timed in video seconds. It is now timed in GPS seconds, at the rate of its clock. That also fixes distance samples of a timelapse, which were already stamped with their capture times. --- .../geotag/geotag_images_from_video.py | 21 +++- mapillary_tools/sample_video.py | 94 +++++++++++---- tests/unit/test_geotag_images_from_video.py | 83 +++++++++++++ tests/unit/test_sample_video.py | 111 +++++++++++++++++- 4 files changed, 278 insertions(+), 31 deletions(-) create mode 100644 tests/unit/test_geotag_images_from_video.py diff --git a/mapillary_tools/geotag/geotag_images_from_video.py b/mapillary_tools/geotag/geotag_images_from_video.py index c20d9f1a..b4984d3b 100644 --- a/mapillary_tools/geotag/geotag_images_from_video.py +++ b/mapillary_tools/geotag/geotag_images_from_video.py @@ -5,6 +5,7 @@ from __future__ import annotations +import dataclasses import logging import sys import typing as T @@ -15,10 +16,11 @@ else: from typing_extensions import override -from .. import types, utils +from .. import geo, types, utils from .base import GeotagImagesFromGeneric from .geotag_images_from_gpx import GeotagImagesFromGPX from .geotag_videos_from_video import GeotagVideosFromVideo +from .video_extractors.gpx import GPXVideoExtractor LOG = logging.getLogger(__name__) @@ -74,7 +76,7 @@ def to_description( ) geotag = GeotagImagesFromGPX( - video_metadata.points, + self._on_capture_timeline(video_metadata.points), use_gpx_start_time=False, use_image_start_time=True, offset_time=self.offset_time, @@ -98,6 +100,21 @@ def to_description( return final_image_metadatas + @classmethod + def _on_capture_timeline( + cls, points: T.Sequence[geo.Point] + ) -> T.Sequence[geo.Point]: + """ + Time the video's GPS points in GPS seconds from the start of the video, + like the samples, which are stamped with their capture times. Each + second of a timelapse spans as many as its GPS clock runs faster. + """ + clock = GPXVideoExtractor._read_video_clock(points) + if clock is None or clock.rate == 1.0: + return points + + return [dataclasses.replace(p, time=p.time * clock.rate) for p in points] + class GeotagImageSamplesFromVideo(GeotagImagesFromGeneric): def __init__( diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py index e945d252..9636c6bb 100644 --- a/mapillary_tools/sample_video.py +++ b/mapillary_tools/sample_video.py @@ -30,6 +30,7 @@ from .exif_write import ExifEdit from .geotag import factory from .geotag.options import SourceOption, SourceType, VideoSyncOption +from .geotag.video_extractors.gpx import GPXVideoExtractor from .geotag.video_extractors.native import NativeVideoExtractor from .mp4 import mp4_sample_parser from .serializer.description import build_capture_time, parse_capture_time @@ -380,14 +381,15 @@ def wip_sample_dir(sample_dir: Path) -> Path: def _gps_clock_start_time( - points: T.Sequence[geo.Point], + points: T.Sequence[geo.Point], rate: float = 1.0 ) -> datetime.datetime | None: """ Map the absolute GPS timestamps at the start of a track back to the video's time 0. - Point times are relative to the start of the video, so subtracting one from - its own absolute timestamp gives the wall clock at which the video started. - Timestamps outside the plausible range are skipped. + Point times are relative to the start of the video, so subtracting one, in + GPS seconds, from its own absolute timestamp gives the wall clock at which + the video started. Each second of video spans **rate** GPS seconds, more + than 1 in a timelapse. Timestamps outside the plausible range are skipped. """ max_start_time = time.time() + _MAX_FUTURE_START_TIME_SECONDS @@ -396,7 +398,7 @@ def _gps_clock_start_time( unix_time = point.get_unix_time() if unix_time is None: continue - start_time = unix_time - point.time + start_time = unix_time - point.time * rate # Written as a negated range check so that NaN is skipped too if not (_MIN_PLAUSIBLE_START_TIME <= start_time <= max_start_time): continue @@ -412,14 +414,44 @@ def _gps_clock_start_time( ) -def _telemetry_start_time(video_path: Path) -> datetime.datetime | None: +def _read_telemetry(video_path: Path) -> T.Sequence[geo.Point]: + """ + Read the GPS points of the video's own telemetry, or none if it has none + """ try: - video_metadata = NativeVideoExtractor(video_path).extract() + return NativeVideoExtractor(video_path).extract().points except exceptions.MapillaryDescriptionError as ex: - LOG.debug("No video telemetry to read the start time from: %s", ex) + LOG.debug("No video telemetry to read the clock from: %s", ex) + except Exception as ex: + # Telemetry is only one way to tell when the frames were captured, so + # a video whose telemetry fails to parse must still be sampled + LOG.warning( + "Unable to read the telemetry of %s: %s", + video_path.name, + ex, + exc_info=LOG.isEnabledFor(logging.DEBUG), + ) + return [] + + +def _telemetry_clock_rate( + video_path: Path, telemetry_points: T.Sequence[geo.Point] +) -> float | None: + """ + GPS seconds per video second of the video's own GPS clock: more than 1 if + the video is a timelapse, or None if it has no GPS clock + """ + clock = GPXVideoExtractor._read_video_clock(telemetry_points) + if clock is None: return None - return _gps_clock_start_time(video_metadata.points) + if clock.rate != 1.0: + LOG.info( + "The GPS clock in %s runs %.1f times faster than the video (timelapse)", + video_path.name, + clock.rate, + ) + return clock.rate def _parse_filename_time(video_path: Path) -> datetime.datetime | None: @@ -530,7 +562,10 @@ def _creation_time_to_start_time( def _extract_video_start_time( - video_path: Path, probe: ffmpeglib.Probe + video_path: Path, + probe: ffmpeglib.Probe, + telemetry_points: T.Sequence[geo.Point], + rate: float, ) -> datetime.datetime | None: """ Determine the wall clock time at which a video started recording. @@ -541,19 +576,7 @@ def _extract_video_start_time( back to the creation time when there is no telemetry to sync against, which is the case for the plain MP4s that get geotagged from a GPX. """ - try: - start_time = _telemetry_start_time(video_path) - except Exception as ex: - # Telemetry is only one way to find the start time, so a video whose - # telemetry fails to parse must still be sampled - LOG.warning( - "Unable to read the start time of %s from its telemetry: %s", - video_path.name, - ex, - exc_info=LOG.isEnabledFor(logging.DEBUG), - ) - start_time = None - + start_time = _gps_clock_start_time(telemetry_points, rate) if start_time is not None: return start_time @@ -569,9 +592,30 @@ def _sample_single_video_by_interval( ) -> None: ffmpeg = ffmpeglib.FFMPEG(constants.FFMPEG_PATH, constants.FFPROBE_PATH) + telemetry_points = _read_telemetry(video_path) + + # Frames are stamped with their capture times, which a timelapse packs + # several seconds of into each second of video: as many as the video's own + # GPS clock says, and for a video without one, --video_duration_ratio + clock_rate = _telemetry_clock_rate(video_path, telemetry_points) + if clock_rate is None: + rate = duration_ratio + else: + rate = clock_rate + if duration_ratio != 1.0: + LOG.warning( + "Ignoring --video_duration_ratio %g for %s, whose own GPS clock runs at %.3gx", + duration_ratio, + video_path.name, + clock_rate, + ) + if start_time is None: start_time = _extract_video_start_time( - video_path, ffmpeglib.Probe(ffmpeg.probe_format_and_streams(video_path)) + video_path, + ffmpeglib.Probe(ffmpeg.probe_format_and_streams(video_path)), + telemetry_points, + rate, ) if start_time is None: raise exceptions.MapillaryVideoError( @@ -586,7 +630,7 @@ def _sample_single_video_by_interval( if sample_paths[0] is None: continue # extract_frames() produces 1-based frame indices so we need to subtract 1 here - seconds = (frame_idx_1based - 1) * sample_interval * duration_ratio + seconds = (frame_idx_1based - 1) * sample_interval * rate timestamp = start_time + datetime.timedelta(seconds=seconds) exif_edit = ExifEdit(sample_paths[0]) exif_edit.add_date_time_original(timestamp) diff --git a/tests/unit/test_geotag_images_from_video.py b/tests/unit/test_geotag_images_from_video.py new file mode 100644 index 00000000..27e5c948 --- /dev/null +++ b/tests/unit/test_geotag_images_from_video.py @@ -0,0 +1,83 @@ +# 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 + +import datetime +import os +import shutil +import typing as T +from pathlib import Path + +import pytest +from mapillary_tools import geo, telemetry, types +from mapillary_tools.exif_write import ExifEdit +from mapillary_tools.geotag.geotag_images_from_video import GeotagImagesFromVideo + +_PWD = Path(os.path.dirname(os.path.abspath(__file__))) + +START_TIME = datetime.datetime(2021, 8, 10, 6, 38, 6, tzinfo=datetime.timezone.utc) + + +def _video_metadata(video_path: Path, rate: float) -> types.VideoMetadata: + # One point per second of video, recorded **rate** seconds apart + points = [ + telemetry.GPSPoint( + time=float(i), + lat=40.0 + i * 0.0001, + lon=-74.0, + alt=None, + angle=None, + epoch_time=START_TIME.timestamp() + rate * i, + fix=None, + precision=None, + ground_speed=None, + ) + for i in range(10) + ] + return types.VideoMetadata( + filename=video_path, + filetype=types.FileType.GOPRO, + points=T.cast(T.List[geo.Point], points), + ) + + +def _sample(sample_dir: Path, idx: int, capture_time: datetime.datetime) -> Path: + sample_path = sample_dir / f"hello_v_{idx:06d}.jpg" + shutil.copyfile(_PWD / "data/test_exif.jpg", sample_path) + exif_edit = ExifEdit(sample_path) + exif_edit.add_date_time_original(capture_time) + exif_edit.add_gps_datetime(capture_time) + exif_edit.write() + return sample_path + + +@pytest.mark.parametrize("rate", [1.0, 10.0]) +def test_samples_are_placed_by_their_capture_times(tmp_path: Path, rate: float): + video_path = tmp_path / "hello.mp4" + sample_dir = tmp_path / "samples" / "hello.mp4" + sample_dir.mkdir(parents=True) + + # A sample every 2 seconds of video, stamped with its capture time, which + # a timelapse spaces **rate** times as far apart + samples = [ + _sample( + sample_dir, k + 1, START_TIME + datetime.timedelta(seconds=2 * k * rate) + ) + for k in range(4) + ] + + metadatas = GeotagImagesFromVideo( + [_video_metadata(video_path, rate)], num_processes=0 + ).to_description(samples) + + lat_by_name = {} + for metadata in metadatas: + assert isinstance(metadata, types.ImageMetadata), metadata + lat_by_name[metadata.filename.name] = metadata.lat + assert lat_by_name == { + sample.name: pytest.approx(40.0 + 2 * k * 0.0001) + for k, sample in enumerate(samples) + } diff --git a/tests/unit/test_sample_video.py b/tests/unit/test_sample_video.py index cf4760b7..052a30cd 100644 --- a/tests/unit/test_sample_video.py +++ b/tests/unit/test_sample_video.py @@ -74,12 +74,16 @@ def setup_mock(monkeypatch): monkeypatch.setattr(ffmpeglib, "FFMPEG", MOCK_FFMPEG) -def _validate_interval(samples: T.Sequence[Path], video_start_time): +def _validate_interval( + samples: T.Sequence[Path], video_start_time, seconds_per_sample: float = 2 +): assert len(samples), "expect samples but got none" for idx, sample in enumerate(sorted(samples)): assert sample.name == f"hello_v_{idx + 1:06d}.jpg" exif = exif_read.ExifRead(sample) - expected_dt = video_start_time + datetime.timedelta(seconds=2 * idx) + expected_dt = video_start_time + datetime.timedelta( + seconds=seconds_per_sample * idx + ) assert exif.extract_capture_time() == expected_dt @@ -208,10 +212,103 @@ def raise_type_error(): samples = sample_dir.join("hello.mp4").listdir() _validate_interval([Path(s) for s in samples], PROBE_START_TIME) - assert "Unable to read the start time of hello.mp4 from its telemetry" in ( - caplog.text + assert "Unable to read the telemetry of hello.mp4" in caplog.text + + +class TestSampleTimelapseByInterval: + """ + Interval samples of a timelapse are stamped with their capture times, which + its GPS clock spaces 10 times as far apart as the frames + """ + + GPS_START_TIME = datetime.datetime( + 2021, 8, 10, 6, 38, 6, tzinfo=datetime.timezone.utc ) + @pytest.fixture + def timelapse_telemetry(self, monkeypatch): + # One point per second of video, recorded 10 seconds apart + points = [ + telemetry.GPSPoint( + time=float(i), + lat=40.0 + i * 0.001, + lon=-74.0, + alt=None, + angle=None, + epoch_time=self.GPS_START_TIME.timestamp() + 10 * i, + fix=None, + precision=None, + ground_speed=None, + ) + for i in range(60) + ] + monkeypatch.setattr( + sample_video, + "NativeVideoExtractor", + lambda video_path: mock.Mock( + extract=lambda: VideoMetadata( + filename=video_path, + filetype=FileType.GOPRO, + points=T.cast(T.List[geo.Point], points), + ) + ), + ) + + def _sample(self, tmpdir: py.path.local, **kwargs) -> list[Path]: + root = _PWD.joinpath("data/mock_sample_video") + sample_dir = tmpdir.mkdir("sampled_video_frames") + sample_video.sample_video( + root.joinpath("videos"), + Path(sample_dir), + video_sample_distance=-1, + video_sample_interval=2, + rerun=True, + **kwargs, + ) + return [Path(s) for s in sample_dir.join("hello.mp4").listdir()] + + def test_capture_times_follow_the_gps_clock( + self, tmpdir, setup_mock, timelapse_telemetry, caplog + ): + _validate_interval( + self._sample(tmpdir), self.GPS_START_TIME, seconds_per_sample=20 + ) + assert "--video_duration_ratio" not in caplog.text + + def test_from_the_given_start_time(self, tmpdir, setup_mock, timelapse_telemetry): + video_start_time = "2020_08_10_14_37_05_023" + _validate_interval( + self._sample(tmpdir, video_start_time=video_start_time), + description.parse_capture_time(video_start_time), + seconds_per_sample=20, + ) + + def test_the_gps_clock_wins_over_the_duration_ratio( + self, tmpdir, setup_mock, timelapse_telemetry, caplog + ): + """ + The ratio is for the videos without a GPS clock, which a batch of + videos may mix with videos that have one + """ + _validate_interval( + self._sample(tmpdir, video_duration_ratio=5), + self.GPS_START_TIME, + seconds_per_sample=20, + ) + assert ( + "Ignoring --video_duration_ratio 5 for hello.mp4, whose own GPS clock runs at 10x" + in caplog.text + ) + + def test_the_duration_ratio_paces_a_video_without_a_gps_clock( + self, tmpdir, setup_mock + ): + _validate_interval( + self._sample(tmpdir, video_duration_ratio=5), + PROBE_START_TIME, + seconds_per_sample=10, + ) + class TestGPSClockStartTime: """Tests for _gps_clock_start_time.""" @@ -296,6 +393,12 @@ def test_uses_the_start_of_a_timelapse(self) -> None: 2021, 8, 10, 12, 38, 24, tzinfo=datetime.timezone.utc ) + def test_maps_a_timelapse_back_at_its_rate(self) -> None: + points = [self._gps_point(float(i), 1628599086.0 + 10 * i) for i in range(1000)] + assert sample_video._gps_clock_start_time(points, rate=10) == datetime.datetime( + 2021, 8, 10, 12, 38, 6, tzinfo=datetime.timezone.utc + ) + def test_no_absolute_timestamps(self) -> None: assert sample_video._gps_clock_start_time(_make_gps_points(3)) is None From 7970382bff4cd631fc6b0ed9f2c0ab61ed8782ed Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Tue, 20 Oct 2026 14:06:17 +0200 Subject: [PATCH 2/5] Keep a timelapse's clock rate when the GPX starts with the video --interpolation_use_gpx_start_time places the GPX from the start of the video instead of by the video's GPS clock. It skipped reading the clock altogether, and with it the clock's rate, so a timelapse got its GPX laid out at 1x: a 5 s, 30x timelapse got 157 s of GPX, and distance sampling picked 5 frames along its first 12 m instead of 148 along the 1.4 km route. The clock is now read in that mode too, and placed at the start of the GPX, so that only its rate is used. A video that is not a timelapse has a rate of 1, and is placed as before. The coverage check still sees such a track as one that no clock placed: the clock gave it only its pace. --- .../geotag/video_extractors/gpx.py | 22 ++++++++++++++----- tests/unit/test_gpx_video_coverage.py | 19 +++++++++++++--- 2 files changed, 32 insertions(+), 9 deletions(-) diff --git a/mapillary_tools/geotag/video_extractors/gpx.py b/mapillary_tools/geotag/video_extractors/gpx.py index 75d4aa16..60f56a38 100644 --- a/mapillary_tools/geotag/video_extractors/gpx.py +++ b/mapillary_tools/geotag/video_extractors/gpx.py @@ -157,11 +157,16 @@ def extract(self) -> types.VideoMetadata: points=gpx_points, ) - clock: VideoClock | None = None - if self.sync_mode is not SyncMode.REBASE: - clock = self._read_video_clock(native_video_metadata.points) - - if clock is not None and gpx_points: + clock = self._read_video_clock(native_video_metadata.points) + + if clock is not None and gpx_points and self.sync_mode is SyncMode.REBASE: + # The GPX is placed to start with the video, and runs on at the rate + # of the video's GPS clock, which in a timelapse packs several GPS + # seconds into each second of video + clock = VideoClock( + video_time=0.0, unix_time=gpx_points[0].time, rate=clock.rate + ) + elif clock is not None and gpx_points: offset = gpx_points[0].time - clock.to_unix_time(0) if abs(offset) > _IMPLAUSIBLE_OFFSET_SECONDS: if noisy and self.sync_mode is SyncMode.SYNC: @@ -193,8 +198,13 @@ def extract(self) -> types.VideoMetadata: p.time = clock.to_video_time(p.time) self._apply_offset_time(gpx_points, rate=1.0 if clock is None else clock.rate) + # Started with the video, the track is checked against it as one that + # no clock placed: the clock gave it only its pace self._check_coverage( - gpx_points, gpx_unix_times, native_video_metadata.points, clock + gpx_points, + gpx_unix_times, + native_video_metadata.points, + None if self.sync_mode is SyncMode.REBASE else clock, ) return dataclasses.replace(native_video_metadata, points=gpx_points) diff --git a/tests/unit/test_gpx_video_coverage.py b/tests/unit/test_gpx_video_coverage.py index 73a6581a..7ad03592 100644 --- a/tests/unit/test_gpx_video_coverage.py +++ b/tests/unit/test_gpx_video_coverage.py @@ -33,7 +33,7 @@ SourceType, VideoSyncOption, ) -from mapillary_tools.geotag.video_extractors.gpx import GPXVideoExtractor +from mapillary_tools.geotag.video_extractors.gpx import GPXVideoExtractor, SyncMode from mapillary_tools.mp4 import construct_mp4_parser as cparser, simple_mp4_builder @@ -321,12 +321,12 @@ class TestTimelapse: # 11 seconds of video spanning 165 seconds of GPS time RATE = 15.0 - def _extract(self, tmp_path, gpx_start: float, gpx_seconds: float): + def _extract(self, tmp_path, gpx_start: float, gpx_seconds: float, **kwargs): video_path = _write_camm_mp4( tmp_path / "timelapse.mp4", _video_track(self.RATE), duration=11.0 ) gpx_path = _write_gpx(tmp_path / "track.gpx", gpx_start, gpx_seconds) - return GPXVideoExtractor(video_path, gpx_path).extract() + return GPXVideoExtractor(video_path, gpx_path, **kwargs).extract() def test_gpx_of_the_whole_recording_is_silent(self, tmp_path, caplog): """At 1x, the track would run on 154 seconds past the video's end.""" @@ -349,6 +349,19 @@ def test_gap_is_reported_in_gps_time(self, tmp_path): self._extract(tmp_path, recording_end + 3600, 60) assert "by 1.0 hours." in str(info.value) + def test_gpx_start_time_keeps_the_rate(self, tmp_path, caplog): + """ + --interpolation_use_gpx_start_time starts the track with the video, + whatever the clock says, and the clock still sets its pace + """ + with caplog.at_level(logging.WARNING): + metadata = self._extract( + tmp_path, A_UNIX_TIME + 3600, 11 * self.RATE, sync_mode=SyncMode.REBASE + ) + assert not _warnings(caplog) + assert metadata.points[0].time == 0 + assert metadata.points[-1].time == pytest.approx(11.0, abs=1e-3) + class TestVideoWithoutGPSClock: """ From 4f69382895c3d61e089f5ee89b6a7ae3a2807e63 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Tue, 20 Oct 2026 14:17:53 +0200 Subject: [PATCH 3/5] Sync a GPX to a video without a GPS clock at --video_duration_ratio A timelapse packs several seconds of capture into each second of video. With a GPS clock of its own, a GPX is synced at the clock's rate. Without one, nothing in the video tells the rate, and the GPX was laid out at 1x, so the frames got only the start of the track: an Insta360 ONE X timelapse, 1.8 s of video with a 110 s GPX, gave 1 frame. --video_duration_ratio, the seconds of real time per second of video, used to apply to interval sampling only, and was marked deprecated. It now paces the GPX of a video without a GPS clock in distance sampling too, whether the GPX is placed by --video_start_time, by --interpolation_use_gpx_start_time or by assuming that it starts with the video. The GPS clock of a video, if it has one, still wins, and says so in a warning when it overrides a ratio. The ratio is required where a track runs on long after the video and nothing else accounts for it. Without a start time, a track more than twice as long as the video was already refused; it is now measured against the time the video spans at the ratio, and the error names the ratio. With --interpolation_use_gpx_start_time, such a track is now refused as well, in interval sampling too: the video is a timelapse, or the track was left running after it or covers more than this video. For a track of less than a day, which may be a GPS logger left running, the error gives the start of the track to pass as --video_start_time instead, which places the video the same way. A longer track is a log of several trips, which that would put the video on the first of: the error asks for the time the video started. With --video_start_time nothing is required, so that one GPX can serve many videos, each placed by its start time. The ratio divides GPS time now, so one that is not positive is rejected. process is unchanged. --- mapillary_tools/commands/sample_video.py | 2 +- mapillary_tools/exceptions.py | 5 +- mapillary_tools/geotag/options.py | 4 + .../geotag/video_extractors/gpx.py | 228 +++++++++++++----- mapillary_tools/sample_video.py | 38 +++ tests/integration/test_video_process.py | 56 ++++- tests/unit/test_gpx_video_sync.py | 155 +++++++++++- tests/unit/test_sample_video.py | 70 ++++++ 8 files changed, 491 insertions(+), 67 deletions(-) diff --git a/mapillary_tools/commands/sample_video.py b/mapillary_tools/commands/sample_video.py index c7dc3e92..12be7671 100644 --- a/mapillary_tools/commands/sample_video.py +++ b/mapillary_tools/commands/sample_video.py @@ -34,7 +34,7 @@ def add_basic_arguments(self, parser: argparse.ArgumentParser): ) group.add_argument( "--video_duration_ratio", - help="[DEPRECATED since v0.10.0] Real time video duration ratio of the under or oversampled video duration. [default: %(default)s]", + help="Seconds of real time per second of video, for videos without a GPS clock of their own to tell, such as 60 for a timelapse that took a frame every 2 seconds and plays at 30 frames per second. A timelapse needs it to be synced to a GPX track. The GPS clock of a video, if it has one, wins. [default: %(default)s]", type=float, default=constants.VIDEO_DURATION_RATIO, required=False, diff --git a/mapillary_tools/exceptions.py b/mapillary_tools/exceptions.py index 4c400f30..dac650ef 100644 --- a/mapillary_tools/exceptions.py +++ b/mapillary_tools/exceptions.py @@ -91,8 +91,9 @@ class MapillaryStationaryVideoError(MapillaryDescriptionError): 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. + Nothing says how a GPX track lines up with a video without a GPS clock: + the track is too long either to assume that it starts with the video, or, + when it is said to, to end with it at the rate the video is said to run. """ diff --git a/mapillary_tools/geotag/options.py b/mapillary_tools/geotag/options.py index 72f635f3..108d59d5 100644 --- a/mapillary_tools/geotag/options.py +++ b/mapillary_tools/geotag/options.py @@ -166,6 +166,10 @@ class VideoSyncOption: # frame at the start of the GPX use_gpx_start_time: bool = False + # --video_duration_ratio: GPS seconds per second of a video without a GPS + # clock of its own to tell, such as a timelapse + duration_ratio: float = 1.0 + SourceOptionSchema = { "type": "object", diff --git a/mapillary_tools/geotag/video_extractors/gpx.py b/mapillary_tools/geotag/video_extractors/gpx.py index 60f56a38..17e0764d 100644 --- a/mapillary_tools/geotag/video_extractors/gpx.py +++ b/mapillary_tools/geotag/video_extractors/gpx.py @@ -38,6 +38,10 @@ # of magnitude. _IMPLAUSIBLE_OFFSET_SECONDS = 24 * 3600 +# A GPX track that lasts longer than this is not that of one recording, such as +# a GPS logger left running after the video, but a log of several trips +_MAX_RECORDING_SECONDS = 24 * 3600 + # A GPS clock that runs this much faster than the video clock is a timelapse. # Normal videos stay well below it, although over a few seconds of GPS # timestamped once per block the estimate can be 20% off @@ -108,32 +112,7 @@ def __init__( @override def extract(self) -> types.VideoMetadata: - try: - gpx_tracks = parse_gpx(self.gpx_path) - except (OSError, UnicodeError, gpxpy.gpx.GPXException) as ex: - raise exceptions.MapillaryInvalidGPXError( - f"{self.gpx_path.name} is not a valid GPX file. " - "Replace it with a valid GPX file." - ) from ex - - if 1 < len(gpx_tracks): - LOG.warning( - f"Found {len(gpx_tracks)} tracks in the GPX file {self.gpx_path}. Will merge points in all the tracks as a single track for interpolation" - ) - - all_points: list[geo.Point] = sum(gpx_tracks, []) - gpx_points = [p for p in all_points if _GPS_EPOCH <= p.time] - if len(gpx_points) < len(all_points): - LOG.warning( - "Ignoring %d of the %d points in the GPX file %s, which are timestamped before 1980", - len(all_points) - len(gpx_points), - len(all_points), - self.gpx_path, - ) - - # The tracks need not be in time order, and interpolation needs the - # merged points to be - gpx_points.sort(key=lambda p: p.time) + gpx_points = self._read_gpx_points() # The first and last Unix time of the track, before it is placed on # the timeline of the video gpx_unix_times = ( @@ -148,8 +127,9 @@ def extract(self) -> types.VideoMetadata: if self.sync_mode is SyncMode.STRICT_SYNC: raise ex start_clock = self._place_without_video_gps(gpx_points) - self._apply_offset_time(gpx_points) - self._check_coverage(gpx_points, gpx_unix_times, [], start_clock) + rate = self._duration_ratio() + self._apply_offset_time(gpx_points, rate=rate) + self._check_coverage(gpx_points, gpx_unix_times, [], start_clock, rate) return types.VideoMetadata( filename=self.video_path, filesize=utils.get_file_size(self.video_path), @@ -187,6 +167,13 @@ def extract(self) -> types.VideoMetadata: # Its GPS has no clock to sync with, as if it had no GPS clock = self._place_without_video_gps(gpx_points) else: + if self._duration_ratio() != 1.0: + LOG.warning( + "Ignoring --video_duration_ratio %g for %s, whose own GPS clock runs at %.3gx", + self._duration_ratio(), + self.video_path, + clock.rate, + ) if clock.rate != 1.0: LOG.info( "The GPS clock in %s runs %.1f times faster than the video (timelapse). Syncing %s accordingly", @@ -197,7 +184,8 @@ def extract(self) -> types.VideoMetadata: for p in gpx_points: p.time = clock.to_video_time(p.time) - self._apply_offset_time(gpx_points, rate=1.0 if clock is None else clock.rate) + rate = self._duration_ratio() if clock is None else clock.rate + self._apply_offset_time(gpx_points, rate=rate) # Started with the video, the track is checked against it as one that # no clock placed: the clock gave it only its pace self._check_coverage( @@ -205,17 +193,51 @@ def extract(self) -> types.VideoMetadata: gpx_unix_times, native_video_metadata.points, None if self.sync_mode is SyncMode.REBASE else clock, + rate, ) return dataclasses.replace(native_video_metadata, points=gpx_points) + def _read_gpx_points(self) -> list[geo.Point]: + """ + Read the points of the GPX file, merged into one track in time order + """ + try: + gpx_tracks = parse_gpx(self.gpx_path) + except (OSError, UnicodeError, gpxpy.gpx.GPXException) as ex: + raise exceptions.MapillaryInvalidGPXError( + f"{self.gpx_path.name} is not a valid GPX file. " + "Replace it with a valid GPX file." + ) from ex + + if 1 < len(gpx_tracks): + LOG.warning( + f"Found {len(gpx_tracks)} tracks in the GPX file {self.gpx_path}. Will merge points in all the tracks as a single track for interpolation" + ) + + all_points: list[geo.Point] = sum(gpx_tracks, []) + gpx_points = [p for p in all_points if _GPS_EPOCH <= p.time] + if len(gpx_points) < len(all_points): + LOG.warning( + "Ignoring %d of the %d points in the GPX file %s, which are timestamped before 1980", + len(all_points) - len(gpx_points), + len(all_points), + self.gpx_path, + ) + + # The tracks need not be in time order, and interpolation needs the + # merged points to be + gpx_points.sort(key=lambda p: p.time) + return gpx_points + def _place_without_video_gps( self, gpx_points: T.Sequence[geo.Point] ) -> VideoClock | None: """ - Place the GPX track on the timeline of a video that has no GPS to sync - it with: by the start time of the video, when sampling was given one, - and otherwise by assuming that the track starts with the video. + Place the GPX track on the timeline of a video that has no GPS clock to + sync it with: by the start time of the video, when sampling was given + one, and otherwise by assuming that the track starts with the video. It + runs at --video_duration_ratio GPS seconds per second of video. Return the clock of the video the track was placed by: the start time, or None if the track was assumed to start with the video. @@ -223,17 +245,34 @@ def _place_without_video_gps( if not gpx_points: return None + ratio = self._duration_ratio() + if self.sync_mode is SyncMode.SYNC and self.video_sync is not None: if self.video_sync.start_time is not None: self._rebase_times( - gpx_points, offset=gpx_points[0].time - self.video_sync.start_time + gpx_points, + offset=(gpx_points[0].time - self.video_sync.start_time) / ratio, + rate=ratio, + ) + return VideoClock( + video_time=0.0, unix_time=self.video_sync.start_time, rate=ratio ) - return VideoClock(video_time=0.0, unix_time=self.video_sync.start_time) - self._check_assumed_start(gpx_points) + self._check_assumed_start(gpx_points, ratio) + elif self.sync_mode is SyncMode.REBASE and self.video_sync is not None: + self._check_given_start(gpx_points, ratio) - self._rebase_times(gpx_points) + self._rebase_times(gpx_points, rate=ratio) return None + def _duration_ratio(self) -> float: + """ + GPS seconds per second of a video without a GPS clock of its own to + tell, from --video_duration_ratio: more than 1 for a timelapse + """ + if self.video_sync is None: + return 1.0 + return self.video_sync.duration_ratio + def _apply_offset_time( self, gpx_points: T.Sequence[geo.Point], rate: float = 1.0 ) -> None: @@ -263,10 +302,12 @@ def _check_coverage( gpx_unix_times: tuple[float, float] | None, video_gps_points: T.Sequence[geo.Point], clock: VideoClock | None, + rate: float, ) -> None: """ Check the GPX track, as placed on the timeline of the video, against - the video, from video time 0 to its mvhd duration. + the video, from video time 0 to its mvhd duration. Each second of video + spans **rate** GPS seconds. A track that misses the video has no position for any moment of it, so it raises when the clock it was placed by says so. A track that covers @@ -288,8 +329,6 @@ def _check_coverage( if video_end <= 0: return - # GPS seconds per video second - rate = 1.0 if clock is None else clock.rate gpx_first = min(p.time for p in gpx_points) gpx_last = max(p.time for p in gpx_points) @@ -329,33 +368,33 @@ def _check_coverage( LOG.warning(message) - def _check_assumed_start(self, gpx_points: T.Sequence[geo.Point]) -> None: + def _check_assumed_start( + self, gpx_points: T.Sequence[geo.Point], ratio: float + ) -> None: """ Check that the GPX track can have started with the video, as it is - assumed to when nothing says when the video starts. + assumed to when nothing says when the video starts. The video spans + **ratio** GPS seconds per second. A track that runs on for longer than the whole video was not recorded for the video alone, and the video could start anywhere along it: raise. One that runs on for less may have been started a little before the video, which puts every frame up to that much too early along it: warn. """ - video_duration = _video_duration(self.video_path) - if video_duration is None: - LOG.debug( - "Unable to read the duration of %s to check the GPX track against", - self.video_path, - ) + durations = self._durations(gpx_points, ratio) + if durations is None: return - gpx_duration = max(p.time for p in gpx_points) - min(p.time for p in gpx_points) - excess = gpx_duration - video_duration + video_span, gpx_duration = durations + excess = gpx_duration - video_span - if video_duration < excess: + if video_span < excess: raise exceptions.MapillaryGPXVideoStartUnknownError( f"The GPX track in {self.gpx_path} lasts {_format_duration(gpx_duration)} " - f"and the video only {_format_duration(video_duration)}, so the video " + f"and {self._describe_video_span(video_span, ratio)}, so the video " "could start anywhere along the track. Specify --video_start_time, or " - "--interpolation_use_gpx_start_time if the track was started with the video" + "--interpolation_use_gpx_start_time if the track was started with the " + "video, and if the video is a timelapse, --video_duration_ratio as well" ) if _LOGGER_TOLERANCE_SECONDS < excess: @@ -370,6 +409,82 @@ def _check_assumed_start(self, gpx_points: T.Sequence[geo.Point]) -> None: _format_duration(excess), ) + def _check_given_start( + self, gpx_points: T.Sequence[geo.Point], ratio: float + ) -> None: + """ + Check a GPX track that --interpolation_use_gpx_start_time says started + with the video, which spans **ratio** GPS seconds per second. A track + that runs on for longer than the whole video was not recorded for it + alone, or belongs to a timelapse, whose rate nothing gives: raise. + """ + durations = self._durations(gpx_points, ratio) + if durations is None: + return + + video_span, gpx_duration = durations + if gpx_duration - video_span <= video_span: + return + + message = ( + f"The GPX track in {self.gpx_path} lasts {_format_duration(gpx_duration)} " + f"and {self._describe_video_span(video_span, ratio)}, so it runs on long " + "after the video ends. " + ) + if gpx_duration < _MAX_RECORDING_SECONDS: + # The start of the track is the start of the video only if the + # track is that of the one recording + message += ( + "If the video is a timelapse, specify --video_duration_ratio. If the " + "GPS logger was started with the video and left running, specify " + f"--video_start_time {build_capture_time(min(p.time for p in gpx_points))}, " + "when the track starts, instead of --interpolation_use_gpx_start_time. " + "If the track covers more than this video, specify the time the video " + "started as --video_start_time instead" + ) + else: + message += ( + "A track that long covers more than this video: specify the time the " + "video started as --video_start_time instead of " + "--interpolation_use_gpx_start_time" + ) + raise exceptions.MapillaryGPXVideoStartUnknownError(message) + + def _describe_video_span(self, video_span: float, ratio: float) -> str: + """ + >>> extractor = GPXVideoExtractor(Path("a.mp4"), Path("a.gpx")) + >>> extractor._describe_video_span(5, 1) + 'the video only 5 seconds' + >>> extractor._describe_video_span(120, 60) + 'the video only 2 seconds, which at --video_duration_ratio 60 span 2.0 minutes' + """ + description = f"the video only {_format_duration(video_span / ratio)}" + if ratio != 1.0: + description += ( + f", which at --video_duration_ratio {ratio:g} span " + f"{_format_duration(video_span)}" + ) + return description + + def _durations( + self, gpx_points: T.Sequence[geo.Point], ratio: float + ) -> tuple[float, float] | None: + """ + The GPS seconds that the video spans at **ratio** GPS seconds per + second, and the duration of the GPX track, or None if the duration of + the video cannot be read + """ + video_duration = _video_duration(self.video_path) + if video_duration is None: + LOG.debug( + "Unable to read the duration of %s to check the GPX track against", + self.video_path, + ) + return None + + gpx_duration = max(p.time for p in gpx_points) - min(p.time for p in gpx_points) + return video_duration * ratio, gpx_duration + def _extract_native_video_metadata(self) -> tuple[types.VideoMetadata, bool]: """ Extract the video metadata, and whether its GPS was rejected as noise @@ -386,14 +501,17 @@ def _extract_native_video_metadata(self) -> tuple[types.VideoMetadata, bool]: return metadata, True @classmethod - def _rebase_times(cls, points: T.Sequence[geo.Point], offset: float = 0.0) -> None: + def _rebase_times( + cls, points: T.Sequence[geo.Point], offset: float = 0.0, rate: float = 1.0 + ) -> None: """ - Rebase point times to start from **offset** + Rebase point times to start from **offset**, at **rate** of their + seconds per second """ if points: first_timestamp = points[0].time for p in points: - p.time = (p.time - first_timestamp) + offset + p.time = (p.time - first_timestamp) / rate + offset @classmethod def _read_video_clock( diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py index 9636c6bb..4ad8cd9d 100644 --- a/mapillary_tools/sample_video.py +++ b/mapillary_tools/sample_video.py @@ -158,6 +158,12 @@ def sample_video( f"Expect either non-negative video_sample_distance or positive video_sample_interval but got {video_sample_distance} and {video_sample_interval} respectively" ) + # Written as a negated comparison so that NaN is rejected too + if not 0 < video_duration_ratio: + raise exceptions.MapillaryBadParameterError( + f"Expect a positive video_duration_ratio but got {video_duration_ratio}" + ) + video_start_time_dt: datetime.datetime | None = None if video_start_time is not None: try: @@ -176,6 +182,7 @@ def sample_video( ), offset_time=interpolation_offset_time, use_gpx_start_time=interpolation_use_gpx_start_time, + duration_ratio=video_duration_ratio, ) geotag_options = _parse_geotag_options( @@ -245,6 +252,8 @@ def sample_video( sample_interval=video_sample_interval, duration_ratio=video_duration_ratio, start_time=video_start_time_dt, + geotag_options=geotag_options, + video_sync=video_sync, ) except ffmpeglib.FFmpegNotFoundError as ex: # fatal error @@ -583,12 +592,38 @@ def _extract_video_start_time( return _creation_time_to_start_time(video_path, probe) +def _check_gpx_start_by_interval( + video_path: Path, + geotag_options: T.Sequence[SourceOption], + video_sync: VideoSyncOption, +) -> None: + """ + Check the GPX tracks that --interpolation_use_gpx_start_time says started + with a video without a GPS clock, as distance sampling does. The geotagging + stage puts the first frame at the start of the track and the others at + --video_duration_ratio, so a track that runs on long after the video was + not recorded for it alone, or needs the ratio. + """ + for option in geotag_options: + if option.source is not SourceType.GPX or option.source_path is None: + continue + gpx_path = option.source_path.resolve(video_path) + if not gpx_path.is_file(): + continue + extractor = GPXVideoExtractor(video_path, gpx_path, video_sync=video_sync) + gpx_points = extractor._read_gpx_points() + if gpx_points: + extractor._check_given_start(gpx_points, video_sync.duration_ratio) + + def _sample_single_video_by_interval( video_path: Path, sample_dir: Path, sample_interval: float, duration_ratio: float, start_time: datetime.datetime | None = None, + geotag_options: T.Sequence[SourceOption] = (), + video_sync: VideoSyncOption | None = None, ) -> None: ffmpeg = ffmpeglib.FFMPEG(constants.FFMPEG_PATH, constants.FFPROBE_PATH) @@ -610,6 +645,9 @@ def _sample_single_video_by_interval( clock_rate, ) + if clock_rate is None and video_sync is not None and video_sync.use_gpx_start_time: + _check_gpx_start_by_interval(video_path, geotag_options, video_sync) + if start_time is None: start_time = _extract_video_start_time( video_path, diff --git a/tests/integration/test_video_process.py b/tests/integration/test_video_process.py index bdd26788..cad2b6b4 100644 --- a/tests/integration/test_video_process.py +++ b/tests/integration/test_video_process.py @@ -6,6 +6,7 @@ import functools import json import os +import re import subprocess import tempfile from pathlib import Path @@ -272,18 +273,34 @@ def test_video_process_by_distance_applies_the_interpolation_offset( assert capture_times[-1] <= "2025_03_14_07_01_33_624", capture_times +def _gpx_until(gpx_file: py.path.local, until: str, path: Path) -> Path: + """Copy the GPX file without its points recorded after **until**""" + text = gpx_file.read_text("utf-8") + trkpts = re.findall(r"", text, flags=re.DOTALL) + for trkpt in trkpts: + [time] = re.findall(r"", trkpt) + if until < time: + text = text.replace(trkpt, "") + path.write_text(text, "utf-8") + return path + + def test_video_process_by_distance_with_the_gpx_start_time( setup_data: py.path.local, ): """ As the README describes for a GPX started 2.5 seconds before the video: - the frames start 2.5 seconds into the track, however long it is. + the frames start 2.5 seconds into the track. """ pytest_skip_if_not_ffmpeg_installed() - gpx_file = setup_data.join("gpx").join("sf_30km_h.gpx") - with tempfile.TemporaryDirectory() as dir: + # Stopped about when the 5.76s video ended + gpx_file = _gpx_until( + setup_data.join("gpx").join("sf_30km_h.gpx"), + "2025-03-14T07:00:10", + Path(dir, "track.gpx"), + ) video_path = _with_creation_time( setup_data.join("videos").join("sample-5s.mp4"), "2025-03-14T07:00:36Z", dir ) @@ -306,6 +323,39 @@ def test_video_process_by_distance_with_the_gpx_start_time( assert capture_times[-1] < "2025_03_14_07_00_09_000", capture_times +def test_video_process_with_the_gpx_start_time_refuses_a_far_longer_gpx( + setup_data: py.path.local, +): + """ + A GPX running on for 1.5 minutes after a 5.76s video was not recorded for + it alone, or the video is a timelapse: the error says what to specify. + """ + pytest_skip_if_not_ffmpeg_installed() + + gpx_file = setup_data.join("gpx").join("sf_30km_h.gpx") + + with tempfile.TemporaryDirectory() as dir: + video_path = _with_creation_time( + setup_data.join("videos").join("sample-5s.mp4"), "2025-03-14T07:00:36Z", dir + ) + with pytest.raises(subprocess.CalledProcessError) as excinfo: + run_video_process( + [ + "--video_sample_distance=3", + "--interpolation_use_gpx_start_time", + "--geotag_source=gpx", + *["--geotag_source_path", str(gpx_file)], + str(video_path), + str(Path(dir, "samples")), + ], + capture_output=True, + text=True, + ) + + assert "--video_duration_ratio" in excinfo.value.stderr + assert "--video_start_time" in excinfo.value.stderr + + def test_video_process_by_distance_with_an_untidy_gpx(setup_data: py.path.local): """ GPX files converted from camera logs can list their tracks out of order and diff --git a/tests/unit/test_gpx_video_sync.py b/tests/unit/test_gpx_video_sync.py index 619e13e6..9ab9a586 100644 --- a/tests/unit/test_gpx_video_sync.py +++ b/tests/unit/test_gpx_video_sync.py @@ -24,7 +24,11 @@ SourceType, VideoSyncOption, ) -from mapillary_tools.geotag.video_extractors.gpx import GPXVideoExtractor, VideoClock +from mapillary_tools.geotag.video_extractors.gpx import ( + GPXVideoExtractor, + SyncMode, + VideoClock, +) _DATA = Path(__file__).resolve().parent.parent.joinpath("data") @@ -89,6 +93,123 @@ def test_a_track_far_longer_than_the_video_is_refused(self, tmp_path): _extract(NO_GPS_VIDEO, gpx, VideoSyncOption()) assert "--video_start_time" in str(excinfo.value) + assert "--video_duration_ratio" in str(excinfo.value) + + def test_gpx_start_time_on_a_track_far_longer_than_the_video_is_refused( + self, tmp_path + ): + """ + It was not recorded for the video alone, or the video is a timelapse, + and nothing says how fast it runs through the track. + """ + gpx = _write_gpx(tmp_path / "track.gpx", seconds=60) + + with pytest.raises(exceptions.MapillaryGPXVideoStartUnknownError) as excinfo: + GPXVideoExtractor( + NO_GPS_VIDEO, gpx, SyncMode.REBASE, video_sync=VideoSyncOption() + ).extract() + + message = str(excinfo.value) + assert "--video_duration_ratio" in message + # For a logger left running, ready to paste: the track starts at GPX_START + assert ( + "started with the video and left running, specify --video_start_time " + "2025_03_14_07_00_00_000, when the track starts" + ) in message + # Not for a track of more than this video, which the start of the track + # would put on the wrong part of it + assert "covers more than this video, specify the time the video started" in ( + message + ) + + def test_gpx_start_time_on_a_log_of_days_offers_no_start_time(self, tmp_path): + """ + A track of several days is a log of several trips, and its start that + of the first: placed there, the video would land on the wrong trip. + """ + gpx = _write_gpx_tracks( + tmp_path / "track.gpx", + [ + ["2025-03-14T07:00:00Z", "2025-03-14T07:00:01Z"], + ["2025-03-16T07:00:00Z", "2025-03-16T07:00:01Z"], + ], + ) + + with pytest.raises(exceptions.MapillaryGPXVideoStartUnknownError) as excinfo: + GPXVideoExtractor( + NO_GPS_VIDEO, gpx, SyncMode.REBASE, video_sync=VideoSyncOption() + ).extract() + + message = str(excinfo.value) + assert "specify the time the video started as --video_start_time" in message + assert "2025_03_14" not in message + assert "left running" not in message + assert "--video_duration_ratio" not in message + + def test_gpx_start_time_on_a_track_as_long_as_the_video_is_silent( + self, tmp_path, caplog + ): + gpx = _write_gpx(tmp_path / "track.gpx", seconds=6) + + with caplog.at_level(logging.WARNING): + GPXVideoExtractor( + NO_GPS_VIDEO, gpx, SyncMode.REBASE, video_sync=VideoSyncOption() + ).extract() + + assert not caplog.text + + def test_the_duration_ratio_paces_a_track_placed_by_the_start_time(self, tmp_path): + """ + One GPX can serve many videos, each placed by its start time, and a + timelapse runs through it at its ratio. + """ + gpx = _write_gpx(tmp_path / "track.gpx", seconds=600) + video_start = GPX_START + datetime.timedelta(seconds=300) + + metadata = _extract( + NO_GPS_VIDEO, + gpx, + VideoSyncOption(start_time=video_start.timestamp(), duration_ratio=10), + ) + + assert metadata.points[300].time == pytest.approx(0) + assert metadata.points[310].time == pytest.approx(1) + # the absolute times, which the frames are stamped with, are untouched + assert metadata.points[310].get_unix_time() == pytest.approx( + video_start.timestamp() + 10 + ) + + @pytest.mark.parametrize("sync_mode", [SyncMode.SYNC, SyncMode.REBASE]) + def test_the_duration_ratio_spreads_a_track_over_the_video( + self, tmp_path, caplog, sync_mode + ): + """ + The 5.76s video spans 57.6s of GPS time at 10 seconds per second, so a + track of 57s starts and ends with it. + """ + gpx = _write_gpx(tmp_path / "track.gpx", seconds=57) + + with caplog.at_level(logging.WARNING): + metadata = GPXVideoExtractor( + NO_GPS_VIDEO, + gpx, + sync_mode, + video_sync=VideoSyncOption(duration_ratio=10), + ).extract() + + assert metadata.points[0].time == 0 + assert metadata.points[-1].time == pytest.approx(5.7) + assert not caplog.text + + def test_a_track_too_long_even_at_the_duration_ratio_is_refused(self, tmp_path): + gpx = _write_gpx(tmp_path / "track.gpx", seconds=600) + + with pytest.raises(exceptions.MapillaryGPXVideoStartUnknownError) as excinfo: + _extract(NO_GPS_VIDEO, gpx, VideoSyncOption(duration_ratio=10)) + + assert "which at --video_duration_ratio 10 span 57.6 seconds" in str( + excinfo.value + ) def test_a_track_a_little_longer_than_the_video_warns(self, tmp_path, caplog): gpx = _write_gpx(tmp_path / "track.gpx", seconds=9) @@ -110,13 +231,18 @@ def test_a_track_as_long_as_the_video_starts_with_it_silently( assert metadata.points[0].time == 0 assert not caplog.text - def test_process_still_assumes_the_track_starts_with_the_video(self, tmp_path): + @pytest.mark.parametrize("sync_mode", [SyncMode.SYNC, SyncMode.REBASE]) + def test_process_still_assumes_the_track_starts_with_the_video( + self, tmp_path, caplog, sync_mode + ): """process has no --video_start_time to offer as a way out.""" gpx = _write_gpx(tmp_path / "track.gpx", seconds=3600) - metadata = _extract(NO_GPS_VIDEO, gpx, video_sync=None) + with caplog.at_level(logging.WARNING): + metadata = GPXVideoExtractor(NO_GPS_VIDEO, gpx, sync_mode).extract() assert metadata.points[0].time == 0 + assert "timelapse" not in caplog.text def test_the_refusal_is_not_handed_on_to_the_next_source(self, tmp_path): """ @@ -162,6 +288,23 @@ def test_the_video_gps_wins_over_the_start_time(self, tmp_path): p.time for p in synced.points ] + def test_the_video_gps_clock_wins_over_the_duration_ratio(self, tmp_path, caplog): + """ + The ratio is for the videos without a GPS clock, which a batch of + videos may mix with videos that have one. + """ + gpx = _write_gpx( + tmp_path / "track.gpx", + seconds=30, + start=_GPS_VIDEO_START, + ) + + synced = _extract(GPS_VIDEO, gpx, video_sync=None) + with_ratio = _extract(GPS_VIDEO, gpx, VideoSyncOption(duration_ratio=10)) + + assert [p.time for p in with_ratio.points] == [p.time for p in synced.points] + assert "Ignoring --video_duration_ratio 10 for" in caplog.text + class TestInterpolationOptions: """ @@ -206,10 +349,10 @@ def test_the_offset_applies_to_a_track_synced_by_the_video_gps(self, tmp_path): def test_use_gpx_start_time_starts_the_track_with_the_video(self, tmp_path): """ - Whatever the start time says, and however long the track: the user - has said that the two started together. + Whatever the start time says: the user has said that the two started + together. """ - gpx = _write_gpx(tmp_path / "track.gpx", seconds=60) + gpx = _write_gpx(tmp_path / "track.gpx", seconds=6) metadata = _extract( NO_GPS_VIDEO, diff --git a/tests/unit/test_sample_video.py b/tests/unit/test_sample_video.py index 052a30cd..4f57557a 100644 --- a/tests/unit/test_sample_video.py +++ b/tests/unit/test_sample_video.py @@ -310,6 +310,65 @@ def test_the_duration_ratio_paces_a_video_without_a_gps_clock( ) +class TestGPXStartTimeByInterval: + """ + With --interpolation_use_gpx_start_time, the geotagging stage puts the + first frame of a video without a GPS clock at the start of the GPX, and + the others at --video_duration_ratio. A track that runs on long after the + video is refused, as in distance sampling. + """ + + # 5.76 seconds long, without GPS + VIDEO = _PWD.parent.joinpath("data", "videos", "sample-5s.mp4") + + @staticmethod + def _write_gpx(path: Path, seconds: int) -> Path: + start = datetime.datetime(2025, 3, 14, 7, 0, 0, tzinfo=datetime.timezone.utc) + trkpts = "".join( + f'" + for i in range(seconds + 1) + ) + path.write_text( + '' + f"{trkpts}" + ) + return path + + def test_a_track_far_longer_than_the_video_is_refused( + self, tmp_path, setup_mock, caplog + ): + video_dir = tmp_path / "videos" + video_dir.mkdir() + shutil.copyfile(self.VIDEO, video_dir / self.VIDEO.name) + gpx = self._write_gpx(tmp_path / "track.gpx", seconds=60) + + with pytest.raises(exceptions.MapillaryVideoSampleError): + sample_video.sample_video( + video_dir, + tmp_path / "samples", + video_sample_distance=-1, + video_sample_interval=2, + geotag_source=["gpx"], + geotag_source_path=gpx, + interpolation_use_gpx_start_time=True, + ) + + assert "--video_duration_ratio" in caplog.text + + def test_a_track_that_ends_with_the_video_at_the_ratio_passes(self, tmp_path): + """The video spans 57.6s of GPS time at 10 seconds per second.""" + gpx = self._write_gpx(tmp_path / "track.gpx", seconds=57) + video_sync = VideoSyncOption(use_gpx_start_time=True, duration_ratio=10) + options = sample_video._parse_geotag_options( + ["gpx"], gpx, None, tmp_path, video_sync + ) + + sample_video._check_gpx_start_by_interval(self.VIDEO, options, video_sync) + + class TestGPSClockStartTime: """Tests for _gps_clock_start_time.""" @@ -908,6 +967,17 @@ def test_negative_distance_raises(self, tmp_path: Path) -> None: ) +@pytest.mark.parametrize("ratio", [0, -1, float("nan")]) +def test_a_duration_ratio_must_be_positive(tmp_path: Path, ratio: float) -> None: + """It divides the GPS time of a video without a GPS clock.""" + with pytest.raises(exceptions.MapillaryBadParameterError): + sample_video.sample_video( + _PWD.joinpath("data/mock_sample_video/videos/hello.mp4"), + tmp_path, + video_duration_ratio=ratio, + ) + + class TestSampleVideoRerun: """Test rerun behavior of sample_video.""" From 22852b1d57433811a0d65ee5a4b289ab230838b8 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Wed, 21 Oct 2026 16:57:22 +0200 Subject: [PATCH 4/5] Time a timelapse's track and interval samples by its GPS timestamps The previous commit timed the GPS points of a timelapse at one rate from its first timestamp. Distance samples are stamped with the GPS time of their video time instead, and the first timestamp of a GoPro MAX timelapse can be about 3 s early: in one, the camera moved 6 m at 8 m/s in the 3.9 s to the next one. On two such timelapses (30x and 15x), the frames that geotagged were placed a median of 36 m and 5 m from where the sampler picked them, and the last 2 and 1 frames were stamped past the end of the track: The image time is 1.801 seconds beyond the GPX end point Time the track by the timestamps, interpolated between them and extrapolated beyond them at the clock's rate. Stamp interval samples of a timelapse the same way, from the GPS time at video time 0, as distance samples are. On those two timelapses, distance samples now all geotag where they were picked. Interval samples (0.5 s) stay where they were with --geotag_source gopro, and with a GPX of the same track they move from a median of 0.6 m and 0.04 m (4.4 m and 1.4 m at most) to within 1 cm. Videos without a timelapse clock are unchanged. --- .../geotag/geotag_images_from_video.py | 76 ++++++++++++++++++- mapillary_tools/sample_video.py | 47 +++++++----- tests/unit/test_geotag_images_from_video.py | 53 +++++++++++-- tests/unit/test_sample_video.py | 40 +++++++--- 4 files changed, 181 insertions(+), 35 deletions(-) diff --git a/mapillary_tools/geotag/geotag_images_from_video.py b/mapillary_tools/geotag/geotag_images_from_video.py index b4984d3b..0adbd7fd 100644 --- a/mapillary_tools/geotag/geotag_images_from_video.py +++ b/mapillary_tools/geotag/geotag_images_from_video.py @@ -5,6 +5,7 @@ from __future__ import annotations +import bisect import dataclasses import logging import sys @@ -26,6 +27,68 @@ LOG = logging.getLogger(__name__) +@dataclasses.dataclass(frozen=True) +class GPSTimeline: + """ + The Unix time of the video's GPS clock along the video: interpolated + between its GPS timestamps, and extrapolated beyond them at the clock's + rate. A timelapse captured its frames at those times, which one rate from + one of them does not follow: the rate drifts from them over the video, and + the first timestamp of a GoPro MAX timelapse can be seconds early. + + >>> timeline = GPSTimeline((0.0, 1.0, 2.0), (100.0, 104.0, 114.0), rate=10.0) + >>> timeline.to_unix_time(0.5) + 102.0 + >>> timeline.to_unix_time(3.0) + 124.0 + """ + + video_times: tuple[float, ...] + unix_times: tuple[float, ...] + # GPS seconds per video second beyond the timestamps + rate: float + + @classmethod + def read(cls, points: T.Sequence[geo.Point], rate: float) -> GPSTimeline | None: + """ + Read the timeline from the GPS timestamps of the points, or None if they + have none + """ + pairs: list[tuple[float, float]] = [] + for p in points: + unix_time = p.get_unix_time() + if unix_time is None: + continue + if pairs and unix_time == pairs[-1][1]: + # GoPro GPS5 stamps a whole block (~1s) of points with one + # timestamp. Keep the first point of the block + continue + pairs.append((p.time, unix_time)) + + if not pairs: + return None + + return cls( + video_times=tuple(v for v, _ in pairs), + unix_times=tuple(u for _, u in pairs), + rate=rate, + ) + + def to_unix_time(self, video_time: float) -> float: + idx = bisect.bisect_right(self.video_times, video_time) + if 0 < idx < len(self.video_times): + v0, v1 = self.video_times[idx - 1], self.video_times[idx] + u0, u1 = self.unix_times[idx - 1], self.unix_times[idx] + return u0 + (u1 - u0) * (video_time - v0) / (v1 - v0) + + # Before the first timestamp or after the last + anchor = 0 if idx == 0 else -1 + return ( + self.unix_times[anchor] + + (video_time - self.video_times[anchor]) * self.rate + ) + + class GeotagImagesFromVideo(GeotagImagesFromGeneric): def __init__( self, @@ -106,14 +169,21 @@ def _on_capture_timeline( ) -> T.Sequence[geo.Point]: """ Time the video's GPS points in GPS seconds from the start of the video, - like the samples, which are stamped with their capture times. Each - second of a timelapse spans as many as its GPS clock runs faster. + like the samples, which are stamped with their capture times. A + timelapse packs several into each second of video, as many as its GPS + timestamps tell. """ clock = GPXVideoExtractor._read_video_clock(points) if clock is None or clock.rate == 1.0: return points - return [dataclasses.replace(p, time=p.time * clock.rate) for p in points] + timeline = GPSTimeline.read(points, clock.rate) + assert timeline is not None, "a GPS clock is read from GPS timestamps" + start = timeline.to_unix_time(0.0) + return [ + dataclasses.replace(p, time=timeline.to_unix_time(p.time) - start) + for p in points + ] class GeotagImageSamplesFromVideo(GeotagImagesFromGeneric): diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py index 4ad8cd9d..16308fbb 100644 --- a/mapillary_tools/sample_video.py +++ b/mapillary_tools/sample_video.py @@ -29,6 +29,7 @@ ) from .exif_write import ExifEdit from .geotag import factory +from .geotag.geotag_images_from_video import GPSTimeline from .geotag.options import SourceOption, SourceType, VideoSyncOption from .geotag.video_extractors.gpx import GPXVideoExtractor from .geotag.video_extractors.native import NativeVideoExtractor @@ -390,15 +391,14 @@ def wip_sample_dir(sample_dir: Path) -> Path: def _gps_clock_start_time( - points: T.Sequence[geo.Point], rate: float = 1.0 + points: T.Sequence[geo.Point], ) -> datetime.datetime | None: """ Map the absolute GPS timestamps at the start of a track back to the video's time 0. - Point times are relative to the start of the video, so subtracting one, in - GPS seconds, from its own absolute timestamp gives the wall clock at which - the video started. Each second of video spans **rate** GPS seconds, more - than 1 in a timelapse. Timestamps outside the plausible range are skipped. + Point times are relative to the start of the video, so subtracting one from + its own absolute timestamp gives the wall clock at which the video started. + Timestamps outside the plausible range are skipped. """ max_start_time = time.time() + _MAX_FUTURE_START_TIME_SECONDS @@ -407,7 +407,7 @@ def _gps_clock_start_time( unix_time = point.get_unix_time() if unix_time is None: continue - start_time = unix_time - point.time * rate + start_time = unix_time - point.time # Written as a negated range check so that NaN is skipped too if not (_MIN_PLAUSIBLE_START_TIME <= start_time <= max_start_time): continue @@ -574,7 +574,6 @@ def _extract_video_start_time( video_path: Path, probe: ffmpeglib.Probe, telemetry_points: T.Sequence[geo.Point], - rate: float, ) -> datetime.datetime | None: """ Determine the wall clock time at which a video started recording. @@ -585,7 +584,7 @@ def _extract_video_start_time( back to the creation time when there is no telemetry to sync against, which is the case for the plain MP4s that get geotagged from a GPX. """ - start_time = _gps_clock_start_time(telemetry_points, rate) + start_time = _gps_clock_start_time(telemetry_points) if start_time is not None: return start_time @@ -630,9 +629,11 @@ def _sample_single_video_by_interval( telemetry_points = _read_telemetry(video_path) # Frames are stamped with their capture times, which a timelapse packs - # several seconds of into each second of video: as many as the video's own - # GPS clock says, and for a video without one, --video_duration_ratio + # several seconds of into each second of video: as the video's own GPS + # timestamps tell, and for a video without a GPS clock, at + # --video_duration_ratio clock_rate = _telemetry_clock_rate(video_path, telemetry_points) + timeline: GPSTimeline | None = None if clock_rate is None: rate = duration_ratio else: @@ -644,17 +645,25 @@ def _sample_single_video_by_interval( video_path.name, clock_rate, ) + if clock_rate != 1.0: + # As distance samples are stamped, and as the geotagging stage + # times the video's track to place the samples along it + timeline = GPSTimeline.read(telemetry_points, clock_rate) if clock_rate is None and video_sync is not None and video_sync.use_gpx_start_time: _check_gpx_start_by_interval(video_path, geotag_options, video_sync) if start_time is None: - start_time = _extract_video_start_time( - video_path, - ffmpeglib.Probe(ffmpeg.probe_format_and_streams(video_path)), - telemetry_points, - rate, - ) + if timeline is not None: + start_time = datetime.datetime.fromtimestamp( + timeline.to_unix_time(0.0), tz=datetime.timezone.utc + ) + else: + start_time = _extract_video_start_time( + video_path, + ffmpeglib.Probe(ffmpeg.probe_format_and_streams(video_path)), + telemetry_points, + ) if start_time is None: raise exceptions.MapillaryVideoError( f"Unable to extract video start time from {video_path}" @@ -668,7 +677,11 @@ def _sample_single_video_by_interval( if sample_paths[0] is None: continue # extract_frames() produces 1-based frame indices so we need to subtract 1 here - seconds = (frame_idx_1based - 1) * sample_interval * rate + video_time = (frame_idx_1based - 1) * sample_interval + if timeline is None: + seconds = video_time * rate + else: + seconds = timeline.to_unix_time(video_time) - timeline.to_unix_time(0.0) timestamp = start_time + datetime.timedelta(seconds=seconds) exif_edit = ExifEdit(sample_paths[0]) exif_edit.add_date_time_original(timestamp) diff --git a/tests/unit/test_geotag_images_from_video.py b/tests/unit/test_geotag_images_from_video.py index 27e5c948..9664ed46 100644 --- a/tests/unit/test_geotag_images_from_video.py +++ b/tests/unit/test_geotag_images_from_video.py @@ -21,8 +21,10 @@ START_TIME = datetime.datetime(2021, 8, 10, 6, 38, 6, tzinfo=datetime.timezone.utc) -def _video_metadata(video_path: Path, rate: float) -> types.VideoMetadata: - # One point per second of video, recorded **rate** seconds apart +def _video_metadata( + video_path: Path, gps_seconds: T.Sequence[float] +) -> types.VideoMetadata: + # One point per second of video, at gps_seconds[i] from START_TIME points = [ telemetry.GPSPoint( time=float(i), @@ -30,12 +32,12 @@ def _video_metadata(video_path: Path, rate: float) -> types.VideoMetadata: lon=-74.0, alt=None, angle=None, - epoch_time=START_TIME.timestamp() + rate * i, + epoch_time=START_TIME.timestamp() + seconds, fix=None, precision=None, ground_speed=None, ) - for i in range(10) + for i, seconds in enumerate(gps_seconds) ] return types.VideoMetadata( filename=video_path, @@ -70,7 +72,7 @@ def test_samples_are_placed_by_their_capture_times(tmp_path: Path, rate: float): ] metadatas = GeotagImagesFromVideo( - [_video_metadata(video_path, rate)], num_processes=0 + [_video_metadata(video_path, [rate * i for i in range(10)])], num_processes=0 ).to_description(samples) lat_by_name = {} @@ -81,3 +83,44 @@ def test_samples_are_placed_by_their_capture_times(tmp_path: Path, rate: float): sample.name: pytest.approx(40.0 + 2 * k * 0.0001) for k, sample in enumerate(samples) } + + +def test_timelapse_samples_are_placed_by_the_gps_timestamps(tmp_path: Path): + """ + Distance samples are stamped with the GPS time of their video time. A + GoPro MAX timelapse can stamp its first point seconds early, and its clock + keeps no one rate, so timed at one rate the track placed the samples off, + and the last ones past its end. + """ + video_path = tmp_path / "hello.mp4" + sample_dir = tmp_path / "samples" / "hello.mp4" + sample_dir.mkdir(parents=True) + + # 10x, but the first timestamp 3 seconds early and the clock speeding up + gps_seconds = [0.0] + [3 + 10 * i + 0.1 * i * i for i in range(1, 10)] + + def stamped(video_time: float) -> datetime.datetime: + i = int(video_time) + seconds = gps_seconds[i] + if i + 1 < len(gps_seconds): + seconds += (gps_seconds[i + 1] - gps_seconds[i]) * (video_time - i) + return START_TIME + datetime.timedelta(seconds=seconds) + + video_times = [0.0, 0.5, 2.0, 4.5, 8.0, 9.0] + samples = [ + _sample(sample_dir, k + 1, stamped(video_time)) + for k, video_time in enumerate(video_times) + ] + + metadatas = GeotagImagesFromVideo( + [_video_metadata(video_path, gps_seconds)], num_processes=0 + ).to_description(samples) + + lat_by_name = {} + for metadata in metadatas: + assert isinstance(metadata, types.ImageMetadata), metadata + lat_by_name[metadata.filename.name] = metadata.lat + assert lat_by_name == { + sample.name: pytest.approx(40.0 + video_time * 0.0001) + for sample, video_time in zip(samples, video_times) + } diff --git a/tests/unit/test_sample_video.py b/tests/unit/test_sample_video.py index 4f57557a..ac432973 100644 --- a/tests/unit/test_sample_video.py +++ b/tests/unit/test_sample_video.py @@ -225,9 +225,9 @@ class TestSampleTimelapseByInterval: 2021, 8, 10, 6, 38, 6, tzinfo=datetime.timezone.utc ) - @pytest.fixture - def timelapse_telemetry(self, monkeypatch): - # One point per second of video, recorded 10 seconds apart + @classmethod + def _mock_telemetry(cls, monkeypatch, gps_seconds: T.Callable[[int], float]): + # One point per second of video, at gps_seconds(i) from GPS_START_TIME points = [ telemetry.GPSPoint( time=float(i), @@ -235,7 +235,7 @@ def timelapse_telemetry(self, monkeypatch): lon=-74.0, alt=None, angle=None, - epoch_time=self.GPS_START_TIME.timestamp() + 10 * i, + epoch_time=cls.GPS_START_TIME.timestamp() + gps_seconds(i), fix=None, precision=None, ground_speed=None, @@ -254,6 +254,11 @@ def timelapse_telemetry(self, monkeypatch): ), ) + @pytest.fixture + def timelapse_telemetry(self, monkeypatch): + # Recorded 10 seconds apart + self._mock_telemetry(monkeypatch, lambda i: 10 * i) + def _sample(self, tmpdir: py.path.local, **kwargs) -> list[Path]: root = _PWD.joinpath("data/mock_sample_video") sample_dir = tmpdir.mkdir("sampled_video_frames") @@ -275,6 +280,27 @@ def test_capture_times_follow_the_gps_clock( ) assert "--video_duration_ratio" not in caplog.text + def test_capture_times_follow_the_gps_timestamps( + self, tmpdir, setup_mock, monkeypatch + ): + """ + Not one rate: the GPS clock of a timelapse drifts from it, and distance + samples are stamped from the timestamps. The geotagging stage places + both kinds of samples along a track timed by them + """ + self._mock_telemetry(monkeypatch, lambda i: 10 * i + 0.1 * i * i) + + samples = sorted(self._sample(tmpdir)) + + assert len(samples) > 2 + for idx, sample in enumerate(samples): + video_time = 2 * idx + assert exif_read.ExifRead( + sample + ).extract_capture_time() == self.GPS_START_TIME + datetime.timedelta( + seconds=10 * video_time + 0.1 * video_time * video_time + ) + def test_from_the_given_start_time(self, tmpdir, setup_mock, timelapse_telemetry): video_start_time = "2020_08_10_14_37_05_023" _validate_interval( @@ -452,12 +478,6 @@ def test_uses_the_start_of_a_timelapse(self) -> None: 2021, 8, 10, 12, 38, 24, tzinfo=datetime.timezone.utc ) - def test_maps_a_timelapse_back_at_its_rate(self) -> None: - points = [self._gps_point(float(i), 1628599086.0 + 10 * i) for i in range(1000)] - assert sample_video._gps_clock_start_time(points, rate=10) == datetime.datetime( - 2021, 8, 10, 12, 38, 6, tzinfo=datetime.timezone.utc - ) - def test_no_absolute_timestamps(self) -> None: assert sample_video._gps_clock_start_time(_make_gps_points(3)) is None From 726482b97e8503b3bbc4220fe435b08a1038991a Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Wed, 21 Oct 2026 18:09:00 +0200 Subject: [PATCH 5/5] Check a GPX by interval only for a video GPX sync reads no clock from The previous commit checks a GPX that --interpolation_use_gpx_start_time says started with a video without a GPS clock. Interval sampling reads the clock from the video's telemetry, which is empty when its GPS is rejected as noise. GPX sync reads the clock from that GPS all the same, and syncs the GPX at that clock's pace, as in process and in distance sampling. So a 1.24 s GoPro MAX 2 clip with noisy GPS and a 12x clock, which main samples with its 12 s GPX, was refused: The GPX track in GS018205.360.gpx lasts 12 seconds and the video only 1.24 seconds, so it runs on long after the video ends. Skip the check when GPX sync reads a clock from the video. The clip samples as on main again. --- mapillary_tools/sample_video.py | 19 +++++++++++++++ tests/unit/test_sample_video.py | 41 +++++++++++++++++++++++++++++++++ 2 files changed, 60 insertions(+) diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py index 16308fbb..1de92705 100644 --- a/mapillary_tools/sample_video.py +++ b/mapillary_tools/sample_video.py @@ -610,11 +610,30 @@ def _check_gpx_start_by_interval( if not gpx_path.is_file(): continue extractor = GPXVideoExtractor(video_path, gpx_path, video_sync=video_sync) + if _gpx_sync_reads_a_clock(extractor): + continue gpx_points = extractor._read_gpx_points() if gpx_points: extractor._check_given_start(gpx_points, video_sync.duration_ratio) +def _gpx_sync_reads_a_clock(extractor: GPXVideoExtractor) -> bool: + """ + Whether GPX sync, in distance sampling and process, reads a GPS clock from + the video. It reads one from GPS rejected as noise too, which interval + sampling has no telemetry from, and syncs the GPX at that clock's pace + """ + try: + native_video_metadata, _ = extractor._extract_native_video_metadata() + except exceptions.MapillaryVideoGPSUnusableError: + return False + except Exception: + # Telemetry that fails to parse is warned about by _read_telemetry(), + # and must not stop the video from being sampled + return False + return extractor._read_video_clock(native_video_metadata.points) is not None + + def _sample_single_video_by_interval( video_path: Path, sample_dir: Path, diff --git a/tests/unit/test_sample_video.py b/tests/unit/test_sample_video.py index ac432973..884be856 100644 --- a/tests/unit/test_sample_video.py +++ b/tests/unit/test_sample_video.py @@ -394,6 +394,47 @@ def test_a_track_that_ends_with_the_video_at_the_ratio_passes(self, tmp_path): sample_video._check_gpx_start_by_interval(self.VIDEO, options, video_sync) + def test_gps_rejected_as_noise_still_has_a_clock(self, tmp_path, monkeypatch): + """ + Interval sampling reads no telemetry from GPS rejected as noise, but + GPX sync reads its clock, and syncs the GPX at that clock's pace, as + for any video with a GPS clock + """ + # A 12x timelapse clock + points = [ + telemetry.GPSPoint( + time=i * 0.1, + lat=37.0, + lon=-122.0, + alt=None, + angle=None, + epoch_time=1741935600.0 + i * 1.2, + fix=None, + precision=None, + ground_speed=None, + ) + for i in range(30) + ] + monkeypatch.setattr( + sample_video.GPXVideoExtractor, + "_extract_native_video_metadata", + lambda _self: ( + VideoMetadata( + filename=self.VIDEO, + filetype=FileType.GOPRO, + points=T.cast(T.List[geo.Point], points), + ), + True, + ), + ) + gpx = self._write_gpx(tmp_path / "track.gpx", seconds=60) + video_sync = VideoSyncOption(use_gpx_start_time=True) + options = sample_video._parse_geotag_options( + ["gpx"], gpx, None, tmp_path, video_sync + ) + + sample_video._check_gpx_start_by_interval(self.VIDEO, options, video_sync) + class TestGPSClockStartTime: """Tests for _gps_clock_start_time."""