From de21f059c0835a0371c259681fe547ae2f39a206 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Tue, 27 Oct 2026 15:56:30 +0100 Subject: [PATCH 1/3] Report why a GPX track was not used when no other source geotags the file A GPX track that misses a file lets the later geotag sources take their turn. When none of them geotagged the file, the error reported was that of the last one to fail, which hid the miss. A video whose own GPS is too noisy, given a GPX of another day, failed as "GPS is too noisy": the very problem the GPX was given to work around. The Desktop Uploader, whose chain runs two GPX sources, exiftool_xml, native and exiftool_runtime, shows only that error. Report the miss instead: "The GPX track in X (...) misses the video Y (...) by 105.1 days. Check that the GPX file belongs to this video, ...". An error that ends the chain, such as "Stationary video", is still reported, as it says something about the file itself, and a file that a later source geotags still only gets the warning it gets now. --- mapillary_tools/geotag/factory.py | 11 ++++- tests/unit/test_gpx_over_noisy_gps.py | 61 +++++++++++++++++++++++++++ tests/unit/test_gpx_video_coverage.py | 24 +++++++++++ 3 files changed, 95 insertions(+), 1 deletion(-) diff --git a/mapillary_tools/geotag/factory.py b/mapillary_tools/geotag/factory.py index 87c0016d4..e11984ecc 100644 --- a/mapillary_tools/geotag/factory.py +++ b/mapillary_tools/geotag/factory.py @@ -289,7 +289,8 @@ def process( 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 + # warn about if a later source geotags them instead, and to report if none + # does gpx_rejections: dict[Path, tuple[SourceOption, types.ErrorMetadata]] = {} # Videos let through to sources that cannot geotag videos (e.g. exif), with @@ -399,6 +400,14 @@ def process( if not reprocessable_paths: break + # Of a file that a GPX track missed, and that no later source geotagged, + # report the miss: why the track given for it was not used says more than + # the later sources failing too, such as "GPS is too noisy" for a video + # whose GPS the track was given to replace + for path, (_, gpx_miss) in gpx_rejections.items(): + if path in pending_errors: + pending_errors[path] = gpx_miss + final_metadatas.extend(pending_errors.values()) unread_paths = [ diff --git a/tests/unit/test_gpx_over_noisy_gps.py b/tests/unit/test_gpx_over_noisy_gps.py index f8c494112..ac586847a 100644 --- a/tests/unit/test_gpx_over_noisy_gps.py +++ b/tests/unit/test_gpx_over_noisy_gps.py @@ -374,6 +374,13 @@ def build_video_geotag(option: SourceOption): GPX_NOT_FOUND = exceptions.MapillaryVideoGPSNotFoundError( "GPX file not found for video" ) +GPX_MISS = exceptions.MapillaryOutsideGPXTrackError( + "The GPX track misses the video by 105.1 days", + image_time="2026_09_11_15_39_41_000", + gpx_start_time="2026_05_29_12_47_06_000", + gpx_end_time="2026_05_29_12_53_00_000", +) +XML_NOT_FOUND = exceptions.MapillaryExifToolXMLNotFoundError("No XML file found") DEFAULT_CHAIN = [SourceType(source) for source in DEFAULT_GEOTAG_SOURCE_OPTIONS] @@ -505,6 +512,60 @@ def test_gpx_that_fails_replaces_the_rejection(self, monkeypatch): assert isinstance(result, types.ErrorMetadata) assert result.error is mismatch + def test_gpx_that_misses_is_reported_over_rejected_gps(self, monkeypatch): + """ + The chain of the Desktop Uploader, with a GPX that misses a video whose + GPS is too noisy: the miss says why the GPX was not used, while "GPS + is too noisy" is what the GPX was given to work around. + """ + results, called = _run_chain( + monkeypatch, + [ + SourceType.GPX, + SourceType.EXIFTOOL_XML, + SourceType.NATIVE, + SourceType.EXIFTOOL_RUNTIME, + ], + { + SourceType.GPX: GPX_MISS, + SourceType.EXIFTOOL_XML: XML_NOT_FOUND, + SourceType.NATIVE: NOISE, + SourceType.EXIFTOOL_RUNTIME: None, + }, + ) + result = _only_result(results) + assert isinstance(result, types.ErrorMetadata) + assert result.error is GPX_MISS + assert called == [SourceType.GPX, SourceType.EXIFTOOL_XML, SourceType.NATIVE] + + @pytest.mark.parametrize( + "later_error", + [NOISE, EMPTY, NOT_FOUND, exceptions.MapillaryExiftoolNotFoundError("no")], + ) + def test_gpx_that_misses_is_reported_over_later_failures( + self, monkeypatch, later_error + ): + results, _ = _run_chain( + monkeypatch, + [SourceType.GPX, SourceType.NATIVE], + {SourceType.GPX: GPX_MISS, SourceType.NATIVE: later_error}, + ) + result = _only_result(results) + assert isinstance(result, types.ErrorMetadata) + assert result.error is GPX_MISS + + def test_gpx_that_misses_does_not_hide_a_final_error(self, monkeypatch): + """An error that ends the chain says something about the file itself""" + stationary = exceptions.MapillaryStationaryVideoError("Stationary") + results, _ = _run_chain( + monkeypatch, + [SourceType.GPX, SourceType.NATIVE], + {SourceType.GPX: GPX_MISS, SourceType.NATIVE: stationary}, + ) + result = _only_result(results) + assert isinstance(result, types.ErrorMetadata) + assert result.error is stationary + def test_not_found_still_falls_through_to_exiftool(self, monkeypatch): """Finding nothing is a verdict on the reader, so retrying is fair.""" results, called = _run_chain( diff --git a/tests/unit/test_gpx_video_coverage.py b/tests/unit/test_gpx_video_coverage.py index 8fa98c199..2e657e30c 100644 --- a/tests/unit/test_gpx_video_coverage.py +++ b/tests/unit/test_gpx_video_coverage.py @@ -527,6 +527,30 @@ def test_video_is_geotagged_by_the_next_source_with_a_warning( assert "misses the video" in warning and "by 2.0 hours" in warning assert warning.endswith("Geotagged it from the native source instead") + def test_the_miss_is_reported_if_no_later_source_geotags_the_video(self, tmp_path): + 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), + ), + # Finds no GPX file to read + SourceOption( + SourceType.GPX, + num_processes=0, + source_path=SourcePathOption(source_path=tmp_path / "none.gpx"), + ), + ] + + [metadata] = factory.process([video_path], options) + + assert isinstance(metadata, types.ErrorMetadata) + assert isinstance(metadata.error, exceptions.MapillaryOutsideGPXTrackError) + assert "misses the video" in str(metadata.error) + assert "by 2.0 hours" in str(metadata.error) + @pytest.mark.parametrize( "gpx_start, source", [(A_UNIX_TIME, SourceType.GPX), (A_UNIX_TIME + 2 * 3600, SourceType.NATIVE)], From 0b078715708b4584fe47cfded9d8127c55552548 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Tue, 27 Oct 2026 16:04:51 +0100 Subject: [PATCH 2/3] Trust the GPS timestamps of a video, never its creation time A video's GPS clock places a GPX track along it, and tells whether the track is of the same recording. Two things kept it from doing so: - A GoPro stamps a placeholder date until its GPS has the time, and a GPS rejected as noise may never get past it. Such a clock was told apart by the creation time of the video, which comes from the camera's own clock and can be anything. A noisy GPS whose clock disagreed with the creation time got any GPX synced from the start of the video, even one recorded months before (#858). - Only a leap of more than a day was taken for the GPS acquiring the time. Judge the timestamps alone. Leave out those before 2016 (the placeholder dates are 2000-01-01 and dates in October 2015) or more than a day in the future, and keep only those after the last leap of more than a minute ahead of the video. What is left is the GPS time, whether or not the positions were too noisy to publish, so it places the GPX, and a GPX of another time misses the video. The creation time is no longer read for that. When the clock has no timestamps, sampling looked for a start time in the GPS timestamps by rules of its own, which leave out only those before 2010. Now that the clock leaves out placeholder dates, that would bring them back, so it reads the same clock. --- .../geotag/video_extractors/gpx.py | 120 +++++++----------- mapillary_tools/sample_video.py | 47 ++----- tests/unit/test_gpx_over_noisy_gps.py | 60 ++++++--- tests/unit/test_gpx_video_timestamps.py | 31 +++++ tests/unit/test_sample_video.py | 22 +++- 5 files changed, 142 insertions(+), 138 deletions(-) diff --git a/mapillary_tools/geotag/video_extractors/gpx.py b/mapillary_tools/geotag/video_extractors/gpx.py index c1a7b1ff4..f7e4b6da8 100644 --- a/mapillary_tools/geotag/video_extractors/gpx.py +++ b/mapillary_tools/geotag/video_extractors/gpx.py @@ -13,6 +13,7 @@ import math import statistics import sys +import time import typing as T from pathlib import Path @@ -25,7 +26,6 @@ from typing_extensions import override from ... import constants, 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 @@ -41,6 +41,21 @@ # of magnitude. _IMPLAUSIBLE_OFFSET_SECONDS = 24 * 3600 +# A video GPS timestamp before this is a placeholder, not the GPS time: until it +# acquires the time, a GoPro stamps 2000-01-01, or a date in October 2015 +_MIN_GPS_UNIX_TIME = datetime.datetime( + 2016, 1, 1, tzinfo=datetime.timezone.utc +).timestamp() + +# Nor can a video GPS timestamp lie further than this in the future +_MAX_GPS_FUTURE_SECONDS = 24 * 3600 + +# A GPS clock that leaps ahead of the video by more than this has just acquired +# the GPS time, as a GoPro does from its placeholder date: only its timestamps +# from there on tell when the video was recorded. Stale first timestamps (see +# _STALE_STEP_RATE) lag by seconds +_MAX_CLOCK_JUMP_SECONDS = 60.0 + # 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 @@ -208,7 +223,7 @@ def extract(self) -> types.VideoMetadata: ) try: - native_video_metadata, noisy = self._extract_native_video_metadata() + native_video_metadata, _ = self._extract_native_video_metadata() except exceptions.MapillaryVideoGPSUnusableError as ex: # The GPX replaces the video's own GPS, so GPS that is missing, # empty or too noisy must not reject the video @@ -228,16 +243,7 @@ def extract(self) -> types.VideoMetadata: ) video_clock = self._read_video_clock(native_video_metadata.points) - clock = self._sync_clock(video_clock, gpx_points, noisy) - if video_clock is not None and clock is None: - LOG.warning( - "The GPS clock of %s, whose GPS was rejected as noise, is %.1f days away from %s. " - "Syncing the %s from the start of the video instead", - self.video_path, - abs(gpx_points[0].time - video_clock.to_unix_time(0)) / 86400, - self.gpx_path, - self.file_format, - ) + clock = self._sync_clock(video_clock, gpx_points) if clock is not None and gpx_points and self.sync_mode is SyncMode.REBASE: # Distance sampling reads the frames by their composition times: # start the track at that of the first frame @@ -841,14 +847,19 @@ def _sync_clock( self, clock: VideoClock | None, gpx_points: T.Sequence[geo.Point], - noisy: bool, ) -> VideoClock | None: """ The clock to place the GPX track by along the video, from the GPS - clock of the video (None if it has none), whose GPS was rejected as - noise if **noisy**. Return None if the track is to be placed as for a - video without a GPS clock. --interpolation_use_gpx_start_time starts - the track at video time 0. + clock of the video (None if it has none). Return None if the track is + to be placed as for a video without a GPS clock. + --interpolation_use_gpx_start_time starts the track at video time 0. + + The GPS clock places the track whether or not the GPS positions were + rejected as noise: its timestamps, which _read_timestamps() keeps only + once they are the GPS time, tell when the video was recorded, and a + track recorded at another time misses the video. The camera's own + clock, such as the creation time of the video, is never consulted: + it can be anything. """ if clock is None or not gpx_points: return clock @@ -861,40 +872,8 @@ def _sync_clock( video_time=0.0, unix_time=gpx_points[0].time, rate=clock.rate ) - # 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 - offset = gpx_points[0].time - clock.to_unix_time(0) - if ( - abs(offset) > _IMPLAUSIBLE_OFFSET_SECONDS - and noisy - and self.sync_mode is SyncMode.SYNC - and not self._clock_matches_creation_time(clock) - ): - return None - return clock - 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 _warn_of_a_long_track(self, gpx_points: T.Sequence[geo.Point]) -> None: """ Warn, in process, of a GPX track assumed to start with the video that @@ -1027,17 +1006,22 @@ def _read_timestamps( ) -> list[tuple[float, float]]: """ Read the (video time, Unix time) pairs of the GPS timestamps of the - points, leaving out those that the others show to be wrong: those out - of order, those before a jump of days (from a placeholder date to the - GPS time), and stale first ones + points, leaving out those that cannot be the GPS time (zero, before + 2016, or in the future), and those that the others show to be wrong: + those out of order, those before a leap of more than a minute (from a + placeholder date to the GPS time), and stale first ones. + + Only the timestamps are judged, not the positions: a GPS can have the + time before it has a position good enough to publish. >>> points = [ - ... telemetry.GPSPoint(time=v, lat=1, lon=1, alt=None, angle=None, epoch_time=u, fix=None, precision=None, ground_speed=None) - ... for v, u in [(0, 992.4), (0.4, 1000.4), (1.4, 1001.4), (1.8, 1001.4), (2.4, 1002.4)] + ... telemetry.GPSPoint(time=v, lat=1, lon=1, alt=None, angle=None, epoch_time=1789141180 + u, fix=None, precision=None, ground_speed=None) + ... for v, u in [(0, -7.6), (0.4, 0.4), (1.4, 1.4), (1.8, 1.4), (2.4, 2.4)] ... ] - >>> GPXVideoExtractor._read_timestamps(points) - [(0.4, 1000.4), (1.4, 1001.4), (2.4, 1002.4)] + >>> [(v, round(u - 1789141180, 1)) for v, u in GPXVideoExtractor._read_timestamps(points)] + [(0.4, 0.4), (1.4, 1.4), (2.4, 2.4)] """ + latest = time.time() + _MAX_GPS_FUTURE_SECONDS pairs: list[tuple[float, float]] = [] for p in video_gps_points: # Video GPS timestamps are stored in whatever epoch their container @@ -1046,6 +1030,11 @@ def _read_timestamps( unix_time = p.get_unix_time() if unix_time is None: continue + if not _MIN_GPS_UNIX_TIME <= unix_time <= latest: + # A placeholder date, or a clock gone wrong + continue + if not math.isfinite(p.time): + 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 @@ -1062,11 +1051,13 @@ def _read_timestamps( if pace is None: return pairs - # GoPro stamps a placeholder date until the GPS acquires the time + # Until it acquires the GPS time, a GoPro stamps a placeholder date, + # which can be one since 2016: keep the timestamps after the last leap + # (see _MAX_CLOCK_JUMP_SECONDS) start = 0 for idx in range(1, len(pairs)): (v0, u0), (v1, u1) = pairs[idx - 1], pairs[idx] - if _IMPLAUSIBLE_OFFSET_SECONDS < (u1 - u0) - (v1 - v0) * pace: + if _MAX_CLOCK_JUMP_SECONDS < (u1 - u0) - (v1 - v0) * pace: start = idx # Stale first timestamps (see _STALE_STEP_RATE). Judged by the steps of @@ -1356,19 +1347,6 @@ def _first_frame_time(video_path: Path) -> float: return 0.0 -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/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py index c3256b203..cf0ae607c 100644 --- a/mapillary_tools/sample_video.py +++ b/mapillary_tools/sample_video.py @@ -11,7 +11,6 @@ import os import re import shutil -import statistics import time import typing as T from contextlib import contextmanager @@ -457,21 +456,6 @@ def wip_sample_dir(sample_dir: Path) -> Path: ) -# GPS clocks that report a time before this have never been set: GoPro writes -# 2000-01-01 until it gets its first fix, for example -_MIN_PLAUSIBLE_START_TIME = datetime.datetime( - 2010, 1, 1, tzinfo=datetime.timezone.utc -).timestamp() - -# Leave room for the clock of the machine running this to be behind -_MAX_FUTURE_START_TIME_SECONDS = 24 * 3600 - -# How many GPS timestamps to take the median of, so that a single bad one cannot -# shift the start time: the Labpano PanoX V2 can record a stale first fix, -# seconds older than the rest. Only the first few are used because in timelapses -# the video clock runs slower than the GPS clock, so the two drift apart -_GPS_CLOCK_SAMPLES = 5 - # Cameras known to stamp the creation time at the end of the recording, as # lowercase (make, model) from the container tags. BlackVue is not listed # because it does not write those tags: blackvue_parser.is_blackvue detects it @@ -499,32 +483,17 @@ def _gps_clock_start_time( 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 from - its own absolute timestamp gives the wall clock at which the video started. - Timestamps outside the plausible range are skipped. + The wall clock time at which the video started, by its GPS clock, or None + if it has none: by the timestamps that GPX sync reads the clock from (see + GPXVideoExtractor._read_timestamps), at the clock's pace, which a timelapse + packs several GPS seconds into each second of video at """ - max_start_time = time.time() + _MAX_FUTURE_START_TIME_SECONDS - - start_times: list[float] = [] - for point in points: - unix_time = point.get_unix_time() - if unix_time is None: - continue - 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 - start_times.append(start_time) - if len(start_times) >= _GPS_CLOCK_SAMPLES: - break - - if not start_times: + clock = GPXVideoExtractor._read_video_clock(points) + if clock is None: return None return datetime.datetime.fromtimestamp( - statistics.median(start_times), tz=datetime.timezone.utc + clock.to_unix_time(0.0), tz=datetime.timezone.utc ) @@ -787,7 +756,7 @@ def _gpx_sync_clock( if not noisy: return None video_clock = extractor._read_video_clock(native_video_metadata.points) - return extractor._sync_clock(video_clock, gpx_points, noisy) + return extractor._sync_clock(video_clock, gpx_points) return None diff --git a/tests/unit/test_gpx_over_noisy_gps.py b/tests/unit/test_gpx_over_noisy_gps.py index ac586847a..0ba2d5ee9 100644 --- a/tests/unit/test_gpx_over_noisy_gps.py +++ b/tests/unit/test_gpx_over_noisy_gps.py @@ -188,12 +188,18 @@ def test_good_fixes_are_preferred_as_the_clock( assert [p.time for p in metadata.points] == pytest.approx([2.0, 4.0]) - def test_noisy_clock_days_away_is_ignored(self, monkeypatch, video_path, gpx_path): + @pytest.mark.parametrize("creation_time", [None, A_UNIX_TIME]) + def test_placeholder_clock_is_ignored( + self, monkeypatch, video_path, gpx_path, creation_time + ): """ - Until it acquires GPS time, a GoPro stamps a placeholder date, which a - GPS rejected as noise may never have got past. Syncing by it would put - the GPX years away from the video. + Until it acquires the GPS time, a GoPro stamps a placeholder date, which + a GPS rejected as noise may never have got past. Such a clock says + nothing of when the video was recorded, so the GPX is synced from the + start of the video, whatever the creation time of the video says. """ + if creation_time is not None: + _write_mvhd(video_path, creation_time=creation_time) points = [ _point(time=i * 0.1, epoch_time=PLACEHOLDER_UNIX_TIME + i * 0.1) for i in range(32) @@ -205,31 +211,47 @@ 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( + def test_noisy_clock_after_its_placeholder_date_places_the_gpx( 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) + """Once the GPS acquires the time, its timestamps are the GPS time""" points = [ _point(time=i * 0.1, epoch_time=PLACEHOLDER_UNIX_TIME + i * 0.1) - for i in range(32) + for i in range(10) + ] + points += [ + _point(time=i * 0.1, epoch_time=A_UNIX_TIME + i * 0.1) + for i in range(10, 32) ] _stub_gopro(monkeypatch, points) metadata = GPXVideoExtractor(video_path, gpx_path).extract() - assert [p.time for p in metadata.points] == [0.0, 2.0] + assert [p.time for p in metadata.points] == pytest.approx([2.0, 4.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 + @pytest.mark.parametrize( + "creation_time", + [ + # In local time, 8 hours behind UTC, as GoPro writes it + A_UNIX_TIME - 8 * 3600, + # A creation time that means nothing: 2016-01-01T01:44:00Z, as a + # GoPro HERO10 recorded in 2022 reports + 1451612640, + # None + None, + ], + ) + def test_noisy_clock_rejects_a_gpx_days_away( + self, video_path, tmp_path, noisy_gopro, video_sync, creation_time ): """ - 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. + The timestamps of GPS rejected as noise are the GPS time all the same, + so a GPX days away from them misses the video, as for good GPS. The + creation time of the video, from the camera's own clock, is not asked. """ - # In local time, 8 hours behind UTC, as GoPro writes it - _write_mvhd(video_path, creation_time=A_UNIX_TIME - 8 * 3600) + if creation_time is not None: + _write_mvhd(video_path, creation_time=creation_time) gpx_path = tmp_path / "another_day.gpx" gpx_path.write_text(_gpx_xml("2026-09-09T15:39:43Z", "2026-09-09T15:39:45Z")) @@ -242,8 +264,7 @@ def test_noisy_clock_syncs_despite_a_meaningless_creation_time( ): """ 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. + time, which does not keep its GPS clock from placing the GPX. """ # 2016-01-01T01:44:00Z _write_mvhd(video_path, creation_time=1451612640) @@ -255,10 +276,7 @@ def test_noisy_clock_syncs_despite_a_meaningless_creation_time( 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. A good - clock days away from the GPX says the GPX misses the video. - """ + """A good clock days away from the GPX says the GPX misses the video""" day = 24 * 3600.0 points = [ _point( diff --git a/tests/unit/test_gpx_video_timestamps.py b/tests/unit/test_gpx_video_timestamps.py index ea3346c69..b98103ab4 100644 --- a/tests/unit/test_gpx_video_timestamps.py +++ b/tests/unit/test_gpx_video_timestamps.py @@ -118,6 +118,37 @@ def test_without_a_placeholder_date(self): assert _read(pairs) == [(2, 2), (3, 3)] + def test_without_timestamps_before_a_leap_of_minutes(self): + """ + A clock that leaps 2 hours ahead of the video has just got the GPS + time, such as from a camera clock in local time + """ + pairs = [(0, -7200), (1, -7199), (2, -7198), (3, 3), (4, 4), (5, 5)] + + assert _read(pairs) == [(3, 3), (4, 4), (5, 5)] + + def test_with_a_step_of_less_than_a_minute(self): + """Further on, a step of seconds is the clock's, not a leap to the time""" + pairs = [(0, 0), (1, 1), (2, 32), (3, 33), (4, 34)] + + assert _read(pairs) == pairs + + def test_without_timestamps_before_2016(self): + """A placeholder date that the GPS never got past""" + before = datetime.datetime(2015, 10, 22, tzinfo=datetime.timezone.utc) + offset = before.timestamp() - START.timestamp() + + assert _read([(v, offset + v) for v in range(5)]) == [] + + def test_without_timestamps_in_the_future(self): + future = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta( + days=2 + ) + offset = future.timestamp() - START.timestamp() + pairs = [(0, 0), (1, 1), (2, 2), (3, offset + 3), (4, offset + 4)] + + assert _read(pairs) == [(0, 0), (1, 1), (2, 2)] + def test_none_without_timestamps(self): points = _points([(0, 0)]) points[0] = geo.Point(time=0, lat=37, lon=-122, alt=None, angle=None) diff --git a/tests/unit/test_sample_video.py b/tests/unit/test_sample_video.py index 3953fdebc..d82c69e5b 100644 --- a/tests/unit/test_sample_video.py +++ b/tests/unit/test_sample_video.py @@ -878,10 +878,11 @@ def test_started_with_the_gpx_at_the_pace_of_the_clock( With --interpolation_use_gpx_start_time, the clock gives only the pace: the samples were all stamped within the first minute of the track """ - # A placeholder date, which the GPX start replaces + # A clock a year away from the GPX, whose start replaces it. A clock + # that never got past a placeholder date before 2016 gives no pace self._mock_noisy_telemetry( monkeypatch, - datetime.datetime(2015, 1, 1, tzinfo=datetime.timezone.utc).timestamp(), + datetime.datetime(2024, 1, 1, tzinfo=datetime.timezone.utc).timestamp(), ) gpx_start = self.GPS_START_TIME + datetime.timedelta(minutes=5) gpx = self._write_gpx(Path(tmpdir) / "track.gpx", gpx_start) @@ -1235,16 +1236,23 @@ def test_median_ignores_a_bad_timestamp(self) -> None: 2021, 8, 10, 12, 38, 6, tzinfo=datetime.timezone.utc ) - def test_uses_the_start_of_a_timelapse(self) -> None: + def test_uses_the_clock_rate_of_a_timelapse(self) -> None: # Each second of this timelapse spans 10 seconds of GPS time, so the - # offset between the two clocks grows by 9 seconds per point. Reading it - # at the start of the track keeps the error to a couple of points - # rather than half the track + # offset between the two clocks grows by 9 seconds per point: mapped + # back at the rate of the clock, every point says the same start points = [self._gps_point(float(i), 1628599086.0 + 10 * i) for i in range(1000)] assert sample_video._gps_clock_start_time(points) == datetime.datetime( - 2021, 8, 10, 12, 38, 24, tzinfo=datetime.timezone.utc + 2021, 8, 10, 12, 38, 6, tzinfo=datetime.timezone.utc ) + def test_skips_a_placeholder_date(self) -> None: + # GoPro stamps a date in October 2015 until it acquires the GPS time + points = [ + self._gps_point(0.0, 1445507624.0), + self._gps_point(1.0, 1445507625.0), + ] + assert sample_video._gps_clock_start_time(points) is None + def test_no_absolute_timestamps(self) -> None: assert sample_video._gps_clock_start_time(_make_gps_points(3)) is None From 44880d590877e3d082665b1a6deec46530a82bfa Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Wed, 28 Oct 2026 10:29:59 +0100 Subject: [PATCH 3/3] Place a GPX along a video without GPS by its creation time, with a warning A video without GPS timestamps of its own has nothing better than its creation time to say when it was recorded. That comes from the camera's clock, which can be off, and marks the start or the end of the recording depending on the camera. Without --video_start_time, a GPX track was placed along such a video: - in process, as if it started with the video, with a warning when it ran on for longer than the video; - in distance sampling the same way, but refused, asking for --video_start_time, when it ran on for more than twice the length of the video. Interval sampling with --video_geotag_source refused it alike, leaving every sample without a location; - in interval sampling with --geotag_source, by the creation time, whether or not the track covered the video there, and not at all without one: "Unable to extract video start time". Now all of them place the track by the creation time if the track covers some of the video there, with a warning that the camera's clock can be off and how to correct it. Where nothing tells whether the creation time marks the start or the end of the recording, the one that the track covers is taken. Otherwise the track is assumed to start with the video, with a warning that the creation time puts the video elsewhere. A track is no longer refused for want of a start time: it is taken to be right. Only --interpolation_use_gpx_start_time still refuses a track too long to end with the video. In process, a track placed by the creation time is clipped to the video, as one placed by a GPS clock is: a 10.8-minute video given a 27-minute GPX that started with it was uploaded with the whole track, which is then stretched over the video as a timelapse's. process cannot tell a timelapse without a GPS clock from a video of part of a track, so the track of such a timelapse, if it covers the creation time, is clipped too, and the warning then says to sample the video with video_process and --video_duration_ratio. In sampling, the warning offers --video_duration_ratio for a track more than twice as long as the video. The reading of the creation time moves from sample_video to geotag/video_start_time.py, so that GPX sync can use it. It reads the mvhd box, so process does not need ffprobe for it. --- mapillary_tools/exceptions.py | 4 +- .../geotag/video_extractors/gpx.py | 187 ++++++++++++--- mapillary_tools/geotag/video_start_time.py | 193 +++++++++++++++ mapillary_tools/sample_video.py | 220 +++++++----------- tests/integration/test_video_process.py | 66 ++++-- tests/unit/test_gpx_over_noisy_gps.py | 28 ++- tests/unit/test_gpx_video_coverage.py | 156 ++++++++++++- tests/unit/test_gpx_video_sync.py | 89 ++++--- tests/unit/test_nmea_video.py | 6 +- tests/unit/test_sample_video.py | 42 +++- 10 files changed, 760 insertions(+), 231 deletions(-) create mode 100644 mapillary_tools/geotag/video_start_time.py diff --git a/mapillary_tools/exceptions.py b/mapillary_tools/exceptions.py index 41f1ea2b6..98257956a 100644 --- a/mapillary_tools/exceptions.py +++ b/mapillary_tools/exceptions.py @@ -99,8 +99,8 @@ class MapillaryStationaryVideoError(MapillaryDescriptionError): class MapillaryGPXVideoStartUnknownError(MapillaryDescriptionError): """ 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. + the track is said to start with the video (--interpolation_use_gpx_start_time) + but is too long to end with it at the rate the video is said to run. """ diff --git a/mapillary_tools/geotag/video_extractors/gpx.py b/mapillary_tools/geotag/video_extractors/gpx.py index f7e4b6da8..8306a7eb7 100644 --- a/mapillary_tools/geotag/video_extractors/gpx.py +++ b/mapillary_tools/geotag/video_extractors/gpx.py @@ -29,7 +29,7 @@ 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, utils as geotag_utils +from .. import options, utils as geotag_utils, video_start_time from .base import BaseVideoExtractor from .native import NativeVideoExtractor @@ -251,7 +251,9 @@ def extract(self) -> types.VideoMetadata: if clock is None: # Its GPS has no clock to sync with, as if it had no GPS - clock = self._place_without_video_gps(gpx_points) + clock = self._place_without_video_gps( + gpx_points, native_video_metadata.make, native_video_metadata.model + ) self._apply_offset_time(gpx_points, rate=self._duration_ratio()) else: if self._duration_ratio() != 1.0: @@ -420,13 +422,20 @@ def _clip_to_video( return _clip_track(gpx_points, video_duration) def _place_without_video_gps( - self, gpx_points: T.Sequence[geo.Point] + self, + gpx_points: T.Sequence[geo.Point], + make: str | None = None, + model: str | None = None, ) -> VideoClock | None: """ 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. + one, or else by the creation time of the video, if the track covers + the video placed by it (see _creation_time_start), or else by assuming + that the track starts with the video. Nothing refuses the track: it is + taken to be right. It runs at --video_duration_ratio GPS seconds per + second of video. **make** and **model** are those of the camera, if + known. 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. @@ -436,29 +445,148 @@ def _place_without_video_gps( 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: + if self.sync_mode is SyncMode.SYNC: + if self.video_sync is not None and self.video_sync.start_time is not None: + start_time: float | None = self.video_sync.start_time + else: + start_time = self._creation_time_start(gpx_points, make, model) + if start_time is not None: self._rebase_times( gpx_points, - offset=(gpx_points[0].time - self.video_sync.start_time) / ratio, + offset=(gpx_points[0].time - start_time) / ratio, rate=ratio, ) - return VideoClock( - video_time=0.0, unix_time=self.video_sync.start_time, rate=ratio - ) - self._check_assumed_start(gpx_points, ratio) + return VideoClock(video_time=0.0, unix_time=start_time, rate=ratio) + if self.video_sync is not None: + self._check_assumed_start(gpx_points, ratio) + else: + self._warn_of_a_long_track(gpx_points) 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, offset=self._gpx_start_video_time(), rate=ratio ) return None - elif self.sync_mode is SyncMode.SYNC: - self._warn_of_a_long_track(gpx_points) self._rebase_times(gpx_points, rate=ratio) return None + def _creation_time_start( + self, + gpx_points: T.Sequence[geo.Point], + make: str | None = None, + model: str | None = None, + ) -> float | None: + """ + The Unix time at which the video started by its creation time, if the + GPX track (in Unix time) covers some of the video placed by it, with a + warning that the creation time comes from the camera's clock, or None, + with a warning when the creation time says a time that the track misses. + + The camera's clock can be off, and a video without GPS timestamps has + no better one. A track that does not cover the video placed by the + creation time is taken to be right instead, and to start with the + video. + """ + creation_time = video_start_time.read_creation_time(self.video_path) + if creation_time is None: + return None + + # As long as the track of a timelapse, which only sampling has + # --video_duration_ratio to tell + durations = self._durations(gpx_points, self._duration_ratio()) + long_track = durations is not None and 2 * durations[0] < durations[1] + timelapse_note = "" + if self.video_sync is None: + # process, which has no --video_start_time + advice = ( + "sample the video with video_process and --video_start_time, or " + f"trim the {self.file_format} to the video" + ) + if long_track and self.clip_to_video: + timelapse_note = ( + ". If the video is a timelapse, whose track is cut to the video " + "here, sample it with video_process and --video_duration_ratio " + "instead" + ) + elif long_track: + advice = ( + "specify --video_duration_ratio if the video is a timelapse, or " + "else the time the video started as --video_start_time" + ) + else: + advice = "specify the time the video started as --video_start_time" + + start = video_start_time.creation_time_to_start_time( + self.video_path, + creation_time, + _video_duration(self.video_path), + make=make, + model=model, + ) + # Where nothing says whether the creation time marks the start or the + # end of the recording, the track says which + readings = [(start.start_time, "start")] + if start.if_end_stamped is not None: + readings.append((start.if_end_stamped, "end")) + + gpx_track = f"The {self.file_format} track in {self.gpx_path}" + for start_time, end in readings: + if self._covers_some_of_the_video(gpx_points, start_time.timestamp()): + marks = ( + f", read as the {end} of the recording," + if start.if_end_stamped is not None + else "" + ) + LOG.warning( + f"{gpx_track} is placed along the video {self.video_path} by the " + f"creation time of the video{marks} which puts its start at " + f"{_isoformat(start_time.timestamp())}. That comes from the camera's " + f"clock, which can be off: if the frames are not where they were " + f"taken, {advice}{timelapse_note}" + ) + return start_time.timestamp() + + gpx_start = min(p.time for p in gpx_points) + gpx_end = max(p.time for p in gpx_points) + LOG.warning( + f"{gpx_track} ({_isoformat(gpx_start)} to {_isoformat(gpx_end)}) misses the " + f"video {self.video_path} placed by its creation time, which puts its start " + f"at {_isoformat(start.start_time.timestamp())} by the camera's clock, so the " + f"track is assumed to start with the video. If it does not, {advice}" + ) + return None + + def _covers_some_of_the_video( + self, gpx_points: T.Sequence[geo.Point], start_time: float + ) -> bool: + """ + Whether a part of the GPX track (in Unix time), between its gaps of + more than constants.MAX_GPX_GAP_SECONDS, overlaps the video started at + **start_time** and looked up along the track at + --interpolation_offset_time from there, at --video_duration_ratio + """ + offset_time = ( + 0.0 if self.interpolation is None else self.interpolation.offset_time + ) + video_start = start_time + offset_time + video_end = ( + video_start + + (_video_duration(self.video_path) or 0.0) * self._duration_ratio() + ) + points = sorted(gpx_points, key=lambda p: p.time) + gaps = track_gaps(points, constants.MAX_GPX_GAP_SECONDS) + firsts = [points[0].time, *(after.time for _, after in gaps)] + lasts = [*(before.time for before, _ in gaps), points[-1].time] + if video_end <= video_start: + return any( + first <= video_start <= last for first, last in zip(firsts, lasts) + ) + return any( + first < video_end and video_start < last + for first, last in zip(firsts, lasts) + ) + def _gpx_start_video_time(self) -> float: """ The video time that --interpolation_use_gpx_start_time starts the GPX @@ -714,14 +842,15 @@ 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. The video spans + Check, in sampling, the GPX track that is assumed to start with the + video 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. + for the video alone, or belongs to a timelapse, and 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. The track is + taken to be right, so neither is refused. """ durations = self._durations(gpx_points, ratio) if durations is None: @@ -734,25 +863,21 @@ def _check_assumed_start( message = ( f"The {self.file_format} track in {self.gpx_path} lasts {_format_duration(gpx_duration)} " f"and {self._describe_video_span(video_span, ratio)}, so the video " - "could start anywhere along the track. " + "could start anywhere along the track, and is assumed to start with it. " ) - # Not --interpolation_use_gpx_start_time, which starts the track - # with the video too, and refuses it for the same reason if gpx_duration < _MAX_RECORDING_SECONDS: message += ( "If the video is a timelapse, specify --video_duration_ratio. If " - "the GPS logger was started with the video and left running, " - "specify --video_start_time " - f"{build_capture_time(min(p.time for p in gpx_points))}, when the " - "track starts. If the track covers more than this video, specify " - "the time the video started as --video_start_time instead" + "the track covers more than this video, specify the time the " + "video started as --video_start_time" ) else: message += ( "A track that long covers more than this video: specify the time " "the video started as --video_start_time" ) - raise exceptions.MapillaryGPXVideoStartUnknownError(message) + LOG.warning(message) + return if _LOGGER_TOLERANCE_SECONDS < excess: LOG.warning( @@ -877,10 +1002,10 @@ def _sync_clock( def _warn_of_a_long_track(self, gpx_points: T.Sequence[geo.Point]) -> None: """ Warn, in process, of a GPX track assumed to start with the video that - runs on for longer than the whole video, which sampling refuses (see + runs on for longer than the whole video, as sampling does (see _check_assumed_start): the video could start anywhere along it. Only - warn, since the track of a timelapse without a GPS clock of its own - runs on past the end of the video too. + warn, since the track is taken to be right, and that of a timelapse + without a GPS clock of its own runs on past the end of the video too. With --interpolation_offset_time, it is the offset that places the video along the track, and one way to place it right. diff --git a/mapillary_tools/geotag/video_start_time.py b/mapillary_tools/geotag/video_start_time.py new file mode 100644 index 000000000..871361a5b --- /dev/null +++ b/mapillary_tools/geotag/video_start_time.py @@ -0,0 +1,193 @@ +# 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. + +""" +The time at which a video started recording, by the creation time in its +metadata. That comes from the camera's own clock, which can be off, and marks +the start or the end of the recording depending on the camera, so it is read +only for a video without GPS timestamps of its own. +""" + +from __future__ import annotations + +import dataclasses +import datetime +import logging +import re +from pathlib import Path + +import construct as C + +from .. import blackvue_parser +from ..camm import camm_parser +from ..mp4 import simple_mp4_parser as sparser +from ..mp4.mp4_sample_parser import MovieBoxParser +from ..serializer.description import build_capture_time + +LOG = logging.getLogger(__name__) + +# Cameras known to stamp the creation time at the end of the recording, as +# lowercase (make, model) from the container tags. BlackVue is not listed +# because it does not write those tags: blackvue_parser.is_blackvue detects it +_END_STAMPING_CAMERAS = { + ("ricoh", "ricoh theta x"), + ("labpano", "panox v2"), +} + +# A date and time in a file name, for example 20230512_101530 (BlackVue, +# Insta360) or 2023_0512_101530 (Viofo), in the camera's local time +_FILENAME_TIME_RE = re.compile( + r"(? datetime.datetime | None: + for match in _FILENAME_TIME_RE.finditer(video_path.stem): + year, month, day, hour, minute, second = (int(g) for g in match.groups()) + try: + return datetime.datetime( + year, month, day, hour, minute, second, tzinfo=datetime.timezone.utc + ) + except ValueError: + continue + + return None + + +def _as_utc(dt: datetime.datetime) -> datetime.datetime: + if dt.tzinfo is None: + return dt.replace(tzinfo=datetime.timezone.utc) + return dt + + +def _is_same_time_in_another_time_zone( + a: datetime.datetime, b: datetime.datetime +) -> bool: + """ + Tell whether two times could be the same moment, one of them possibly in + local time labelled as UTC + """ + delta = abs((_as_utc(a) - _as_utc(b)).total_seconds()) + if _MAX_UTC_OFFSET_SECONDS + _FILENAME_TIME_TOLERANCE_SECONDS < delta: + return False + remainder = delta % _UTC_OFFSET_GRANULARITY_SECONDS + return ( + min(remainder, _UTC_OFFSET_GRANULARITY_SECONDS - remainder) + <= _FILENAME_TIME_TOLERANCE_SECONDS + ) + + +def _is_end_stamping_camera( + video_path: Path, make: str | None, model: str | None +) -> bool: + if make is not None and model is not None: + if (make.strip().lower(), model.strip().lower()) in _END_STAMPING_CAMERAS: + return True + + try: + with video_path.open("rb") as fp: + return blackvue_parser.is_blackvue(fp) + except Exception as ex: + LOG.debug("Unable to tell whether %s is a BlackVue video: %s", video_path, ex) + return False + + +@dataclasses.dataclass(frozen=True) +class StartTime: + # When the video started, by its creation time + start_time: datetime.datetime + # When it started if the camera stamped the creation time at the other end + # of the recording, where nothing said which end the creation time marks + # and start_time assumes the start; else None + if_end_stamped: datetime.datetime | None = None + + +def creation_time_to_start_time( + video_path: Path, + creation_time: datetime.datetime, + duration: float | None, + make: str | None = None, + model: str | None = None, +) -> StartTime: + """ + Determine the wall clock time at which a video started recording from the + creation time in its metadata, the video lasting **duration** seconds. + + Cameras disagree on what the creation time marks. Many stamp the end of the + recording: BlackVue, Viofo, Vantrue and most other dashcams, the Ricoh Theta + X and the Labpano PanoX V2. Others stamp the start: Sony, Insta360 and the + dashcam in the bug report that motivated reading this at all. Nothing in the + metadata says which, so look for evidence: a camera known to stamp the end, + or a time in the file name that matches only one of the two. Without any, + assume the start, and give the alternative for the caller to warn about. + """ + if duration is None: + LOG.warning( + "Unable to read the duration of %s, so assuming its creation time %s marks the start of the recording", + video_path.name, + creation_time, + ) + return StartTime(creation_time) + + start_time_if_end_stamped = creation_time - datetime.timedelta(seconds=duration) + + if _is_end_stamping_camera(video_path, make, model): + return StartTime(start_time_if_end_stamped) + + filename_time = _parse_filename_time(video_path) + if filename_time is not None: + matches_start = _is_same_time_in_another_time_zone(filename_time, creation_time) + matches_end = _is_same_time_in_another_time_zone( + filename_time, start_time_if_end_stamped + ) + if matches_end and not matches_start: + return StartTime(start_time_if_end_stamped) + if matches_start and not matches_end: + return StartTime(creation_time) + + return StartTime(creation_time, if_end_stamped=start_time_if_end_stamped) + + +def describe_end_stamped_alternative( + video_path: Path, start_time: StartTime, how_to_specify: str +) -> str | None: + """ + The warning that the creation time was assumed to mark the start of the + recording, which **how_to_specify** says how to correct, or None if + evidence said which end it marks + """ + if start_time.if_end_stamped is None: + return None + seconds = (start_time.start_time - start_time.if_end_stamped).total_seconds() + return ( + f"Assuming the creation time {start_time.start_time} of {video_path.name} marks " + "the start of the recording. If the camera stamps the end instead, as most " + f"dashcams do, the recording started {seconds:.1f} seconds earlier: " + f"{how_to_specify} {build_capture_time(start_time.if_end_stamped)} to use that" + ) + + +def read_creation_time(video_path: Path) -> datetime.datetime | None: + """ + The creation time of the video, from its mvhd box, or None if it is unset + or cannot be read. Read without ffprobe, which process does not need. + """ + 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 + unix_time = camm_parser._extract_creation_time(moov) + if unix_time is None: + return None + return datetime.datetime.fromtimestamp(unix_time, tz=datetime.timezone.utc) diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py index cf0ae607c..71669dc53 100644 --- a/mapillary_tools/sample_video.py +++ b/mapillary_tools/sample_video.py @@ -9,7 +9,6 @@ import datetime import logging import os -import re import shutil import time import typing as T @@ -18,17 +17,14 @@ import humanize -from . import ( - blackvue_parser, - constants, - exceptions, - ffmpeg as ffmpeglib, - geo, - types, - utils, -) +from . import constants, exceptions, ffmpeg as ffmpeglib, geo, types, utils from .exif_write import ExifEdit -from .geotag import factory, geotag_images_from_gpx_file, geotag_images_from_nmea_file +from .geotag import ( + factory, + geotag_images_from_gpx_file, + geotag_images_from_nmea_file, + video_start_time, +) from .geotag.geotag_images_from_gpx import check_within_track from .geotag.options import ( InterpolationOption, @@ -269,11 +265,10 @@ def sample_video( except ValueError as ex: raise exceptions.MapillaryBadParameterError(str(ex)) - # Only a start time the user gave. A video needs one to place a GPX only if - # it has no GPS clock, and then its start time would come from its creation - # time, which cameras stamp at the start or at the end of the recording, some - # in local time. Telling which takes evidence that a video may not carry, and - # a GPX placed by a wrong guess is off by a whole video length or a time zone + # Only a start time the user gave. A video needs one to place a GPX exactly + # only if it has no GPS clock. Without it, GPX sync places the GPX by the + # creation time of the video, with a warning, as cameras stamp it at the + # start or at the end of the recording, some in local time video_sync = VideoSyncOption( start_time=( video_start_time_dt.timestamp() if video_start_time_dt is not None else None @@ -456,29 +451,6 @@ def wip_sample_dir(sample_dir: Path) -> Path: ) -# Cameras known to stamp the creation time at the end of the recording, as -# lowercase (make, model) from the container tags. BlackVue is not listed -# because it does not write those tags: blackvue_parser.is_blackvue detects it -_END_STAMPING_CAMERAS = { - ("ricoh", "ricoh theta x"), - ("labpano", "panox v2"), -} - -# A date and time in a file name, for example 20230512_101530 (BlackVue, -# Insta360) or 2023_0512_101530 (Viofo), in the camera's local time -_FILENAME_TIME_RE = re.compile( - r"(? datetime.datetime | None: @@ -537,111 +509,31 @@ def _telemetry_clock( return clock -def _parse_filename_time(video_path: Path) -> datetime.datetime | None: - for match in _FILENAME_TIME_RE.finditer(video_path.stem): - year, month, day, hour, minute, second = (int(g) for g in match.groups()) - try: - return datetime.datetime( - year, month, day, hour, minute, second, tzinfo=datetime.timezone.utc - ) - except ValueError: - continue - - return None - - -def _as_utc(dt: datetime.datetime) -> datetime.datetime: - if dt.tzinfo is None: - return dt.replace(tzinfo=datetime.timezone.utc) - return dt - - -def _is_same_time_in_another_time_zone( - a: datetime.datetime, b: datetime.datetime -) -> bool: - """ - Tell whether two times could be the same moment, one of them possibly in - local time labelled as UTC - """ - delta = abs((_as_utc(a) - _as_utc(b)).total_seconds()) - if _MAX_UTC_OFFSET_SECONDS + _FILENAME_TIME_TOLERANCE_SECONDS < delta: - return False - remainder = delta % _UTC_OFFSET_GRANULARITY_SECONDS - return ( - min(remainder, _UTC_OFFSET_GRANULARITY_SECONDS - remainder) - <= _FILENAME_TIME_TOLERANCE_SECONDS - ) - - -def _is_end_stamping_camera(video_path: Path, probe: ffmpeglib.Probe) -> bool: - make = probe.probe_format_tag("make") - model = probe.probe_format_tag("model") - if make is not None and model is not None: - if (make.strip().lower(), model.strip().lower()) in _END_STAMPING_CAMERAS: - return True - - try: - with video_path.open("rb") as fp: - return blackvue_parser.is_blackvue(fp) - except Exception as ex: - LOG.debug("Unable to tell whether %s is a BlackVue video: %s", video_path, ex) - return False - - def _creation_time_to_start_time( video_path: Path, probe: ffmpeglib.Probe ) -> datetime.datetime | None: """ Determine the wall clock time at which a video started recording from the - creation time in its metadata. - - Cameras disagree on what the creation time marks. Many stamp the end of the - recording: BlackVue, Viofo, Vantrue and most other dashcams, the Ricoh Theta - X and the Labpano PanoX V2. Others stamp the start: Sony, Insta360 and the - dashcam in the bug report that motivated reading this at all. Nothing in the - metadata says which, so look for evidence: a camera known to stamp the end, - or a time in the file name that matches only one of the two. Without any, - assume the start and warn about the alternative. + creation time in its metadata, as read by ffprobe (see + video_start_time.creation_time_to_start_time) """ creation_time = probe.probe_video_creation_time() if creation_time is None: return None - duration = probe.probe_video_duration() - if duration is None: - LOG.warning( - "Unable to read the duration of %s, so assuming its creation time %s marks the start of the recording", - video_path.name, - creation_time, - ) - return creation_time - - start_time_if_end_stamped = creation_time - datetime.timedelta(seconds=duration) - - if _is_end_stamping_camera(video_path, probe): - return start_time_if_end_stamped - - filename_time = _parse_filename_time(video_path) - if filename_time is not None: - matches_start = _is_same_time_in_another_time_zone(filename_time, creation_time) - matches_end = _is_same_time_in_another_time_zone( - filename_time, start_time_if_end_stamped - ) - if matches_end and not matches_start: - return start_time_if_end_stamped - if matches_start and not matches_end: - return creation_time - - LOG.warning( - "Assuming the creation time %s of %s marks the start of the recording. " - "If the camera stamps the end instead, as most dashcams do, the recording started %.1f seconds earlier: " - "specify --video_start_time %s to use that", + start_time = video_start_time.creation_time_to_start_time( + video_path, creation_time, - video_path.name, - duration, - build_capture_time(start_time_if_end_stamped), + probe.probe_video_duration(), + make=probe.probe_format_tag("make"), + model=probe.probe_format_tag("model"), ) - return creation_time + warning = video_start_time.describe_end_stamped_alternative( + video_path, start_time, "specify --video_start_time" + ) + if warning is not None: + LOG.warning(warning) + return start_time.start_time def _extract_video_start_time( @@ -706,6 +598,62 @@ def _check_gpx_start_by_interval( extractor._check_given_start(gpx_points, video_sync.duration_ratio) +def _track_file_start_time( + video_path: Path, + geotag_options: T.Sequence[SourceOption], + video_sync: VideoSyncOption | None, + make: str | None = None, + model: str | None = None, +) -> datetime.datetime | None: + """ + The start time of a video without a GPS clock, by the GPX (or NMEA) track + of the first source that geotags images, which the geotagging stage places + the interval samples along by their capture times: as GPX sync places a + track on such a video (see GPXVideoExtractor._place_without_video_gps), by + the creation time of the video if the track covers it, or else by the + start of the track. None to leave it to the creation time alone: without + such a track, or when --interpolation_use_gpx_start_time starts the track + with the video, which the geotagging stage does. + """ + for option in geotag_options: + extractor_class = _TRACK_FILE_EXTRACTORS.get(option.source) + if extractor_class is None or option.source_path is None: + continue + if ( + option.filetypes is not None + and types.FileType.IMAGE not in option.filetypes + ): + continue + if option.interpolation is not None and option.interpolation.use_gpx_start_time: + return None + gpx_path = option.source_path.resolve(video_path) + if not gpx_path.is_file(): + continue + extractor = extractor_class( + video_path, + gpx_path, + video_sync=video_sync, + interpolation=option.interpolation, + ) + try: + gpx_points = extractor._read_gpx_points() + except Exception as ex: + # The geotagging stage reports what keeps it from reading the GPX + LOG.debug("No start time from %s for %s: %s", gpx_path, video_path.name, ex) + return None + if not gpx_points: + return None + start_time = extractor._creation_time_start(gpx_points, make, model) + if start_time is None: + start_time = min(p.time for p in gpx_points) + extractor._check_assumed_start( + gpx_points, 1.0 if video_sync is None else video_sync.duration_ratio + ) + return datetime.datetime.fromtimestamp(start_time, tz=datetime.timezone.utc) + + return None + + def _gpx_sync_reads_a_clock(extractor: GPXVideoExtractor) -> bool: """ Whether GPX sync, in distance sampling and process, reads a GPS clock from @@ -850,7 +798,13 @@ def _sample_single_video_by_interval( clock.to_unix_time(0.0), tz=datetime.timezone.utc ) else: - start_time = _extract_video_start_time(video_path, probe, telemetry_points) + start_time = _track_file_start_time( + video_path, + geotag_options, + video_sync, + make=probe.probe_format_tag("make"), + model=probe.probe_format_tag("model"), + ) or _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}" diff --git a/tests/integration/test_video_process.py b/tests/integration/test_video_process.py index eda8338b5..4e75caaee 100644 --- a/tests/integration/test_video_process.py +++ b/tests/integration/test_video_process.py @@ -3,6 +3,8 @@ # 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 functools import json import os @@ -490,36 +492,62 @@ def test_video_process_by_distance_places_a_gpx_by_the_video_start_time( assert capture_times[-1] < "2025_03_14_07_00_36_000", capture_times -def test_video_process_by_distance_refuses_to_guess_where_the_video_starts( +@pytest.mark.parametrize( + "sampling", + [ + ["--video_sample_distance=3"], + ["--video_sample_distance=-1", "--video_sample_interval=2"], + ], +) +@pytest.mark.parametrize( + "creation_time, first, last", + [ + # Within the GPX, which is placed by it + ("2025-03-14T07:00:36Z", "2025_03_14_07_00_36_000", "2025_03_14_07_00_42_000"), + # Hours after the GPX ends, which is taken to start with the video + ("2025-03-14T12:00:00Z", "2025_03_14_07_00_00_000", "2025_03_14_07_00_06_000"), + # None, which interval sampling failed without + (None, "2025_03_14_07_00_00_000", "2025_03_14_07_00_06_000"), + ], +) +def test_video_process_places_a_gpx_by_the_creation_time( setup_data: py.path.local, + sampling: list[str], + creation_time: str | None, + first: str, + last: str, ): """ - Without --video_start_time, a GPX far longer than the video could start - anywhere along it, so assuming that it starts with the video is a guess. + Without --video_start_time, the GPX is placed along a video without GPS + of its own by the creation time of the video, if it covers the video + there, and is otherwise assumed to start with the video. Neither refuses + the GPX, which is taken to be right. """ 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 ex: - run_video_process( - [ - "--video_sample_distance=3", - "--geotag_source=gpx", - *["--geotag_source_path", str(gpx_file)], - str(video_path), - str(Path(dir, "samples")), - ], - capture_output=True, - text=True, + video_path = str(setup_data.join("videos").join("sample-5s.mp4")) + if creation_time is not None: + video_path = str( + _with_creation_time(py.path.local(video_path), creation_time, dir) ) + descs = run_video_process_for_descs( + [ + *sampling, + "--geotag_source=gpx", + *["--geotag_source_path", str(gpx_file)], + str(video_path), + str(Path(dir, "samples")), + ] + ) - assert 7 == ex.value.returncode, ex.value.stderr - assert "specify --video_start_time" in ex.value.stderr.lower(), ex.value.stderr + assert descs + assert all("error" not in desc for desc in descs), descs + capture_times = sorted(desc["MAPCaptureTime"] for desc in descs) + assert first <= capture_times[0], capture_times + assert capture_times[-1] < last, capture_times def test_video_process_by_distance_with_a_gpx_that_misses_the_video( diff --git a/tests/unit/test_gpx_over_noisy_gps.py b/tests/unit/test_gpx_over_noisy_gps.py index 0ba2d5ee9..bc9f64083 100644 --- a/tests/unit/test_gpx_over_noisy_gps.py +++ b/tests/unit/test_gpx_over_noisy_gps.py @@ -17,6 +17,7 @@ from __future__ import annotations +import logging import typing as T from pathlib import Path @@ -188,18 +189,13 @@ def test_good_fixes_are_preferred_as_the_clock( assert [p.time for p in metadata.points] == pytest.approx([2.0, 4.0]) - @pytest.mark.parametrize("creation_time", [None, A_UNIX_TIME]) - def test_placeholder_clock_is_ignored( - self, monkeypatch, video_path, gpx_path, creation_time - ): + def test_placeholder_clock_is_ignored(self, monkeypatch, video_path, gpx_path): """ Until it acquires the GPS time, a GoPro stamps a placeholder date, which a GPS rejected as noise may never have got past. Such a clock says nothing of when the video was recorded, so the GPX is synced from the - start of the video, whatever the creation time of the video says. + start of the video, which has no creation time either. """ - if creation_time is not None: - _write_mvhd(video_path, creation_time=creation_time) points = [ _point(time=i * 0.1, epoch_time=PLACEHOLDER_UNIX_TIME + i * 0.1) for i in range(32) @@ -211,6 +207,24 @@ def test_placeholder_clock_is_ignored( assert [p.time for p in metadata.points] == [0.0, 2.0] assert (metadata.make, metadata.model) == ("GoPro", "MAX2") + def test_placeholder_clock_leaves_the_gpx_to_the_creation_time( + self, monkeypatch, video_path, gpx_path, caplog + ): + """As for a video without GPS, with a warning that it can be off""" + _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) + + with caplog.at_level(logging.WARNING): + metadata = GPXVideoExtractor(video_path, gpx_path).extract() + + assert [p.time for p in metadata.points] == pytest.approx([2.0, 4.0]) + assert "by the creation time of the video" in caplog.text + assert "camera's clock, which can be off" in caplog.text + def test_noisy_clock_after_its_placeholder_date_places_the_gpx( self, monkeypatch, video_path, gpx_path ): diff --git a/tests/unit/test_gpx_video_coverage.py b/tests/unit/test_gpx_video_coverage.py index 2e657e30c..22228048b 100644 --- a/tests/unit/test_gpx_video_coverage.py +++ b/tests/unit/test_gpx_video_coverage.py @@ -59,6 +59,9 @@ # Captured 2018-06-08T20:24:11Z at 45.5169, -122.5728 IMAGE = _DATA.joinpath("images", "DSC00001.JPG") +# A_UNIX_TIME to the second, as an mvhd creation time is written +WHOLE_SECOND = float(int(A_UNIX_TIME)) + # A 10s video track starting at A_UNIX_TIME, in a 12s video VIDEO_DURATION = 12.0 @@ -388,7 +391,9 @@ def test_gpx_start_time_keeps_the_rate(self, tmp_path, caplog): 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. + Assumed to start with the video, it can only fall short of its end. The + creation time of the video places the track only where the track covers + the video placed by it. """ def _extract( @@ -417,7 +422,9 @@ def test_start_time_within_the_track_is_silent(self, tmp_path, caplog): "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) + video_path = _write_camm_mp4( + tmp_path / "video.mp4", video_track, creation_time=None + ) 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) @@ -425,11 +432,149 @@ def test_assumed_start_warns_of_a_short_track(self, tmp_path, caplog, video_trac 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) + video_path = _write_camm_mp4( + tmp_path / "video.mp4", UNTIMED_TRACK, creation_time=None + ) with caplog.at_level(logging.WARNING): self._extract(tmp_path, video_path, A_UNIX_TIME, VIDEO_DURATION) assert not _warnings(caplog) + @pytest.mark.parametrize( + "sync", [None, VideoSyncOption()], ids=["process", "sampling"] + ) + def test_placed_by_the_creation_time_within_the_track(self, tmp_path, caplog, sync): + """ + A track recorded from 6 seconds before the 12 seconds of video to 2 + seconds after them, which the creation time (taken as the start) + places the video in + """ + video_path = _write_camm_mp4( + tmp_path / "video.mp4", [], creation_time=WHOLE_SECOND + ) + gpx_path = _write_gpx(tmp_path / "track.gpx", WHOLE_SECOND - 6, 20) + with caplog.at_level(logging.WARNING): + metadata = GPXVideoExtractor( + video_path, gpx_path, video_sync=sync + ).extract() + + assert metadata.points[0].time == pytest.approx(-6.0) + placed = [w for w in _warnings(caplog) if "by the creation time" in w] + assert len(placed) == 1, _warnings(caplog) + assert "puts its start at 2026-08-12T08:26:27+00:00" in placed[0] + assert "comes from the camera's clock, which can be off" in placed[0] + assert "--video_start_time" in placed[0] + assert "--video_duration_ratio" not in placed[0] + + def test_a_track_placed_by_the_creation_time_is_clipped_in_process( + self, tmp_path, caplog + ): + """ + Uploaded with the video, a track that runs on past its end is clipped + to it, so that it is not stretched over the video as a timelapse's. + process cannot tell a timelapse, whose track that clips, and warns + """ + video_path = _write_camm_mp4( + tmp_path / "video.mp4", [], creation_time=WHOLE_SECOND + ) + # 27 minutes from the start of the video, as a logger left running + gpx_path = _write_gpx(tmp_path / "track.gpx", WHOLE_SECOND, 27 * 60) + with caplog.at_level(logging.WARNING): + metadata = GPXVideoExtractor( + video_path, gpx_path, clip_to_video=True + ).extract() + + assert metadata.points[0].time == pytest.approx(0.0) + assert metadata.points[-1].time == pytest.approx(VIDEO_DURATION) + [placed] = [w for w in _warnings(caplog) if "by the creation time" in w] + assert placed.endswith( + "trim the GPX to the video. If the video is a timelapse, whose track is " + "cut to the video here, sample it with video_process and " + "--video_duration_ratio instead" + ) + + def test_a_short_track_placed_by_the_creation_time_says_nothing_of_a_timelapse( + self, tmp_path, caplog + ): + video_path = _write_camm_mp4( + tmp_path / "video.mp4", [], creation_time=WHOLE_SECOND + ) + gpx_path = _write_gpx(tmp_path / "track.gpx", WHOLE_SECOND - 6, 20) + with caplog.at_level(logging.WARNING): + metadata = GPXVideoExtractor( + video_path, gpx_path, clip_to_video=True + ).extract() + + assert metadata.points[0].time == pytest.approx(0.0) + assert metadata.points[-1].time == pytest.approx(VIDEO_DURATION) + [placed] = [w for w in _warnings(caplog) if "by the creation time" in w] + assert "timelapse" not in placed + + def test_placed_by_the_creation_time_in_a_long_track_in_sampling( + self, tmp_path, caplog + ): + """ + Where --video_duration_ratio tells a timelapse, which the warning + offers + """ + video_path = _write_camm_mp4( + tmp_path / "video.mp4", [], creation_time=WHOLE_SECOND + ) + gpx_path = _write_gpx(tmp_path / "track.gpx", WHOLE_SECOND - 60, 132) + with caplog.at_level(logging.WARNING): + metadata = GPXVideoExtractor( + video_path, gpx_path, video_sync=VideoSyncOption() + ).extract() + + assert metadata.points[0].time == pytest.approx(-60.0) + [placed] = [w for w in _warnings(caplog) if "by the creation time" in w] + assert ( + "specify --video_duration_ratio if the video is a timelapse, or else " + "the time the video started as --video_start_time" + ) in placed + + def test_the_creation_time_as_the_end_when_the_track_says_so( + self, tmp_path, caplog + ): + """ + Nothing says whether this camera stamps the creation time at the start + or the end of the recording: the track, which covers the video only if + it ended then, says the end + """ + video_path = _write_camm_mp4( + tmp_path / "video.mp4", [], creation_time=WHOLE_SECOND + VIDEO_DURATION + ) + gpx_path = _write_gpx(tmp_path / "track.gpx", WHOLE_SECOND - 60, 60 + 10) + with caplog.at_level(logging.WARNING): + metadata = GPXVideoExtractor( + video_path, gpx_path, video_sync=VideoSyncOption() + ).extract() + + assert metadata.points[0].time == pytest.approx(-60.0) + [placed] = [w for w in _warnings(caplog) if "by the creation time" in w] + assert "read as the end of the recording" in placed + + @pytest.mark.parametrize( + "sync", [None, VideoSyncOption()], ids=["process", "sampling"] + ) + def test_a_creation_time_that_the_track_misses_is_not_used( + self, tmp_path, caplog, sync + ): + """The track is taken to be right, and to start with the video""" + video_path = _write_camm_mp4( + tmp_path / "video.mp4", [], creation_time=WHOLE_SECOND + 30 * 24 * 3600 + ) + gpx_path = _write_gpx(tmp_path / "track.gpx", WHOLE_SECOND, VIDEO_DURATION) + with caplog.at_level(logging.WARNING): + metadata = GPXVideoExtractor( + video_path, gpx_path, video_sync=sync + ).extract() + + assert metadata.points[0].time == 0.0 + [warning] = _warnings(caplog) + assert "misses the video" in warning + assert "placed by its creation time" in warning + assert "the track is assumed to start with the video" in warning + class TestInterpolationOffset: """ @@ -445,7 +590,10 @@ def _extract( offset_time: float, use_gpx_start_time: bool = False, ) -> types.VideoMetadata: - video_path = _write_camm_mp4(tmp_path / "video.mp4", video_track) + # Without a creation time, which would place the track otherwise + 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, VIDEO_DURATION) return GPXVideoExtractor( video_path, diff --git a/tests/unit/test_gpx_video_sync.py b/tests/unit/test_gpx_video_sync.py index e3b23723f..ccc9f7568 100644 --- a/tests/unit/test_gpx_video_sync.py +++ b/tests/unit/test_gpx_video_sync.py @@ -99,30 +99,31 @@ def test_the_track_is_placed_by_the_video_start_time(self, tmp_path, caplog): video_start.timestamp() ) - def test_a_track_far_longer_than_the_video_is_refused(self, tmp_path): - """The video could start anywhere along a track a minute long.""" + def test_a_track_far_longer_than_the_video_starts_with_it_with_a_warning( + self, tmp_path, caplog + ): + """ + The video could start anywhere along a track a minute long, but the + track is taken to be right, so it is not refused. + """ gpx = _write_gpx(tmp_path / "track.gpx", seconds=60) - with pytest.raises(exceptions.MapillaryGPXVideoStartUnknownError) as excinfo: - _extract(NO_GPS_VIDEO, gpx, VideoSyncOption()) + with caplog.at_level(logging.WARNING): + metadata = _extract(NO_GPS_VIDEO, gpx, VideoSyncOption()) - 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 + assert metadata.points[0].time == 0 + assert "is assumed to start with it" in caplog.text + assert "--video_duration_ratio" in caplog.text assert "covers more than this video, specify the time the video started" in ( - message + caplog.text ) - # Which starts the track with the video too, and refuses it alike - assert "--interpolation_use_gpx_start_time" not in message + # Which starts the track with the video too + assert "--interpolation_use_gpx_start_time" not in caplog.text - def test_a_log_of_days_offers_no_start_time(self, tmp_path): + def test_a_log_of_days_starts_with_the_video_with_a_warning(self, tmp_path, caplog): """ - 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. + A track of several days is a log of several trips, which the video + lands on the first of: warn that it needs its start time. """ gpx = _write_gpx_tracks( tmp_path / "track.gpx", @@ -132,13 +133,15 @@ def test_a_log_of_days_offers_no_start_time(self, tmp_path): ], ) - with pytest.raises(exceptions.MapillaryGPXVideoStartUnknownError) as excinfo: - _extract(NO_GPS_VIDEO, gpx, VideoSyncOption()) + with caplog.at_level(logging.WARNING): + metadata = _extract(NO_GPS_VIDEO, gpx, VideoSyncOption()) - 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 "--interpolation_use_gpx_start_time" not in message + assert metadata.points[0].time == 0 + assert "specify the time the video started as --video_start_time" in ( + caplog.text + ) + assert "--video_duration_ratio" not in caplog.text + assert "--interpolation_use_gpx_start_time" not in caplog.text def test_gpx_start_time_on_a_track_far_longer_than_the_video_is_refused( self, tmp_path @@ -248,15 +251,14 @@ def test_the_duration_ratio_spreads_a_track_over_the_video( assert metadata.points[-1].time == pytest.approx(start + 5.7) assert not caplog.text - def test_a_track_too_long_even_at_the_duration_ratio_is_refused(self, tmp_path): + def test_a_track_too_long_even_at_the_duration_ratio_warns(self, tmp_path, caplog): 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)) + with caplog.at_level(logging.WARNING): + metadata = _extract(NO_GPS_VIDEO, gpx, VideoSyncOption(duration_ratio=10)) - assert "which at --video_duration_ratio 10 span 57.6 seconds" in str( - excinfo.value - ) + assert metadata.points[0].time == 0 + assert "which at --video_duration_ratio 10 span 57.6 seconds" in caplog.text def test_a_track_a_little_longer_than_the_video_warns(self, tmp_path, caplog): gpx = _write_gpx(tmp_path / "track.gpx", seconds=9) @@ -291,10 +293,36 @@ def test_process_still_assumes_the_track_starts_with_the_video( assert metadata.points[0].time == 0 assert "--video_duration_ratio" not in caplog.text + def test_a_track_far_longer_than_the_video_geotags_it_before_the_next_source( + self, tmp_path + ): + """ + The video has no GPS, so a native source after the GPX can only fail: + the track geotags it, assumed to start with it. + """ + gpx = _write_gpx(tmp_path / "track.gpx", seconds=60) + video_sync = VideoSyncOption() + + [metadata] = factory.process( + [NO_GPS_VIDEO], + [ + SourceOption( + SourceType.GPX, + source_path=SourcePathOption(source_path=gpx), + video_sync=video_sync, + ), + SourceOption(SourceType.NATIVE, video_sync=video_sync), + ], + ) + + assert isinstance(metadata, types.VideoMetadata) + assert metadata.points[0].time == 0 + def test_the_refusal_is_not_handed_on_to_the_next_source(self, tmp_path): """ The video has no GPS, so a native source after the GPX can only fail, - and its error would hide why the GPX was refused. + and its error would hide why --interpolation_use_gpx_start_time could + not start the track with the video. """ gpx = _write_gpx(tmp_path / "track.gpx", seconds=60) video_sync = VideoSyncOption() @@ -305,6 +333,7 @@ def test_the_refusal_is_not_handed_on_to_the_next_source(self, tmp_path): SourceOption( SourceType.GPX, source_path=SourcePathOption(source_path=gpx), + interpolation=InterpolationOption(use_gpx_start_time=True), video_sync=video_sync, ), SourceOption(SourceType.NATIVE, video_sync=video_sync), diff --git a/tests/unit/test_nmea_video.py b/tests/unit/test_nmea_video.py index e3ccb01da..577bbfc5f 100644 --- a/tests/unit/test_nmea_video.py +++ b/tests/unit/test_nmea_video.py @@ -180,15 +180,15 @@ def test_a_malformed_sentence_is_skipped(self, tmp_path): for nmea_point, gpx_point in zip(from_nmea, from_gpx): assert nmea_point == pytest.approx(gpx_point, abs=1e-6) - def test_messages_name_the_nmea_file(self, tmp_path): + def test_messages_name_the_nmea_file(self, tmp_path, caplog): nmea = _write_nmea(tmp_path / "track.nmea", seconds=60) - with pytest.raises(exceptions.MapillaryGPXVideoStartUnknownError) as excinfo: + with caplog.at_level(logging.WARNING): NMEAVideoExtractor( NO_GPS_VIDEO, nmea, video_sync=VideoSyncOption() ).extract() - assert str(excinfo.value).startswith(f"The NMEA track in {nmea} lasts") + assert f"The NMEA track in {nmea} lasts" in caplog.text class TestInvalidNMEAFile: diff --git a/tests/unit/test_sample_video.py b/tests/unit/test_sample_video.py index d82c69e5b..b5098d6dd 100644 --- a/tests/unit/test_sample_video.py +++ b/tests/unit/test_sample_video.py @@ -28,6 +28,7 @@ sample_video, telemetry, ) +from mapillary_tools.geotag import video_start_time from mapillary_tools.geotag.options import ( InterpolationOption, SourceOption, @@ -1007,7 +1008,10 @@ def test_none_left_out_of_several_videos_started_with_the_gpx( assert len(list(sample_dir.iterdir())) == num_samples def test_not_by_a_placeholder_clock(self, tmpdir, setup_mock, monkeypatch): - """Which GPX sync does not place the GPX by either""" + """ + Which GPX sync does not place the GPX by either: the video has no + creation time to place it by, so the track starts with it + """ self._mock_noisy_telemetry( monkeypatch, datetime.datetime(2015, 1, 1, tzinfo=datetime.timezone.utc).timestamp(), @@ -1016,7 +1020,41 @@ def test_not_by_a_placeholder_clock(self, tmpdir, setup_mock, monkeypatch): _validate_interval( self._sample(tmpdir, geotag_source=["gpx"], geotag_source_path=gpx), - PROBE_START_TIME, + self.GPS_START_TIME, + ) + + def test_by_the_creation_time_within_the_gpx(self, tmpdir, setup_mock, monkeypatch): + """As GPX sync places the GPX on a video without a GPS clock""" + self._mock_noisy_telemetry( + monkeypatch, + datetime.datetime(2015, 1, 1, tzinfo=datetime.timezone.utc).timestamp(), + ) + creation_time = self.GPS_START_TIME + datetime.timedelta(seconds=100) + monkeypatch.setattr( + video_start_time, "read_creation_time", lambda path: creation_time + ) + gpx = self._write_gpx(Path(tmpdir) / "track.gpx", self.GPS_START_TIME) + + _validate_interval( + self._sample(tmpdir, geotag_source=["gpx"], geotag_source_path=gpx), + creation_time, + ) + + def test_with_the_gpx_if_it_misses_the_creation_time( + self, tmpdir, setup_mock, monkeypatch + ): + self._mock_noisy_telemetry( + monkeypatch, + datetime.datetime(2015, 1, 1, tzinfo=datetime.timezone.utc).timestamp(), + ) + monkeypatch.setattr( + video_start_time, "read_creation_time", lambda path: PROBE_START_TIME + ) + gpx = self._write_gpx(Path(tmpdir) / "track.gpx", self.GPS_START_TIME) + + _validate_interval( + self._sample(tmpdir, geotag_source=["gpx"], geotag_source_path=gpx), + self.GPS_START_TIME, ) def test_not_without_a_gpx(self, tmpdir, setup_mock, monkeypatch):