From 9f04bede7cc62e4636e0d18fdfd36a8819577141 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Tue, 20 Oct 2026 13:57:37 +0200 Subject: [PATCH] Reject GPX tracks that miss the video, as placed on its timeline Check a GPX track against the video once it is placed on the video's timeline, by the video's GPS clock (including a timelapse's), by --video_start_time, or by assuming that it starts with the video, and after --interpolation_offset_time: - A track that misses the video raises MapillaryOutsideGPXTrackError, whatever the gap: a GPX file from another day or another clip, an epoch mix-up, or naive GPX timestamps read in the wrong time zone. Before, the video was synced to a track with no position for any moment of it, and process exited 0, with a warning only for an offset over a day. The message names both files, their times and the gap, in GPS time. - A track that covers only part of the video warns with how much of it is left uncovered, allowing 2 s of GPS time for a once-a-second logger started and stopped with the camera. - If the duration cannot be read, a gap warns up to a day and raises beyond. A track assumed to start with the video only warns. MapillaryOutsideGPXTrackError is now a geotagging error, so the next geotag source gets its turn, for images too, and a warning says so when that source geotags the file: one per video, and one per GPX source for images. With GPX as the only or last source the error is final, as before. The error is made picklable, since video geotagging raises it in a worker process. The GPX module keeps one duration formatter. It rounds instead of truncating, since timestamps often fall a hair short of a whole unit, which humanize reported as "4 seconds" for 4.999 and "an hour" for 1 h 59 min. Tests: - test_good_clock_days_away_is_still_used now expects the rejection. - Two GPX sync tests used a GPX 8 hours before the sample video's GPS, which started at 23:44:42Z, not 15:44:47Z as their comment said. - Two integration tests paired the 2019 GoPro sample video with a 2025 GPX file and asserted success. They now shift the GPX to the video's time, and a new test asserts that the unrelated pairing fails. - video_process with a GPX two hours before --video_start_time still fails with exit 7, now with MapillaryOutsideGPXTrackError. --- mapillary_tools/exceptions.py | 12 +- mapillary_tools/geotag/factory.py | 55 ++ .../geotag/video_extractors/gpx.py | 155 +++++- tests/integration/test_process.py | 58 ++- tests/integration/test_video_process.py | 8 +- tests/unit/test_gpx_over_noisy_gps.py | 12 +- tests/unit/test_gpx_video_coverage.py | 472 ++++++++++++++++++ tests/unit/test_gpx_video_sync.py | 14 +- 8 files changed, 742 insertions(+), 44 deletions(-) create mode 100644 tests/unit/test_gpx_video_coverage.py diff --git a/mapillary_tools/exceptions.py b/mapillary_tools/exceptions.py index ecf2ad40..4c400f30 100644 --- a/mapillary_tools/exceptions.py +++ b/mapillary_tools/exceptions.py @@ -96,7 +96,9 @@ class MapillaryGPXVideoStartUnknownError(MapillaryDescriptionError): """ -class MapillaryOutsideGPXTrackError(MapillaryDescriptionError): +# A geotagging error, so that the next geotag source gets its turn: a GPX track +# that misses the capture time says nothing about the file itself +class MapillaryOutsideGPXTrackError(MapillaryGeoTaggingError): def __init__( self, message: str, image_time: str, gpx_start_time: str, gpx_end_time: str ): @@ -105,6 +107,14 @@ def __init__( self.gpx_start_time = gpx_start_time self.gpx_end_time = gpx_end_time + def __reduce__(self): + # Pickle with every argument, so that the error survives the trip back + # from a worker process (video geotagging raises it in one) + return ( + self.__class__, + (self.args[0], self.image_time, self.gpx_start_time, self.gpx_end_time), + ) + class MapillaryDuplicationError(MapillaryDescriptionError): def __init__( diff --git a/mapillary_tools/geotag/factory.py b/mapillary_tools/geotag/factory.py index c25d3de9..9cdd8479 100644 --- a/mapillary_tools/geotag/factory.py +++ b/mapillary_tools/geotag/factory.py @@ -162,6 +162,10 @@ def process( # overturn it by reading the same bad data differently rejected_gps_paths: set[Path] = set() + # Paths whose GPX track missed them, with the source that rejected them, to + # warn about if a later source geotags them instead + gpx_rejections: dict[Path, tuple[SourceOption, types.ErrorMetadata]] = {} + for option in options: if option.source in EXTERNAL_GPS_SOURCES: option_paths = reprocessable_paths @@ -197,11 +201,17 @@ def process( else: video_metadata_or_errors = [] + rescued: list[ + tuple[types.MetadataOrError, SourceOption, types.ErrorMetadata] + ] = [] + for metadata in image_metadata_or_errors + video_metadata_or_errors: if isinstance(metadata, types.ErrorMetadata) and _is_reprocessable( metadata ): # Leave what it is for the next geotag process + if isinstance(metadata.error, exceptions.MapillaryOutsideGPXTrackError): + gpx_rejections[metadata.filename] = (option, metadata) if isinstance( metadata.error, ( @@ -221,6 +231,13 @@ def process( final_metadatas.append(metadata) reprocessable_paths.remove(metadata.filename) pending_errors.pop(metadata.filename, None) + rejection = gpx_rejections.pop(metadata.filename, None) + if rejection is not None and not isinstance( + metadata, types.ErrorMetadata + ): + rescued.append((metadata, *rejection)) + + _warn_geotagged_despite_gpx(rescued, option) # Quit if no more paths to process if not reprocessable_paths: @@ -231,6 +248,44 @@ def process( return final_metadatas +def _warn_geotagged_despite_gpx( + rescued: T.Sequence[ + tuple[types.MetadataOrError, SourceOption, types.ErrorMetadata] + ], + option: SourceOption, +) -> None: + """ + Warn about the files whose GPX track missed them, but which the source + **option** geotagged instead, so that a GPX that does not belong to them + does not go unnoticed: one warning per video, and one per GPX source for + images, of which a GPX can miss many. + """ + images_by_gpx_option: dict[int, tuple[SourceOption, list[Path]]] = {} + + for metadata, gpx_option, rejection in rescued: + if isinstance(metadata, types.VideoMetadata): + LOG.warning( + f"{rejection.error}. Geotagged it from the {option.source.value} source instead" + ) + else: + images_by_gpx_option.setdefault(id(gpx_option), (gpx_option, []))[1].append( + metadata.filename + ) + + for gpx_option, image_paths in images_by_gpx_option.values(): + gpx_path = gpx_option.source_path and gpx_option.source_path.source_path + gpx_track = f"the GPX track in {gpx_path}" if gpx_path else "the GPX track" + if len(image_paths) == 1: + images, them = "1 image falls", "it" + else: + images, them = f"{len(image_paths)} images fall", "them" + LOG.warning( + f"{images} outside {gpx_track}. " + f"Geotagged {them} from the {option.source.value} source instead" + ) + LOG.debug("Images outside %s: %s", gpx_track, [p.name for p in image_paths]) + + def _is_reprocessable(metadata: types.MetadataOrError) -> bool: if isinstance(metadata, types.ErrorMetadata): if isinstance( diff --git a/mapillary_tools/geotag/video_extractors/gpx.py b/mapillary_tools/geotag/video_extractors/gpx.py index 8fcc7fb4..75d4aa16 100644 --- a/mapillary_tools/geotag/video_extractors/gpx.py +++ b/mapillary_tools/geotag/video_extractors/gpx.py @@ -16,7 +16,6 @@ import construct as C import gpxpy -import humanize if sys.version_info >= (3, 12): from typing import override @@ -25,6 +24,7 @@ from ... import exceptions, geo, telemetry, types, utils from ...mp4 import construct_mp4_parser as cparser, simple_mp4_parser as sparser +from ...serializer.description import build_capture_time from .. import options from ..utils import parse_gpx from .base import BaseVideoExtractor @@ -33,9 +33,9 @@ LOG = logging.getLogger(__name__) -# A GPX track and the video it is synced against should overlap in time. Warn -# above a day, which no legitimate pairing needs and an epoch mix-up exceeds by -# orders of magnitude. +# No camera clock or time zone mistake puts a GPX track and the video it +# belongs to more than a day apart, while an epoch mix-up exceeds it by orders +# of magnitude. _IMPLAUSIBLE_OFFSET_SECONDS = 24 * 3600 # A GPS clock that runs this much faster than the video clock is a timelapse. @@ -48,10 +48,11 @@ # 1969-12-31T15:59:59Z, on points that had no valid time _GPS_EPOCH = datetime.datetime(1980, 1, 6, tzinfo=datetime.timezone.utc).timestamp() -# How much longer than a video a GPX track may last without a warning, when it -# is assumed to start with the video. A logger that records once a second, -# started and stopped with the camera, can run on for a second at each end. -_ASSUMED_START_TOLERANCE_SECONDS = 2.0 +# How far a GPX track may run past either end of a video, when it is assumed +# to start with the video, or fall short of them, without a warning. A logger +# that records once a second, started and stopped with the camera, can be a +# second off at each end. +_LOGGER_TOLERANCE_SECONDS = 2.0 class SyncMode(enum.Enum): @@ -133,6 +134,11 @@ def extract(self) -> types.VideoMetadata: # The tracks need not be in time order, and interpolation needs the # merged points to be gpx_points.sort(key=lambda p: p.time) + # The first and last Unix time of the track, before it is placed on + # the timeline of the video + gpx_unix_times = ( + (gpx_points[0].time, gpx_points[-1].time) if gpx_points else None + ) try: native_video_metadata, noisy = self._extract_native_video_metadata() @@ -141,8 +147,9 @@ def extract(self) -> types.VideoMetadata: # empty or too noisy must not reject the video if self.sync_mode is SyncMode.STRICT_SYNC: raise ex - self._place_without_video_gps(gpx_points) + 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) return types.VideoMetadata( filename=self.video_path, filesize=utils.get_file_size(self.video_path), @@ -170,19 +177,10 @@ def extract(self) -> types.VideoMetadata: self.gpx_path, ) clock = None - else: - LOG.warning( - "Syncing %s against %s requires an offset of %.0f seconds (%.1f days). " - "The GPX file probably does not belong to this video", - self.video_path, - self.gpx_path, - offset, - offset / 86400, - ) if clock is None: # Its GPS has no clock to sync with, as if it had no GPS - self._place_without_video_gps(gpx_points) + clock = self._place_without_video_gps(gpx_points) else: if clock.rate != 1.0: LOG.info( @@ -195,27 +193,36 @@ 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) + self._check_coverage( + gpx_points, gpx_unix_times, native_video_metadata.points, clock + ) return dataclasses.replace(native_video_metadata, points=gpx_points) - def _place_without_video_gps(self, gpx_points: T.Sequence[geo.Point]) -> None: + 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. + + 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. """ if not gpx_points: - return + return None 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 ) - return + return VideoClock(video_time=0.0, unix_time=self.video_sync.start_time) self._check_assumed_start(gpx_points) self._rebase_times(gpx_points) + return None def _apply_offset_time( self, gpx_points: T.Sequence[geo.Point], rate: float = 1.0 @@ -240,6 +247,78 @@ def _apply_offset_time( if isinstance(p, telemetry.CAMMGPSPoint) and 0 < p.epoch_time: p.epoch_time -= offset_time + def _check_coverage( + self, + gpx_points: T.Sequence[geo.Point], + gpx_unix_times: tuple[float, float] | None, + video_gps_points: T.Sequence[geo.Point], + clock: VideoClock | None, + ) -> None: + """ + Check the GPX track, as placed on the timeline of the video, against + the video, from video time 0 to its mvhd duration. + + 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 + only part of the video warns, and one that covers all of it is silent. + + Without the duration of the video, the video is known only up to its + last GPS point, and a track that starts after it may still overlap the + video. Then a gap only warns, unless it is too large for any clock or + time zone mistake to explain. A track assumed to start with the video + has no clock to say that it misses the video, so it only warns. + """ + if not gpx_points or gpx_unix_times is None: + return + + video_duration = _video_duration(self.video_path) + video_end = max((p.time for p in video_gps_points), default=0.0) + if video_duration is not None: + video_end = max(video_end, video_duration) + 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) + + gpx_track = f"The GPX track in {self.gpx_path} ({_isoformat(gpx_unix_times[0])} to {_isoformat(gpx_unix_times[1])})" + if clock is None: + video = f"the video {self.video_path}" + else: + video = f"the video {self.video_path} ({_isoformat(clock.to_unix_time(0))} to {_isoformat(clock.to_unix_time(video_end))})" + + # In GPS seconds, which is where clock and time zone mistakes are made + gap = max(gpx_first - video_end, -gpx_last) * rate + if gap <= 0: + # In video seconds + uncovered = max(gpx_first, 0.0) + max(video_end - gpx_last, 0.0) + if _LOGGER_TOLERANCE_SECONDS < uncovered * rate: + LOG.warning( + f"{gpx_track} covers only part of {video}: " + f"{_format_duration(uncovered)} of the video fall outside the track" + ) + return + + message = ( + f"{gpx_track} misses {video} by {_format_duration(gap)}. Check that " + "the GPX file belongs to this video, then the camera clock and the " + "time zone of the GPX timestamps" + ) + + if clock is not None and ( + video_duration is not None or _IMPLAUSIBLE_OFFSET_SECONDS < gap + ): + raise exceptions.MapillaryOutsideGPXTrackError( + message, + image_time=build_capture_time(clock.to_unix_time(0)), + gpx_start_time=build_capture_time(gpx_unix_times[0]), + gpx_end_time=build_capture_time(gpx_unix_times[1]), + ) + + LOG.warning(message) + def _check_assumed_start(self, gpx_points: T.Sequence[geo.Point]) -> None: """ Check that the GPX track can have started with the video, as it is @@ -269,7 +348,7 @@ def _check_assumed_start(self, gpx_points: T.Sequence[geo.Point]) -> None: "--interpolation_use_gpx_start_time if the track was started with the video" ) - if _ASSUMED_START_TOLERANCE_SECONDS < excess: + if _LOGGER_TOLERANCE_SECONDS < excess: LOG.warning( "Assuming that the GPX track in %s starts with the video %s, although " "the track lasts %s longer. If it was started first, the frames are " @@ -424,9 +503,33 @@ def _video_duration(video_path: Path) -> float | None: def _format_duration(seconds: float) -> str: """ - >>> _format_duration(5) + Format a duration in the largest unit it fills, rounded rather than + truncated, since timestamps often fall a hair short of a whole unit + + >>> _format_duration(0.089) + '0.089 seconds' + >>> _format_duration(1) + '1 second' + >>> _format_duration(4.9991) '5 seconds' + >>> _format_duration(90) + '1.5 minutes' + >>> _format_duration(7199.999) + '2.0 hours' >>> _format_duration(43474) - '12 hours' + '12.1 hours' + >>> _format_duration(3 * 86400 - 0.001) + '3.0 days' """ - return humanize.naturaldelta(datetime.timedelta(seconds=seconds)) + for unit, size in (("days", 86400), ("hours", 3600), ("minutes", 60)): + if seconds >= size: + return f"{seconds / size:.1f} {unit}" + # Significant digits, so that a gap under a second does not read as 0 + text = f"{seconds:.3g}" + return f"{text} second" if text == "1" else f"{text} seconds" + + +def _isoformat(unix_time: float) -> str: + return datetime.datetime.fromtimestamp( + unix_time, tz=datetime.timezone.utc + ).isoformat() diff --git a/tests/integration/test_process.py b/tests/integration/test_process.py index 83574937..5d1a81bb 100644 --- a/tests/integration/test_process.py +++ b/tests/integration/test_process.py @@ -9,6 +9,7 @@ import subprocess from pathlib import Path +import gpxpy import py.path import pytest @@ -623,11 +624,32 @@ def test_process_video_geotag_source_gpx_not_found(setup_data: py.path.local): assert descs[0]["error"]["type"] == "MapillaryVideoGPSNotFoundError" +# The GPS track of gopro_data/max-360mode.mp4 starts at 2019-11-18T23:44:42.59Z +_GOPRO_MAX_GPS_START = datetime.datetime( + 2019, 11, 18, 23, 44, 40, tzinfo=datetime.timezone.utc +) + + +def _copy_gpx_shifted_to( + src: py.path.local, dst: py.path.local, start: datetime.datetime +) -> None: + """Copy a GPX file with its times shifted to begin at start.""" + with open(src) as fp: + gpx = gpxpy.parse(fp) + start_time = gpx.get_time_bounds().start_time + assert start_time is not None + gpx.adjust_time(start - start_time) + dst.write(gpx.to_xml()) + + def test_process_video_geotag_source_with_gopro_gpx_specified( setup_data: py.path.local, ): video_path = setup_data.join("gopro_data").join("max-360mode.mp4") - gpx_file = setup_data.join("gpx").join("sf_30km_h.gpx") + gpx_file = setup_data.join("gpx").join("max-360mode.gpx") + _copy_gpx_shifted_to( + setup_data.join("gpx").join("sf_30km_h.gpx"), gpx_file, _GOPRO_MAX_GPS_START + ) descs = run_process_for_descs( [ @@ -645,6 +667,33 @@ def test_process_video_geotag_source_with_gopro_gpx_specified( assert len(descs[0]["MAPGPSTrack"]) > 0 +def test_process_video_geotag_source_with_gpx_outside_video( + setup_data: py.path.local, +): + """A GPX file recorded at another time than the video must not sync.""" + video_path = setup_data.join("gopro_data").join("max-360mode.mp4") + # Recorded in 2025, five years after the video + gpx_file = setup_data.join("gpx").join("sf_30km_h.gpx") + + descs = run_process_for_descs( + [ + *[ + "--video_geotag_source", + json.dumps({"source": "gpx", "source_path": str(gpx_file)}), + ], + str(video_path), + ] + ) + + assert len(descs) == 1 + assert descs[0]["error"]["type"] == "MapillaryOutsideGPXTrackError" + assert descs[0]["error"]["vars"] == { + "image_time": "2019_11_18_23_44_42_590", + "gpx_start_time": "2025_03_14_07_00_00_000", + "gpx_end_time": "2025_03_14_07_01_33_624", + } + + def test_process_geotag_with_gpx_pattern_not_found(setup_data: py.path.local): video_path = setup_data.join("gopro_data").join("max-360mode.mp4") @@ -661,8 +710,11 @@ def test_process_geotag_with_gpx_pattern_not_found(setup_data: py.path.local): def test_process_geotag_with_gpx_pattern(setup_data: py.path.local): video_path = setup_data.join("gopro_data").join("max-360mode.mp4") - gpx_file = setup_data.join("gpx").join("sf_30km_h.gpx") - gpx_file.copy(setup_data.join("gopro_data").join("max-360mode.gpx")) + _copy_gpx_shifted_to( + setup_data.join("gpx").join("sf_30km_h.gpx"), + setup_data.join("gopro_data").join("max-360mode.gpx"), + _GOPRO_MAX_GPS_START, + ) descs = run_process_for_descs( [ diff --git a/tests/integration/test_video_process.py b/tests/integration/test_video_process.py index 076724d1..bdd26788 100644 --- a/tests/integration/test_video_process.py +++ b/tests/integration/test_video_process.py @@ -207,7 +207,10 @@ def test_video_process_by_distance_refuses_to_guess_where_the_video_starts( def test_video_process_by_distance_with_a_gpx_that_misses_the_video( setup_data: py.path.local, ): - """No frame falls within the track: the command fails, instead of passing with no frames.""" + """ + The track ends two hours before the video starts: the command fails, saying + so, instead of passing with no frames. + """ pytest_skip_if_not_ffmpeg_installed() video_path = setup_data.join("videos").join("sample-5s.mp4") @@ -231,7 +234,8 @@ def test_video_process_by_distance_with_a_gpx_that_misses_the_video( assert not list(Path(dir, "samples").rglob("*.jpg")) assert 7 == ex.value.returncode, ex.value.stderr - assert "none of its frames falls within its GPS track" in ex.value.stderr + assert "MapillaryOutsideGPXTrackError" in ex.value.stderr + assert "misses the video" in ex.value.stderr def test_video_process_by_distance_applies_the_interpolation_offset( diff --git a/tests/unit/test_gpx_over_noisy_gps.py b/tests/unit/test_gpx_over_noisy_gps.py index ac4c82d8..846dc3d5 100644 --- a/tests/unit/test_gpx_over_noisy_gps.py +++ b/tests/unit/test_gpx_over_noisy_gps.py @@ -173,10 +173,13 @@ 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_good_clock_days_away_is_still_used( + def test_good_clock_days_away_rejects_the_gpx( self, monkeypatch, video_path, gpx_path ): - """Only a clock rejected with its noisy track is second-guessed.""" + """ + Only a clock rejected with its noisy track is second-guessed. A good + clock days away from the GPX says the GPX misses the video. + """ day = 24 * 3600.0 points = [ _point( @@ -189,9 +192,8 @@ def test_good_clock_days_away_is_still_used( ] _stub_gopro(monkeypatch, points) - metadata = GPXVideoExtractor(video_path, gpx_path).extract() - - assert metadata.points[0].time == pytest.approx(2 * day + 2.0) + with pytest.raises(exceptions.MapillaryOutsideGPXTrackError): + GPXVideoExtractor(video_path, gpx_path).extract() class TestEmptyGPSFallsBack: diff --git a/tests/unit/test_gpx_video_coverage.py b/tests/unit/test_gpx_video_coverage.py new file mode 100644 index 00000000..73a6581a --- /dev/null +++ b/tests/unit/test_gpx_video_coverage.py @@ -0,0 +1,472 @@ +# 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. + +""" +Tests for checking a GPX track, as placed on the timeline of a video, against +the video. + +A GPX track that misses the video has no position for any moment of it, so +it fails with MapillaryOutsideGPXTrackError, a geotagging error that lets the +next geotag source take its turn, with a warning. A track that covers only part +of the video warns. +""" + +from __future__ import annotations + +import datetime +import io +import logging +import pickle +import shutil +import typing as T +from pathlib import Path + +import pytest +from mapillary_tools import exceptions, geo, telemetry, types, uploader +from mapillary_tools.camm import camm_builder +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 +from mapillary_tools.mp4 import construct_mp4_parser as cparser, simple_mp4_builder + + +# 2026-08-12T08:26:27Z, in GPS time and in Unix time +A_GPS_TIME = 1470558405.9798455 +A_UNIX_TIME = 1786523187.9798455 + +# Seconds between the mp4 epoch (1904-01-01) and the Unix epoch +MP4_UNIX_DELTA = 2082844800 + +_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") +# Captured 2018-06-08T20:24:11Z at 45.5169, -122.5728 +IMAGE = _DATA.joinpath("images", "DSC00001.JPG") + +# A 10s video track starting at A_UNIX_TIME, in a 12s video +VIDEO_DURATION = 12.0 + + +def _camm_point(time: float, epoch_time: float) -> telemetry.CAMMGPSPoint: + return telemetry.CAMMGPSPoint( + time=time, + lat=37.8436443, + lon=14.9886571, + alt=1202.345, + angle=None, + epoch_time=epoch_time, + gps_fix_type=3, + horizontal_accuracy=0.0, + vertical_accuracy=0.0, + velocity_east=0.0, + velocity_north=0.0, + velocity_up=0.0, + speed_accuracy=0.0, + ) + + +def _video_track(rate: float = 1.0) -> list[telemetry.CAMMGPSPoint]: + """A GPS point every second of video, its clock running rate times faster""" + return [ + _camm_point(time=float(t), epoch_time=A_UNIX_TIME + rate * t) for t in range(11) + ] + + +# GPS without timestamps, so the GPX can only be assumed to start with the video +UNTIMED_TRACK = [ + geo.Point(time=float(t), lat=37.0, lon=14.0, alt=None, angle=None) + for t in range(11) +] + + +def _write_camm_mp4( + path: Path, + points: T.Sequence[geo.Point], + duration: float = VIDEO_DURATION, + creation_time: float | None = A_UNIX_TIME + 600, +) -> Path: + """ + Write points as a CAMM track into an empty mp4, the way the uploader does. + + duration is the duration of the video in seconds, or 0 for unknown. + creation_time is the Unix time to put in mvhd, or None to leave it unset. + """ + mp4_creation_time = ( + 0 if creation_time is None else int(creation_time) + MP4_UNIX_DELTA + ) + mvhd: cparser.BoxDict = { + "type": b"mvhd", + "data": { + "creation_time": mp4_creation_time, + "modification_time": mp4_creation_time, + "timescale": 1000, + "duration": int(duration * 1000), + }, + } + src = cparser.MP4WithoutSTBLBuilderConstruct.build_boxlist( + [ + {"type": b"ftyp", "data": b"test"}, + {"type": b"moov", "data": [mvhd]}, + ] + ) + metadata = types.VideoMetadata( + Path(""), filetype=types.FileType.CAMM, points=list(points) + ) + camm_info = uploader.VideoUploader.prepare_camm_info(metadata) + target_fp = simple_mp4_builder.transform_mp4( + io.BytesIO(src), camm_builder.camm_sample_generator2(camm_info) + ) + path.write_bytes(target_fp.read()) + return path + + +def _write_gpx(path: Path, start: float, seconds: float, step: float = 1.0) -> Path: + """A GPX track from start, a point every step seconds for seconds""" + count = int(round(seconds / step)) + trkpts = "".join( + f'' + for i in range(count + 1) + ) + path.write_text( + '' + '' + f"{trkpts}" + ) + return path + + +def _isoformat(unix_time: float) -> str: + return ( + datetime.datetime.fromtimestamp(unix_time, datetime.timezone.utc) + .isoformat(timespec="microseconds") + .replace("+00:00", "Z") + ) + + +def _warnings(caplog) -> list[str]: + return [r.getMessage() for r in caplog.records if r.levelno >= logging.WARNING] + + +class TestVideoWithGPSClock: + """The video's own GPS clock places the track, and says when it misses.""" + + def _extract( + self, + tmp_path: Path, + gpx_start: float, + gpx_seconds: float, + video_track: T.Sequence[geo.Point] | None = None, + duration: float = VIDEO_DURATION, + video_sync: VideoSyncOption | None = None, + ) -> types.VideoMetadata: + video_path = _write_camm_mp4( + tmp_path / "video.mp4", + _video_track() if video_track is None else video_track, + duration, + ) + gpx_path = _write_gpx(tmp_path / "track.gpx", gpx_start, gpx_seconds) + return GPXVideoExtractor(video_path, gpx_path, video_sync=video_sync).extract() + + @pytest.mark.parametrize( + "start, seconds", + [ + # Starts 30s before the video and ends after it + (A_UNIX_TIME - 30, 60), + # Started a second into the video, and ends with it + (A_UNIX_TIME + 1, 11), + # A long recording that spans the video + (A_UNIX_TIME - 3 * 24 * 3600, 6 * 24 * 3600), + ], + ) + def test_covering_gpx_is_silent(self, tmp_path, caplog, start, seconds): + with caplog.at_level(logging.WARNING): + self._extract(tmp_path, start, seconds) + assert not _warnings(caplog) + + @pytest.mark.parametrize( + "start, seconds, uncovered", + [ + # Ends 5s into the video + (A_UNIX_TIME - 30, 35, "7 seconds"), + # Starts 5s into the video + (A_UNIX_TIME + 5, 60, "5 seconds"), + # Starts after the video's own GPS gives out, before the video ends + (A_UNIX_TIME + 11, 20, "11 seconds"), + ], + ) + def test_partial_cover_warns(self, tmp_path, caplog, start, seconds, uncovered): + with caplog.at_level(logging.WARNING): + self._extract(tmp_path, start, seconds) + [warning] = _warnings(caplog) + assert "covers only part of" in warning + assert f"{uncovered} of the video fall outside the track" in warning + assert "video.mp4" in warning and "track.gpx" in warning + + @pytest.mark.parametrize( + "start", + [ + # Ended a minute before the video started + A_UNIX_TIME - 70, + # Starts after the video ends, though within a minute of it + A_UNIX_TIME + 13, + # Naive GPX timestamps read in a time zone 2h off + A_UNIX_TIME + 2 * 3600, + # A GPX file from another day + A_UNIX_TIME - 3 * 24 * 3600, + ], + ) + def test_missing_the_video_raises(self, tmp_path, start): + with pytest.raises(exceptions.MapillaryOutsideGPXTrackError) as info: + self._extract(tmp_path, start, 10) + message = str(info.value) + assert "video.mp4" in message and "track.gpx" in message + assert "belongs to this video" in message + assert "time zone" in message + + @pytest.mark.parametrize( + "gap, reported", + [ + # The GPX of the next clip, starting just after the video ends + (0.089, "by 0.089 seconds."), + (1, "by 1 second."), + (70, "by 1.2 minutes."), + (2 * 3600, "by 2.0 hours."), + (3 * 24 * 3600, "by 3.0 days."), + ], + ) + def test_gap_is_reported_in_a_readable_unit(self, tmp_path, gap, reported): + with pytest.raises(exceptions.MapillaryOutsideGPXTrackError) as info: + self._extract(tmp_path, A_UNIX_TIME + VIDEO_DURATION + gap, 10) + assert reported in str(info.value) + + def test_error_vars_name_the_video_start_and_the_gpx(self, tmp_path): + with pytest.raises(exceptions.MapillaryOutsideGPXTrackError) as info: + self._extract(tmp_path, A_UNIX_TIME + 2 * 3600, 10) + assert info.value.image_time == "2026_08_12_08_26_27_979" + assert info.value.gpx_start_time == "2026_08_12_10_26_27_979" + assert info.value.gpx_end_time == "2026_08_12_10_26_37_979" + + @pytest.mark.parametrize( + "start", + [ + A_UNIX_TIME - 70, + A_UNIX_TIME + 2 * 3600, + # Starts after the video's own GPS gives out, which the video + # itself may not + A_UNIX_TIME + 60, + ], + ) + def test_without_duration_a_gap_warns(self, tmp_path, caplog, start): + with caplog.at_level(logging.WARNING): + self._extract(tmp_path, start, 10, duration=0) + [warning] = _warnings(caplog) + assert "misses" in warning + + @pytest.mark.parametrize( + "start", [A_UNIX_TIME + 2 * 24 * 3600, A_UNIX_TIME - 3 * 24 * 3600] + ) + def test_without_duration_a_gap_of_days_raises(self, tmp_path, start): + with pytest.raises(exceptions.MapillaryOutsideGPXTrackError): + self._extract(tmp_path, start, 10, duration=0) + + def test_epoch_mixup_raises(self, tmp_path): + """Video timestamps left in GPS time sync ten years off.""" + video_track = [ + _camm_point(time=float(t), epoch_time=A_GPS_TIME + t) for t in range(11) + ] + video_path = _write_camm_mp4( + tmp_path / "video.mp4", video_track, creation_time=None + ) + gpx_path = _write_gpx(tmp_path / "track.gpx", A_UNIX_TIME, 10) + with pytest.raises(exceptions.MapillaryOutsideGPXTrackError) as info: + GPXVideoExtractor(video_path, gpx_path).extract() + assert "by 3657.0 days." in str(info.value) + + def test_the_interpolation_offset_is_part_of_the_placement(self, tmp_path, caplog): + """ + The geotagging stage looks the frames up 5s later along the track, so + a track that covers the video exactly leaves its last 5s uncovered. + """ + with caplog.at_level(logging.WARNING): + self._extract( + tmp_path, + A_UNIX_TIME, + VIDEO_DURATION, + video_sync=VideoSyncOption(offset_time=5), + ) + [warning] = _warnings(caplog) + assert "5 seconds of the video fall outside the track" in warning + + def test_error_survives_a_worker_process(self, tmp_path): + """Videos are geotagged in a process pool, so the error gets pickled.""" + with pytest.raises(exceptions.MapillaryOutsideGPXTrackError) as info: + self._extract(tmp_path, A_UNIX_TIME + 2 * 3600, 10) + + unpickled = pickle.loads(pickle.dumps(info.value)) + + assert str(unpickled) == str(info.value) + assert vars(unpickled) == vars(info.value) + + +class TestTimelapse: + """The check is made on the placement the GPS clock gives the track.""" + + # 11 seconds of video spanning 165 seconds of GPS time + RATE = 15.0 + + def _extract(self, tmp_path, gpx_start: float, gpx_seconds: float): + 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() + + 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.""" + with caplog.at_level(logging.WARNING): + metadata = self._extract(tmp_path, A_UNIX_TIME, 11 * self.RATE) + assert not _warnings(caplog) + assert metadata.points[-1].time == pytest.approx(11.0, abs=1e-3) + + def test_gpx_of_half_the_recording_warns(self, tmp_path, caplog): + with caplog.at_level(logging.WARNING): + # The first 6 of the video's 11 seconds + self._extract(tmp_path, A_UNIX_TIME, 6 * self.RATE) + [warning] = _warnings(caplog) + assert "5 seconds of the video fall outside the track" in warning + + def test_gap_is_reported_in_gps_time(self, tmp_path): + """The GPX starts an hour after the recording ended.""" + recording_end = A_UNIX_TIME + 11 * self.RATE + with pytest.raises(exceptions.MapillaryOutsideGPXTrackError) as info: + self._extract(tmp_path, recording_end + 3600, 60) + assert "by 1.0 hours." in str(info.value) + + +class TestVideoWithoutGPSClock: + """ + Placed by --video_start_time, the track can be found to miss the video. + Assumed to start with the video, it can only fall short of its end. + """ + + def _extract( + self, tmp_path, video_path: Path, gpx_start: float, gpx_seconds: float, **sync + ): + gpx_path = _write_gpx(tmp_path / "track.gpx", gpx_start, gpx_seconds) + return GPXVideoExtractor( + video_path, gpx_path, video_sync=VideoSyncOption(**sync) + ).extract() + + def test_start_time_two_hours_off_raises(self, tmp_path): + with pytest.raises(exceptions.MapillaryOutsideGPXTrackError) as info: + self._extract( + tmp_path, NO_GPS_VIDEO, A_UNIX_TIME, 30, start_time=A_UNIX_TIME + 7200 + ) + assert "by 2.0 hours." in str(info.value) + + def test_start_time_within_the_track_is_silent(self, tmp_path, caplog): + with caplog.at_level(logging.WARNING): + self._extract( + tmp_path, NO_GPS_VIDEO, A_UNIX_TIME, 30, start_time=A_UNIX_TIME + 10 + ) + assert not _warnings(caplog) + + @pytest.mark.parametrize( + "video_track", [[], UNTIMED_TRACK], ids=["no GPS", "GPS without timestamps"] + ) + def test_assumed_start_warns_of_a_short_track(self, tmp_path, caplog, video_track): + video_path = _write_camm_mp4(tmp_path / "video.mp4", video_track) + with caplog.at_level(logging.WARNING): + # Covers the first 6 of the video's 12 seconds + self._extract(tmp_path, video_path, A_UNIX_TIME, 6) + [warning] = _warnings(caplog) + assert "6 seconds of the video fall outside the track" in warning + + def test_assumed_start_is_silent_on_a_covering_track(self, tmp_path, caplog): + video_path = _write_camm_mp4(tmp_path / "video.mp4", UNTIMED_TRACK) + with caplog.at_level(logging.WARNING): + self._extract(tmp_path, video_path, A_UNIX_TIME, VIDEO_DURATION) + assert not _warnings(caplog) + + +class TestFallThrough: + """ + A GPX that misses a file lets the next geotag source take its turn, with a + warning when that source geotags the file, so the GPX does not go unnoticed. + """ + + def test_video_is_geotagged_by_the_next_source_with_a_warning( + self, tmp_path, caplog + ): + video_path = _write_camm_mp4(tmp_path / "video.mp4", _video_track()) + gpx_path = _write_gpx(tmp_path / "track.gpx", A_UNIX_TIME + 2 * 3600, 10) + options = [ + SourceOption( + SourceType.GPX, + num_processes=0, + source_path=SourcePathOption(source_path=gpx_path), + ), + SourceOption(SourceType.NATIVE, num_processes=0), + ] + + with caplog.at_level(logging.WARNING): + [metadata] = factory.process([video_path], options) + + assert isinstance(metadata, types.VideoMetadata) + assert [p.time for p in metadata.points] == [float(t) for t in range(11)] + [warning] = _warnings(caplog) + assert "misses the video" in warning and "by 2.0 hours" in warning + assert warning.endswith("Geotagged it from the native source instead") + + def _process_images( + self, tmp_path: Path, sources: list[SourceType], count: int = 1 + ) -> list[types.MetadataOrError]: + image_paths = [] + for i in range(count): + image_path = tmp_path / f"image{i}.jpg" + shutil.copyfile(IMAGE, image_path) + image_paths.append(image_path) + # A track recorded years after the images + gpx_path = _write_gpx(tmp_path / "walk.gpx", A_UNIX_TIME, 10) + options = [ + SourceOption( + source, + num_processes=0, + source_path=SourcePathOption(source_path=gpx_path), + ) + if source is SourceType.GPX + else SourceOption(source, num_processes=0) + for source in sources + ] + return factory.process(image_paths, options) + + def test_images_are_geotagged_by_the_next_source_with_one_warning( + self, tmp_path, caplog + ): + with caplog.at_level(logging.WARNING): + metadatas = self._process_images( + tmp_path, [SourceType.GPX, SourceType.EXIF], count=3 + ) + + assert all(isinstance(m, types.ImageMetadata) for m in metadatas) + assert {(round(m.lat, 4), round(m.lon, 4)) for m in metadatas} == { + (45.5169, -122.5728) + } + [warning] = _warnings(caplog) + assert warning == ( + f"3 images fall outside the GPX track in {tmp_path / 'walk.gpx'}. " + "Geotagged them from the exif source instead" + ) + + def test_gpx_as_the_only_source_fails_the_image(self, tmp_path): + [metadata] = self._process_images(tmp_path, [SourceType.GPX]) + + assert isinstance(metadata, types.ErrorMetadata) + assert isinstance(metadata.error, exceptions.MapillaryOutsideGPXTrackError) diff --git a/tests/unit/test_gpx_video_sync.py b/tests/unit/test_gpx_video_sync.py index 42ebebfa..619e13e6 100644 --- a/tests/unit/test_gpx_video_sync.py +++ b/tests/unit/test_gpx_video_sync.py @@ -30,8 +30,12 @@ # No GPS of its own, and 5.76 seconds long NO_GPS_VIDEO = _DATA.joinpath("videos", "sample-5s.mp4") -# GPS of its own, recorded from 2019-11-18T15:44:47Z +# GPS of its own, recorded from 2019-11-18T23:44:42.59Z to 23:44:54.64Z GPS_VIDEO = _DATA.joinpath("gopro_data", "max-360mode.mp4") +# Shortly before the GPS track of GPS_VIDEO starts +_GPS_VIDEO_START = datetime.datetime( + 2019, 11, 18, 23, 44, 40, tzinfo=datetime.timezone.utc +) GPX_START = datetime.datetime(2025, 3, 14, 7, 0, 0, tzinfo=datetime.timezone.utc) @@ -147,9 +151,7 @@ def test_the_video_gps_wins_over_the_start_time(self, tmp_path): gpx = _write_gpx( tmp_path / "track.gpx", seconds=30, - start=datetime.datetime( - 2019, 11, 18, 15, 44, 40, tzinfo=datetime.timezone.utc - ), + start=_GPS_VIDEO_START, ) far_off = GPX_START.timestamp() @@ -192,9 +194,7 @@ def test_the_offset_applies_to_a_track_synced_by_the_video_gps(self, tmp_path): gpx = _write_gpx( tmp_path / "track.gpx", seconds=30, - start=datetime.datetime( - 2019, 11, 18, 15, 44, 40, tzinfo=datetime.timezone.utc - ), + start=_GPS_VIDEO_START, ) synced = _extract(GPS_VIDEO, gpx, VideoSyncOption())