From 484d281c06a859fc5aea5acf169da0aabdc82e57 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Tue, 20 Oct 2026 14:01:22 +0200 Subject: [PATCH 01/12] 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 Date: Wed, 21 Oct 2026 16:57:22 +0200 Subject: [PATCH 02/12] 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 Date: Tue, 20 Oct 2026 14:06:17 +0200 Subject: [PATCH 03/12] 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: """ From e7ccf01a8162a516aec8fd14b7b5d6091aa758be Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Tue, 20 Oct 2026 14:17:53 +0200 Subject: [PATCH 04/12] 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 c7dc3e922..12be76714 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 4c400f309..dac650efc 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 72f635f3c..108d59d52 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 60f56a389..17e0764d1 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 d555f6844..16308fbbc 100644 --- a/mapillary_tools/sample_video.py +++ b/mapillary_tools/sample_video.py @@ -159,6 +159,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: @@ -177,6 +183,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( @@ -246,6 +253,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 @@ -582,12 +591,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) @@ -615,6 +650,9 @@ def _sample_single_video_by_interval( # 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: if timeline is not None: start_time = datetime.datetime.fromtimestamp( diff --git a/tests/integration/test_video_process.py b/tests/integration/test_video_process.py index bdd26788b..cad2b6b4a 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 619e13e60..9ab9a5861 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 fd646f3e9..ac432973b 100644 --- a/tests/unit/test_sample_video.py +++ b/tests/unit/test_sample_video.py @@ -336,6 +336,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.""" @@ -928,6 +987,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 049fc4c8a7602f7037f51dab2c2a153a59ef7d16 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Wed, 21 Oct 2026 18:09:00 +0200 Subject: [PATCH 05/12] 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 16308fbbc..1de927054 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 ac432973b..884be8569 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.""" From a02168e1ff921e0b42299b6afdf7085866dc59fb Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Tue, 20 Oct 2026 16:00:39 +0200 Subject: [PATCH 06/12] Update the README on geotagging videos with a GPX It said that a GPX works with interval sampling only, which has not been true since distance sampling honours --geotag_source, and that the start time of a video is always read from its GPS or its creation time, which distance sampling does not do. It now says how each kind of video is placed along a GPX, that --video_start_time applies to every video in a command, so one GPX serves several videos run one at a time, how --video_duration_ratio paces a timelapse without a GPS clock, and what to pass instead of --interpolation_use_gpx_start_time for a GPS logger left running. The examples sample one video, and the --interpolation_use_gpx_start_time one drops the interval sampling options it no longer needs. --- README.md | 42 +++++++++++++++++++++++++++++------------- 1 file changed, 29 insertions(+), 13 deletions(-) diff --git a/README.md b/README.md index 655c1c91a..93ea96218 100644 --- a/README.md +++ b/README.md @@ -333,20 +333,34 @@ It is used to locate the images along the GPS tracks. mapillary_tools process MY_IMAGE_DIR --geotag_source "gpx" --geotag_source_path MY_EXTERNAL_GPS.gpx ``` -To geotag videos with a GPX file, video start time is required to locate the sample images along the GPS tracks. -It is read from the video's own GPS track when it has one, and otherwise from the video creation time. -Cameras disagree on whether the creation time marks the start or the end of the recording (most dashcams write the end), -so mapillary_tools uses the camera model or a date and time in the file name to tell which, and assumes the start when neither says. -Use `--video_start_time` to override it, in UTC, when the sample images end up one video duration off along the track. +To geotag videos with a GPX file, the sample images are located along the GPS track by the time they were captured. +A video whose own GPS records the time is synced to the track by that clock. For a video without one, specify its start +time, in UTC, with `--video_start_time`. That start time applies to every video in the command, so to place several +videos along one GPX file, run it once per video, each with its own start time: ```sh -# Geotagging with GPX works with interval-based sampling only, -# the two options --video_sample_distance -1 --video_sample_interval 2 are therefore required -# to switch from the default distance-based sampling to the legacy interval-based sampling -mapillary_tools video_process MY_VIDEO_DIR \ +mapillary_tools video_process MY_VIDEO.mp4 \ --geotag_source "gpx" \ --geotag_source_path MY_EXTERNAL_GPS.gpx \ - --video_sample_distance -1 --video_sample_interval 2 + --video_start_time 2025_03_14_07_00_30_000 +``` + +Without it, distance-based sampling (the default) assumes that the track starts with the video, and refuses a track +more than twice as long as the video. Interval-based sampling (`--video_sample_distance -1 --video_sample_interval N`) +reads the start time from the video creation time instead. Cameras disagree on whether the creation time marks the start +or the end of the recording (most dashcams write the end), so mapillary_tools uses the camera model or a date and time +in the file name to tell which, and assumes the start when neither says. + +A timelapse video packs several seconds of capture into each second of video. Its GPS clock, if it has one, tells how +many. If it has none, specify how many with `--video_duration_ratio`, for instance 60 for a timelapse that took a frame +every 2 seconds and plays at 30 frames per second: + +```sh +mapillary_tools video_process MY_TIMELAPSE.mp4 \ + --geotag_source "gpx" \ + --geotag_source_path MY_EXTERNAL_GPS.gpx \ + --video_start_time 2025_03_14_07_00_30_000 \ + --video_duration_ratio 60 ``` Ideally, the GPS device and the capture device should use the same clock to get the timestamps synchronized. @@ -363,15 +377,17 @@ mapillary_tools process MY_IMAGE_DIR \ Another option `--interpolation_use_gpx_start_time` moves your images to align with the beginning of the GPS track. This is useful when you can confirm that you start GPS recording and capturing at the same time, or with a known delay. +For a video without a GPS clock, a track that lasts more than twice as long as the video spans at its +`--video_duration_ratio` is then refused. If the GPS logger was left running after the video, pass the start of the +track as `--video_start_time` instead, which places the video the same way: the error prints it. ```sh # Start capturing 2.5 seconds after start GPS recording -mapillary_tools video_process MY_VIDEO_DIR \ +mapillary_tools video_process MY_VIDEO.mp4 \ --geotag_source "gpx" \ --geotag_source_path MY_EXTERNAL_GPS.gpx \ --interpolation_use_gpx_start_time \ - --interpolation_offset_time 2.5 \ - --video_sample_distance -1 --video_sample_interval 2 + --interpolation_offset_time 2.5 ``` ## New video geotagging features (experimental) From c5bce1966b6c280f7bd30a0352ec82deb5a4e579 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Tue, 20 Oct 2026 15:21:53 +0200 Subject: [PATCH 07/12] Sample a single video stream by interval, as by distance Interval sampling mapped every video stream of a video to one output, and ffmpeg numbers the frames of all of them in one sequence. A GoPro MAX .360 video has two video streams, so its frames came out interleaved, and frame i, taken at (i - 1) // 2 intervals, was stamped (i - 1) intervals into the video. Half of them were stamped past its end. On a 5-minute .360 video sampled every second, the frames that could be geotagged were placed a median of 73 m from where they were taken, and the other half failed. Interval sampling now samples the video stream with the highest resolution, as distance sampling does. A video with a single video stream, which is nearly every video, is sampled as before and its samples keep their names. --- mapillary_tools/sample_video.py | 36 ++++++++++++++++++++++++++------- tests/unit/test_sample_video.py | 34 +++++++++++++++++++++++++++++++ 2 files changed, 63 insertions(+), 7 deletions(-) diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py index 1de927054..d5e15ff2e 100644 --- a/mapillary_tools/sample_video.py +++ b/mapillary_tools/sample_video.py @@ -634,6 +634,24 @@ def _gpx_sync_reads_a_clock(extractor: GPXVideoExtractor) -> bool: return extractor._read_video_clock(native_video_metadata.points) is not None +def _interval_stream_specifier(probe: ffmpeglib.Probe) -> str: + """ + The video stream to sample by interval: the one with the highest + resolution, as for sampling by distance. ffmpeg numbers the frames of all + the streams it samples in one sequence, so with two (a GoPro MAX .360 video + has two) each frame would be numbered, and stamped, as if twice as late. + A video with a single video stream is sampled as "v", which names its + samples as it always has. + """ + video_streams = probe.probe_video_streams() + if len(video_streams) <= 1: + return "v" + + video_stream = probe.probe_video_with_max_resolution() + assert video_stream is not None + return str(video_stream["index"]) + + def _sample_single_video_by_interval( video_path: Path, sample_dir: Path, @@ -672,25 +690,29 @@ def _sample_single_video_by_interval( 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) + probe = ffmpeglib.Probe(ffmpeg.probe_format_and_streams(video_path)) + if start_time is None: 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, - ) + start_time = _extract_video_start_time(video_path, probe, telemetry_points) if start_time is None: raise exceptions.MapillaryVideoError( f"Unable to extract video start time from {video_path}" ) + stream_specifier = _interval_stream_specifier(probe) + with wip_dir_context(wip_sample_dir(sample_dir), sample_dir) as wip_dir: - ffmpeg.extract_frames_by_interval(video_path, wip_dir, sample_interval) - frame_samples = ffmpeglib.FFMPEG.sort_selected_samples(wip_dir, video_path) + ffmpeg.extract_frames_by_interval( + video_path, wip_dir, sample_interval, stream_specifier=stream_specifier + ) + frame_samples = ffmpeglib.FFMPEG.sort_selected_samples( + wip_dir, video_path, selected_stream_specifiers=[stream_specifier] + ) for frame_idx_1based, sample_paths in frame_samples: assert len(sample_paths) == 1 if sample_paths[0] is None: diff --git a/tests/unit/test_sample_video.py b/tests/unit/test_sample_video.py index 884be8569..a4cf6d476 100644 --- a/tests/unit/test_sample_video.py +++ b/tests/unit/test_sample_video.py @@ -135,6 +135,40 @@ def test_sample_video_with_start_time(tmpdir: py.path.local, setup_mock): _validate_interval([Path(s) for s in samples], video_start_time) +def test_sample_the_video_stream_with_the_highest_resolution( + tmpdir: py.path.local, setup_mock +): + """ + By interval as by distance: ffmpeg would number the frames of two streams + in one sequence, and they would be stamped as if twice as late. + """ + root = _PWD.joinpath("data/mock_sample_video") + probe = json.loads(root.joinpath("videos", "hello.mp4").read_text()) + [video_stream] = [s for s in probe["streams"] if s["codec_type"] == "video"] + probe["streams"].append({**video_stream, "index": 2, "width": 5760}) + video_dir = Path(tmpdir.mkdir("videos")) + video_dir.joinpath("hello.mp4").write_text(json.dumps(probe)) + sample_dir = tmpdir.mkdir("sampled_video_frames") + + sample_video.sample_video( + video_dir, + Path(sample_dir), + video_sample_distance=-1, + video_sample_interval=2, + rerun=True, + ) + + samples = sorted(Path(s) for s in sample_dir.join("hello.mp4").listdir()) + assert len(samples), "expect samples but got none" + for idx, sample in enumerate(samples): + assert sample.name == f"hello_2_{idx + 1:06d}.jpg" + assert exif_read.ExifRead( + sample + ).extract_capture_time() == PROBE_START_TIME + datetime.timedelta( + seconds=2 * idx + ) + + def test_sample_video_from_gps_clock(tmpdir: py.path.local, setup_mock, monkeypatch): """A video's own GPS clock wins over the container's creation time.""" root = _PWD.joinpath("data/mock_sample_video") From 04aee80493b534f78dacff232d786e3cb8219283 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Tue, 20 Oct 2026 16:41:00 +0200 Subject: [PATCH 08/12] Judge a noisy GPS clock by the video's creation time, not by the GPX When a video's GPS is rejected as noise, the GPX is still synced by the clock of its unfiltered points, unless that clock is more than a day from the GPX: GoPro stamps a placeholder date (2015-10-22 and similar) until it acquires GPS time. Judged against the GPX, though, the clock took the blame for a GPX that does not belong to the video. A GoPro HERO8 clip whose noisy clock is real, given a GPX from another day, had that track placed from its first frame, and process exited 0, while the same GPX on a video with good GPS is rejected for missing it. A noisy clock that starts within a day of the mvhd creation time of the video is now taken as real, and places the GPX as for any video with GPS, so a GPX that misses the video raises MapillaryOutsideGPXTrackError. A day absorbs a creation time written in local time (GoPro) and whether it marks the start or the end of the recording. A clock is still a placeholder only when it is more than a day from the GPX too: some cameras write a creation time that means nothing next to a real GPS clock (a GoPro HERO10 recorded in 2022 reports 2016-01-01), and a GPX that agrees with the clock keeps it, as before. Without a readable creation time, nothing changes. The creation time is read by the same function that decides the CAMM GPS epoch. --- .../geotag/video_extractors/gpx.py | 49 +++++++++-- tests/unit/test_gpx_over_noisy_gps.py | 87 ++++++++++++++++++- 2 files changed, 127 insertions(+), 9 deletions(-) diff --git a/mapillary_tools/geotag/video_extractors/gpx.py b/mapillary_tools/geotag/video_extractors/gpx.py index 17e0764d1..61701affa 100644 --- a/mapillary_tools/geotag/video_extractors/gpx.py +++ b/mapillary_tools/geotag/video_extractors/gpx.py @@ -23,7 +23,9 @@ from typing_extensions import override from ... import exceptions, geo, telemetry, types, utils +from ...camm import camm_parser from ...mp4 import construct_mp4_parser as cparser, simple_mp4_parser as sparser +from ...mp4.mp4_sample_parser import MovieBoxParser from ...serializer.description import build_capture_time from .. import options from ..utils import parse_gpx @@ -149,11 +151,18 @@ def extract(self) -> types.VideoMetadata: 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: - # GPS rejected as noise may not have GPS time yet: GoPro - # stamps a placeholder date until it acquires it. That GPS - # is what the GPX replaces, so do not let its clock place - # the GPX days away from the video + # GPS rejected as noise may not have GPS time yet: GoPro + # stamps a placeholder date until it acquires it. That GPS is + # what the GPX replaces, so do not let a placeholder clock + # place the GPX days away from the video. A clock that agrees + # with the creation time of the video is no placeholder: it + # places the GPX as for any video with GPS, and a GPX days + # away from it misses the video + if ( + noisy + and self.sync_mode is SyncMode.SYNC + and not self._clock_matches_creation_time(clock) + ): LOG.warning( "The GPS clock of %s, whose GPS was rejected as noise, is %.1f days away from %s. " "Syncing the GPX from the start of the video instead", @@ -485,6 +494,23 @@ def _durations( 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 _clock_matches_creation_time(self, clock: VideoClock) -> bool: + """ + Whether the GPS clock of the video starts within a day of the mvhd + creation time of the video, or False if that cannot be read. + + A day absorbs any time zone the creation time is written in (GoPro + writes local time), and whether it marks the start or the end of the + recording. Some cameras write a creation time that means nothing (a + GoPro HERO10 recorded in 2022 reports 2016-01-01), so a clock that does + not match it may still be real. + """ + creation_time = _video_creation_time(self.video_path) + if creation_time is None: + return False + gap = abs(clock.to_unix_time(0) - creation_time) + return gap <= _IMPLAUSIBLE_OFFSET_SECONDS + def _extract_native_video_metadata(self) -> tuple[types.VideoMetadata, bool]: """ Extract the video metadata, and whether its GPS was rejected as noise @@ -629,6 +655,19 @@ def _video_duration(video_path: Path) -> float | None: return mvhd.duration / mvhd.timescale +def _video_creation_time(video_path: Path) -> float | None: + """ + The creation time of the video as Unix time, from its mvhd box, or None if + it is unset or cannot be read. + """ + try: + moov = MovieBoxParser.parse_file(video_path) + except (OSError, sparser.ParsingError, C.ConstructError) as ex: + LOG.debug("Failed to read the creation time of %s: %s", video_path, ex) + return None + return camm_parser._extract_creation_time(moov) + + def _format_duration(seconds: float) -> str: """ Format a duration in the largest unit it fills, rounded rather than diff --git a/tests/unit/test_gpx_over_noisy_gps.py b/tests/unit/test_gpx_over_noisy_gps.py index 846dc3d53..c6a7c2ec1 100644 --- a/tests/unit/test_gpx_over_noisy_gps.py +++ b/tests/unit/test_gpx_over_noisy_gps.py @@ -24,10 +24,11 @@ from mapillary_tools import exceptions, types from mapillary_tools.geotag import factory -from mapillary_tools.geotag.options import SourceOption, SourceType +from mapillary_tools.geotag.options import SourceOption, SourceType, VideoSyncOption from mapillary_tools.geotag.video_extractors.gpx import GPXVideoExtractor, SyncMode from mapillary_tools.geotag.video_extractors.native import NativeVideoExtractor from mapillary_tools.gpmf import gpmf_gps_filter, gpmf_parser +from mapillary_tools.mp4 import construct_mp4_parser as cparser from mapillary_tools.process_geotag_properties import DEFAULT_GEOTAG_SOURCE_OPTIONS from mapillary_tools.telemetry import GPSFix, GPSPoint @@ -35,6 +36,13 @@ # 2026-09-11T15:39:41Z A_UNIX_TIME = 1789141181.0 +# 2015-10-22T09:53:44Z, a placeholder date GoPro stamps until it acquires GPS +# time +PLACEHOLDER_UNIX_TIME = 1445507624.0 + +# Seconds between the mp4 epoch (1904-01-01) and the Unix epoch +MP4_UNIX_DELTA = 2082844800 + def _gpx_xml(*times: str) -> str: points = "\n".join( @@ -85,6 +93,31 @@ def gpx_path(tmp_path: Path) -> Path: return path +def _write_mvhd(video_path: Path, creation_time: float, duration: float = 3.2): + """ + Make the video an mp4 with only an mvhd box, which is all of it that is + read when the GPMF parser is stubbed out. + """ + mp4_time = int(creation_time) + MP4_UNIX_DELTA + mvhd: cparser.BoxDict = { + "type": b"mvhd", + "data": { + "creation_time": mp4_time, + "modification_time": mp4_time, + "timescale": 1000, + "duration": int(duration * 1000), + }, + } + video_path.write_bytes( + cparser.MP4WithoutSTBLBuilderConstruct.build_boxlist( + [ + {"type": b"ftyp", "data": b"test"}, + {"type": b"moov", "data": [mvhd]}, + ] + ) + ) + + def _stub_gopro(monkeypatch: pytest.MonkeyPatch, points: list[GPSPoint]): info = gpmf_parser.GoProInfo(gps=points, make="GoPro", model="MAX2") monkeypatch.setattr(gpmf_parser, "extract_gopro_info", lambda *a, **kw: info) @@ -161,10 +194,9 @@ def test_noisy_clock_days_away_is_ignored(self, monkeypatch, video_path, gpx_pat GPS rejected as noise may never have got past. Syncing by it would put the GPX years away from the video. """ - # 2015-10-22T09:53:44Z - placeholder = 1445507624.0 points = [ - _point(time=i * 0.1, epoch_time=placeholder + i * 0.1) for i in range(32) + _point(time=i * 0.1, epoch_time=PLACEHOLDER_UNIX_TIME + i * 0.1) + for i in range(32) ] _stub_gopro(monkeypatch, points) @@ -173,6 +205,53 @@ def test_noisy_clock_days_away_is_ignored(self, monkeypatch, video_path, gpx_pat assert [p.time for p in metadata.points] == [0.0, 2.0] assert (metadata.make, metadata.model) == ("GoPro", "MAX2") + def test_noisy_clock_far_from_creation_time_is_ignored( + self, monkeypatch, video_path, gpx_path + ): + """A noisy clock days away from the creation time is ignored too.""" + _write_mvhd(video_path, creation_time=A_UNIX_TIME) + points = [ + _point(time=i * 0.1, epoch_time=PLACEHOLDER_UNIX_TIME + i * 0.1) + for i in range(32) + ] + _stub_gopro(monkeypatch, points) + + metadata = GPXVideoExtractor(video_path, gpx_path).extract() + + assert [p.time for p in metadata.points] == [0.0, 2.0] + + @pytest.mark.parametrize("video_sync", [None, VideoSyncOption()]) + def test_noisy_clock_at_creation_time_rejects_a_gpx_days_away( + self, video_path, tmp_path, noisy_gopro, video_sync + ): + """ + A noisy clock that agrees with the creation time of the video is real, + so a GPX days away from it misses the video, as for good GPS. + """ + # In local time, 8 hours behind UTC, as GoPro writes it + _write_mvhd(video_path, creation_time=A_UNIX_TIME - 8 * 3600) + gpx_path = tmp_path / "another_day.gpx" + gpx_path.write_text(_gpx_xml("2026-09-09T15:39:43Z", "2026-09-09T15:39:45Z")) + + extractor = GPXVideoExtractor(video_path, gpx_path, video_sync=video_sync) + with pytest.raises(exceptions.MapillaryOutsideGPXTrackError): + extractor.extract() + + def test_noisy_clock_syncs_despite_a_meaningless_creation_time( + self, video_path, gpx_path, noisy_gopro + ): + """ + A GoPro HERO10 recorded in 2022 reports 2016-01-01 as its creation + time, so a noisy clock that disagrees with it but agrees with the GPX is + still real. + """ + # 2016-01-01T01:44:00Z + _write_mvhd(video_path, creation_time=1451612640) + + metadata = GPXVideoExtractor(video_path, gpx_path).extract() + + assert [p.time for p in metadata.points] == pytest.approx([2.0, 4.0]) + def test_good_clock_days_away_rejects_the_gpx( self, monkeypatch, video_path, gpx_path ): From b03ea9bfbd14695cb2d181d31a8d4b71f8e94af4 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Tue, 20 Oct 2026 15:36:36 +0200 Subject: [PATCH 09/12] Anchor the GPS clock of a video at the median of its timestamps A GPX is synced to a video by the video's GPS clock, which was anchored at its first timestamp. That one can be off the rest: the first GPS block of a GoPro timelapse can be stamped 3 s early, the PanoX V2 can record a stale first fix 2 s old, and GoPro stamps a placeholder date until it acquires GPS time. The whole GPX then lands that much early or late. The clock is now anchored, still at the video time of its first timestamp, at the median of all its timestamps mapped back to that time at the clock's rate. Syncing the tracks of 17 sample videos back to them, the median error drops from 0.63 to 0.02 s and from 0.81 to 0.02 s on two GoPro MAX videos, from 2.2 to 0.02 s on the PanoX V2, and to 0.011 s or less on the GoPro timelapses. No median gets worse by more than 4 ms. A noisy GoPro video whose clock starts on the placeholder date is now synced by its clock, not from the start of the video. --- .../geotag/video_extractors/gpx.py | 19 ++++++++++++------- tests/integration/test_process.py | 4 +++- tests/unit/test_gpx_video_clock.py | 17 +++++++++++++++++ 3 files changed, 32 insertions(+), 8 deletions(-) diff --git a/mapillary_tools/geotag/video_extractors/gpx.py b/mapillary_tools/geotag/video_extractors/gpx.py index 61701affa..2f1f92499 100644 --- a/mapillary_tools/geotag/video_extractors/gpx.py +++ b/mapillary_tools/geotag/video_extractors/gpx.py @@ -571,10 +571,6 @@ def _read_video_clock( if not pairs: return None - # Anchored at its own video time, which is not 0 when the video's GPS - # starts late (e.g. GoPro drops noisy fixes at the start of a recording) - video_time, unix_time = pairs[0] - rate = cls._estimate_clock_rate(pairs) # One timestamp far off (e.g. a placeholder date before GPS time is # acquired) can pull the least-squares slope far off, so a timelapse @@ -582,15 +578,24 @@ def _read_video_clock( # too coarse to measure the rate, but not to tell a timelapse from a # normal video step_rate = cls._median_step_rate(pairs) - if ( + if not ( rate is not None and _MIN_TIMELAPSE_RATE < rate and step_rate is not None and _MIN_TIMELAPSE_RATE < step_rate ): - return VideoClock(video_time=video_time, unix_time=unix_time, rate=rate) + rate = 1.0 - return VideoClock(video_time=video_time, unix_time=unix_time) + # Anchored at its own video time, which is not 0 when the video's GPS + # starts late (e.g. GoPro drops noisy fixes at the start of a recording), + # and at the median of all its timestamps mapped back to that time, not + # at the first alone: that can be seconds off the rest (the first GPS + # block of a GoPro timelapse, a stale first fix), and some GPS clocks + # wander by a second or so against the video + video_time = pairs[0][0] + unix_time = statistics.median(u - (v - video_time) * rate for v, u in pairs) + + return VideoClock(video_time=video_time, unix_time=unix_time, rate=rate) @classmethod def _estimate_clock_rate( diff --git a/tests/integration/test_process.py b/tests/integration/test_process.py index 5d1a81bb0..747181706 100644 --- a/tests/integration/test_process.py +++ b/tests/integration/test_process.py @@ -688,7 +688,9 @@ def test_process_video_geotag_source_with_gpx_outside_video( assert len(descs) == 1 assert descs[0]["error"]["type"] == "MapillaryOutsideGPXTrackError" assert descs[0]["error"]["vars"] == { - "image_time": "2019_11_18_23_44_42_590", + # Its first GPS timestamp is 23:44:42.590, a second before the rest + # place the start of the video + "image_time": "2019_11_18_23_44_43_613", "gpx_start_time": "2025_03_14_07_00_00_000", "gpx_end_time": "2025_03_14_07_01_33_624", } diff --git a/tests/unit/test_gpx_video_clock.py b/tests/unit/test_gpx_video_clock.py index 00313707c..eb95bc69e 100644 --- a/tests/unit/test_gpx_video_clock.py +++ b/tests/unit/test_gpx_video_clock.py @@ -110,6 +110,23 @@ def test_block_timestamps_anchor_at_the_block_start(self): def test_one_jump_does_not_make_a_timelapse(self, pairs): assert _read(pairs).rate == 1.0 + @pytest.mark.parametrize("rate", [1.0, 30.0]) + def test_a_first_timestamp_off_the_rest_does_not_anchor_the_clock(self, rate): + """The first GPS block of a GoPro timelapse can be stamped 3s early.""" + pairs = [(0, A_UNIX_TIME - 3)] + [ + (i / rate, A_UNIX_TIME + i) for i in range(1, 60) + ] + clock = _read(pairs) + assert clock.rate == pytest.approx(rate, rel=0.05) + assert clock.to_unix_time(0) == pytest.approx(A_UNIX_TIME, abs=0.2) + + def test_a_placeholder_date_does_not_anchor_the_clock(self): + """GoPro stamps a placeholder date until it acquires GPS time.""" + pairs = [(i, PLACEHOLDER_UNIX_TIME + i) for i in range(5)] + [ + (i, A_UNIX_TIME + i) for i in range(5, 15) + ] + assert _read(pairs).to_unix_time(0) == A_UNIX_TIME + def test_no_timestamps(self): points = [ geo.Point(time=i, lat=1, lon=1, alt=None, angle=None) for i in range(3) From 76c8b25f357d0686794e23b560b0c79a59f3f1a1 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Wed, 21 Oct 2026 11:24:44 +0200 Subject: [PATCH 10/12] Stamp distance samples of a timelapse within their GPS track Distance sampling keeps frames up to 1 ms of video past either end of the GPS track, and extrapolates their timestamps. In a timelapse, 1 ms of video spans as many ms of GPS time as the video is faster than real time, so such a frame was stamped up to that far past the end of a GPX track. The geotagging stage allows a frame only 1 ms past it, so the frame failed, video_process exited with an error, and video_process_and_upload uploaded nothing. On a 30x GoPro MAX timelapse whose GPX ends with its last frame, the frame came out 23 ms past the end. Those frames are now stamped at the first or last timestamp of the track. --- mapillary_tools/sample_video.py | 15 ++++++++++ tests/unit/test_sample_video.py | 52 +++++++++++++++++++++++++++++++++ 2 files changed, 67 insertions(+) diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py index d5e15ff2e..d816d1b00 100644 --- a/mapillary_tools/sample_video.py +++ b/mapillary_tools/sample_video.py @@ -883,6 +883,15 @@ def _sample_single_video_by_distance( ) sorted_sample_indices = sorted(sample_points_by_frame_idx.keys()) + # The GPS timestamps of the track, which frames are stamped within + track_unix_times = [ + unix_time + for unix_time in (p.get_unix_time() for p in video_metadata.points) + if unix_time is not None + ] + first_unix_time = min(track_unix_times, default=None) + last_unix_time = max(track_unix_times, default=None) + # Frames at points with an absolute timestamp are timestamped from it # below, so only the others need the start time if start_time is None and any( @@ -933,6 +942,12 @@ def _sample_single_video_by_distance( # Try to use the GPS timestamp if available (for timelapse videos) gps_unix_time = interp.get_unix_time() if gps_unix_time is not None: + assert first_unix_time is not None and last_unix_time is not None + # Frames up to 1 ms of video past either end of the track are + # kept, and their timestamps extrapolated. In a timelapse, 1 ms + # of video spans several ms of GPS time, and the geotagging + # stage allows a frame only 1 ms past the end of a GPX track + gps_unix_time = min(max(gps_unix_time, first_unix_time), last_unix_time) timestamp = datetime.datetime.fromtimestamp( gps_unix_time, tz=datetime.timezone.utc ) diff --git a/tests/unit/test_sample_video.py b/tests/unit/test_sample_video.py index a4cf6d476..c1627d0f8 100644 --- a/tests/unit/test_sample_video.py +++ b/tests/unit/test_sample_video.py @@ -1435,6 +1435,58 @@ def test_timestamps_from_gps_clock(self, tmp_path: Path) -> None: gps_start_time + datetime.timedelta(seconds=idx) ) + def test_timelapse_frames_past_the_end_are_stamped_within_the_track( + self, tmp_path: Path + ) -> None: + """ + The last frame is 0.5 ms of video past the end of the track, which a + 30x timelapse stretches to 15 ms of GPS time. Stamped there, it would + fall outside the GPX in the geotagging stage, which allows 1 ms. + """ + video_dir = tmp_path / "videos" + video_dir.mkdir() + video_file = video_dir / "test.mp4" + video_file.touch() + output_dir = tmp_path / "output" + + gps_start_time = datetime.datetime( + 2021, 8, 10, 6, 38, 6, tzinfo=datetime.timezone.utc + ) + gps_points = [ + telemetry.GPSPoint( + time=time, + lat=40.0 + time * 0.001, + lon=-74.0, + alt=None, + angle=None, + epoch_time=gps_start_time.timestamp() + 30 * time, + fix=None, + precision=None, + ground_speed=None, + ) + for time in [*range(9), 8.9995] + ] + mocks = self._setup_mocks(tmp_path, video_file, gps_points=gps_points) + + with ( + mocks["patches"]["ffmpeg_cls"], + mocks["patches"]["geotag_cls"], + mocks["patches"]["moov_parse"], + ): + sample_video.sample_video( + video_import_path=video_file, + import_path=output_dir, + video_sample_distance=0.0, + ) + + frames = sorted((output_dir / "test.mp4").glob("*.jpg")) + capture_times = [ + exif_read.ExifRead(frame).extract_capture_time() for frame in frames + ] + assert capture_times == [ + gps_start_time + datetime.timedelta(seconds=30 * idx) for idx in range(9) + ] + [gps_start_time + datetime.timedelta(seconds=269.985)] + class TestFailuresAreReportedTogether: """ From fef4e6052f327d241101e1a194374fff0286ba72 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Wed, 21 Oct 2026 12:43:06 +0200 Subject: [PATCH 11/12] Clip a GPX track that a clock placed to the video it geotags process uploaded a video geotagged from a GPX with the whole GPX track. Its points before the start of the video were dropped when the CAMM track was built, which left the frames before the first point inside the video without a position. Its points after the end of the video were uploaded as they were, although a GPX of a longer trip can run on for an hour past a clip of a few minutes. None of those points is the position of a frame of the video. When the GPS clock of the video placed the track, so that it is known where the video ends along it, GPXVideoExtractor now clips it to the video, from 0 to the mvhd duration, with a point interpolated at either end that the track runs past. Only process asks for that: sampling reads the track only within the video. A track that is only assumed to start with the video, or said to by --interpolation_use_gpx_start_time, is left as it is. It may be the track of a timelapse without a GPS clock, which runs on past the end of the video at the pace of the GPS clock, and nothing says where the video ends along it. --- .../geotag/geotag_videos_from_gpx.py | 7 +- .../geotag/video_extractors/gpx.py | 50 ++++++ tests/integration/test_upload.py | 51 ++++++ tests/unit/test_gpx_video_clip.py | 150 ++++++++++++++++++ 4 files changed, 257 insertions(+), 1 deletion(-) create mode 100644 tests/unit/test_gpx_video_clip.py diff --git a/mapillary_tools/geotag/geotag_videos_from_gpx.py b/mapillary_tools/geotag/geotag_videos_from_gpx.py index 05aeecc96..342731eca 100644 --- a/mapillary_tools/geotag/geotag_videos_from_gpx.py +++ b/mapillary_tools/geotag/geotag_videos_from_gpx.py @@ -47,7 +47,12 @@ def _generate_video_extractors( if source_path.is_file(): results.append( GPXVideoExtractor( - video_path, source_path, video_sync=self.video_sync + video_path, + source_path, + video_sync=self.video_sync, + # The track that process uploads with the video. + # Sampling reads its track only within the video + clip_to_video=self.video_sync is None, ) ) else: diff --git a/mapillary_tools/geotag/video_extractors/gpx.py b/mapillary_tools/geotag/video_extractors/gpx.py index 2f1f92499..43b2ae58c 100644 --- a/mapillary_tools/geotag/video_extractors/gpx.py +++ b/mapillary_tools/geotag/video_extractors/gpx.py @@ -9,6 +9,7 @@ import datetime import enum import logging +import math import statistics import sys import typing as T @@ -104,11 +105,14 @@ def __init__( gpx_path: Path, sync_mode: SyncMode = SyncMode.SYNC, video_sync: options.VideoSyncOption | None = None, + clip_to_video: bool = False, ): self.video_path = video_path self.gpx_path = gpx_path self.sync_mode = sync_mode self.video_sync = video_sync + # Clip the track to the video when a clock placed it (see _clip_to_video) + self.clip_to_video = clip_to_video if video_sync is not None and video_sync.use_gpx_start_time: self.sync_mode = SyncMode.REBASE @@ -132,6 +136,8 @@ def extract(self) -> types.VideoMetadata: rate = self._duration_ratio() self._apply_offset_time(gpx_points, rate=rate) self._check_coverage(gpx_points, gpx_unix_times, [], start_clock, rate) + if self.clip_to_video and start_clock is not None: + gpx_points = self._clip_to_video(gpx_points) return types.VideoMetadata( filename=self.video_path, filesize=utils.get_file_size(self.video_path), @@ -204,6 +210,8 @@ def extract(self) -> types.VideoMetadata: None if self.sync_mode is SyncMode.REBASE else clock, rate, ) + if self.clip_to_video and clock is not None: + gpx_points = self._clip_to_video(gpx_points) return dataclasses.replace(native_video_metadata, points=gpx_points) @@ -239,6 +247,18 @@ def _read_gpx_points(self) -> list[geo.Point]: gpx_points.sort(key=lambda p: p.time) return gpx_points + def _clip_to_video(self, gpx_points: list[geo.Point]) -> list[geo.Point]: + """ + Clip a GPX track that a clock placed on the timeline of the video to + the video. Its points outside the video are the positions of none of + its frames, and a GPX of a longer trip can run on for long after it. + + A track that is only assumed to start with the video is left as it is: + it may be that of a timelapse, which runs on past the end of the video + at the pace of the GPS clock rather than that of the video. + """ + return _clip_track(gpx_points, _video_duration(self.video_path)) + def _place_without_video_gps( self, gpx_points: T.Sequence[geo.Point] ) -> VideoClock | None: @@ -638,6 +658,36 @@ def _median_step_rate(cls, pairs: T.Sequence[tuple[float, float]]) -> float | No return statistics.median(steps) if steps else None +def _clip_track( + points: T.Sequence[geo.Point], duration: float | None +) -> list[geo.Point]: + """ + Clip a track, sorted by time, to the video, from time 0 to duration (None + if unknown), with a point interpolated at either end that the track runs + past, so that the frames up to that end keep their position + + >>> track = [geo.Point(time=t, lat=t, lon=0, alt=None, angle=None) for t in (-2, 2, 6, 10)] + >>> [(p.time, p.lat) for p in _clip_track(track, 8.0)] + [(0.0, 0.0), (2, 2), (6, 6), (8.0, 8.0)] + >>> [p.time for p in _clip_track(track, None)] + [0.0, 2, 6, 10] + >>> [p.time for p in _clip_track(track, 10.0)] + [0.0, 2, 6, 10] + >>> _clip_track(track[:1], 8.0) + [] + """ + if not points: + return [] + + end = math.inf if duration is None else duration + clipped = [p for p in points if 0 <= p.time <= end] + if points[0].time < 0 < points[-1].time and (not clipped or 0 < clipped[0].time): + clipped.insert(0, geo.interpolate(points, 0.0)) + if points[0].time < end < points[-1].time and clipped[-1].time < end: + clipped.append(geo.interpolate(points, end)) + return clipped + + def _video_duration(video_path: Path) -> float | None: """ The duration of the video in seconds, from its mvhd box, or None if it diff --git a/tests/integration/test_upload.py b/tests/integration/test_upload.py index 0cb7de83d..3bfa1e357 100644 --- a/tests/integration/test_upload.py +++ b/tests/integration/test_upload.py @@ -3,12 +3,15 @@ # This source code is licensed under the BSD license found in the # LICENSE file in the root directory of this source tree. +import datetime import json import subprocess from pathlib import Path import py.path import pytest +from mapillary_tools.geotag.video_extractors.gpx import GPXVideoExtractor +from mapillary_tools.geotag.video_extractors.native import NativeVideoExtractor from .fixtures import ( assert_contains_image_descs, @@ -36,6 +39,54 @@ def test_upload_everything(setup_data: py.path.local, setup_upload: py.path.loca assert_contains_image_descs(descs, uploaded_descs) +@pytest.mark.usefixtures("setup_config") +def test_upload_video_with_gpx_running_past_both_ends( + setup_data: py.path.local, setup_upload: py.path.local +): + """ + The uploaded track keeps only the video, with a point interpolated at + either end, from a GPX of a longer trip + """ + video_path = setup_data.join("gopro_data").join("max-360mode.mp4") + duration = 12.137125 + points = NativeVideoExtractor(Path(video_path)).extract().points + clock = GPXVideoExtractor._read_video_clock(points) + assert clock is not None + video_start = clock.to_unix_time(0) + # From 12.5 seconds before the video to about 47 seconds after it + trkpts = "".join( + f'" + for i in range(61) + ) + gpx_file = setup_data.join("track.gpx") + gpx_file.write( + '\n' + '' + f"{trkpts}\n" + ) + + run_process_and_upload_for_descs( + [ + *["--geotag_source", "gpx"], + *["--geotag_source_path", str(gpx_file)], + str(video_path), + ] + ) + + [[desc]] = extract_all_uploaded_descs(Path(setup_upload)) + track = desc["MAPGPSTrack"] + # The GPX points recorded from 0.5 to 11.5 seconds, and one at either end + assert len(track) == 14 + first, last = track[0], track[-1] + assert first[0] == 0 + assert first[2] == pytest.approx(37.0 + 12.5 * 0.0001, abs=1e-7) + assert first[5] == pytest.approx(video_start, abs=1e-3) + assert last[0] == pytest.approx(duration * 1000, abs=1) + assert last[5] == pytest.approx(video_start + duration, abs=1e-3) + + @pytest.mark.usefixtures("setup_config") def test_upload_image_dir(setup_data: py.path.local, setup_upload: py.path.local): descs = run_process_and_upload_for_descs(["--file_types=image", str(setup_data)]) diff --git a/tests/unit/test_gpx_video_clip.py b/tests/unit/test_gpx_video_clip.py new file mode 100644 index 000000000..5d377cc77 --- /dev/null +++ b/tests/unit/test_gpx_video_clip.py @@ -0,0 +1,150 @@ +# 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. + +""" +A GPX track that a clock placed on the timeline of a video is clipped to the +video, with a point interpolated at either end, when it geotags the video. +One that is only assumed to start with the video is not. +""" + +from __future__ import annotations + +import datetime +from pathlib import Path + +import pytest +from mapillary_tools import types +from mapillary_tools.geotag import factory +from mapillary_tools.geotag.options import ( + SourceOption, + SourcePathOption, + SourceType, + VideoSyncOption, +) +from mapillary_tools.geotag.video_extractors.gpx import GPXVideoExtractor, SyncMode +from mapillary_tools.geotag.video_extractors.native import NativeVideoExtractor + +_DATA = Path(__file__).resolve().parent.parent.joinpath("data") + +# No GPS of its own, and 5.76 seconds long +NO_GPS_VIDEO = _DATA.joinpath("videos", "sample-5s.mp4") +NO_GPS_VIDEO_DURATION = 5.759 + +# GPS of its own, and 12.137125 seconds long +GPS_VIDEO = _DATA.joinpath("gopro_data", "max-360mode.mp4") +GPS_VIDEO_DURATION = 12.137125 + + +def _gps_video_start() -> float: + """The Unix time of the first frame of GPS_VIDEO, by its GPS clock""" + points = NativeVideoExtractor(GPS_VIDEO).extract().points + clock = GPXVideoExtractor._read_video_clock(points) + assert clock is not None + return clock.to_unix_time(0) + + +def _write_gpx(path: Path, start: float, seconds: int) -> Path: + """A GPX track with a point every second from start, heading north at 10 m/s.""" + points = "".join( + f'" + for i in range(seconds + 1) + ) + path.write_text( + '\n' + '' + f"{points}\n" + ) + return path + + +def _times(metadata: types.VideoMetadata) -> list[float]: + return [p.time for p in metadata.points] + + +class TestClockPlacedTrack: + def _gpx(self, tmp_path: Path) -> Path: + # From 12.5 seconds before the video to about 47 seconds after it + return _write_gpx(tmp_path / "track.gpx", _gps_video_start() - 12.5, 60) + + def test_is_clipped_to_the_video(self, tmp_path): + metadata = GPXVideoExtractor( + GPS_VIDEO, self._gpx(tmp_path), clip_to_video=True + ).extract() + + times = _times(metadata) + assert times[0] == 0 + assert times[-1] == pytest.approx(GPS_VIDEO_DURATION) + # The GPX points recorded during the video, from 0.5 to 11.5 seconds + assert len(times) == 2 + 12 + assert all(0 <= t <= GPS_VIDEO_DURATION for t in times) + + def test_the_ends_are_interpolated(self, tmp_path): + metadata = GPXVideoExtractor( + GPS_VIDEO, self._gpx(tmp_path), clip_to_video=True + ).extract() + + first, last = metadata.points[0], metadata.points[-1] + assert first.get_unix_time() == pytest.approx(_gps_video_start()) + assert first.lat == pytest.approx(37.0 + 12.5 * 10 / 111_320, abs=1e-7) + assert last.get_unix_time() == pytest.approx( + _gps_video_start() + GPS_VIDEO_DURATION + ) + + def test_by_the_start_time_is_clipped_to_the_video(self, tmp_path): + start = datetime.datetime(2025, 3, 14, 7, 0, 0, tzinfo=datetime.timezone.utc) + gpx = _write_gpx(tmp_path / "track.gpx", start.timestamp(), 60) + + metadata = GPXVideoExtractor( + NO_GPS_VIDEO, + gpx, + video_sync=VideoSyncOption(start_time=start.timestamp() + 30), + clip_to_video=True, + ).extract() + + times = _times(metadata) + assert times[0] == 0 + assert times[-1] == pytest.approx(NO_GPS_VIDEO_DURATION, abs=1e-3) + + def test_the_gpx_source_clips_it(self, tmp_path): + [metadata] = factory.process( + [GPS_VIDEO], + [ + SourceOption( + SourceType.GPX, + num_processes=0, + source_path=SourcePathOption(source_path=self._gpx(tmp_path)), + ) + ], + ) + + assert isinstance(metadata, types.VideoMetadata) + assert _times(metadata)[0] == 0 + assert _times(metadata)[-1] == pytest.approx(GPS_VIDEO_DURATION) + + +class TestTrackAssumedToStartWithTheVideo: + """ + Without a clock to say where the video ends along the track, a track that + runs on past it may be that of a timelapse, at the pace of the GPS clock + """ + + def test_without_gps_is_not_clipped(self, tmp_path): + start = datetime.datetime(2025, 3, 14, 7, 0, 0, tzinfo=datetime.timezone.utc) + gpx = _write_gpx(tmp_path / "track.gpx", start.timestamp(), 60) + + metadata = GPXVideoExtractor(NO_GPS_VIDEO, gpx, clip_to_video=True).extract() + + assert _times(metadata) == [float(t) for t in range(61)] + + def test_said_to_by_gpx_start_time_is_not_clipped(self, tmp_path): + gpx = _write_gpx(tmp_path / "track.gpx", _gps_video_start() - 12.5, 60) + + metadata = GPXVideoExtractor( + GPS_VIDEO, gpx, sync_mode=SyncMode.REBASE, clip_to_video=True + ).extract() + + assert _times(metadata) == [float(t) for t in range(61)] From cea3bc33ea7ecaea72987a09b7f759d686b64055 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Wed, 21 Oct 2026 18:08:48 +0200 Subject: [PATCH 12/12] Clip a track started with the video, which the GPS clock paces The previous commit clips a GPX track whenever a clock gives its pace along the video, so that where the video ends along it is known. Since the GPS clock's rate is kept for --interpolation_use_gpx_start_time (#853), that includes a track the option starts with the video: it is clipped as any other, and a GPS logger left running past the end of the video uploads only the part of the track that the video covers. Say so in the docstrings, which said a clock had to place the track, and test it among the tracks a clock paces. --- .../geotag/video_extractors/gpx.py | 22 ++++++++----- tests/unit/test_gpx_video_clip.py | 31 ++++++++++++------- 2 files changed, 33 insertions(+), 20 deletions(-) diff --git a/mapillary_tools/geotag/video_extractors/gpx.py b/mapillary_tools/geotag/video_extractors/gpx.py index 43b2ae58c..37bf2582a 100644 --- a/mapillary_tools/geotag/video_extractors/gpx.py +++ b/mapillary_tools/geotag/video_extractors/gpx.py @@ -111,7 +111,8 @@ def __init__( self.gpx_path = gpx_path self.sync_mode = sync_mode self.video_sync = video_sync - # Clip the track to the video when a clock placed it (see _clip_to_video) + # Clip the track to the video when a clock gives its pace along the + # video (see _clip_to_video) self.clip_to_video = clip_to_video if video_sync is not None and video_sync.use_gpx_start_time: self.sync_mode = SyncMode.REBASE @@ -249,13 +250,18 @@ def _read_gpx_points(self) -> list[geo.Point]: def _clip_to_video(self, gpx_points: list[geo.Point]) -> list[geo.Point]: """ - Clip a GPX track that a clock placed on the timeline of the video to - the video. Its points outside the video are the positions of none of - its frames, and a GPX of a longer trip can run on for long after it. - - A track that is only assumed to start with the video is left as it is: - it may be that of a timelapse, which runs on past the end of the video - at the pace of the GPS clock rather than that of the video. + Clip a GPX track to the video when a clock gives its pace along the + video, so that where the video ends along it is known: the GPS clock + of the video, which places the track too unless + --interpolation_use_gpx_start_time starts it with the video, or the + --video_start_time of sampling. Its points outside the video are the + positions of none of its frames, and a GPX of a longer trip can run on + for long after it. + + A track that is only assumed to start with a video without a GPS clock + is left as it is: it may be that of a timelapse, which runs on past + the end of the video at the pace of the GPS clock rather than that of + the video. """ return _clip_track(gpx_points, _video_duration(self.video_path)) diff --git a/tests/unit/test_gpx_video_clip.py b/tests/unit/test_gpx_video_clip.py index 5d377cc77..e6d713dbd 100644 --- a/tests/unit/test_gpx_video_clip.py +++ b/tests/unit/test_gpx_video_clip.py @@ -4,9 +4,9 @@ # LICENSE file in the root directory of this source tree. """ -A GPX track that a clock placed on the timeline of a video is clipped to the -video, with a point interpolated at either end, when it geotags the video. -One that is only assumed to start with the video is not. +A GPX track whose pace along a video a clock gives is clipped to the video, +with a point interpolated at either end, when it geotags the video. One that +is only assumed to start with a video without a GPS clock is not. """ from __future__ import annotations @@ -109,6 +109,22 @@ def test_by_the_start_time_is_clipped_to_the_video(self, tmp_path): assert times[0] == 0 assert times[-1] == pytest.approx(NO_GPS_VIDEO_DURATION, abs=1e-3) + def test_started_with_the_video_is_clipped(self, tmp_path): + """ + --interpolation_use_gpx_start_time places the track, and the GPS clock + gives its pace + """ + metadata = GPXVideoExtractor( + GPS_VIDEO, + self._gpx(tmp_path), + sync_mode=SyncMode.REBASE, + clip_to_video=True, + ).extract() + + times = _times(metadata) + assert times[:-1] == [float(t) for t in range(13)] + assert times[-1] == pytest.approx(GPS_VIDEO_DURATION) + def test_the_gpx_source_clips_it(self, tmp_path): [metadata] = factory.process( [GPS_VIDEO], @@ -139,12 +155,3 @@ def test_without_gps_is_not_clipped(self, tmp_path): metadata = GPXVideoExtractor(NO_GPS_VIDEO, gpx, clip_to_video=True).extract() assert _times(metadata) == [float(t) for t in range(61)] - - def test_said_to_by_gpx_start_time_is_not_clipped(self, tmp_path): - gpx = _write_gpx(tmp_path / "track.gpx", _gps_video_start() - 12.5, 60) - - metadata = GPXVideoExtractor( - GPS_VIDEO, gpx, sync_mode=SyncMode.REBASE, clip_to_video=True - ).extract() - - assert _times(metadata) == [float(t) for t in range(61)]