diff --git a/README.md b/README.md index 655c1c91a..93ea96218 100644 --- a/README.md +++ b/README.md @@ -333,20 +333,34 @@ It is used to locate the images along the GPS tracks. mapillary_tools process MY_IMAGE_DIR --geotag_source "gpx" --geotag_source_path MY_EXTERNAL_GPS.gpx ``` -To geotag videos with a GPX file, video start time is required to locate the sample images along the GPS tracks. -It is read from the video's own GPS track when it has one, and otherwise from the video creation time. -Cameras disagree on whether the creation time marks the start or the end of the recording (most dashcams write the end), -so mapillary_tools uses the camera model or a date and time in the file name to tell which, and assumes the start when neither says. -Use `--video_start_time` to override it, in UTC, when the sample images end up one video duration off along the track. +To geotag videos with a GPX file, the sample images are located along the GPS track by the time they were captured. +A video whose own GPS records the time is synced to the track by that clock. For a video without one, specify its start +time, in UTC, with `--video_start_time`. That start time applies to every video in the command, so to place several +videos along one GPX file, run it once per video, each with its own start time: ```sh -# Geotagging with GPX works with interval-based sampling only, -# the two options --video_sample_distance -1 --video_sample_interval 2 are therefore required -# to switch from the default distance-based sampling to the legacy interval-based sampling -mapillary_tools video_process MY_VIDEO_DIR \ +mapillary_tools video_process MY_VIDEO.mp4 \ --geotag_source "gpx" \ --geotag_source_path MY_EXTERNAL_GPS.gpx \ - --video_sample_distance -1 --video_sample_interval 2 + --video_start_time 2025_03_14_07_00_30_000 +``` + +Without it, distance-based sampling (the default) assumes that the track starts with the video, and refuses a track +more than twice as long as the video. Interval-based sampling (`--video_sample_distance -1 --video_sample_interval N`) +reads the start time from the video creation time instead. Cameras disagree on whether the creation time marks the start +or the end of the recording (most dashcams write the end), so mapillary_tools uses the camera model or a date and time +in the file name to tell which, and assumes the start when neither says. + +A timelapse video packs several seconds of capture into each second of video. Its GPS clock, if it has one, tells how +many. If it has none, specify how many with `--video_duration_ratio`, for instance 60 for a timelapse that took a frame +every 2 seconds and plays at 30 frames per second: + +```sh +mapillary_tools video_process MY_TIMELAPSE.mp4 \ + --geotag_source "gpx" \ + --geotag_source_path MY_EXTERNAL_GPS.gpx \ + --video_start_time 2025_03_14_07_00_30_000 \ + --video_duration_ratio 60 ``` Ideally, the GPS device and the capture device should use the same clock to get the timestamps synchronized. @@ -363,15 +377,17 @@ mapillary_tools process MY_IMAGE_DIR \ Another option `--interpolation_use_gpx_start_time` moves your images to align with the beginning of the GPS track. This is useful when you can confirm that you start GPS recording and capturing at the same time, or with a known delay. +For a video without a GPS clock, a track that lasts more than twice as long as the video spans at its +`--video_duration_ratio` is then refused. If the GPS logger was left running after the video, pass the start of the +track as `--video_start_time` instead, which places the video the same way: the error prints it. ```sh # Start capturing 2.5 seconds after start GPS recording -mapillary_tools video_process MY_VIDEO_DIR \ +mapillary_tools video_process MY_VIDEO.mp4 \ --geotag_source "gpx" \ --geotag_source_path MY_EXTERNAL_GPS.gpx \ --interpolation_use_gpx_start_time \ - --interpolation_offset_time 2.5 \ - --video_sample_distance -1 --video_sample_interval 2 + --interpolation_offset_time 2.5 ``` ## New video geotagging features (experimental) diff --git a/mapillary_tools/commands/sample_video.py b/mapillary_tools/commands/sample_video.py index c7dc3e922..12be76714 100644 --- a/mapillary_tools/commands/sample_video.py +++ b/mapillary_tools/commands/sample_video.py @@ -34,7 +34,7 @@ def add_basic_arguments(self, parser: argparse.ArgumentParser): ) group.add_argument( "--video_duration_ratio", - help="[DEPRECATED since v0.10.0] Real time video duration ratio of the under or oversampled video duration. [default: %(default)s]", + help="Seconds of real time per second of video, for videos without a GPS clock of their own to tell, such as 60 for a timelapse that took a frame every 2 seconds and plays at 30 frames per second. A timelapse needs it to be synced to a GPX track. The GPS clock of a video, if it has one, wins. [default: %(default)s]", type=float, default=constants.VIDEO_DURATION_RATIO, required=False, diff --git a/mapillary_tools/exceptions.py b/mapillary_tools/exceptions.py index 4c400f309..dac650efc 100644 --- a/mapillary_tools/exceptions.py +++ b/mapillary_tools/exceptions.py @@ -91,8 +91,9 @@ class MapillaryStationaryVideoError(MapillaryDescriptionError): class MapillaryGPXVideoStartUnknownError(MapillaryDescriptionError): """ - Nothing says where along a GPX track a video without GPS starts, and the - track is too long to assume that it starts with the video. + Nothing says how a GPX track lines up with a video without a GPS clock: + the track is too long either to assume that it starts with the video, or, + when it is said to, to end with it at the rate the video is said to run. """ diff --git a/mapillary_tools/geotag/geotag_images_from_video.py b/mapillary_tools/geotag/geotag_images_from_video.py index c20d9f1a3..0adbd7fd9 100644 --- a/mapillary_tools/geotag/geotag_images_from_video.py +++ b/mapillary_tools/geotag/geotag_images_from_video.py @@ -5,6 +5,8 @@ from __future__ import annotations +import bisect +import dataclasses import logging import sys import typing as T @@ -15,15 +17,78 @@ else: from typing_extensions import override -from .. import types, utils +from .. import geo, types, utils from .base import GeotagImagesFromGeneric from .geotag_images_from_gpx import GeotagImagesFromGPX from .geotag_videos_from_video import GeotagVideosFromVideo +from .video_extractors.gpx import GPXVideoExtractor LOG = logging.getLogger(__name__) +@dataclasses.dataclass(frozen=True) +class GPSTimeline: + """ + The Unix time of the video's GPS clock along the video: interpolated + between its GPS timestamps, and extrapolated beyond them at the clock's + rate. A timelapse captured its frames at those times, which one rate from + one of them does not follow: the rate drifts from them over the video, and + the first timestamp of a GoPro MAX timelapse can be seconds early. + + >>> timeline = GPSTimeline((0.0, 1.0, 2.0), (100.0, 104.0, 114.0), rate=10.0) + >>> timeline.to_unix_time(0.5) + 102.0 + >>> timeline.to_unix_time(3.0) + 124.0 + """ + + video_times: tuple[float, ...] + unix_times: tuple[float, ...] + # GPS seconds per video second beyond the timestamps + rate: float + + @classmethod + def read(cls, points: T.Sequence[geo.Point], rate: float) -> GPSTimeline | None: + """ + Read the timeline from the GPS timestamps of the points, or None if they + have none + """ + pairs: list[tuple[float, float]] = [] + for p in points: + unix_time = p.get_unix_time() + if unix_time is None: + continue + if pairs and unix_time == pairs[-1][1]: + # GoPro GPS5 stamps a whole block (~1s) of points with one + # timestamp. Keep the first point of the block + continue + pairs.append((p.time, unix_time)) + + if not pairs: + return None + + return cls( + video_times=tuple(v for v, _ in pairs), + unix_times=tuple(u for _, u in pairs), + rate=rate, + ) + + def to_unix_time(self, video_time: float) -> float: + idx = bisect.bisect_right(self.video_times, video_time) + if 0 < idx < len(self.video_times): + v0, v1 = self.video_times[idx - 1], self.video_times[idx] + u0, u1 = self.unix_times[idx - 1], self.unix_times[idx] + return u0 + (u1 - u0) * (video_time - v0) / (v1 - v0) + + # Before the first timestamp or after the last + anchor = 0 if idx == 0 else -1 + return ( + self.unix_times[anchor] + + (video_time - self.video_times[anchor]) * self.rate + ) + + class GeotagImagesFromVideo(GeotagImagesFromGeneric): def __init__( self, @@ -74,7 +139,7 @@ def to_description( ) geotag = GeotagImagesFromGPX( - video_metadata.points, + self._on_capture_timeline(video_metadata.points), use_gpx_start_time=False, use_image_start_time=True, offset_time=self.offset_time, @@ -98,6 +163,28 @@ def to_description( return final_image_metadatas + @classmethod + def _on_capture_timeline( + cls, points: T.Sequence[geo.Point] + ) -> T.Sequence[geo.Point]: + """ + Time the video's GPS points in GPS seconds from the start of the video, + like the samples, which are stamped with their capture times. A + timelapse packs several into each second of video, as many as its GPS + timestamps tell. + """ + clock = GPXVideoExtractor._read_video_clock(points) + if clock is None or clock.rate == 1.0: + return points + + timeline = GPSTimeline.read(points, clock.rate) + assert timeline is not None, "a GPS clock is read from GPS timestamps" + start = timeline.to_unix_time(0.0) + return [ + dataclasses.replace(p, time=timeline.to_unix_time(p.time) - start) + for p in points + ] + class GeotagImageSamplesFromVideo(GeotagImagesFromGeneric): def __init__( diff --git a/mapillary_tools/geotag/geotag_videos_from_gpx.py b/mapillary_tools/geotag/geotag_videos_from_gpx.py index 05aeecc96..342731eca 100644 --- a/mapillary_tools/geotag/geotag_videos_from_gpx.py +++ b/mapillary_tools/geotag/geotag_videos_from_gpx.py @@ -47,7 +47,12 @@ def _generate_video_extractors( if source_path.is_file(): results.append( GPXVideoExtractor( - video_path, source_path, video_sync=self.video_sync + video_path, + source_path, + video_sync=self.video_sync, + # The track that process uploads with the video. + # Sampling reads its track only within the video + clip_to_video=self.video_sync is None, ) ) else: diff --git a/mapillary_tools/geotag/options.py b/mapillary_tools/geotag/options.py index 72f635f3c..108d59d52 100644 --- a/mapillary_tools/geotag/options.py +++ b/mapillary_tools/geotag/options.py @@ -166,6 +166,10 @@ class VideoSyncOption: # frame at the start of the GPX use_gpx_start_time: bool = False + # --video_duration_ratio: GPS seconds per second of a video without a GPS + # clock of its own to tell, such as a timelapse + duration_ratio: float = 1.0 + SourceOptionSchema = { "type": "object", diff --git a/mapillary_tools/geotag/video_extractors/gpx.py b/mapillary_tools/geotag/video_extractors/gpx.py index 75d4aa168..37bf2582a 100644 --- a/mapillary_tools/geotag/video_extractors/gpx.py +++ b/mapillary_tools/geotag/video_extractors/gpx.py @@ -9,6 +9,7 @@ import datetime import enum import logging +import math import statistics import sys import typing as T @@ -23,7 +24,9 @@ from typing_extensions import override from ... import exceptions, geo, telemetry, types, utils +from ...camm import camm_parser from ...mp4 import construct_mp4_parser as cparser, simple_mp4_parser as sparser +from ...mp4.mp4_sample_parser import MovieBoxParser from ...serializer.description import build_capture_time from .. import options from ..utils import parse_gpx @@ -38,6 +41,10 @@ # of magnitude. _IMPLAUSIBLE_OFFSET_SECONDS = 24 * 3600 +# A GPX track that lasts longer than this is not that of one recording, such as +# a GPS logger left running after the video, but a log of several trips +_MAX_RECORDING_SECONDS = 24 * 3600 + # A GPS clock that runs this much faster than the video clock is a timelapse. # Normal videos stay well below it, although over a few seconds of GPS # timestamped once per block the estimate can be 20% off @@ -98,42 +105,21 @@ def __init__( gpx_path: Path, sync_mode: SyncMode = SyncMode.SYNC, video_sync: options.VideoSyncOption | None = None, + clip_to_video: bool = False, ): self.video_path = video_path self.gpx_path = gpx_path self.sync_mode = sync_mode self.video_sync = video_sync + # Clip the track to the video when a clock gives its pace along the + # video (see _clip_to_video) + self.clip_to_video = clip_to_video if video_sync is not None and video_sync.use_gpx_start_time: self.sync_mode = SyncMode.REBASE @override def extract(self) -> types.VideoMetadata: - try: - gpx_tracks = parse_gpx(self.gpx_path) - except (OSError, UnicodeError, gpxpy.gpx.GPXException) as ex: - raise exceptions.MapillaryInvalidGPXError( - f"{self.gpx_path.name} is not a valid GPX file. " - "Replace it with a valid GPX file." - ) from ex - - if 1 < len(gpx_tracks): - LOG.warning( - f"Found {len(gpx_tracks)} tracks in the GPX file {self.gpx_path}. Will merge points in all the tracks as a single track for interpolation" - ) - - all_points: list[geo.Point] = sum(gpx_tracks, []) - gpx_points = [p for p in all_points if _GPS_EPOCH <= p.time] - if len(gpx_points) < len(all_points): - LOG.warning( - "Ignoring %d of the %d points in the GPX file %s, which are timestamped before 1980", - len(all_points) - len(gpx_points), - len(all_points), - self.gpx_path, - ) - - # The tracks need not be in time order, and interpolation needs the - # merged points to be - gpx_points.sort(key=lambda p: p.time) + gpx_points = self._read_gpx_points() # The first and last Unix time of the track, before it is placed on # the timeline of the video gpx_unix_times = ( @@ -148,8 +134,11 @@ def extract(self) -> types.VideoMetadata: if self.sync_mode is SyncMode.STRICT_SYNC: raise ex start_clock = self._place_without_video_gps(gpx_points) - self._apply_offset_time(gpx_points) - self._check_coverage(gpx_points, gpx_unix_times, [], start_clock) + rate = self._duration_ratio() + self._apply_offset_time(gpx_points, rate=rate) + self._check_coverage(gpx_points, gpx_unix_times, [], start_clock, rate) + if self.clip_to_video and start_clock is not None: + gpx_points = self._clip_to_video(gpx_points) return types.VideoMetadata( filename=self.video_path, filesize=utils.get_file_size(self.video_path), @@ -157,18 +146,30 @@ def extract(self) -> types.VideoMetadata: points=gpx_points, ) - clock: VideoClock | None = None - if self.sync_mode is not SyncMode.REBASE: - clock = self._read_video_clock(native_video_metadata.points) + clock = self._read_video_clock(native_video_metadata.points) - if clock is not None and gpx_points: + if clock is not None and gpx_points and self.sync_mode is SyncMode.REBASE: + # The GPX is placed to start with the video, and runs on at the rate + # of the video's GPS clock, which in a timelapse packs several GPS + # seconds into each second of video + clock = VideoClock( + video_time=0.0, unix_time=gpx_points[0].time, rate=clock.rate + ) + elif clock is not None and gpx_points: offset = gpx_points[0].time - clock.to_unix_time(0) if abs(offset) > _IMPLAUSIBLE_OFFSET_SECONDS: - if noisy and self.sync_mode is SyncMode.SYNC: - # GPS rejected as noise may not have GPS time yet: GoPro - # stamps a placeholder date until it acquires it. That GPS - # is what the GPX replaces, so do not let its clock place - # the GPX days away from the video + # GPS rejected as noise may not have GPS time yet: GoPro + # stamps a placeholder date until it acquires it. That GPS is + # what the GPX replaces, so do not let a placeholder clock + # place the GPX days away from the video. A clock that agrees + # with the creation time of the video is no placeholder: it + # places the GPX as for any video with GPS, and a GPX days + # away from it misses the video + if ( + noisy + and self.sync_mode is SyncMode.SYNC + and not self._clock_matches_creation_time(clock) + ): LOG.warning( "The GPS clock of %s, whose GPS was rejected as noise, is %.1f days away from %s. " "Syncing the GPX from the start of the video instead", @@ -182,6 +183,13 @@ def extract(self) -> types.VideoMetadata: # Its GPS has no clock to sync with, as if it had no GPS clock = self._place_without_video_gps(gpx_points) else: + if self._duration_ratio() != 1.0: + LOG.warning( + "Ignoring --video_duration_ratio %g for %s, whose own GPS clock runs at %.3gx", + self._duration_ratio(), + self.video_path, + clock.rate, + ) if clock.rate != 1.0: LOG.info( "The GPS clock in %s runs %.1f times faster than the video (timelapse). Syncing %s accordingly", @@ -192,20 +200,79 @@ def extract(self) -> types.VideoMetadata: for p in gpx_points: p.time = clock.to_video_time(p.time) - self._apply_offset_time(gpx_points, rate=1.0 if clock is None else clock.rate) + rate = self._duration_ratio() if clock is None else clock.rate + self._apply_offset_time(gpx_points, rate=rate) + # Started with the video, the track is checked against it as one that + # no clock placed: the clock gave it only its pace self._check_coverage( - gpx_points, gpx_unix_times, native_video_metadata.points, clock + gpx_points, + gpx_unix_times, + native_video_metadata.points, + None if self.sync_mode is SyncMode.REBASE else clock, + rate, ) + if self.clip_to_video and clock is not None: + gpx_points = self._clip_to_video(gpx_points) return dataclasses.replace(native_video_metadata, points=gpx_points) + def _read_gpx_points(self) -> list[geo.Point]: + """ + Read the points of the GPX file, merged into one track in time order + """ + try: + gpx_tracks = parse_gpx(self.gpx_path) + except (OSError, UnicodeError, gpxpy.gpx.GPXException) as ex: + raise exceptions.MapillaryInvalidGPXError( + f"{self.gpx_path.name} is not a valid GPX file. " + "Replace it with a valid GPX file." + ) from ex + + if 1 < len(gpx_tracks): + LOG.warning( + f"Found {len(gpx_tracks)} tracks in the GPX file {self.gpx_path}. Will merge points in all the tracks as a single track for interpolation" + ) + + all_points: list[geo.Point] = sum(gpx_tracks, []) + gpx_points = [p for p in all_points if _GPS_EPOCH <= p.time] + if len(gpx_points) < len(all_points): + LOG.warning( + "Ignoring %d of the %d points in the GPX file %s, which are timestamped before 1980", + len(all_points) - len(gpx_points), + len(all_points), + self.gpx_path, + ) + + # The tracks need not be in time order, and interpolation needs the + # merged points to be + gpx_points.sort(key=lambda p: p.time) + return gpx_points + + def _clip_to_video(self, gpx_points: list[geo.Point]) -> list[geo.Point]: + """ + Clip a GPX track to the video when a clock gives its pace along the + video, so that where the video ends along it is known: the GPS clock + of the video, which places the track too unless + --interpolation_use_gpx_start_time starts it with the video, or the + --video_start_time of sampling. Its points outside the video are the + positions of none of its frames, and a GPX of a longer trip can run on + for long after it. + + A track that is only assumed to start with a video without a GPS clock + is left as it is: it may be that of a timelapse, which runs on past + the end of the video at the pace of the GPS clock rather than that of + the video. + """ + return _clip_track(gpx_points, _video_duration(self.video_path)) + def _place_without_video_gps( self, gpx_points: T.Sequence[geo.Point] ) -> VideoClock | None: """ - Place the GPX track on the timeline of a video that has no GPS to sync - it with: by the start time of the video, when sampling was given one, - and otherwise by assuming that the track starts with the video. + Place the GPX track on the timeline of a video that has no GPS clock to + sync it with: by the start time of the video, when sampling was given + one, and otherwise by assuming that the track starts with the video. It + runs at --video_duration_ratio GPS seconds per second of video. Return the clock of the video the track was placed by: the start time, or None if the track was assumed to start with the video. @@ -213,17 +280,34 @@ def _place_without_video_gps( if not gpx_points: return None + ratio = self._duration_ratio() + if self.sync_mode is SyncMode.SYNC and self.video_sync is not None: if self.video_sync.start_time is not None: self._rebase_times( - gpx_points, offset=gpx_points[0].time - self.video_sync.start_time + gpx_points, + offset=(gpx_points[0].time - self.video_sync.start_time) / ratio, + rate=ratio, ) - return VideoClock(video_time=0.0, unix_time=self.video_sync.start_time) - self._check_assumed_start(gpx_points) + return VideoClock( + video_time=0.0, unix_time=self.video_sync.start_time, rate=ratio + ) + self._check_assumed_start(gpx_points, ratio) + elif self.sync_mode is SyncMode.REBASE and self.video_sync is not None: + self._check_given_start(gpx_points, ratio) - self._rebase_times(gpx_points) + self._rebase_times(gpx_points, rate=ratio) return None + def _duration_ratio(self) -> float: + """ + GPS seconds per second of a video without a GPS clock of its own to + tell, from --video_duration_ratio: more than 1 for a timelapse + """ + if self.video_sync is None: + return 1.0 + return self.video_sync.duration_ratio + def _apply_offset_time( self, gpx_points: T.Sequence[geo.Point], rate: float = 1.0 ) -> None: @@ -253,10 +337,12 @@ def _check_coverage( gpx_unix_times: tuple[float, float] | None, video_gps_points: T.Sequence[geo.Point], clock: VideoClock | None, + rate: float, ) -> None: """ Check the GPX track, as placed on the timeline of the video, against - the video, from video time 0 to its mvhd duration. + the video, from video time 0 to its mvhd duration. Each second of video + spans **rate** GPS seconds. A track that misses the video has no position for any moment of it, so it raises when the clock it was placed by says so. A track that covers @@ -278,8 +364,6 @@ def _check_coverage( if video_end <= 0: return - # GPS seconds per video second - rate = 1.0 if clock is None else clock.rate gpx_first = min(p.time for p in gpx_points) gpx_last = max(p.time for p in gpx_points) @@ -319,33 +403,33 @@ def _check_coverage( LOG.warning(message) - def _check_assumed_start(self, gpx_points: T.Sequence[geo.Point]) -> None: + def _check_assumed_start( + self, gpx_points: T.Sequence[geo.Point], ratio: float + ) -> None: """ Check that the GPX track can have started with the video, as it is - assumed to when nothing says when the video starts. + assumed to when nothing says when the video starts. The video spans + **ratio** GPS seconds per second. A track that runs on for longer than the whole video was not recorded for the video alone, and the video could start anywhere along it: raise. One that runs on for less may have been started a little before the video, which puts every frame up to that much too early along it: warn. """ - video_duration = _video_duration(self.video_path) - if video_duration is None: - LOG.debug( - "Unable to read the duration of %s to check the GPX track against", - self.video_path, - ) + durations = self._durations(gpx_points, ratio) + if durations is None: return - gpx_duration = max(p.time for p in gpx_points) - min(p.time for p in gpx_points) - excess = gpx_duration - video_duration + video_span, gpx_duration = durations + excess = gpx_duration - video_span - if video_duration < excess: + if video_span < excess: raise exceptions.MapillaryGPXVideoStartUnknownError( f"The GPX track in {self.gpx_path} lasts {_format_duration(gpx_duration)} " - f"and the video only {_format_duration(video_duration)}, so the video " + f"and {self._describe_video_span(video_span, ratio)}, so the video " "could start anywhere along the track. Specify --video_start_time, or " - "--interpolation_use_gpx_start_time if the track was started with the video" + "--interpolation_use_gpx_start_time if the track was started with the " + "video, and if the video is a timelapse, --video_duration_ratio as well" ) if _LOGGER_TOLERANCE_SECONDS < excess: @@ -360,6 +444,99 @@ def _check_assumed_start(self, gpx_points: T.Sequence[geo.Point]) -> None: _format_duration(excess), ) + def _check_given_start( + self, gpx_points: T.Sequence[geo.Point], ratio: float + ) -> None: + """ + Check a GPX track that --interpolation_use_gpx_start_time says started + with the video, which spans **ratio** GPS seconds per second. A track + that runs on for longer than the whole video was not recorded for it + alone, or belongs to a timelapse, whose rate nothing gives: raise. + """ + durations = self._durations(gpx_points, ratio) + if durations is None: + return + + video_span, gpx_duration = durations + if gpx_duration - video_span <= video_span: + return + + message = ( + f"The GPX track in {self.gpx_path} lasts {_format_duration(gpx_duration)} " + f"and {self._describe_video_span(video_span, ratio)}, so it runs on long " + "after the video ends. " + ) + if gpx_duration < _MAX_RECORDING_SECONDS: + # The start of the track is the start of the video only if the + # track is that of the one recording + message += ( + "If the video is a timelapse, specify --video_duration_ratio. If the " + "GPS logger was started with the video and left running, specify " + f"--video_start_time {build_capture_time(min(p.time for p in gpx_points))}, " + "when the track starts, instead of --interpolation_use_gpx_start_time. " + "If the track covers more than this video, specify the time the video " + "started as --video_start_time instead" + ) + else: + message += ( + "A track that long covers more than this video: specify the time the " + "video started as --video_start_time instead of " + "--interpolation_use_gpx_start_time" + ) + raise exceptions.MapillaryGPXVideoStartUnknownError(message) + + def _describe_video_span(self, video_span: float, ratio: float) -> str: + """ + >>> extractor = GPXVideoExtractor(Path("a.mp4"), Path("a.gpx")) + >>> extractor._describe_video_span(5, 1) + 'the video only 5 seconds' + >>> extractor._describe_video_span(120, 60) + 'the video only 2 seconds, which at --video_duration_ratio 60 span 2.0 minutes' + """ + description = f"the video only {_format_duration(video_span / ratio)}" + if ratio != 1.0: + description += ( + f", which at --video_duration_ratio {ratio:g} span " + f"{_format_duration(video_span)}" + ) + return description + + def _durations( + self, gpx_points: T.Sequence[geo.Point], ratio: float + ) -> tuple[float, float] | None: + """ + The GPS seconds that the video spans at **ratio** GPS seconds per + second, and the duration of the GPX track, or None if the duration of + the video cannot be read + """ + video_duration = _video_duration(self.video_path) + if video_duration is None: + LOG.debug( + "Unable to read the duration of %s to check the GPX track against", + self.video_path, + ) + return None + + gpx_duration = max(p.time for p in gpx_points) - min(p.time for p in gpx_points) + return video_duration * ratio, gpx_duration + + def _clock_matches_creation_time(self, clock: VideoClock) -> bool: + """ + Whether the GPS clock of the video starts within a day of the mvhd + creation time of the video, or False if that cannot be read. + + A day absorbs any time zone the creation time is written in (GoPro + writes local time), and whether it marks the start or the end of the + recording. Some cameras write a creation time that means nothing (a + GoPro HERO10 recorded in 2022 reports 2016-01-01), so a clock that does + not match it may still be real. + """ + creation_time = _video_creation_time(self.video_path) + if creation_time is None: + return False + gap = abs(clock.to_unix_time(0) - creation_time) + return gap <= _IMPLAUSIBLE_OFFSET_SECONDS + def _extract_native_video_metadata(self) -> tuple[types.VideoMetadata, bool]: """ Extract the video metadata, and whether its GPS was rejected as noise @@ -376,14 +553,17 @@ def _extract_native_video_metadata(self) -> tuple[types.VideoMetadata, bool]: return metadata, True @classmethod - def _rebase_times(cls, points: T.Sequence[geo.Point], offset: float = 0.0) -> None: + def _rebase_times( + cls, points: T.Sequence[geo.Point], offset: float = 0.0, rate: float = 1.0 + ) -> None: """ - Rebase point times to start from **offset** + Rebase point times to start from **offset**, at **rate** of their + seconds per second """ if points: first_timestamp = points[0].time for p in points: - p.time = (p.time - first_timestamp) + offset + p.time = (p.time - first_timestamp) / rate + offset @classmethod def _read_video_clock( @@ -417,10 +597,6 @@ def _read_video_clock( if not pairs: return None - # Anchored at its own video time, which is not 0 when the video's GPS - # starts late (e.g. GoPro drops noisy fixes at the start of a recording) - video_time, unix_time = pairs[0] - rate = cls._estimate_clock_rate(pairs) # One timestamp far off (e.g. a placeholder date before GPS time is # acquired) can pull the least-squares slope far off, so a timelapse @@ -428,15 +604,24 @@ def _read_video_clock( # too coarse to measure the rate, but not to tell a timelapse from a # normal video step_rate = cls._median_step_rate(pairs) - if ( + if not ( rate is not None and _MIN_TIMELAPSE_RATE < rate and step_rate is not None and _MIN_TIMELAPSE_RATE < step_rate ): - return VideoClock(video_time=video_time, unix_time=unix_time, rate=rate) + rate = 1.0 + + # Anchored at its own video time, which is not 0 when the video's GPS + # starts late (e.g. GoPro drops noisy fixes at the start of a recording), + # and at the median of all its timestamps mapped back to that time, not + # at the first alone: that can be seconds off the rest (the first GPS + # block of a GoPro timelapse, a stale first fix), and some GPS clocks + # wander by a second or so against the video + video_time = pairs[0][0] + unix_time = statistics.median(u - (v - video_time) * rate for v, u in pairs) - return VideoClock(video_time=video_time, unix_time=unix_time) + return VideoClock(video_time=video_time, unix_time=unix_time, rate=rate) @classmethod def _estimate_clock_rate( @@ -479,6 +664,36 @@ def _median_step_rate(cls, pairs: T.Sequence[tuple[float, float]]) -> float | No return statistics.median(steps) if steps else None +def _clip_track( + points: T.Sequence[geo.Point], duration: float | None +) -> list[geo.Point]: + """ + Clip a track, sorted by time, to the video, from time 0 to duration (None + if unknown), with a point interpolated at either end that the track runs + past, so that the frames up to that end keep their position + + >>> track = [geo.Point(time=t, lat=t, lon=0, alt=None, angle=None) for t in (-2, 2, 6, 10)] + >>> [(p.time, p.lat) for p in _clip_track(track, 8.0)] + [(0.0, 0.0), (2, 2), (6, 6), (8.0, 8.0)] + >>> [p.time for p in _clip_track(track, None)] + [0.0, 2, 6, 10] + >>> [p.time for p in _clip_track(track, 10.0)] + [0.0, 2, 6, 10] + >>> _clip_track(track[:1], 8.0) + [] + """ + if not points: + return [] + + end = math.inf if duration is None else duration + clipped = [p for p in points if 0 <= p.time <= end] + if points[0].time < 0 < points[-1].time and (not clipped or 0 < clipped[0].time): + clipped.insert(0, geo.interpolate(points, 0.0)) + if points[0].time < end < points[-1].time and clipped[-1].time < end: + clipped.append(geo.interpolate(points, end)) + return clipped + + def _video_duration(video_path: Path) -> float | None: """ The duration of the video in seconds, from its mvhd box, or None if it @@ -501,6 +716,19 @@ def _video_duration(video_path: Path) -> float | None: return mvhd.duration / mvhd.timescale +def _video_creation_time(video_path: Path) -> float | None: + """ + The creation time of the video as Unix time, from its mvhd box, or None if + it is unset or cannot be read. + """ + try: + moov = MovieBoxParser.parse_file(video_path) + except (OSError, sparser.ParsingError, C.ConstructError) as ex: + LOG.debug("Failed to read the creation time of %s: %s", video_path, ex) + return None + return camm_parser._extract_creation_time(moov) + + def _format_duration(seconds: float) -> str: """ Format a duration in the largest unit it fills, rounded rather than diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py index e945d2526..d816d1b00 100644 --- a/mapillary_tools/sample_video.py +++ b/mapillary_tools/sample_video.py @@ -29,7 +29,9 @@ ) from .exif_write import ExifEdit from .geotag import factory +from .geotag.geotag_images_from_video import GPSTimeline from .geotag.options import SourceOption, SourceType, VideoSyncOption +from .geotag.video_extractors.gpx import GPXVideoExtractor from .geotag.video_extractors.native import NativeVideoExtractor from .mp4 import mp4_sample_parser from .serializer.description import build_capture_time, parse_capture_time @@ -157,6 +159,12 @@ def sample_video( f"Expect either non-negative video_sample_distance or positive video_sample_interval but got {video_sample_distance} and {video_sample_interval} respectively" ) + # Written as a negated comparison so that NaN is rejected too + if not 0 < video_duration_ratio: + raise exceptions.MapillaryBadParameterError( + f"Expect a positive video_duration_ratio but got {video_duration_ratio}" + ) + video_start_time_dt: datetime.datetime | None = None if video_start_time is not None: try: @@ -175,6 +183,7 @@ def sample_video( ), offset_time=interpolation_offset_time, use_gpx_start_time=interpolation_use_gpx_start_time, + duration_ratio=video_duration_ratio, ) geotag_options = _parse_geotag_options( @@ -244,6 +253,8 @@ def sample_video( sample_interval=video_sample_interval, duration_ratio=video_duration_ratio, start_time=video_start_time_dt, + geotag_options=geotag_options, + video_sync=video_sync, ) except ffmpeglib.FFmpegNotFoundError as ex: # fatal error @@ -412,14 +423,44 @@ def _gps_clock_start_time( ) -def _telemetry_start_time(video_path: Path) -> datetime.datetime | None: +def _read_telemetry(video_path: Path) -> T.Sequence[geo.Point]: + """ + Read the GPS points of the video's own telemetry, or none if it has none + """ try: - video_metadata = NativeVideoExtractor(video_path).extract() + return NativeVideoExtractor(video_path).extract().points except exceptions.MapillaryDescriptionError as ex: - LOG.debug("No video telemetry to read the start time from: %s", ex) + LOG.debug("No video telemetry to read the clock from: %s", ex) + except Exception as ex: + # Telemetry is only one way to tell when the frames were captured, so + # a video whose telemetry fails to parse must still be sampled + LOG.warning( + "Unable to read the telemetry of %s: %s", + video_path.name, + ex, + exc_info=LOG.isEnabledFor(logging.DEBUG), + ) + return [] + + +def _telemetry_clock_rate( + video_path: Path, telemetry_points: T.Sequence[geo.Point] +) -> float | None: + """ + GPS seconds per video second of the video's own GPS clock: more than 1 if + the video is a timelapse, or None if it has no GPS clock + """ + clock = GPXVideoExtractor._read_video_clock(telemetry_points) + if clock is None: return None - return _gps_clock_start_time(video_metadata.points) + if clock.rate != 1.0: + LOG.info( + "The GPS clock in %s runs %.1f times faster than the video (timelapse)", + video_path.name, + clock.rate, + ) + return clock.rate def _parse_filename_time(video_path: Path) -> datetime.datetime | None: @@ -530,7 +571,9 @@ def _creation_time_to_start_time( def _extract_video_start_time( - video_path: Path, probe: ffmpeglib.Probe + video_path: Path, + probe: ffmpeglib.Probe, + telemetry_points: T.Sequence[geo.Point], ) -> datetime.datetime | None: """ Determine the wall clock time at which a video started recording. @@ -541,52 +584,145 @@ def _extract_video_start_time( back to the creation time when there is no telemetry to sync against, which is the case for the plain MP4s that get geotagged from a GPX. """ - try: - start_time = _telemetry_start_time(video_path) - except Exception as ex: - # Telemetry is only one way to find the start time, so a video whose - # telemetry fails to parse must still be sampled - LOG.warning( - "Unable to read the start time of %s from its telemetry: %s", - video_path.name, - ex, - exc_info=LOG.isEnabledFor(logging.DEBUG), - ) - start_time = None - + start_time = _gps_clock_start_time(telemetry_points) if start_time is not None: return start_time return _creation_time_to_start_time(video_path, probe) +def _check_gpx_start_by_interval( + video_path: Path, + geotag_options: T.Sequence[SourceOption], + video_sync: VideoSyncOption, +) -> None: + """ + Check the GPX tracks that --interpolation_use_gpx_start_time says started + with a video without a GPS clock, as distance sampling does. The geotagging + stage puts the first frame at the start of the track and the others at + --video_duration_ratio, so a track that runs on long after the video was + not recorded for it alone, or needs the ratio. + """ + for option in geotag_options: + if option.source is not SourceType.GPX or option.source_path is None: + continue + gpx_path = option.source_path.resolve(video_path) + if not gpx_path.is_file(): + continue + extractor = GPXVideoExtractor(video_path, gpx_path, video_sync=video_sync) + if _gpx_sync_reads_a_clock(extractor): + continue + gpx_points = extractor._read_gpx_points() + if gpx_points: + extractor._check_given_start(gpx_points, video_sync.duration_ratio) + + +def _gpx_sync_reads_a_clock(extractor: GPXVideoExtractor) -> bool: + """ + Whether GPX sync, in distance sampling and process, reads a GPS clock from + the video. It reads one from GPS rejected as noise too, which interval + sampling has no telemetry from, and syncs the GPX at that clock's pace + """ + try: + native_video_metadata, _ = extractor._extract_native_video_metadata() + except exceptions.MapillaryVideoGPSUnusableError: + return False + except Exception: + # Telemetry that fails to parse is warned about by _read_telemetry(), + # and must not stop the video from being sampled + return False + return extractor._read_video_clock(native_video_metadata.points) is not None + + +def _interval_stream_specifier(probe: ffmpeglib.Probe) -> str: + """ + The video stream to sample by interval: the one with the highest + resolution, as for sampling by distance. ffmpeg numbers the frames of all + the streams it samples in one sequence, so with two (a GoPro MAX .360 video + has two) each frame would be numbered, and stamped, as if twice as late. + A video with a single video stream is sampled as "v", which names its + samples as it always has. + """ + video_streams = probe.probe_video_streams() + if len(video_streams) <= 1: + return "v" + + video_stream = probe.probe_video_with_max_resolution() + assert video_stream is not None + return str(video_stream["index"]) + + def _sample_single_video_by_interval( video_path: Path, sample_dir: Path, sample_interval: float, duration_ratio: float, start_time: datetime.datetime | None = None, + geotag_options: T.Sequence[SourceOption] = (), + video_sync: VideoSyncOption | None = None, ) -> None: ffmpeg = ffmpeglib.FFMPEG(constants.FFMPEG_PATH, constants.FFPROBE_PATH) + telemetry_points = _read_telemetry(video_path) + + # Frames are stamped with their capture times, which a timelapse packs + # several seconds of into each second of video: as the video's own GPS + # timestamps tell, and for a video without a GPS clock, at + # --video_duration_ratio + clock_rate = _telemetry_clock_rate(video_path, telemetry_points) + timeline: GPSTimeline | None = None + if clock_rate is None: + rate = duration_ratio + else: + rate = clock_rate + if duration_ratio != 1.0: + LOG.warning( + "Ignoring --video_duration_ratio %g for %s, whose own GPS clock runs at %.3gx", + duration_ratio, + video_path.name, + clock_rate, + ) + if clock_rate != 1.0: + # As distance samples are stamped, and as the geotagging stage + # times the video's track to place the samples along it + timeline = GPSTimeline.read(telemetry_points, clock_rate) + + if clock_rate is None and video_sync is not None and video_sync.use_gpx_start_time: + _check_gpx_start_by_interval(video_path, geotag_options, video_sync) + + probe = ffmpeglib.Probe(ffmpeg.probe_format_and_streams(video_path)) + if start_time is None: - start_time = _extract_video_start_time( - video_path, ffmpeglib.Probe(ffmpeg.probe_format_and_streams(video_path)) - ) + if timeline is not None: + start_time = datetime.datetime.fromtimestamp( + timeline.to_unix_time(0.0), tz=datetime.timezone.utc + ) + else: + start_time = _extract_video_start_time(video_path, probe, telemetry_points) if start_time is None: raise exceptions.MapillaryVideoError( f"Unable to extract video start time from {video_path}" ) + stream_specifier = _interval_stream_specifier(probe) + with wip_dir_context(wip_sample_dir(sample_dir), sample_dir) as wip_dir: - ffmpeg.extract_frames_by_interval(video_path, wip_dir, sample_interval) - frame_samples = ffmpeglib.FFMPEG.sort_selected_samples(wip_dir, video_path) + ffmpeg.extract_frames_by_interval( + video_path, wip_dir, sample_interval, stream_specifier=stream_specifier + ) + frame_samples = ffmpeglib.FFMPEG.sort_selected_samples( + wip_dir, video_path, selected_stream_specifiers=[stream_specifier] + ) for frame_idx_1based, sample_paths in frame_samples: assert len(sample_paths) == 1 if sample_paths[0] is None: continue # extract_frames() produces 1-based frame indices so we need to subtract 1 here - seconds = (frame_idx_1based - 1) * sample_interval * duration_ratio + video_time = (frame_idx_1based - 1) * sample_interval + if timeline is None: + seconds = video_time * rate + else: + seconds = timeline.to_unix_time(video_time) - timeline.to_unix_time(0.0) timestamp = start_time + datetime.timedelta(seconds=seconds) exif_edit = ExifEdit(sample_paths[0]) exif_edit.add_date_time_original(timestamp) @@ -747,6 +883,15 @@ def _sample_single_video_by_distance( ) sorted_sample_indices = sorted(sample_points_by_frame_idx.keys()) + # The GPS timestamps of the track, which frames are stamped within + track_unix_times = [ + unix_time + for unix_time in (p.get_unix_time() for p in video_metadata.points) + if unix_time is not None + ] + first_unix_time = min(track_unix_times, default=None) + last_unix_time = max(track_unix_times, default=None) + # Frames at points with an absolute timestamp are timestamped from it # below, so only the others need the start time if start_time is None and any( @@ -797,6 +942,12 @@ def _sample_single_video_by_distance( # Try to use the GPS timestamp if available (for timelapse videos) gps_unix_time = interp.get_unix_time() if gps_unix_time is not None: + assert first_unix_time is not None and last_unix_time is not None + # Frames up to 1 ms of video past either end of the track are + # kept, and their timestamps extrapolated. In a timelapse, 1 ms + # of video spans several ms of GPS time, and the geotagging + # stage allows a frame only 1 ms past the end of a GPX track + gps_unix_time = min(max(gps_unix_time, first_unix_time), last_unix_time) timestamp = datetime.datetime.fromtimestamp( gps_unix_time, tz=datetime.timezone.utc ) diff --git a/tests/integration/test_process.py b/tests/integration/test_process.py index 5d1a81bb0..747181706 100644 --- a/tests/integration/test_process.py +++ b/tests/integration/test_process.py @@ -688,7 +688,9 @@ def test_process_video_geotag_source_with_gpx_outside_video( assert len(descs) == 1 assert descs[0]["error"]["type"] == "MapillaryOutsideGPXTrackError" assert descs[0]["error"]["vars"] == { - "image_time": "2019_11_18_23_44_42_590", + # Its first GPS timestamp is 23:44:42.590, a second before the rest + # place the start of the video + "image_time": "2019_11_18_23_44_43_613", "gpx_start_time": "2025_03_14_07_00_00_000", "gpx_end_time": "2025_03_14_07_01_33_624", } diff --git a/tests/integration/test_upload.py b/tests/integration/test_upload.py index 0cb7de83d..3bfa1e357 100644 --- a/tests/integration/test_upload.py +++ b/tests/integration/test_upload.py @@ -3,12 +3,15 @@ # This source code is licensed under the BSD license found in the # LICENSE file in the root directory of this source tree. +import datetime import json import subprocess from pathlib import Path import py.path import pytest +from mapillary_tools.geotag.video_extractors.gpx import GPXVideoExtractor +from mapillary_tools.geotag.video_extractors.native import NativeVideoExtractor from .fixtures import ( assert_contains_image_descs, @@ -36,6 +39,54 @@ def test_upload_everything(setup_data: py.path.local, setup_upload: py.path.loca assert_contains_image_descs(descs, uploaded_descs) +@pytest.mark.usefixtures("setup_config") +def test_upload_video_with_gpx_running_past_both_ends( + setup_data: py.path.local, setup_upload: py.path.local +): + """ + The uploaded track keeps only the video, with a point interpolated at + either end, from a GPX of a longer trip + """ + video_path = setup_data.join("gopro_data").join("max-360mode.mp4") + duration = 12.137125 + points = NativeVideoExtractor(Path(video_path)).extract().points + clock = GPXVideoExtractor._read_video_clock(points) + assert clock is not None + video_start = clock.to_unix_time(0) + # From 12.5 seconds before the video to about 47 seconds after it + trkpts = "".join( + f'" + for i in range(61) + ) + gpx_file = setup_data.join("track.gpx") + gpx_file.write( + '\n' + '' + f"{trkpts}\n" + ) + + run_process_and_upload_for_descs( + [ + *["--geotag_source", "gpx"], + *["--geotag_source_path", str(gpx_file)], + str(video_path), + ] + ) + + [[desc]] = extract_all_uploaded_descs(Path(setup_upload)) + track = desc["MAPGPSTrack"] + # The GPX points recorded from 0.5 to 11.5 seconds, and one at either end + assert len(track) == 14 + first, last = track[0], track[-1] + assert first[0] == 0 + assert first[2] == pytest.approx(37.0 + 12.5 * 0.0001, abs=1e-7) + assert first[5] == pytest.approx(video_start, abs=1e-3) + assert last[0] == pytest.approx(duration * 1000, abs=1) + assert last[5] == pytest.approx(video_start + duration, abs=1e-3) + + @pytest.mark.usefixtures("setup_config") def test_upload_image_dir(setup_data: py.path.local, setup_upload: py.path.local): descs = run_process_and_upload_for_descs(["--file_types=image", str(setup_data)]) diff --git a/tests/integration/test_video_process.py b/tests/integration/test_video_process.py index bdd26788b..cad2b6b4a 100644 --- a/tests/integration/test_video_process.py +++ b/tests/integration/test_video_process.py @@ -6,6 +6,7 @@ import functools import json import os +import re import subprocess import tempfile from pathlib import Path @@ -272,18 +273,34 @@ def test_video_process_by_distance_applies_the_interpolation_offset( assert capture_times[-1] <= "2025_03_14_07_01_33_624", capture_times +def _gpx_until(gpx_file: py.path.local, until: str, path: Path) -> Path: + """Copy the GPX file without its points recorded after **until**""" + text = gpx_file.read_text("utf-8") + trkpts = re.findall(r"", text, flags=re.DOTALL) + for trkpt in trkpts: + [time] = re.findall(r"", trkpt) + if until < time: + text = text.replace(trkpt, "") + path.write_text(text, "utf-8") + return path + + def test_video_process_by_distance_with_the_gpx_start_time( setup_data: py.path.local, ): """ As the README describes for a GPX started 2.5 seconds before the video: - the frames start 2.5 seconds into the track, however long it is. + the frames start 2.5 seconds into the track. """ pytest_skip_if_not_ffmpeg_installed() - gpx_file = setup_data.join("gpx").join("sf_30km_h.gpx") - with tempfile.TemporaryDirectory() as dir: + # Stopped about when the 5.76s video ended + gpx_file = _gpx_until( + setup_data.join("gpx").join("sf_30km_h.gpx"), + "2025-03-14T07:00:10", + Path(dir, "track.gpx"), + ) video_path = _with_creation_time( setup_data.join("videos").join("sample-5s.mp4"), "2025-03-14T07:00:36Z", dir ) @@ -306,6 +323,39 @@ def test_video_process_by_distance_with_the_gpx_start_time( assert capture_times[-1] < "2025_03_14_07_00_09_000", capture_times +def test_video_process_with_the_gpx_start_time_refuses_a_far_longer_gpx( + setup_data: py.path.local, +): + """ + A GPX running on for 1.5 minutes after a 5.76s video was not recorded for + it alone, or the video is a timelapse: the error says what to specify. + """ + pytest_skip_if_not_ffmpeg_installed() + + gpx_file = setup_data.join("gpx").join("sf_30km_h.gpx") + + with tempfile.TemporaryDirectory() as dir: + video_path = _with_creation_time( + setup_data.join("videos").join("sample-5s.mp4"), "2025-03-14T07:00:36Z", dir + ) + with pytest.raises(subprocess.CalledProcessError) as excinfo: + run_video_process( + [ + "--video_sample_distance=3", + "--interpolation_use_gpx_start_time", + "--geotag_source=gpx", + *["--geotag_source_path", str(gpx_file)], + str(video_path), + str(Path(dir, "samples")), + ], + capture_output=True, + text=True, + ) + + assert "--video_duration_ratio" in excinfo.value.stderr + assert "--video_start_time" in excinfo.value.stderr + + def test_video_process_by_distance_with_an_untidy_gpx(setup_data: py.path.local): """ GPX files converted from camera logs can list their tracks out of order and diff --git a/tests/unit/test_geotag_images_from_video.py b/tests/unit/test_geotag_images_from_video.py new file mode 100644 index 000000000..9664ed464 --- /dev/null +++ b/tests/unit/test_geotag_images_from_video.py @@ -0,0 +1,126 @@ +# Copyright (c) Meta Platforms, Inc. and affiliates. +# +# This source code is licensed under the BSD license found in the +# LICENSE file in the root directory of this source tree. + +from __future__ import annotations + +import datetime +import os +import shutil +import typing as T +from pathlib import Path + +import pytest +from mapillary_tools import geo, telemetry, types +from mapillary_tools.exif_write import ExifEdit +from mapillary_tools.geotag.geotag_images_from_video import GeotagImagesFromVideo + +_PWD = Path(os.path.dirname(os.path.abspath(__file__))) + +START_TIME = datetime.datetime(2021, 8, 10, 6, 38, 6, tzinfo=datetime.timezone.utc) + + +def _video_metadata( + video_path: Path, gps_seconds: T.Sequence[float] +) -> types.VideoMetadata: + # One point per second of video, at gps_seconds[i] from START_TIME + points = [ + telemetry.GPSPoint( + time=float(i), + lat=40.0 + i * 0.0001, + lon=-74.0, + alt=None, + angle=None, + epoch_time=START_TIME.timestamp() + seconds, + fix=None, + precision=None, + ground_speed=None, + ) + for i, seconds in enumerate(gps_seconds) + ] + return types.VideoMetadata( + filename=video_path, + filetype=types.FileType.GOPRO, + points=T.cast(T.List[geo.Point], points), + ) + + +def _sample(sample_dir: Path, idx: int, capture_time: datetime.datetime) -> Path: + sample_path = sample_dir / f"hello_v_{idx:06d}.jpg" + shutil.copyfile(_PWD / "data/test_exif.jpg", sample_path) + exif_edit = ExifEdit(sample_path) + exif_edit.add_date_time_original(capture_time) + exif_edit.add_gps_datetime(capture_time) + exif_edit.write() + return sample_path + + +@pytest.mark.parametrize("rate", [1.0, 10.0]) +def test_samples_are_placed_by_their_capture_times(tmp_path: Path, rate: float): + video_path = tmp_path / "hello.mp4" + sample_dir = tmp_path / "samples" / "hello.mp4" + sample_dir.mkdir(parents=True) + + # A sample every 2 seconds of video, stamped with its capture time, which + # a timelapse spaces **rate** times as far apart + samples = [ + _sample( + sample_dir, k + 1, START_TIME + datetime.timedelta(seconds=2 * k * rate) + ) + for k in range(4) + ] + + metadatas = GeotagImagesFromVideo( + [_video_metadata(video_path, [rate * i for i in range(10)])], num_processes=0 + ).to_description(samples) + + lat_by_name = {} + for metadata in metadatas: + assert isinstance(metadata, types.ImageMetadata), metadata + lat_by_name[metadata.filename.name] = metadata.lat + assert lat_by_name == { + sample.name: pytest.approx(40.0 + 2 * k * 0.0001) + for k, sample in enumerate(samples) + } + + +def test_timelapse_samples_are_placed_by_the_gps_timestamps(tmp_path: Path): + """ + Distance samples are stamped with the GPS time of their video time. A + GoPro MAX timelapse can stamp its first point seconds early, and its clock + keeps no one rate, so timed at one rate the track placed the samples off, + and the last ones past its end. + """ + video_path = tmp_path / "hello.mp4" + sample_dir = tmp_path / "samples" / "hello.mp4" + sample_dir.mkdir(parents=True) + + # 10x, but the first timestamp 3 seconds early and the clock speeding up + gps_seconds = [0.0] + [3 + 10 * i + 0.1 * i * i for i in range(1, 10)] + + def stamped(video_time: float) -> datetime.datetime: + i = int(video_time) + seconds = gps_seconds[i] + if i + 1 < len(gps_seconds): + seconds += (gps_seconds[i + 1] - gps_seconds[i]) * (video_time - i) + return START_TIME + datetime.timedelta(seconds=seconds) + + video_times = [0.0, 0.5, 2.0, 4.5, 8.0, 9.0] + samples = [ + _sample(sample_dir, k + 1, stamped(video_time)) + for k, video_time in enumerate(video_times) + ] + + metadatas = GeotagImagesFromVideo( + [_video_metadata(video_path, gps_seconds)], num_processes=0 + ).to_description(samples) + + lat_by_name = {} + for metadata in metadatas: + assert isinstance(metadata, types.ImageMetadata), metadata + lat_by_name[metadata.filename.name] = metadata.lat + assert lat_by_name == { + sample.name: pytest.approx(40.0 + video_time * 0.0001) + for sample, video_time in zip(samples, video_times) + } diff --git a/tests/unit/test_gpx_over_noisy_gps.py b/tests/unit/test_gpx_over_noisy_gps.py index 846dc3d53..c6a7c2ec1 100644 --- a/tests/unit/test_gpx_over_noisy_gps.py +++ b/tests/unit/test_gpx_over_noisy_gps.py @@ -24,10 +24,11 @@ from mapillary_tools import exceptions, types from mapillary_tools.geotag import factory -from mapillary_tools.geotag.options import SourceOption, SourceType +from mapillary_tools.geotag.options import SourceOption, SourceType, VideoSyncOption from mapillary_tools.geotag.video_extractors.gpx import GPXVideoExtractor, SyncMode from mapillary_tools.geotag.video_extractors.native import NativeVideoExtractor from mapillary_tools.gpmf import gpmf_gps_filter, gpmf_parser +from mapillary_tools.mp4 import construct_mp4_parser as cparser from mapillary_tools.process_geotag_properties import DEFAULT_GEOTAG_SOURCE_OPTIONS from mapillary_tools.telemetry import GPSFix, GPSPoint @@ -35,6 +36,13 @@ # 2026-09-11T15:39:41Z A_UNIX_TIME = 1789141181.0 +# 2015-10-22T09:53:44Z, a placeholder date GoPro stamps until it acquires GPS +# time +PLACEHOLDER_UNIX_TIME = 1445507624.0 + +# Seconds between the mp4 epoch (1904-01-01) and the Unix epoch +MP4_UNIX_DELTA = 2082844800 + def _gpx_xml(*times: str) -> str: points = "\n".join( @@ -85,6 +93,31 @@ def gpx_path(tmp_path: Path) -> Path: return path +def _write_mvhd(video_path: Path, creation_time: float, duration: float = 3.2): + """ + Make the video an mp4 with only an mvhd box, which is all of it that is + read when the GPMF parser is stubbed out. + """ + mp4_time = int(creation_time) + MP4_UNIX_DELTA + mvhd: cparser.BoxDict = { + "type": b"mvhd", + "data": { + "creation_time": mp4_time, + "modification_time": mp4_time, + "timescale": 1000, + "duration": int(duration * 1000), + }, + } + video_path.write_bytes( + cparser.MP4WithoutSTBLBuilderConstruct.build_boxlist( + [ + {"type": b"ftyp", "data": b"test"}, + {"type": b"moov", "data": [mvhd]}, + ] + ) + ) + + def _stub_gopro(monkeypatch: pytest.MonkeyPatch, points: list[GPSPoint]): info = gpmf_parser.GoProInfo(gps=points, make="GoPro", model="MAX2") monkeypatch.setattr(gpmf_parser, "extract_gopro_info", lambda *a, **kw: info) @@ -161,10 +194,9 @@ def test_noisy_clock_days_away_is_ignored(self, monkeypatch, video_path, gpx_pat GPS rejected as noise may never have got past. Syncing by it would put the GPX years away from the video. """ - # 2015-10-22T09:53:44Z - placeholder = 1445507624.0 points = [ - _point(time=i * 0.1, epoch_time=placeholder + i * 0.1) for i in range(32) + _point(time=i * 0.1, epoch_time=PLACEHOLDER_UNIX_TIME + i * 0.1) + for i in range(32) ] _stub_gopro(monkeypatch, points) @@ -173,6 +205,53 @@ def test_noisy_clock_days_away_is_ignored(self, monkeypatch, video_path, gpx_pat assert [p.time for p in metadata.points] == [0.0, 2.0] assert (metadata.make, metadata.model) == ("GoPro", "MAX2") + def test_noisy_clock_far_from_creation_time_is_ignored( + self, monkeypatch, video_path, gpx_path + ): + """A noisy clock days away from the creation time is ignored too.""" + _write_mvhd(video_path, creation_time=A_UNIX_TIME) + points = [ + _point(time=i * 0.1, epoch_time=PLACEHOLDER_UNIX_TIME + i * 0.1) + for i in range(32) + ] + _stub_gopro(monkeypatch, points) + + metadata = GPXVideoExtractor(video_path, gpx_path).extract() + + assert [p.time for p in metadata.points] == [0.0, 2.0] + + @pytest.mark.parametrize("video_sync", [None, VideoSyncOption()]) + def test_noisy_clock_at_creation_time_rejects_a_gpx_days_away( + self, video_path, tmp_path, noisy_gopro, video_sync + ): + """ + A noisy clock that agrees with the creation time of the video is real, + so a GPX days away from it misses the video, as for good GPS. + """ + # In local time, 8 hours behind UTC, as GoPro writes it + _write_mvhd(video_path, creation_time=A_UNIX_TIME - 8 * 3600) + gpx_path = tmp_path / "another_day.gpx" + gpx_path.write_text(_gpx_xml("2026-09-09T15:39:43Z", "2026-09-09T15:39:45Z")) + + extractor = GPXVideoExtractor(video_path, gpx_path, video_sync=video_sync) + with pytest.raises(exceptions.MapillaryOutsideGPXTrackError): + extractor.extract() + + def test_noisy_clock_syncs_despite_a_meaningless_creation_time( + self, video_path, gpx_path, noisy_gopro + ): + """ + A GoPro HERO10 recorded in 2022 reports 2016-01-01 as its creation + time, so a noisy clock that disagrees with it but agrees with the GPX is + still real. + """ + # 2016-01-01T01:44:00Z + _write_mvhd(video_path, creation_time=1451612640) + + metadata = GPXVideoExtractor(video_path, gpx_path).extract() + + assert [p.time for p in metadata.points] == pytest.approx([2.0, 4.0]) + def test_good_clock_days_away_rejects_the_gpx( self, monkeypatch, video_path, gpx_path ): diff --git a/tests/unit/test_gpx_video_clip.py b/tests/unit/test_gpx_video_clip.py new file mode 100644 index 000000000..e6d713dbd --- /dev/null +++ b/tests/unit/test_gpx_video_clip.py @@ -0,0 +1,157 @@ +# Copyright (c) Meta Platforms, Inc. and affiliates. +# +# This source code is licensed under the BSD license found in the +# LICENSE file in the root directory of this source tree. + +""" +A GPX track whose pace along a video a clock gives is clipped to the video, +with a point interpolated at either end, when it geotags the video. One that +is only assumed to start with a video without a GPS clock is not. +""" + +from __future__ import annotations + +import datetime +from pathlib import Path + +import pytest +from mapillary_tools import types +from mapillary_tools.geotag import factory +from mapillary_tools.geotag.options import ( + SourceOption, + SourcePathOption, + SourceType, + VideoSyncOption, +) +from mapillary_tools.geotag.video_extractors.gpx import GPXVideoExtractor, SyncMode +from mapillary_tools.geotag.video_extractors.native import NativeVideoExtractor + +_DATA = Path(__file__).resolve().parent.parent.joinpath("data") + +# No GPS of its own, and 5.76 seconds long +NO_GPS_VIDEO = _DATA.joinpath("videos", "sample-5s.mp4") +NO_GPS_VIDEO_DURATION = 5.759 + +# GPS of its own, and 12.137125 seconds long +GPS_VIDEO = _DATA.joinpath("gopro_data", "max-360mode.mp4") +GPS_VIDEO_DURATION = 12.137125 + + +def _gps_video_start() -> float: + """The Unix time of the first frame of GPS_VIDEO, by its GPS clock""" + points = NativeVideoExtractor(GPS_VIDEO).extract().points + clock = GPXVideoExtractor._read_video_clock(points) + assert clock is not None + return clock.to_unix_time(0) + + +def _write_gpx(path: Path, start: float, seconds: int) -> Path: + """A GPX track with a point every second from start, heading north at 10 m/s.""" + points = "".join( + f'" + for i in range(seconds + 1) + ) + path.write_text( + '\n' + '' + f"{points}\n" + ) + return path + + +def _times(metadata: types.VideoMetadata) -> list[float]: + return [p.time for p in metadata.points] + + +class TestClockPlacedTrack: + def _gpx(self, tmp_path: Path) -> Path: + # From 12.5 seconds before the video to about 47 seconds after it + return _write_gpx(tmp_path / "track.gpx", _gps_video_start() - 12.5, 60) + + def test_is_clipped_to_the_video(self, tmp_path): + metadata = GPXVideoExtractor( + GPS_VIDEO, self._gpx(tmp_path), clip_to_video=True + ).extract() + + times = _times(metadata) + assert times[0] == 0 + assert times[-1] == pytest.approx(GPS_VIDEO_DURATION) + # The GPX points recorded during the video, from 0.5 to 11.5 seconds + assert len(times) == 2 + 12 + assert all(0 <= t <= GPS_VIDEO_DURATION for t in times) + + def test_the_ends_are_interpolated(self, tmp_path): + metadata = GPXVideoExtractor( + GPS_VIDEO, self._gpx(tmp_path), clip_to_video=True + ).extract() + + first, last = metadata.points[0], metadata.points[-1] + assert first.get_unix_time() == pytest.approx(_gps_video_start()) + assert first.lat == pytest.approx(37.0 + 12.5 * 10 / 111_320, abs=1e-7) + assert last.get_unix_time() == pytest.approx( + _gps_video_start() + GPS_VIDEO_DURATION + ) + + def test_by_the_start_time_is_clipped_to_the_video(self, tmp_path): + start = datetime.datetime(2025, 3, 14, 7, 0, 0, tzinfo=datetime.timezone.utc) + gpx = _write_gpx(tmp_path / "track.gpx", start.timestamp(), 60) + + metadata = GPXVideoExtractor( + NO_GPS_VIDEO, + gpx, + video_sync=VideoSyncOption(start_time=start.timestamp() + 30), + clip_to_video=True, + ).extract() + + times = _times(metadata) + assert times[0] == 0 + assert times[-1] == pytest.approx(NO_GPS_VIDEO_DURATION, abs=1e-3) + + def test_started_with_the_video_is_clipped(self, tmp_path): + """ + --interpolation_use_gpx_start_time places the track, and the GPS clock + gives its pace + """ + metadata = GPXVideoExtractor( + GPS_VIDEO, + self._gpx(tmp_path), + sync_mode=SyncMode.REBASE, + clip_to_video=True, + ).extract() + + times = _times(metadata) + assert times[:-1] == [float(t) for t in range(13)] + assert times[-1] == pytest.approx(GPS_VIDEO_DURATION) + + def test_the_gpx_source_clips_it(self, tmp_path): + [metadata] = factory.process( + [GPS_VIDEO], + [ + SourceOption( + SourceType.GPX, + num_processes=0, + source_path=SourcePathOption(source_path=self._gpx(tmp_path)), + ) + ], + ) + + assert isinstance(metadata, types.VideoMetadata) + assert _times(metadata)[0] == 0 + assert _times(metadata)[-1] == pytest.approx(GPS_VIDEO_DURATION) + + +class TestTrackAssumedToStartWithTheVideo: + """ + Without a clock to say where the video ends along the track, a track that + runs on past it may be that of a timelapse, at the pace of the GPS clock + """ + + def test_without_gps_is_not_clipped(self, tmp_path): + start = datetime.datetime(2025, 3, 14, 7, 0, 0, tzinfo=datetime.timezone.utc) + gpx = _write_gpx(tmp_path / "track.gpx", start.timestamp(), 60) + + metadata = GPXVideoExtractor(NO_GPS_VIDEO, gpx, clip_to_video=True).extract() + + assert _times(metadata) == [float(t) for t in range(61)] diff --git a/tests/unit/test_gpx_video_clock.py b/tests/unit/test_gpx_video_clock.py index 00313707c..eb95bc69e 100644 --- a/tests/unit/test_gpx_video_clock.py +++ b/tests/unit/test_gpx_video_clock.py @@ -110,6 +110,23 @@ def test_block_timestamps_anchor_at_the_block_start(self): def test_one_jump_does_not_make_a_timelapse(self, pairs): assert _read(pairs).rate == 1.0 + @pytest.mark.parametrize("rate", [1.0, 30.0]) + def test_a_first_timestamp_off_the_rest_does_not_anchor_the_clock(self, rate): + """The first GPS block of a GoPro timelapse can be stamped 3s early.""" + pairs = [(0, A_UNIX_TIME - 3)] + [ + (i / rate, A_UNIX_TIME + i) for i in range(1, 60) + ] + clock = _read(pairs) + assert clock.rate == pytest.approx(rate, rel=0.05) + assert clock.to_unix_time(0) == pytest.approx(A_UNIX_TIME, abs=0.2) + + def test_a_placeholder_date_does_not_anchor_the_clock(self): + """GoPro stamps a placeholder date until it acquires GPS time.""" + pairs = [(i, PLACEHOLDER_UNIX_TIME + i) for i in range(5)] + [ + (i, A_UNIX_TIME + i) for i in range(5, 15) + ] + assert _read(pairs).to_unix_time(0) == A_UNIX_TIME + def test_no_timestamps(self): points = [ geo.Point(time=i, lat=1, lon=1, alt=None, angle=None) for i in range(3) diff --git a/tests/unit/test_gpx_video_coverage.py b/tests/unit/test_gpx_video_coverage.py index 73a6581a8..7ad03592e 100644 --- a/tests/unit/test_gpx_video_coverage.py +++ b/tests/unit/test_gpx_video_coverage.py @@ -33,7 +33,7 @@ SourceType, VideoSyncOption, ) -from mapillary_tools.geotag.video_extractors.gpx import GPXVideoExtractor +from mapillary_tools.geotag.video_extractors.gpx import GPXVideoExtractor, SyncMode from mapillary_tools.mp4 import construct_mp4_parser as cparser, simple_mp4_builder @@ -321,12 +321,12 @@ class TestTimelapse: # 11 seconds of video spanning 165 seconds of GPS time RATE = 15.0 - def _extract(self, tmp_path, gpx_start: float, gpx_seconds: float): + def _extract(self, tmp_path, gpx_start: float, gpx_seconds: float, **kwargs): video_path = _write_camm_mp4( tmp_path / "timelapse.mp4", _video_track(self.RATE), duration=11.0 ) gpx_path = _write_gpx(tmp_path / "track.gpx", gpx_start, gpx_seconds) - return GPXVideoExtractor(video_path, gpx_path).extract() + return GPXVideoExtractor(video_path, gpx_path, **kwargs).extract() def test_gpx_of_the_whole_recording_is_silent(self, tmp_path, caplog): """At 1x, the track would run on 154 seconds past the video's end.""" @@ -349,6 +349,19 @@ def test_gap_is_reported_in_gps_time(self, tmp_path): self._extract(tmp_path, recording_end + 3600, 60) assert "by 1.0 hours." in str(info.value) + def test_gpx_start_time_keeps_the_rate(self, tmp_path, caplog): + """ + --interpolation_use_gpx_start_time starts the track with the video, + whatever the clock says, and the clock still sets its pace + """ + with caplog.at_level(logging.WARNING): + metadata = self._extract( + tmp_path, A_UNIX_TIME + 3600, 11 * self.RATE, sync_mode=SyncMode.REBASE + ) + assert not _warnings(caplog) + assert metadata.points[0].time == 0 + assert metadata.points[-1].time == pytest.approx(11.0, abs=1e-3) + class TestVideoWithoutGPSClock: """ diff --git a/tests/unit/test_gpx_video_sync.py b/tests/unit/test_gpx_video_sync.py index 619e13e60..9ab9a5861 100644 --- a/tests/unit/test_gpx_video_sync.py +++ b/tests/unit/test_gpx_video_sync.py @@ -24,7 +24,11 @@ SourceType, VideoSyncOption, ) -from mapillary_tools.geotag.video_extractors.gpx import GPXVideoExtractor, VideoClock +from mapillary_tools.geotag.video_extractors.gpx import ( + GPXVideoExtractor, + SyncMode, + VideoClock, +) _DATA = Path(__file__).resolve().parent.parent.joinpath("data") @@ -89,6 +93,123 @@ def test_a_track_far_longer_than_the_video_is_refused(self, tmp_path): _extract(NO_GPS_VIDEO, gpx, VideoSyncOption()) assert "--video_start_time" in str(excinfo.value) + assert "--video_duration_ratio" in str(excinfo.value) + + def test_gpx_start_time_on_a_track_far_longer_than_the_video_is_refused( + self, tmp_path + ): + """ + It was not recorded for the video alone, or the video is a timelapse, + and nothing says how fast it runs through the track. + """ + gpx = _write_gpx(tmp_path / "track.gpx", seconds=60) + + with pytest.raises(exceptions.MapillaryGPXVideoStartUnknownError) as excinfo: + GPXVideoExtractor( + NO_GPS_VIDEO, gpx, SyncMode.REBASE, video_sync=VideoSyncOption() + ).extract() + + message = str(excinfo.value) + assert "--video_duration_ratio" in message + # For a logger left running, ready to paste: the track starts at GPX_START + assert ( + "started with the video and left running, specify --video_start_time " + "2025_03_14_07_00_00_000, when the track starts" + ) in message + # Not for a track of more than this video, which the start of the track + # would put on the wrong part of it + assert "covers more than this video, specify the time the video started" in ( + message + ) + + def test_gpx_start_time_on_a_log_of_days_offers_no_start_time(self, tmp_path): + """ + A track of several days is a log of several trips, and its start that + of the first: placed there, the video would land on the wrong trip. + """ + gpx = _write_gpx_tracks( + tmp_path / "track.gpx", + [ + ["2025-03-14T07:00:00Z", "2025-03-14T07:00:01Z"], + ["2025-03-16T07:00:00Z", "2025-03-16T07:00:01Z"], + ], + ) + + with pytest.raises(exceptions.MapillaryGPXVideoStartUnknownError) as excinfo: + GPXVideoExtractor( + NO_GPS_VIDEO, gpx, SyncMode.REBASE, video_sync=VideoSyncOption() + ).extract() + + message = str(excinfo.value) + assert "specify the time the video started as --video_start_time" in message + assert "2025_03_14" not in message + assert "left running" not in message + assert "--video_duration_ratio" not in message + + def test_gpx_start_time_on_a_track_as_long_as_the_video_is_silent( + self, tmp_path, caplog + ): + gpx = _write_gpx(tmp_path / "track.gpx", seconds=6) + + with caplog.at_level(logging.WARNING): + GPXVideoExtractor( + NO_GPS_VIDEO, gpx, SyncMode.REBASE, video_sync=VideoSyncOption() + ).extract() + + assert not caplog.text + + def test_the_duration_ratio_paces_a_track_placed_by_the_start_time(self, tmp_path): + """ + One GPX can serve many videos, each placed by its start time, and a + timelapse runs through it at its ratio. + """ + gpx = _write_gpx(tmp_path / "track.gpx", seconds=600) + video_start = GPX_START + datetime.timedelta(seconds=300) + + metadata = _extract( + NO_GPS_VIDEO, + gpx, + VideoSyncOption(start_time=video_start.timestamp(), duration_ratio=10), + ) + + assert metadata.points[300].time == pytest.approx(0) + assert metadata.points[310].time == pytest.approx(1) + # the absolute times, which the frames are stamped with, are untouched + assert metadata.points[310].get_unix_time() == pytest.approx( + video_start.timestamp() + 10 + ) + + @pytest.mark.parametrize("sync_mode", [SyncMode.SYNC, SyncMode.REBASE]) + def test_the_duration_ratio_spreads_a_track_over_the_video( + self, tmp_path, caplog, sync_mode + ): + """ + The 5.76s video spans 57.6s of GPS time at 10 seconds per second, so a + track of 57s starts and ends with it. + """ + gpx = _write_gpx(tmp_path / "track.gpx", seconds=57) + + with caplog.at_level(logging.WARNING): + metadata = GPXVideoExtractor( + NO_GPS_VIDEO, + gpx, + sync_mode, + video_sync=VideoSyncOption(duration_ratio=10), + ).extract() + + assert metadata.points[0].time == 0 + assert metadata.points[-1].time == pytest.approx(5.7) + assert not caplog.text + + def test_a_track_too_long_even_at_the_duration_ratio_is_refused(self, tmp_path): + gpx = _write_gpx(tmp_path / "track.gpx", seconds=600) + + with pytest.raises(exceptions.MapillaryGPXVideoStartUnknownError) as excinfo: + _extract(NO_GPS_VIDEO, gpx, VideoSyncOption(duration_ratio=10)) + + assert "which at --video_duration_ratio 10 span 57.6 seconds" in str( + excinfo.value + ) def test_a_track_a_little_longer_than_the_video_warns(self, tmp_path, caplog): gpx = _write_gpx(tmp_path / "track.gpx", seconds=9) @@ -110,13 +231,18 @@ def test_a_track_as_long_as_the_video_starts_with_it_silently( assert metadata.points[0].time == 0 assert not caplog.text - def test_process_still_assumes_the_track_starts_with_the_video(self, tmp_path): + @pytest.mark.parametrize("sync_mode", [SyncMode.SYNC, SyncMode.REBASE]) + def test_process_still_assumes_the_track_starts_with_the_video( + self, tmp_path, caplog, sync_mode + ): """process has no --video_start_time to offer as a way out.""" gpx = _write_gpx(tmp_path / "track.gpx", seconds=3600) - metadata = _extract(NO_GPS_VIDEO, gpx, video_sync=None) + with caplog.at_level(logging.WARNING): + metadata = GPXVideoExtractor(NO_GPS_VIDEO, gpx, sync_mode).extract() assert metadata.points[0].time == 0 + assert "timelapse" not in caplog.text def test_the_refusal_is_not_handed_on_to_the_next_source(self, tmp_path): """ @@ -162,6 +288,23 @@ def test_the_video_gps_wins_over_the_start_time(self, tmp_path): p.time for p in synced.points ] + def test_the_video_gps_clock_wins_over_the_duration_ratio(self, tmp_path, caplog): + """ + The ratio is for the videos without a GPS clock, which a batch of + videos may mix with videos that have one. + """ + gpx = _write_gpx( + tmp_path / "track.gpx", + seconds=30, + start=_GPS_VIDEO_START, + ) + + synced = _extract(GPS_VIDEO, gpx, video_sync=None) + with_ratio = _extract(GPS_VIDEO, gpx, VideoSyncOption(duration_ratio=10)) + + assert [p.time for p in with_ratio.points] == [p.time for p in synced.points] + assert "Ignoring --video_duration_ratio 10 for" in caplog.text + class TestInterpolationOptions: """ @@ -206,10 +349,10 @@ def test_the_offset_applies_to_a_track_synced_by_the_video_gps(self, tmp_path): def test_use_gpx_start_time_starts_the_track_with_the_video(self, tmp_path): """ - Whatever the start time says, and however long the track: the user - has said that the two started together. + Whatever the start time says: the user has said that the two started + together. """ - gpx = _write_gpx(tmp_path / "track.gpx", seconds=60) + gpx = _write_gpx(tmp_path / "track.gpx", seconds=6) metadata = _extract( NO_GPS_VIDEO, diff --git a/tests/unit/test_sample_video.py b/tests/unit/test_sample_video.py index cf4760b73..c1627d0f8 100644 --- a/tests/unit/test_sample_video.py +++ b/tests/unit/test_sample_video.py @@ -74,12 +74,16 @@ def setup_mock(monkeypatch): monkeypatch.setattr(ffmpeglib, "FFMPEG", MOCK_FFMPEG) -def _validate_interval(samples: T.Sequence[Path], video_start_time): +def _validate_interval( + samples: T.Sequence[Path], video_start_time, seconds_per_sample: float = 2 +): assert len(samples), "expect samples but got none" for idx, sample in enumerate(sorted(samples)): assert sample.name == f"hello_v_{idx + 1:06d}.jpg" exif = exif_read.ExifRead(sample) - expected_dt = video_start_time + datetime.timedelta(seconds=2 * idx) + expected_dt = video_start_time + datetime.timedelta( + seconds=seconds_per_sample * idx + ) assert exif.extract_capture_time() == expected_dt @@ -131,6 +135,40 @@ def test_sample_video_with_start_time(tmpdir: py.path.local, setup_mock): _validate_interval([Path(s) for s in samples], video_start_time) +def test_sample_the_video_stream_with_the_highest_resolution( + tmpdir: py.path.local, setup_mock +): + """ + By interval as by distance: ffmpeg would number the frames of two streams + in one sequence, and they would be stamped as if twice as late. + """ + root = _PWD.joinpath("data/mock_sample_video") + probe = json.loads(root.joinpath("videos", "hello.mp4").read_text()) + [video_stream] = [s for s in probe["streams"] if s["codec_type"] == "video"] + probe["streams"].append({**video_stream, "index": 2, "width": 5760}) + video_dir = Path(tmpdir.mkdir("videos")) + video_dir.joinpath("hello.mp4").write_text(json.dumps(probe)) + sample_dir = tmpdir.mkdir("sampled_video_frames") + + sample_video.sample_video( + video_dir, + Path(sample_dir), + video_sample_distance=-1, + video_sample_interval=2, + rerun=True, + ) + + samples = sorted(Path(s) for s in sample_dir.join("hello.mp4").listdir()) + assert len(samples), "expect samples but got none" + for idx, sample in enumerate(samples): + assert sample.name == f"hello_2_{idx + 1:06d}.jpg" + assert exif_read.ExifRead( + sample + ).extract_capture_time() == PROBE_START_TIME + datetime.timedelta( + seconds=2 * idx + ) + + def test_sample_video_from_gps_clock(tmpdir: py.path.local, setup_mock, monkeypatch): """A video's own GPS clock wins over the container's creation time.""" root = _PWD.joinpath("data/mock_sample_video") @@ -208,10 +246,229 @@ def raise_type_error(): samples = sample_dir.join("hello.mp4").listdir() _validate_interval([Path(s) for s in samples], PROBE_START_TIME) - assert "Unable to read the start time of hello.mp4 from its telemetry" in ( - caplog.text + assert "Unable to read the telemetry of hello.mp4" in caplog.text + + +class TestSampleTimelapseByInterval: + """ + Interval samples of a timelapse are stamped with their capture times, which + its GPS clock spaces 10 times as far apart as the frames + """ + + GPS_START_TIME = datetime.datetime( + 2021, 8, 10, 6, 38, 6, tzinfo=datetime.timezone.utc ) + @classmethod + def _mock_telemetry(cls, monkeypatch, gps_seconds: T.Callable[[int], float]): + # One point per second of video, at gps_seconds(i) from GPS_START_TIME + points = [ + telemetry.GPSPoint( + time=float(i), + lat=40.0 + i * 0.001, + lon=-74.0, + alt=None, + angle=None, + epoch_time=cls.GPS_START_TIME.timestamp() + gps_seconds(i), + fix=None, + precision=None, + ground_speed=None, + ) + for i in range(60) + ] + monkeypatch.setattr( + sample_video, + "NativeVideoExtractor", + lambda video_path: mock.Mock( + extract=lambda: VideoMetadata( + filename=video_path, + filetype=FileType.GOPRO, + points=T.cast(T.List[geo.Point], points), + ) + ), + ) + + @pytest.fixture + def timelapse_telemetry(self, monkeypatch): + # Recorded 10 seconds apart + self._mock_telemetry(monkeypatch, lambda i: 10 * i) + + def _sample(self, tmpdir: py.path.local, **kwargs) -> list[Path]: + root = _PWD.joinpath("data/mock_sample_video") + sample_dir = tmpdir.mkdir("sampled_video_frames") + sample_video.sample_video( + root.joinpath("videos"), + Path(sample_dir), + video_sample_distance=-1, + video_sample_interval=2, + rerun=True, + **kwargs, + ) + return [Path(s) for s in sample_dir.join("hello.mp4").listdir()] + + def test_capture_times_follow_the_gps_clock( + self, tmpdir, setup_mock, timelapse_telemetry, caplog + ): + _validate_interval( + self._sample(tmpdir), self.GPS_START_TIME, seconds_per_sample=20 + ) + assert "--video_duration_ratio" not in caplog.text + + def test_capture_times_follow_the_gps_timestamps( + self, tmpdir, setup_mock, monkeypatch + ): + """ + Not one rate: the GPS clock of a timelapse drifts from it, and distance + samples are stamped from the timestamps. The geotagging stage places + both kinds of samples along a track timed by them + """ + self._mock_telemetry(monkeypatch, lambda i: 10 * i + 0.1 * i * i) + + samples = sorted(self._sample(tmpdir)) + + assert len(samples) > 2 + for idx, sample in enumerate(samples): + video_time = 2 * idx + assert exif_read.ExifRead( + sample + ).extract_capture_time() == self.GPS_START_TIME + datetime.timedelta( + seconds=10 * video_time + 0.1 * video_time * video_time + ) + + def test_from_the_given_start_time(self, tmpdir, setup_mock, timelapse_telemetry): + video_start_time = "2020_08_10_14_37_05_023" + _validate_interval( + self._sample(tmpdir, video_start_time=video_start_time), + description.parse_capture_time(video_start_time), + seconds_per_sample=20, + ) + + def test_the_gps_clock_wins_over_the_duration_ratio( + self, tmpdir, setup_mock, timelapse_telemetry, caplog + ): + """ + The ratio is for the videos without a GPS clock, which a batch of + videos may mix with videos that have one + """ + _validate_interval( + self._sample(tmpdir, video_duration_ratio=5), + self.GPS_START_TIME, + seconds_per_sample=20, + ) + assert ( + "Ignoring --video_duration_ratio 5 for hello.mp4, whose own GPS clock runs at 10x" + in caplog.text + ) + + def test_the_duration_ratio_paces_a_video_without_a_gps_clock( + self, tmpdir, setup_mock + ): + _validate_interval( + self._sample(tmpdir, video_duration_ratio=5), + PROBE_START_TIME, + seconds_per_sample=10, + ) + + +class TestGPXStartTimeByInterval: + """ + With --interpolation_use_gpx_start_time, the geotagging stage puts the + first frame of a video without a GPS clock at the start of the GPX, and + the others at --video_duration_ratio. A track that runs on long after the + video is refused, as in distance sampling. + """ + + # 5.76 seconds long, without GPS + VIDEO = _PWD.parent.joinpath("data", "videos", "sample-5s.mp4") + + @staticmethod + def _write_gpx(path: Path, seconds: int) -> Path: + start = datetime.datetime(2025, 3, 14, 7, 0, 0, tzinfo=datetime.timezone.utc) + trkpts = "".join( + f'" + for i in range(seconds + 1) + ) + path.write_text( + '' + f"{trkpts}" + ) + return path + + def test_a_track_far_longer_than_the_video_is_refused( + self, tmp_path, setup_mock, caplog + ): + video_dir = tmp_path / "videos" + video_dir.mkdir() + shutil.copyfile(self.VIDEO, video_dir / self.VIDEO.name) + gpx = self._write_gpx(tmp_path / "track.gpx", seconds=60) + + with pytest.raises(exceptions.MapillaryVideoSampleError): + sample_video.sample_video( + video_dir, + tmp_path / "samples", + video_sample_distance=-1, + video_sample_interval=2, + geotag_source=["gpx"], + geotag_source_path=gpx, + interpolation_use_gpx_start_time=True, + ) + + assert "--video_duration_ratio" in caplog.text + + def test_a_track_that_ends_with_the_video_at_the_ratio_passes(self, tmp_path): + """The video spans 57.6s of GPS time at 10 seconds per second.""" + gpx = self._write_gpx(tmp_path / "track.gpx", seconds=57) + video_sync = VideoSyncOption(use_gpx_start_time=True, duration_ratio=10) + options = sample_video._parse_geotag_options( + ["gpx"], gpx, None, tmp_path, video_sync + ) + + sample_video._check_gpx_start_by_interval(self.VIDEO, options, video_sync) + + def test_gps_rejected_as_noise_still_has_a_clock(self, tmp_path, monkeypatch): + """ + Interval sampling reads no telemetry from GPS rejected as noise, but + GPX sync reads its clock, and syncs the GPX at that clock's pace, as + for any video with a GPS clock + """ + # A 12x timelapse clock + points = [ + telemetry.GPSPoint( + time=i * 0.1, + lat=37.0, + lon=-122.0, + alt=None, + angle=None, + epoch_time=1741935600.0 + i * 1.2, + fix=None, + precision=None, + ground_speed=None, + ) + for i in range(30) + ] + monkeypatch.setattr( + sample_video.GPXVideoExtractor, + "_extract_native_video_metadata", + lambda _self: ( + VideoMetadata( + filename=self.VIDEO, + filetype=FileType.GOPRO, + points=T.cast(T.List[geo.Point], points), + ), + True, + ), + ) + gpx = self._write_gpx(tmp_path / "track.gpx", seconds=60) + video_sync = VideoSyncOption(use_gpx_start_time=True) + options = sample_video._parse_geotag_options( + ["gpx"], gpx, None, tmp_path, video_sync + ) + + sample_video._check_gpx_start_by_interval(self.VIDEO, options, video_sync) + class TestGPSClockStartTime: """Tests for _gps_clock_start_time.""" @@ -805,6 +1062,17 @@ def test_negative_distance_raises(self, tmp_path: Path) -> None: ) +@pytest.mark.parametrize("ratio", [0, -1, float("nan")]) +def test_a_duration_ratio_must_be_positive(tmp_path: Path, ratio: float) -> None: + """It divides the GPS time of a video without a GPS clock.""" + with pytest.raises(exceptions.MapillaryBadParameterError): + sample_video.sample_video( + _PWD.joinpath("data/mock_sample_video/videos/hello.mp4"), + tmp_path, + video_duration_ratio=ratio, + ) + + class TestSampleVideoRerun: """Test rerun behavior of sample_video.""" @@ -1167,6 +1435,58 @@ def test_timestamps_from_gps_clock(self, tmp_path: Path) -> None: gps_start_time + datetime.timedelta(seconds=idx) ) + def test_timelapse_frames_past_the_end_are_stamped_within_the_track( + self, tmp_path: Path + ) -> None: + """ + The last frame is 0.5 ms of video past the end of the track, which a + 30x timelapse stretches to 15 ms of GPS time. Stamped there, it would + fall outside the GPX in the geotagging stage, which allows 1 ms. + """ + video_dir = tmp_path / "videos" + video_dir.mkdir() + video_file = video_dir / "test.mp4" + video_file.touch() + output_dir = tmp_path / "output" + + gps_start_time = datetime.datetime( + 2021, 8, 10, 6, 38, 6, tzinfo=datetime.timezone.utc + ) + gps_points = [ + telemetry.GPSPoint( + time=time, + lat=40.0 + time * 0.001, + lon=-74.0, + alt=None, + angle=None, + epoch_time=gps_start_time.timestamp() + 30 * time, + fix=None, + precision=None, + ground_speed=None, + ) + for time in [*range(9), 8.9995] + ] + mocks = self._setup_mocks(tmp_path, video_file, gps_points=gps_points) + + with ( + mocks["patches"]["ffmpeg_cls"], + mocks["patches"]["geotag_cls"], + mocks["patches"]["moov_parse"], + ): + sample_video.sample_video( + video_import_path=video_file, + import_path=output_dir, + video_sample_distance=0.0, + ) + + frames = sorted((output_dir / "test.mp4").glob("*.jpg")) + capture_times = [ + exif_read.ExifRead(frame).extract_capture_time() for frame in frames + ] + assert capture_times == [ + gps_start_time + datetime.timedelta(seconds=30 * idx) for idx in range(9) + ] + [gps_start_time + datetime.timedelta(seconds=269.985)] + class TestFailuresAreReportedTogether: """