From 484d281c06a859fc5aea5acf169da0aabdc82e57 Mon Sep 17 00:00:00 2001
From: Caglar Pir <caglarp@meta.com>
Date: Tue, 20 Oct 2026 14:01:22 +0200
Subject: [PATCH 1/3] 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 c20d9f1a3..b4984d3b9 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 e945d2526..9636c6bbd 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 000000000..27e5c9486
--- /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 cf4760b73..052a30cda 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 cf7d178e080f14edec4e071557943c2e8248b1a9 Mon Sep 17 00:00:00 2001
From: Caglar Pir <caglarp@meta.com>
Date: Wed, 21 Oct 2026 16:57:22 +0200
Subject: [PATCH 2/3] 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 b4984d3b9..0adbd7fd9 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 9636c6bbd..d555f6844 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
@@ -381,15 +382,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
 
@@ -398,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 * 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
@@ -565,7 +565,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.
@@ -576,7 +575,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
 
@@ -595,9 +594,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:
@@ -609,14 +610,22 @@ 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 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}"
@@ -630,7 +639,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 27e5c9486..9664ed464 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 052a30cda..fd646f3e9 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(
@@ -393,12 +419,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 a8c31474e3463fbe9b29491d1c7702404f972a08 Mon Sep 17 00:00:00 2001
From: Caglar Pir <caglarp@meta.com>
Date: Tue, 20 Oct 2026 14:06:17 +0200
Subject: [PATCH 3/3] 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 75d4aa168..60f56a389 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 73a6581a8..7ad03592e 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:
     """