diff --git a/mapillary_tools/commands/sample_video.py b/mapillary_tools/commands/sample_video.py index c7dc3e92..12be7671 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 4c400f30..dac650ef 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 c20d9f1a..0adbd7fd 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/options.py b/mapillary_tools/geotag/options.py index 72f635f3..108d59d5 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 75d4aa16..17e0764d 100644 --- a/mapillary_tools/geotag/video_extractors/gpx.py +++ b/mapillary_tools/geotag/video_extractors/gpx.py @@ -38,6 +38,10 @@ # of magnitude. _IMPLAUSIBLE_OFFSET_SECONDS = 24 * 3600 +# A GPX track that lasts longer than this is not that of one recording, such as +# a GPS logger left running after the video, but a log of several trips +_MAX_RECORDING_SECONDS = 24 * 3600 + # A GPS clock that runs this much faster than the video clock is a timelapse. # Normal videos stay well below it, although over a few seconds of GPS # timestamped once per block the estimate can be 20% off @@ -108,32 +112,7 @@ def __init__( @override def extract(self) -> types.VideoMetadata: - try: - gpx_tracks = parse_gpx(self.gpx_path) - except (OSError, UnicodeError, gpxpy.gpx.GPXException) as ex: - raise exceptions.MapillaryInvalidGPXError( - f"{self.gpx_path.name} is not a valid GPX file. " - "Replace it with a valid GPX file." - ) from ex - - if 1 < len(gpx_tracks): - LOG.warning( - f"Found {len(gpx_tracks)} tracks in the GPX file {self.gpx_path}. Will merge points in all the tracks as a single track for interpolation" - ) - - all_points: list[geo.Point] = sum(gpx_tracks, []) - gpx_points = [p for p in all_points if _GPS_EPOCH <= p.time] - if len(gpx_points) < len(all_points): - LOG.warning( - "Ignoring %d of the %d points in the GPX file %s, which are timestamped before 1980", - len(all_points) - len(gpx_points), - len(all_points), - self.gpx_path, - ) - - # The tracks need not be in time order, and interpolation needs the - # merged points to be - gpx_points.sort(key=lambda p: p.time) + gpx_points = self._read_gpx_points() # The first and last Unix time of the track, before it is placed on # the timeline of the video gpx_unix_times = ( @@ -148,8 +127,9 @@ def extract(self) -> types.VideoMetadata: if self.sync_mode is SyncMode.STRICT_SYNC: raise ex start_clock = self._place_without_video_gps(gpx_points) - self._apply_offset_time(gpx_points) - self._check_coverage(gpx_points, gpx_unix_times, [], start_clock) + rate = self._duration_ratio() + self._apply_offset_time(gpx_points, rate=rate) + self._check_coverage(gpx_points, gpx_unix_times, [], start_clock, rate) return types.VideoMetadata( filename=self.video_path, filesize=utils.get_file_size(self.video_path), @@ -157,11 +137,16 @@ def extract(self) -> types.VideoMetadata: points=gpx_points, ) - clock: VideoClock | None = None - if self.sync_mode is not SyncMode.REBASE: - clock = self._read_video_clock(native_video_metadata.points) + 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: @@ -182,6 +167,13 @@ def extract(self) -> types.VideoMetadata: # Its GPS has no clock to sync with, as if it had no GPS clock = self._place_without_video_gps(gpx_points) else: + if self._duration_ratio() != 1.0: + LOG.warning( + "Ignoring --video_duration_ratio %g for %s, whose own GPS clock runs at %.3gx", + self._duration_ratio(), + self.video_path, + clock.rate, + ) if clock.rate != 1.0: LOG.info( "The GPS clock in %s runs %.1f times faster than the video (timelapse). Syncing %s accordingly", @@ -192,20 +184,60 @@ 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, ) return dataclasses.replace(native_video_metadata, points=gpx_points) + def _read_gpx_points(self) -> list[geo.Point]: + """ + Read the points of the GPX file, merged into one track in time order + """ + try: + gpx_tracks = parse_gpx(self.gpx_path) + except (OSError, UnicodeError, gpxpy.gpx.GPXException) as ex: + raise exceptions.MapillaryInvalidGPXError( + f"{self.gpx_path.name} is not a valid GPX file. " + "Replace it with a valid GPX file." + ) from ex + + if 1 < len(gpx_tracks): + LOG.warning( + f"Found {len(gpx_tracks)} tracks in the GPX file {self.gpx_path}. Will merge points in all the tracks as a single track for interpolation" + ) + + all_points: list[geo.Point] = sum(gpx_tracks, []) + gpx_points = [p for p in all_points if _GPS_EPOCH <= p.time] + if len(gpx_points) < len(all_points): + LOG.warning( + "Ignoring %d of the %d points in the GPX file %s, which are timestamped before 1980", + len(all_points) - len(gpx_points), + len(all_points), + self.gpx_path, + ) + + # The tracks need not be in time order, and interpolation needs the + # merged points to be + gpx_points.sort(key=lambda p: p.time) + return gpx_points + def _place_without_video_gps( self, gpx_points: T.Sequence[geo.Point] ) -> VideoClock | None: """ - Place the GPX track on the timeline of a video that has no GPS to sync - it with: by the start time of the video, when sampling was given one, - and otherwise by assuming that the track starts with the video. + Place the GPX track on the timeline of a video that has no GPS clock to + sync it with: by the start time of the video, when sampling was given + one, and otherwise by assuming that the track starts with the video. It + runs at --video_duration_ratio GPS seconds per second of video. Return the clock of the video the track was placed by: the start time, or None if the track was assumed to start with the video. @@ -213,17 +245,34 @@ def _place_without_video_gps( if not gpx_points: return None + ratio = self._duration_ratio() + if self.sync_mode is SyncMode.SYNC and self.video_sync is not None: if self.video_sync.start_time is not None: self._rebase_times( - gpx_points, offset=gpx_points[0].time - self.video_sync.start_time + gpx_points, + offset=(gpx_points[0].time - self.video_sync.start_time) / ratio, + rate=ratio, ) - return VideoClock(video_time=0.0, unix_time=self.video_sync.start_time) - 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 +302,12 @@ def _check_coverage( gpx_unix_times: tuple[float, float] | None, video_gps_points: T.Sequence[geo.Point], clock: VideoClock | None, + rate: float, ) -> None: """ Check the GPX track, as placed on the timeline of the video, against - the video, from video time 0 to its mvhd duration. + the video, from video time 0 to its mvhd duration. Each second of video + spans **rate** GPS seconds. A track that misses the video has no position for any moment of it, so it raises when the clock it was placed by says so. A track that covers @@ -278,8 +329,6 @@ def _check_coverage( if video_end <= 0: return - # GPS seconds per video second - rate = 1.0 if clock is None else clock.rate gpx_first = min(p.time for p in gpx_points) gpx_last = max(p.time for p in gpx_points) @@ -319,33 +368,33 @@ def _check_coverage( LOG.warning(message) - def _check_assumed_start(self, gpx_points: T.Sequence[geo.Point]) -> None: + def _check_assumed_start( + self, gpx_points: T.Sequence[geo.Point], ratio: float + ) -> None: """ Check that the GPX track can have started with the video, as it is - assumed to when nothing says when the video starts. + assumed to when nothing says when the video starts. The video spans + **ratio** GPS seconds per second. A track that runs on for longer than the whole video was not recorded for the video alone, and the video could start anywhere along it: raise. One that runs on for less may have been started a little before the video, which puts every frame up to that much too early along it: warn. """ - video_duration = _video_duration(self.video_path) - if video_duration is None: - LOG.debug( - "Unable to read the duration of %s to check the GPX track against", - self.video_path, - ) + durations = self._durations(gpx_points, ratio) + if durations is None: return - gpx_duration = max(p.time for p in gpx_points) - min(p.time for p in gpx_points) - excess = gpx_duration - video_duration + video_span, gpx_duration = durations + excess = gpx_duration - video_span - if video_duration < excess: + if video_span < excess: raise exceptions.MapillaryGPXVideoStartUnknownError( f"The GPX track in {self.gpx_path} lasts {_format_duration(gpx_duration)} " - f"and the video only {_format_duration(video_duration)}, so the video " + f"and {self._describe_video_span(video_span, ratio)}, so the video " "could start anywhere along the track. Specify --video_start_time, or " - "--interpolation_use_gpx_start_time if the track was started with the video" + "--interpolation_use_gpx_start_time if the track was started with the " + "video, and if the video is a timelapse, --video_duration_ratio as well" ) if _LOGGER_TOLERANCE_SECONDS < excess: @@ -360,6 +409,82 @@ def _check_assumed_start(self, gpx_points: T.Sequence[geo.Point]) -> None: _format_duration(excess), ) + def _check_given_start( + self, gpx_points: T.Sequence[geo.Point], ratio: float + ) -> None: + """ + Check a GPX track that --interpolation_use_gpx_start_time says started + with the video, which spans **ratio** GPS seconds per second. A track + that runs on for longer than the whole video was not recorded for it + alone, or belongs to a timelapse, whose rate nothing gives: raise. + """ + durations = self._durations(gpx_points, ratio) + if durations is None: + return + + video_span, gpx_duration = durations + if gpx_duration - video_span <= video_span: + return + + message = ( + f"The GPX track in {self.gpx_path} lasts {_format_duration(gpx_duration)} " + f"and {self._describe_video_span(video_span, ratio)}, so it runs on long " + "after the video ends. " + ) + if gpx_duration < _MAX_RECORDING_SECONDS: + # The start of the track is the start of the video only if the + # track is that of the one recording + message += ( + "If the video is a timelapse, specify --video_duration_ratio. If the " + "GPS logger was started with the video and left running, specify " + f"--video_start_time {build_capture_time(min(p.time for p in gpx_points))}, " + "when the track starts, instead of --interpolation_use_gpx_start_time. " + "If the track covers more than this video, specify the time the video " + "started as --video_start_time instead" + ) + else: + message += ( + "A track that long covers more than this video: specify the time the " + "video started as --video_start_time instead of " + "--interpolation_use_gpx_start_time" + ) + raise exceptions.MapillaryGPXVideoStartUnknownError(message) + + def _describe_video_span(self, video_span: float, ratio: float) -> str: + """ + >>> extractor = GPXVideoExtractor(Path("a.mp4"), Path("a.gpx")) + >>> extractor._describe_video_span(5, 1) + 'the video only 5 seconds' + >>> extractor._describe_video_span(120, 60) + 'the video only 2 seconds, which at --video_duration_ratio 60 span 2.0 minutes' + """ + description = f"the video only {_format_duration(video_span / ratio)}" + if ratio != 1.0: + description += ( + f", which at --video_duration_ratio {ratio:g} span " + f"{_format_duration(video_span)}" + ) + return description + + def _durations( + self, gpx_points: T.Sequence[geo.Point], ratio: float + ) -> tuple[float, float] | None: + """ + The GPS seconds that the video spans at **ratio** GPS seconds per + second, and the duration of the GPX track, or None if the duration of + the video cannot be read + """ + video_duration = _video_duration(self.video_path) + if video_duration is None: + LOG.debug( + "Unable to read the duration of %s to check the GPX track against", + self.video_path, + ) + return None + + gpx_duration = max(p.time for p in gpx_points) - min(p.time for p in gpx_points) + return video_duration * ratio, gpx_duration + def _extract_native_video_metadata(self) -> tuple[types.VideoMetadata, bool]: """ Extract the video metadata, and whether its GPS was rejected as noise @@ -376,14 +501,17 @@ def _extract_native_video_metadata(self) -> tuple[types.VideoMetadata, bool]: return metadata, True @classmethod - def _rebase_times(cls, points: T.Sequence[geo.Point], offset: float = 0.0) -> None: + def _rebase_times( + cls, points: T.Sequence[geo.Point], offset: float = 0.0, rate: float = 1.0 + ) -> None: """ - Rebase point times to start from **offset** + Rebase point times to start from **offset**, at **rate** of their + seconds per second """ if points: first_timestamp = points[0].time for p in points: - p.time = (p.time - first_timestamp) + offset + p.time = (p.time - first_timestamp) / rate + offset @classmethod def _read_video_clock( diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py index e945d252..1de92705 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,38 +584,105 @@ 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 _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) + 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, + ffmpeglib.Probe(ffmpeg.probe_format_and_streams(video_path)), + telemetry_points, + ) if start_time is None: raise exceptions.MapillaryVideoError( f"Unable to extract video start time from {video_path}" @@ -586,7 +696,11 @@ def _sample_single_video_by_interval( if sample_paths[0] is None: continue # extract_frames() produces 1-based frame indices so we need to subtract 1 here - seconds = (frame_idx_1based - 1) * sample_interval * 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) diff --git a/tests/integration/test_video_process.py b/tests/integration/test_video_process.py index bdd26788..cad2b6b4 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 00000000..9664ed46 --- /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_video_coverage.py b/tests/unit/test_gpx_video_coverage.py index 73a6581a..7ad03592 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 619e13e6..9ab9a586 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 cf4760b7..884be856 100644 --- a/tests/unit/test_sample_video.py +++ b/tests/unit/test_sample_video.py @@ -74,12 +74,16 @@ def setup_mock(monkeypatch): monkeypatch.setattr(ffmpeglib, "FFMPEG", MOCK_FFMPEG) -def _validate_interval(samples: T.Sequence[Path], video_start_time): +def _validate_interval( + samples: T.Sequence[Path], video_start_time, seconds_per_sample: float = 2 +): assert len(samples), "expect samples but got none" for idx, sample in enumerate(sorted(samples)): assert sample.name == f"hello_v_{idx + 1:06d}.jpg" exif = exif_read.ExifRead(sample) - expected_dt = video_start_time + datetime.timedelta(seconds=2 * idx) + expected_dt = video_start_time + datetime.timedelta( + seconds=seconds_per_sample * idx + ) assert exif.extract_capture_time() == expected_dt @@ -208,10 +212,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 +1028,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."""