diff --git a/mapillary_tools/exceptions.py b/mapillary_tools/exceptions.py index 5c1f112c..ecf2ad40 100644 --- a/mapillary_tools/exceptions.py +++ b/mapillary_tools/exceptions.py @@ -89,6 +89,13 @@ class MapillaryStationaryVideoError(MapillaryDescriptionError): pass +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. + """ + + class MapillaryOutsideGPXTrackError(MapillaryDescriptionError): def __init__( self, message: str, image_time: str, gpx_start_time: str, gpx_end_time: str diff --git a/mapillary_tools/geotag/factory.py b/mapillary_tools/geotag/factory.py index 95b8ccbd..c25d3de9 100644 --- a/mapillary_tools/geotag/factory.py +++ b/mapillary_tools/geotag/factory.py @@ -22,11 +22,19 @@ geotag_videos_from_gpx, geotag_videos_from_video, ) -from .options import InterpolationOption, SOURCE_TYPE_ALIAS, SourceOption, SourceType +from .options import ( + InterpolationOption, + SOURCE_TYPE_ALIAS, + SourceOption, + SourcePathOption, + SourceType, +) LOG = logging.getLogger(__name__) +_ALL_VIDEO_TYPES = {types.FileType.VIDEO, *types.NATIVE_VIDEO_FILETYPES} + # Sources that read a GPS track from outside the video file, as opposed to # re-reading the telemetry embedded in it. (NMEA cannot geotag videos yet: it # skips them, and they keep the error they had) @@ -63,6 +71,73 @@ def parse_source_option(source: str) -> list[SourceOption]: return [SourceOption(SourceType(SOURCE_TYPE_ALIAS.get(s, s))) for s in sources] +def parse_source_options( + geotag_source: T.Sequence[str], + video_geotag_source: T.Sequence[str], + geotag_source_path: Path | None, +) -> list[SourceOption]: + """ + Turn the raw --geotag_source / --video_geotag_source / --geotag_source_path + arguments into the option list that process() consumes. + """ + parsed_options: list[SourceOption] = [] + + if video_geotag_source and geotag_source: + LOG.warning( + "Video source options will be processed BEFORE the generic source options" + ) + + for s in video_geotag_source: + for video_option in parse_source_option(s): + video_option.filetypes = types.combine_filetype_filters( + video_option.filetypes, {types.FileType.VIDEO} + ) + parsed_options.append(video_option) + + for s in geotag_source: + parsed_options.extend(parse_source_option(s)) + + if geotag_source_path is not None: + for parsed_option in parsed_options: + if parsed_option.source_path is None: + parsed_option.source_path = SourcePathOption( + source_path=Path(geotag_source_path) + ) + else: + source_path_option = parsed_option.source_path + if source_path_option.source_path is None: + source_path_option.source_path = Path(geotag_source_path) + else: + LOG.warning( + "The option --geotag_source_path is ignored for source %s", + parsed_option, + ) + + return parsed_options + + +def can_geotag_videos(option: SourceOption) -> bool: + """ + Whether process() can read positions for videos from the source: it has a + video extractor, and its filetypes let videos through. process() returns no + metadata at all for a video that no source can read. + + Examples: + >>> can_geotag_videos(SourceOption(SourceType.GPX)) + True + >>> can_geotag_videos(SourceOption(SourceType.EXIF)) + False + >>> can_geotag_videos( + ... SourceOption(SourceType.NATIVE, filetypes={types.FileType.IMAGE}) + ... ) + False + """ + if option.filetypes is not None and not option.filetypes & _ALL_VIDEO_TYPES: + return False + + return _build_video_geotag(option) is not None + + def process( # Collection: ABC for sized iterable container classes paths: T.Iterable[Path], @@ -179,14 +254,12 @@ def _filter_images_and_videos( image_paths = [] video_paths = [] - ALL_VIDEO_TYPES = {types.FileType.VIDEO, *types.NATIVE_VIDEO_FILETYPES} - if filetypes is None: include_images = True include_videos = True else: include_images = types.FileType.IMAGE in filetypes - include_videos = bool(filetypes & ALL_VIDEO_TYPES) + include_videos = bool(filetypes & _ALL_VIDEO_TYPES) for path in paths: if utils.is_image_file(path): @@ -332,7 +405,9 @@ def _build_video_geotag(option: SourceOption) -> base.GeotagVideosFromGeneric | elif option.source is SourceType.GPX: return geotag_videos_from_gpx.GeotagVideosFromGPX( - source_path=option.source_path, num_processes=option.num_processes + source_path=option.source_path, + num_processes=option.num_processes, + video_sync=option.video_sync, ) elif option.source is SourceType.NMEA: diff --git a/mapillary_tools/geotag/geotag_videos_from_gpx.py b/mapillary_tools/geotag/geotag_videos_from_gpx.py index 563af205..05aeecc9 100644 --- a/mapillary_tools/geotag/geotag_videos_from_gpx.py +++ b/mapillary_tools/geotag/geotag_videos_from_gpx.py @@ -29,11 +29,13 @@ def __init__( self, source_path: options.SourcePathOption | None = None, num_processes: int | None = None, + video_sync: options.VideoSyncOption | None = None, ): super().__init__(num_processes=num_processes) if source_path is None: source_path = options.SourcePathOption(pattern="%g.gpx") self.source_path = source_path + self.video_sync = video_sync @override def _generate_video_extractors( @@ -43,7 +45,11 @@ def _generate_video_extractors( for video_path in video_paths: source_path = self.source_path.resolve(video_path) if source_path.is_file(): - results.append(GPXVideoExtractor(video_path, source_path)) + results.append( + GPXVideoExtractor( + video_path, source_path, video_sync=self.video_sync + ) + ) else: results.append( types.describe_error_metadata( diff --git a/mapillary_tools/geotag/options.py b/mapillary_tools/geotag/options.py index 91e169b9..72f635f3 100644 --- a/mapillary_tools/geotag/options.py +++ b/mapillary_tools/geotag/options.py @@ -51,6 +51,9 @@ class SourceOption: interpolation: InterpolationOption | None = None + # Set by sample_video only, and not in the JSON schema + video_sync: VideoSyncOption | None = None + @classmethod def from_dict(cls, data: dict[str, T.Any]) -> SourceOption: validate_option(data) @@ -143,6 +146,27 @@ class InterpolationOption: use_gpx_start_time: bool = False +@dataclasses.dataclass +class VideoSyncOption: + """ + How to line up a GPX track with a video that is being sampled, so that its + frames are picked where the geotagging stage of video_process will place + them. process geotags videos without it, and places a GPX as it always has. + """ + + # Unix time of the first frame of the video (--video_start_time), to place + # a GPX by when the video has no GPS of its own to sync it with + start_time: float | None = None + + # --interpolation_offset_time: seconds the geotagging stage adds to the + # capture time of a frame to find its position along the GPX + offset_time: float = 0.0 + + # --interpolation_use_gpx_start_time: the geotagging stage puts the first + # frame at the start of the GPX + use_gpx_start_time: bool = False + + SourceOptionSchema = { "type": "object", "properties": { diff --git a/mapillary_tools/geotag/video_extractors/gpx.py b/mapillary_tools/geotag/video_extractors/gpx.py index 560ab784..8fcc7fb4 100644 --- a/mapillary_tools/geotag/video_extractors/gpx.py +++ b/mapillary_tools/geotag/video_extractors/gpx.py @@ -6,6 +6,7 @@ from __future__ import annotations import dataclasses +import datetime import enum import logging import statistics @@ -13,14 +14,18 @@ import typing as T from pathlib import Path +import construct as C import gpxpy +import humanize if sys.version_info >= (3, 12): from typing import override else: from typing_extensions import override -from ... import exceptions, geo, types, utils +from ... import exceptions, geo, telemetry, types, utils +from ...mp4 import construct_mp4_parser as cparser, simple_mp4_parser as sparser +from .. import options from ..utils import parse_gpx from .base import BaseVideoExtractor from .native import NativeVideoExtractor @@ -38,6 +43,16 @@ # timestamped once per block the estimate can be 20% off _MIN_TIMELAPSE_RATE = 1.5 +# No GPS fix can be timestamped before the GPS epoch, 1980-01-06. GPX files +# converted from camera logs can carry earlier times, such as +# 1969-12-31T15:59:59Z, on points that had no valid time +_GPS_EPOCH = datetime.datetime(1980, 1, 6, tzinfo=datetime.timezone.utc).timestamp() + +# How much longer than a video a GPX track may last without a warning, when it +# is assumed to start with the video. A logger that records once a second, +# started and stopped with the camera, can run on for a second at each end. +_ASSUMED_START_TOLERANCE_SECONDS = 2.0 + class SyncMode(enum.Enum): # Sync by video GPS timestamps if found, otherwise rebase @@ -77,11 +92,18 @@ def to_unix_time(self, video_time: float) -> float: class GPXVideoExtractor(BaseVideoExtractor): def __init__( - self, video_path: Path, gpx_path: Path, sync_mode: SyncMode = SyncMode.SYNC + self, + video_path: Path, + gpx_path: Path, + sync_mode: SyncMode = SyncMode.SYNC, + video_sync: options.VideoSyncOption | None = None, ): self.video_path = video_path self.gpx_path = gpx_path self.sync_mode = sync_mode + self.video_sync = video_sync + if video_sync is not None and video_sync.use_gpx_start_time: + self.sync_mode = SyncMode.REBASE @override def extract(self) -> types.VideoMetadata: @@ -98,7 +120,19 @@ def extract(self) -> types.VideoMetadata: 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" ) - gpx_points: T.Sequence[geo.Point] = sum(gpx_tracks, []) + 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) try: native_video_metadata, noisy = self._extract_native_video_metadata() @@ -107,7 +141,8 @@ def extract(self) -> types.VideoMetadata: # empty or too noisy must not reject the video if self.sync_mode is SyncMode.STRICT_SYNC: raise ex - self._rebase_times(gpx_points) + self._place_without_video_gps(gpx_points) + self._apply_offset_time(gpx_points) return types.VideoMetadata( filename=self.video_path, filesize=utils.get_file_size(self.video_path), @@ -146,7 +181,8 @@ def extract(self) -> types.VideoMetadata: ) if clock is None: - self._rebase_times(gpx_points) + # Its GPS has no clock to sync with, as if it had no GPS + self._place_without_video_gps(gpx_points) else: if clock.rate != 1.0: LOG.info( @@ -158,8 +194,93 @@ 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) + return dataclasses.replace(native_video_metadata, points=gpx_points) + def _place_without_video_gps(self, gpx_points: T.Sequence[geo.Point]) -> 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. + """ + if not gpx_points: + return + + if self.sync_mode is SyncMode.SYNC and self.video_sync is not None: + if self.video_sync.start_time is not None: + self._rebase_times( + gpx_points, offset=gpx_points[0].time - self.video_sync.start_time + ) + return + self._check_assumed_start(gpx_points) + + self._rebase_times(gpx_points) + + def _apply_offset_time( + self, gpx_points: T.Sequence[geo.Point], rate: float = 1.0 + ) -> None: + """ + Shift the track back by --interpolation_offset_time, which the + geotagging stage adds to the capture time of a frame to find its + position along the GPX. The offset is in GPS seconds, of which a + timelapse packs **rate** into each second of video. + + The absolute times shift with the track, so that a frame is stamped + with its capture time, and the geotagging stage, adding the offset back, + finds the position the frame was picked by. + """ + if self.video_sync is None or not self.video_sync.offset_time: + return + + offset_time = self.video_sync.offset_time + for p in gpx_points: + p.time -= offset_time / rate + # parse_gpx() returns CAMMGPSPoint, with epoch_time 0 for no time + if isinstance(p, telemetry.CAMMGPSPoint) and 0 < p.epoch_time: + p.epoch_time -= offset_time + + def _check_assumed_start(self, gpx_points: T.Sequence[geo.Point]) -> None: + """ + Check that the GPX track can have started with the video, as it is + assumed to when nothing says when the video starts. + + 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, + ) + return + + gpx_duration = max(p.time for p in gpx_points) - min(p.time for p in gpx_points) + excess = gpx_duration - video_duration + + if video_duration < 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 " + "could start anywhere along the track. Specify --video_start_time, or " + "--interpolation_use_gpx_start_time if the track was started with the video" + ) + + if _ASSUMED_START_TOLERANCE_SECONDS < excess: + LOG.warning( + "Assuming that the GPX track in %s starts with the video %s, although " + "the track lasts %s longer. If it was started first, the frames are " + "placed up to that much too early along it. Specify --video_start_time " + "to place them exactly, or --interpolation_use_gpx_start_time if the " + "track was started with the video", + self.gpx_path, + self.video_path, + _format_duration(excess), + ) + def _extract_native_video_metadata(self) -> tuple[types.VideoMetadata, bool]: """ Extract the video metadata, and whether its GPS was rejected as noise @@ -277,3 +398,35 @@ def _median_step_rate(cls, pairs: T.Sequence[tuple[float, float]]) -> float | No if v1 < v2 ] return statistics.median(steps) if steps else None + + +def _video_duration(video_path: Path) -> float | None: + """ + The duration of the video in seconds, from its mvhd box, or None if it + cannot be read. + """ + try: + with video_path.open("rb") as fp: + data = sparser.parse_box_data_first(fp, [b"moov", b"mvhd"]) + if data is None: + return None + mvhd = cparser.MovieHeaderBox.parse(data) + except (OSError, sparser.ParsingError, C.ConstructError) as ex: + LOG.debug("Failed to read the duration of %s: %s", video_path, ex) + return None + + # All 1s means the duration is unknown + if not mvhd.timescale or mvhd.duration in (0, 0xFFFFFFFF, 0xFFFFFFFFFFFFFFFF): + return None + + return mvhd.duration / mvhd.timescale + + +def _format_duration(seconds: float) -> str: + """ + >>> _format_duration(5) + '5 seconds' + >>> _format_duration(43474) + '12 hours' + """ + return humanize.naturaldelta(datetime.timedelta(seconds=seconds)) diff --git a/mapillary_tools/process_geotag_properties.py b/mapillary_tools/process_geotag_properties.py index 41d8e8a2..af49f68b 100644 --- a/mapillary_tools/process_geotag_properties.py +++ b/mapillary_tools/process_geotag_properties.py @@ -14,13 +14,8 @@ from tqdm import tqdm from . import constants, exceptions, exif_write, types, utils -from .geotag.factory import parse_source_option, process -from .geotag.options import ( - InterpolationOption, - SourceOption, - SourcePathOption, - SourceType, -) +from .geotag.factory import parse_source_options, process +from .geotag.options import InterpolationOption, SourceType from .serializer.description import ( DescriptionJSONSerializer, validate_and_fail_metadata, @@ -44,47 +39,6 @@ def _normalize_import_paths(import_path: Path | T.Sequence[Path]) -> T.Sequence[ return import_paths -def _parse_source_options( - geotag_source: list[str], - video_geotag_source: list[str], - geotag_source_path: Path | None, -) -> list[SourceOption]: - parsed_options: list[SourceOption] = [] - - if video_geotag_source and geotag_source: - LOG.warning( - "Video source options will be processed BEFORE the generic source options" - ) - - for s in video_geotag_source: - for video_option in parse_source_option(s): - video_option.filetypes = types.combine_filetype_filters( - video_option.filetypes, {types.FileType.VIDEO} - ) - parsed_options.append(video_option) - - for s in geotag_source: - parsed_options.extend(parse_source_option(s)) - - if geotag_source_path is not None: - for parsed_option in parsed_options: - if parsed_option.source_path is None: - parsed_option.source_path = SourcePathOption( - source_path=Path(geotag_source_path) - ) - else: - source_path_option = parsed_option.source_path - if source_path_option.source_path is None: - source_path_option.source_path = Path(geotag_source_path) - else: - LOG.warning( - "The option --geotag_source_path is ignored for source %s", - parsed_option, - ) - - return parsed_options - - def process_geotag_properties( import_path: Path | T.Sequence[Path], filetypes: set[types.FileType] | None, @@ -115,7 +69,7 @@ def process_geotag_properties( if not geotag_source and not video_geotag_source: geotag_source = [*DEFAULT_GEOTAG_SOURCE_OPTIONS] - options = _parse_source_options( + options = parse_source_options( geotag_source=geotag_source or [], video_geotag_source=video_geotag_source or [], geotag_source_path=geotag_source_path, diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py index b0d32652..ba4d11e3 100644 --- a/mapillary_tools/sample_video.py +++ b/mapillary_tools/sample_video.py @@ -14,9 +14,12 @@ from contextlib import contextmanager from pathlib import Path +import humanize + from . import constants, exceptions, ffmpeg as ffmpeglib, geo, types, utils from .exif_write import ExifEdit -from .geotag import geotag_videos_from_video +from .geotag import factory +from .geotag.options import SourceOption, SourceType, VideoSyncOption from .mp4 import mp4_sample_parser from .serializer.description import parse_capture_time @@ -50,6 +53,73 @@ def xor(a: bool, b: bool): return bool(a) ^ bool(b) +def _parse_geotag_options( + geotag_source: list[str] | None, + geotag_source_path: Path | None, + video_geotag_source: list[str] | None, + video_import_path: Path, + video_sync: VideoSyncOption | None = None, +) -> list[SourceOption]: + """ + Resolve which sources distance sampling may take its positions from. + + Defaults to the video's own telemetry rather than to + DEFAULT_GEOTAG_SOURCE_OPTIONS: the default chain continues on to + exiftool_runtime, which reads the same telemetry through a parser that + cannot see every field the noise filter rejects on, so routing sampling + through it would accept tracks the native parser refuses. + + Every option carries video_sync, for a GPX source to place its track where + the geotagging stage will place the frames. + """ + options = _parse_video_sources( + geotag_source, geotag_source_path, video_geotag_source, video_import_path + ) + + for option in options: + option.video_sync = video_sync + + return options + + +def _parse_video_sources( + geotag_source: list[str] | None, + geotag_source_path: Path | None, + video_geotag_source: list[str] | None, + video_import_path: Path, +) -> list[SourceOption]: + if not geotag_source and not video_geotag_source: + return [SourceOption(SourceType.NATIVE)] + + # Without an explicit path, every source gets the video import path, as + # process_geotag_properties() gives it when geotagging the frames: that is + # how exiftool_xml finds the XML files in the video directory. It does not + # find a GPX beside the video, which the geotagging stage could not find + # either; a GPX needs --geotag_source_path + if geotag_source_path is None: + geotag_source_path = video_import_path + + options = factory.parse_source_options( + geotag_source=geotag_source or [], + video_geotag_source=video_geotag_source or [], + geotag_source_path=geotag_source_path, + ) + + # exif, nmea and the legacy gopro/blackvue/camm sources have nothing to read + # a video with; in video_process they are there to geotag the frames. Sample + # by the sources that can read videos, or by the video's own GPS, as before, + # if none of them can + video_options = [option for option in options if factory.can_geotag_videos(option)] + if not video_options: + LOG.info( + "None of the geotag sources %s can read GPS from videos. Sampling by the GPS in the videos", + ",".join(option.source.value for option in options), + ) + return [SourceOption(SourceType.NATIVE)] + + return video_options + + def sample_video( video_import_path: Path, import_path: Path, @@ -61,6 +131,13 @@ def sample_video( video_start_time: str | None = None, skip_sample_errors: bool = False, rerun: bool = False, + # Absent when called from the sample_video command, which does not register + # the process command's arguments + geotag_source: list[str] | None = None, + geotag_source_path: Path | None = None, + video_geotag_source: list[str] | None = None, + interpolation_offset_time: float = 0.0, + interpolation_use_gpx_start_time: bool = False, ) -> None: video_dir, video_list = _normalize_path(video_import_path, skip_subfolders) @@ -76,6 +153,25 @@ def sample_video( except ValueError as ex: raise exceptions.MapillaryBadParameterError(str(ex)) + # Only a start time the user gave. The one probed from the container is + # creation_time - duration, which is a whole video length early for cameras + # whose creation_time marks the start of the recording + video_sync = VideoSyncOption( + start_time=( + video_start_time_dt.timestamp() if video_start_time_dt is not None else None + ), + offset_time=interpolation_offset_time, + use_gpx_start_time=interpolation_use_gpx_start_time, + ) + + geotag_options = _parse_geotag_options( + geotag_source, + geotag_source_path, + video_geotag_source, + video_import_path, + video_sync, + ) + if rerun: for video_path in video_list: # Example: @@ -123,6 +219,7 @@ def sample_video( sample_dir, sample_distance=video_sample_distance, start_time=video_start_time_dt, + geotag_options=geotag_options, ) else: assert 0 < video_sample_interval, ( @@ -328,12 +425,38 @@ def _sample_video_stream_by_distance( } +def _describe_track_time_range(points: T.Sequence[geo.Point]) -> str: + """ + >>> p = lambda t: geo.Point(time=t, lat=0, lon=0, alt=None, angle=None) + >>> _describe_track_time_range([p(-7200), p(-3600)]) + 'which ends an hour before the video starts' + >>> _describe_track_time_range([p(600), p(700)]) + 'which starts 10 minutes after the video starts' + >>> _describe_track_time_range([p(0.01), p(0.02)]) + 'which covers 0.010 to 0.020 seconds of the video' + """ + start = min(p.time for p in points) + end = max(p.time for p in points) + + if end < 0: + return f"which ends {humanize.naturaldelta(-end)} before the video starts" + + if 1 <= start: + return f"which starts {humanize.naturaldelta(start)} after the video starts" + + return f"which covers {start:.3f} to {end:.3f} seconds of the video" + + def _sample_single_video_by_distance( video_path: Path, sample_dir: Path, sample_distance: float, start_time: datetime.datetime | None = None, + geotag_options: T.Sequence[SourceOption] | None = None, ) -> None: + if geotag_options is None: + geotag_options = [SourceOption(SourceType.NATIVE)] + ffmpeg = ffmpeglib.FFMPEG(constants.FFMPEG_PATH, constants.FFPROBE_PATH) probe = ffmpeglib.Probe(ffmpeg.probe_format_and_streams(video_path)) @@ -347,9 +470,13 @@ def _sample_single_video_by_distance( LOG.info("Extracting video metdata") - video_metadatas = geotag_videos_from_video.GeotagVideosFromVideo().to_description( - [video_path] - ) + # Go through the factory rather than reading the video's own telemetry + # directly, so that --geotag_source is honoured here as well: a GPX is the + # documented answer for a camera whose embedded GPS is unusable, and + # distance sampling needs positions just as much as geotagging does. + video_metadatas = factory.process([video_path], geotag_options) + # Every option can read videos (see _parse_geotag_options()), so process() + # returns metadata for the video even if all of them fail assert len(video_metadatas) == 1, "expect 1 video metadata" video_metadata = video_metadatas[0] @@ -361,6 +488,11 @@ def _sample_single_video_by_distance( if isinstance(video_metadata, types.ErrorMetadata): raise video_metadata.error + # Only a video path was passed in, so only video metadata can come back + assert isinstance(video_metadata, types.VideoMetadata), ( + f"expect VideoMetadata but got {type(video_metadata).__name__}" + ) + if not video_metadata.points: raise exceptions.MapillaryVideoGPSNotFoundError("No GPS points found") @@ -380,10 +512,13 @@ def _sample_single_video_by_distance( sample_points_by_frame_idx = _sample_video_stream_by_distance( video_metadata.points, video_track_parser, sample_distance ) + # A track that misses the video, such as a GPX from another trip or one + # placed by a wrong --video_start_time, selects no frames, and an empty + # sample directory would pass for success if not sample_points_by_frame_idx: - # A track that covers none of the video, e.g. a GPX from another trip raise exceptions.MapillaryVideoError( - f"No video frames found within the GPS track from {video_metadata.points[0].time:.3f}s to {video_metadata.points[-1].time:.3f}s" + "none of its frames falls within its GPS track, " + + _describe_track_time_range(video_metadata.points) ) sorted_sample_indices = sorted(sample_points_by_frame_idx.keys()) diff --git a/tests/integration/test_video_process.py b/tests/integration/test_video_process.py index 055ad9be..076724d1 100644 --- a/tests/integration/test_video_process.py +++ b/tests/integration/test_video_process.py @@ -5,6 +5,7 @@ import functools import json +import os import subprocess import tempfile from pathlib import Path @@ -31,6 +32,11 @@ run_sample_video = functools.partial(run_command, command="sample_video") +# Not run_process_for_descs() for a run expected to fail: on Windows it would +# remove its description file while still holding it open, and the +# PermissionError would replace the CalledProcessError +run_video_process = functools.partial(run_command, command="video_process") + def test_sample_video_relpath(): pytest_skip_if_not_ffmpeg_installed() @@ -92,6 +98,259 @@ def test_sample_video_by_distance_without_usable_gps(setup_data: py.path.local): ) +def test_video_process_by_distance_with_an_image_only_geotag_source( + setup_data: py.path.local, +): + """ + exif has nothing to read a video with; it geotags the frames. The video is + sampled by its own GPS, as it was before --geotag_source reached sampling, + instead of failing an assert. + """ + pytest_skip_if_not_ffmpeg_installed() + + video_path = setup_data.join("gopro_data").join("max-360mode.mp4") + + with tempfile.TemporaryDirectory() as dir: + descs = run_video_process_for_descs( + [ + "--video_sample_distance=6", + "--geotag_source=exif", + str(video_path), + dir, + ] + ) + + assert descs + assert all("error" not in desc for desc in descs), descs + + +def _with_creation_time(video: py.path.local, creation_time: str, dir: str) -> Path: + """A copy of the video whose container records the creation time.""" + copy = Path(dir, video.basename) + subprocess.run( + [ + # as pytest_skip_if_not_ffmpeg_installed() finds it + os.getenv("MAPILLARY_TOOLS_FFMPEG_PATH", "ffmpeg"), + *["-loglevel", "error", "-y", "-i", str(video)], + *["-c", "copy", "-metadata", f"creation_time={creation_time}"], + str(copy), + ], + check=True, + ) + return copy + + +def test_video_process_by_distance_places_a_gpx_by_the_video_start_time( + setup_data: py.path.local, +): + """ + The video has no GPS of its own. The GPX was placed as if it started with + the video, so the frames were picked from, and stamped with, its first + seconds, whatever --video_start_time said. + """ + pytest_skip_if_not_ffmpeg_installed() + + video_path = setup_data.join("videos").join("sample-5s.mp4") + gpx_file = setup_data.join("gpx").join("sf_30km_h.gpx") + + with tempfile.TemporaryDirectory() as dir: + descs = run_video_process_for_descs( + [ + "--video_sample_distance=3", + *["--video_start_time", "2025_03_14_07_00_30_000"], + "--geotag_source=gpx", + *["--geotag_source_path", str(gpx_file)], + str(video_path), + dir, + ] + ) + + assert descs + assert all("error" not in desc for desc in descs), descs + capture_times = sorted(desc["MAPCaptureTime"] for desc in descs) + assert "2025_03_14_07_00_30_000" <= capture_times[0], capture_times + assert capture_times[-1] < "2025_03_14_07_00_36_000", capture_times + + +def test_video_process_by_distance_refuses_to_guess_where_the_video_starts( + setup_data: py.path.local, +): + """ + Without --video_start_time, a GPX far longer than the video could start + anywhere along it, so assuming that it starts with the video is a guess. + """ + pytest_skip_if_not_ffmpeg_installed() + + gpx_file = setup_data.join("gpx").join("sf_30km_h.gpx") + + with tempfile.TemporaryDirectory() as dir: + video_path = _with_creation_time( + setup_data.join("videos").join("sample-5s.mp4"), "2025-03-14T07:00:36Z", dir + ) + with pytest.raises(subprocess.CalledProcessError) as ex: + run_video_process( + [ + "--video_sample_distance=3", + "--geotag_source=gpx", + *["--geotag_source_path", str(gpx_file)], + str(video_path), + str(Path(dir, "samples")), + ], + capture_output=True, + text=True, + ) + + assert 7 == ex.value.returncode, ex.value.stderr + assert "specify --video_start_time" in ex.value.stderr.lower(), ex.value.stderr + + +def test_video_process_by_distance_with_a_gpx_that_misses_the_video( + setup_data: py.path.local, +): + """No frame falls within the track: the command fails, instead of passing with no frames.""" + pytest_skip_if_not_ffmpeg_installed() + + video_path = setup_data.join("videos").join("sample-5s.mp4") + gpx_file = setup_data.join("gpx").join("sf_30km_h.gpx") + + with tempfile.TemporaryDirectory() as dir: + with pytest.raises(subprocess.CalledProcessError) as ex: + run_video_process( + [ + "--video_sample_distance=3", + *["--video_start_time", "2025_03_14_09_00_00_000"], + "--geotag_source=gpx", + *["--geotag_source_path", str(gpx_file)], + str(video_path), + str(Path(dir, "samples")), + ], + capture_output=True, + text=True, + ) + # sampling creates the sample directory up front, but writes no frames + assert not list(Path(dir, "samples").rglob("*.jpg")) + + assert 7 == ex.value.returncode, ex.value.stderr + assert "none of its frames falls within its GPS track" in ex.value.stderr + + +def test_video_process_by_distance_applies_the_interpolation_offset( + setup_data: py.path.local, +): + """ + The geotagging stage adds --interpolation_offset_time to each frame's time. + Sampling ignored it, so it picked the last frames by positions within the + GPX that the offset then pushed past its end, and they were rejected. + """ + pytest_skip_if_not_ffmpeg_installed() + + video_path = setup_data.join("videos").join("sample-5s.mp4") + # the track ends at 07:01:33.624 + gpx_file = setup_data.join("gpx").join("sf_30km_h.gpx") + + with tempfile.TemporaryDirectory() as dir: + descs = run_video_process_for_descs( + [ + "--video_sample_distance=3", + *["--video_start_time", "2025_03_14_07_01_28_000"], + *["--interpolation_offset_time", "2"], + "--geotag_source=gpx", + *["--geotag_source_path", str(gpx_file)], + str(video_path), + dir, + ] + ) + + assert descs + assert all("error" not in desc for desc in descs), descs + capture_times = sorted(desc["MAPCaptureTime"] for desc in descs) + assert "2025_03_14_07_01_30_000" <= capture_times[0], capture_times + assert capture_times[-1] <= "2025_03_14_07_01_33_624", capture_times + + +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. + """ + 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 + ) + descs = run_video_process_for_descs( + [ + "--video_sample_distance=3", + "--interpolation_use_gpx_start_time", + *["--interpolation_offset_time", "2.5"], + "--geotag_source=gpx", + *["--geotag_source_path", str(gpx_file)], + str(video_path), + str(Path(dir, "samples")), + ] + ) + + assert descs + assert all("error" not in desc for desc in descs), descs + capture_times = sorted(desc["MAPCaptureTime"] for desc in descs) + assert capture_times[0] == "2025_03_14_07_00_02_500", capture_times + assert capture_times[-1] < "2025_03_14_07_00_09_000", capture_times + + +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 + carry tracks timestamped 1969. Merged as they were, the points were out of + time order, and sampling crashed with a traceback. + """ + pytest_skip_if_not_ffmpeg_installed() + + def track(times): + return "".join( + f'' + for i, time in times + ) + + later = [(i, f"2025-03-14T07:00:{s:02d}Z") for i, s in enumerate(range(10, 20), 10)] + earlier = [(i, f"2025-03-14T07:00:{s:02d}Z") for i, s in enumerate(range(0, 10))] + junk = [(0, "1969-12-31T15:59:59Z")] * 3 + + video_path = setup_data.join("videos").join("sample-5s.mp4") + + with tempfile.TemporaryDirectory() as dir: + gpx_file = Path(dir, "untidy.gpx") + gpx_file.write_text( + '\n' + '' + + "".join( + f"{track(t)}" + for t in [later, junk, earlier] + ) + + "\n" + ) + descs = run_video_process_for_descs( + [ + "--video_sample_distance=3", + *["--video_start_time", "2025_03_14_07_00_00_000"], + "--geotag_source=gpx", + *["--geotag_source_path", str(gpx_file)], + str(video_path), + str(Path(dir, "samples")), + ] + ) + + assert descs + assert all("error" not in desc for desc in descs), descs + capture_times = sorted(desc["MAPCaptureTime"] for desc in descs) + assert "2025_03_14_07_00_00_000" <= capture_times[0], capture_times + assert capture_times[-1] < "2025_03_14_07_00_06_000", capture_times + + def test_video_process_processes_the_videos_that_sampled(setup_data: py.path.local): """ hero8.mp4's GPS is all noise, so it cannot be sampled by distance. That diff --git a/tests/unit/test_gpx_video_sync.py b/tests/unit/test_gpx_video_sync.py new file mode 100644 index 00000000..42ebebfa --- /dev/null +++ b/tests/unit/test_gpx_video_sync.py @@ -0,0 +1,351 @@ +# 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. + +""" +How GPXVideoExtractor lines up a GPX track with the video it geotags, when +distance sampling hands it a VideoSyncOption, and that process, which does +not, places a GPX as it always has. +""" + +from __future__ import annotations + +import datetime +import logging +from pathlib import Path + +import pytest +from mapillary_tools import exceptions, 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, VideoClock + +_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") +# GPS of its own, recorded from 2019-11-18T15:44:47Z +GPS_VIDEO = _DATA.joinpath("gopro_data", "max-360mode.mp4") + +GPX_START = datetime.datetime(2025, 3, 14, 7, 0, 0, tzinfo=datetime.timezone.utc) + + +def _write_gpx(path: Path, seconds: int, start=GPX_START) -> Path: + """A GPX track with a point every second, heading north at 10 m/s.""" + points = [] + for i in range(seconds + 1): + time = (start + datetime.timedelta(seconds=i)).strftime("%Y-%m-%dT%H:%M:%SZ") + lat = 37.0 + i * 10 / 111_320 + points.append( + f'10' + ) + path.write_text( + '\n' + '' + f"{''.join(points)}\n" + ) + return path + + +def _extract(video: Path, gpx: Path, video_sync: VideoSyncOption | None): + return GPXVideoExtractor(video, gpx, video_sync=video_sync).extract() + + +class TestVideoWithoutGPS: + def test_the_track_is_placed_by_the_video_start_time(self, tmp_path): + """ + Not by assuming the track starts with the video: a GPX of a whole drive + then put every clip on its first few minutes. + """ + gpx = _write_gpx(tmp_path / "track.gpx", seconds=60) + video_start = GPX_START + datetime.timedelta(seconds=30) + + metadata = _extract( + NO_GPS_VIDEO, gpx, VideoSyncOption(start_time=video_start.timestamp()) + ) + + assert metadata.points[0].time == pytest.approx(-30) + assert metadata.points[30].time == pytest.approx(0) + # the absolute times, which the frames are stamped with, are untouched + assert metadata.points[30].get_unix_time() == pytest.approx( + video_start.timestamp() + ) + + def test_a_track_far_longer_than_the_video_is_refused(self, tmp_path): + """The video could start anywhere along a track a minute long.""" + gpx = _write_gpx(tmp_path / "track.gpx", seconds=60) + + with pytest.raises(exceptions.MapillaryGPXVideoStartUnknownError) as excinfo: + _extract(NO_GPS_VIDEO, gpx, VideoSyncOption()) + + assert "--video_start_time" 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) + + with caplog.at_level(logging.WARNING): + metadata = _extract(NO_GPS_VIDEO, gpx, VideoSyncOption()) + + assert metadata.points[0].time == 0 + assert "--video_start_time" in caplog.text + + def test_a_track_as_long_as_the_video_starts_with_it_silently( + self, tmp_path, caplog + ): + gpx = _write_gpx(tmp_path / "track.gpx", seconds=6) + + with caplog.at_level(logging.WARNING): + metadata = _extract(NO_GPS_VIDEO, gpx, VideoSyncOption()) + + assert metadata.points[0].time == 0 + assert not caplog.text + + def test_process_still_assumes_the_track_starts_with_the_video(self, tmp_path): + """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) + + assert metadata.points[0].time == 0 + + def test_the_refusal_is_not_handed_on_to_the_next_source(self, tmp_path): + """ + The video has no GPS, so a native source after the GPX can only fail, + and its error would hide why the GPX was refused. + """ + gpx = _write_gpx(tmp_path / "track.gpx", seconds=60) + video_sync = VideoSyncOption() + + [metadata] = factory.process( + [NO_GPS_VIDEO], + [ + SourceOption( + SourceType.GPX, + source_path=SourcePathOption(source_path=gpx), + video_sync=video_sync, + ), + SourceOption(SourceType.NATIVE, video_sync=video_sync), + ], + ) + + assert isinstance(metadata, types.ErrorMetadata) + assert isinstance(metadata.error, exceptions.MapillaryGPXVideoStartUnknownError) + + +class TestVideoWithGPS: + def test_the_video_gps_wins_over_the_start_time(self, tmp_path): + """ + Its own GPS clock syncs the track, as before; --video_start_time only + places a GPX on a video that has nothing to sync it with. + """ + gpx = _write_gpx( + tmp_path / "track.gpx", + seconds=30, + start=datetime.datetime( + 2019, 11, 18, 15, 44, 40, tzinfo=datetime.timezone.utc + ), + ) + far_off = GPX_START.timestamp() + + synced = _extract(GPS_VIDEO, gpx, video_sync=None) + with_start_time = _extract(GPS_VIDEO, gpx, VideoSyncOption(start_time=far_off)) + + assert [p.time for p in with_start_time.points] == [ + p.time for p in synced.points + ] + + +class TestInterpolationOptions: + """ + The geotagging stage of video_process applies --interpolation_offset_time + and --interpolation_use_gpx_start_time to the frames, so sampling applies + them too, or it picks frames by positions the frames do not end up at. + """ + + def test_the_offset_shifts_the_track_and_its_times(self, tmp_path): + """ + The frame at video time t is at track time start + t + offset, and is + stamped start + t, which the geotagging stage adds the offset to. + """ + gpx = _write_gpx(tmp_path / "track.gpx", seconds=60) + video_start = GPX_START + datetime.timedelta(seconds=30) + + metadata = _extract( + NO_GPS_VIDEO, + gpx, + VideoSyncOption(start_time=video_start.timestamp(), offset_time=2), + ) + + # the point recorded at start + 2 is where the first frame is + assert metadata.points[32].time == pytest.approx(0) + assert metadata.points[32].get_unix_time() == pytest.approx( + video_start.timestamp() + ) + + def test_the_offset_applies_to_a_track_synced_by_the_video_gps(self, tmp_path): + gpx = _write_gpx( + tmp_path / "track.gpx", + seconds=30, + start=datetime.datetime( + 2019, 11, 18, 15, 44, 40, tzinfo=datetime.timezone.utc + ), + ) + + synced = _extract(GPS_VIDEO, gpx, VideoSyncOption()) + shifted = _extract(GPS_VIDEO, gpx, VideoSyncOption(offset_time=2)) + + assert [p.time for p in shifted.points] == pytest.approx( + [p.time - 2 for p in synced.points] + ) + + 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. + """ + gpx = _write_gpx(tmp_path / "track.gpx", seconds=60) + + metadata = _extract( + NO_GPS_VIDEO, + gpx, + VideoSyncOption( + start_time=(GPX_START + datetime.timedelta(seconds=30)).timestamp(), + use_gpx_start_time=True, + ), + ) + + assert metadata.points[0].time == 0 + + def test_process_applies_neither(self, tmp_path): + """process passes no VideoSyncOption, and ignores both for videos.""" + gpx = _write_gpx(tmp_path / "track.gpx", seconds=6) + + metadata = _extract(NO_GPS_VIDEO, gpx, video_sync=None) + + assert metadata.points[0].time == 0 + assert metadata.points[0].get_unix_time() == GPX_START.timestamp() + + +def _write_gpx_tracks(path: Path, tracks: list[list[str]]) -> Path: + """A GPX file with one track per list of times, heading north at 10 m/s.""" + trks = [] + for i, times in enumerate(tracks): + points = "".join( + f'' + f"10" + for j, time in enumerate(times) + ) + trks.append(f"{points}") + path.write_text( + '\n' + '' + f"{''.join(trks)}\n" + ) + return path + + +class TestUntidyGPX: + """ + GPX files converted from camera logs can list their tracks out of order, + and carry whole tracks timestamped 1969-12-31T15:59:59Z. Merged as they + were, the points were out of time order, and sampling crashed in the + interpolator. + """ + + JUNK = "1969-12-31T15:59:59Z" + LATER = ["2025-03-14T07:00:10Z", "2025-03-14T07:00:11Z"] + EARLIER = ["2025-03-14T07:00:00Z", "2025-03-14T07:00:01Z"] + + def _gpx(self, tmp_path): + return _write_gpx_tracks( + tmp_path / "track.gpx", [self.LATER, [self.JUNK] * 3, self.EARLIER] + ) + + @pytest.mark.parametrize( + "video_sync", + [ + pytest.param(None, id="process"), + pytest.param( + VideoSyncOption(start_time=GPX_START.timestamp()), id="sampling" + ), + ], + ) + def test_points_before_1980_are_dropped_and_the_rest_sorted( + self, tmp_path, video_sync + ): + metadata = _extract(NO_GPS_VIDEO, self._gpx(tmp_path), video_sync) + + unix_times = [p.get_unix_time() for p in metadata.points] + assert unix_times == sorted(unix_times) + assert len(unix_times) == 4 + assert unix_times[0] == GPX_START.timestamp() + + def test_use_gpx_start_time_starts_at_the_earliest_point(self, tmp_path): + """Not at the first point in the file, nor at a junk one.""" + metadata = _extract( + NO_GPS_VIDEO, self._gpx(tmp_path), VideoSyncOption(use_gpx_start_time=True) + ) + + assert metadata.points[0].time == 0 + assert metadata.points[0].get_unix_time() == GPX_START.timestamp() + + +class TestVideoGPSWithoutAClock: + """ + A video whose GPS has no usable clock, such as GPS rejected as noise with a + placeholder date, has nothing to sync a GPX with, as if it had no GPS. + """ + + def test_the_track_is_placed_by_the_video_start_time(self, tmp_path, monkeypatch): + monkeypatch.setattr( + GPXVideoExtractor, + "_read_video_clock", + classmethod(lambda cls, points: None), + ) + gpx = _write_gpx(tmp_path / "track.gpx", seconds=60) + video_start = GPX_START + datetime.timedelta(seconds=30) + + metadata = _extract( + GPS_VIDEO, gpx, VideoSyncOption(start_time=video_start.timestamp()) + ) + + assert metadata.points[30].time == pytest.approx(0) + + +class TestTimelapseOffset: + def test_the_offset_is_in_gps_seconds(self, tmp_path, monkeypatch): + """ + A timelapse packs rate GPS seconds into each video second, so an offset + of 2 GPS seconds moves the track by 2 / rate seconds of video. + """ + start = datetime.datetime( + 2019, 11, 18, 15, 44, 40, tzinfo=datetime.timezone.utc + ) + monkeypatch.setattr( + GPXVideoExtractor, + "_read_video_clock", + classmethod( + lambda cls, points: VideoClock( + video_time=0.0, unix_time=start.timestamp(), rate=10.0 + ) + ), + ) + gpx = _write_gpx(tmp_path / "track.gpx", seconds=30, start=start) + + synced = _extract(GPS_VIDEO, gpx, VideoSyncOption()) + shifted = _extract(GPS_VIDEO, gpx, VideoSyncOption(offset_time=2)) + + assert [p.time for p in shifted.points] == pytest.approx( + [p.time - 0.2 for p in synced.points] + ) + # and the frames are stamped 2 GPS seconds back, for the geotagging + # stage to add them again + assert [p.get_unix_time() for p in shifted.points] == pytest.approx( + [p.get_unix_time() - 2 for p in synced.points] + ) diff --git a/tests/unit/test_sample_video.py b/tests/unit/test_sample_video.py index 9f1b9299..464476ca 100644 --- a/tests/unit/test_sample_video.py +++ b/tests/unit/test_sample_video.py @@ -23,7 +23,7 @@ geo, sample_video, ) -from mapillary_tools.geotag import geotag_videos_from_video +from mapillary_tools.geotag.options import SourceType, VideoSyncOption from mapillary_tools.mp4 import mp4_sample_parser from mapillary_tools.serializer import description from mapillary_tools.types import describe_error_metadata, FileType, VideoMetadata @@ -357,9 +357,9 @@ def unreadable_gps(self, monkeypatch): """Every GPS read fails the way a fully filtered noisy track does.""" error = exceptions.MapillaryGPSNoiseError("GPS is too noisy") monkeypatch.setattr( - geotag_videos_from_video.GeotagVideosFromVideo, - "to_description", - lambda _self, paths: [ + sample_video.factory, + "process", + lambda paths, options: [ describe_error_metadata( error, filename=paths[0], filetype=FileType.GOPRO ) @@ -424,9 +424,9 @@ def test_the_summary_counts_it_by_its_own_type( def test_an_empty_track_is_also_an_error(self, tmpdir, setup_mock, monkeypatch): """Previously an assert, so it vanished under `python -O`.""" monkeypatch.setattr( - geotag_videos_from_video.GeotagVideosFromVideo, - "to_description", - lambda _self, paths: [ + sample_video.factory, + "process", + lambda paths, options: [ VideoMetadata( filename=paths[0], filesize=0, filetype=FileType.GOPRO, points=[] ) @@ -440,9 +440,9 @@ def test_a_missing_video_stream_is_also_an_error( self, tmpdir, setup_mock, monkeypatch ): monkeypatch.setattr( - geotag_videos_from_video.GeotagVideosFromVideo, - "to_description", - lambda _self, paths: [ + sample_video.factory, + "process", + lambda paths, options: [ VideoMetadata( filename=paths[0], filesize=0, @@ -600,11 +600,9 @@ def fake_extract_frames( mock_ffmpeg_class, ) - mock_geotag_instance = mock.MagicMock() - mock_geotag_instance.to_description.return_value = [video_metadata] patches["geotag_cls"] = mock.patch( - "mapillary_tools.sample_video.geotag_videos_from_video.GeotagVideosFromVideo", - return_value=mock_geotag_instance, + "mapillary_tools.sample_video.factory.process", + return_value=[video_metadata], ) patches["moov_parse"] = mock.patch.object( @@ -755,7 +753,8 @@ def test_no_frame_within_the_track(self, tmp_path: Path) -> None: ): with pytest.raises( exceptions.MapillaryVideoError, - match="No video frames found within the GPS track from 100.000s to 109.000s", + # how humanize words the time varies between its versions + match="none of its frames falls within its GPS track, which starts .+ after the video starts", ): sample_video._sample_single_video_by_distance( video_file, sample_dir, sample_distance=0.0 @@ -920,3 +919,288 @@ def boom(*args, **kwargs): ) assert "--skip_sample_errors" not in str(excinfo.value) + + +# --------------------------------------------------------------------------- +# Distance sampling honours --geotag_source +# --------------------------------------------------------------------------- + + +class TestGeotagSourceReachesDistanceSampling: + """ + Distance sampling read the video's own telemetry directly, bypassing the + geotag factory, so --geotag_source was ignored: a GPX attached to rescue a + camera whose embedded GPS is unusable never got a chance to supply the + positions the sampler picks frames with. + """ + + def test_no_sources_requested_means_native_only(self): + """ + Not DEFAULT_GEOTAG_SOURCE_OPTIONS: that chain continues on to + exiftool_runtime, whose parser cannot see every field the noise filter + rejects on, so it would accept tracks the native parser refuses. + """ + options = sample_video._parse_geotag_options( + None, None, None, Path("/data/v.mp4") + ) + + assert [option.source for option in options] == [SourceType.NATIVE] + + def test_requested_source_is_passed_through(self): + options = sample_video._parse_geotag_options( + ["gpx"], Path("/data/track.gpx"), None, Path("/data/v.mp4") + ) + + assert [option.source for option in options] == [SourceType.GPX] + assert options[0].source_path is not None + assert options[0].source_path.source_path == Path("/data/track.gpx") + + def test_video_geotag_source_is_honoured_too(self): + options = sample_video._parse_geotag_options( + None, Path("/data/track.gpx"), ["gpx"], Path("/data/v.mp4") + ) + + assert [option.source for option in options] == [SourceType.GPX] + + @pytest.mark.parametrize("source", ["exiftool_xml", "gpx"]) + def test_without_a_path_sources_get_the_video_import_path(self, source): + """ + As when the frames are geotagged. That is how exiftool_xml finds the + XML files in the video directory; a GPX gets no sidecar lookup. + """ + options = sample_video._parse_geotag_options( + [source], None, None, Path("/data/videos") + ) + + assert options[0].source_path is not None + assert options[0].source_path.source_path == Path("/data/videos") + + def test_chained_sources_are_preserved(self): + options = sample_video._parse_geotag_options( + ["native", "gpx"], Path("/data/track.gpx"), None, Path("/data/v.mp4") + ) + + assert [option.source for option in options] == [ + SourceType.NATIVE, + SourceType.GPX, + ] + + def test_the_sampler_asks_the_factory_for_them( + self, tmpdir, setup_mock, monkeypatch + ): + """The options reach factory.process() rather than being dropped.""" + seen: list = [] + points = _make_gps_points(4, time_step=1.0) + + def fake_process(paths, options): + seen.append(list(options)) + return [ + VideoMetadata( + filename=paths[0], + filesize=1, + filetype=FileType.GOPRO, + points=points, + ) + ] + + monkeypatch.setattr(sample_video.factory, "process", fake_process) + # stop after the metadata is read; the rest needs real ffmpeg + monkeypatch.setattr( + ffmpeglib.Probe, "probe_video_with_max_resolution", lambda _self: None + ) + + with pytest.raises(exceptions.MapillaryVideoError): + sample_video.sample_video( + _PWD.joinpath("data/mock_sample_video/videos/hello.mp4"), + Path(tmpdir), + video_sample_distance=2, + rerun=True, + geotag_source=["gpx"], + geotag_source_path=Path("/data/track.gpx"), + ) + + assert len(seen) == 1 + assert [option.source for option in seen[0]] == [SourceType.GPX] + + +class TestSourcesThatCannotReadVideos: + """ + exif, nmea and the legacy gopro/blackvue/camm sources have no video + extractor, so process() returned no metadata for the video and the sampler + failed an assert, for a video it used to sample by its own GPS. + """ + + @pytest.mark.parametrize("source", ["exif", "nmea", "gopro", "blackvue", "camm"]) + def test_sampling_falls_back_to_the_video_gps(self, source): + options = sample_video._parse_geotag_options( + [source], Path("/data/track"), None, Path("/data/v.mp4") + ) + + assert [option.source for option in options] == [SourceType.NATIVE] + + def test_they_are_dropped_from_a_chain(self): + options = sample_video._parse_geotag_options( + ["exif", "gpx"], Path("/data/track.gpx"), None, Path("/data/v.mp4") + ) + + assert [option.source for option in options] == [SourceType.GPX] + + def test_a_source_limited_to_images_is_dropped(self): + options = sample_video._parse_geotag_options( + ['{"source": "gpx", "filetypes": ["image"]}'], + Path("/data/track.gpx"), + None, + Path("/data/v.mp4"), + ) + + assert [option.source for option in options] == [SourceType.NATIVE] + + +class TestVideoStartTimeReachesTheGPX: + """ + A GPX on a video without GPS was assumed to start with the video, and + --video_start_time was ignored, so every clip of a drive sampled the first + minutes of its GPX, and was stamped with their times. + """ + + def test_every_option_carries_it(self): + video_sync = VideoSyncOption(start_time=1_700_000_000.0) + + for sources in [None, ["gpx"], ["exif"]]: + options = sample_video._parse_geotag_options( + sources, Path("/data/track.gpx"), None, Path("/data/v.mp4"), video_sync + ) + assert all(option.video_sync is video_sync for option in options) + + def test_sample_video_passes_it_on(self, tmpdir, setup_mock, monkeypatch): + seen: list = [] + + def fake_process(paths, options): + seen.extend(options) + return [ + VideoMetadata( + filename=paths[0], + filesize=1, + filetype=FileType.VIDEO, + points=_make_gps_points(4, time_step=1.0), + ) + ] + + monkeypatch.setattr(sample_video.factory, "process", fake_process) + # stop after the metadata is read; the rest needs real ffmpeg + monkeypatch.setattr( + ffmpeglib.Probe, "probe_video_with_max_resolution", lambda _self: None + ) + + with pytest.raises(exceptions.MapillaryVideoError): + sample_video.sample_video( + _PWD.joinpath("data/mock_sample_video/videos/hello.mp4"), + Path(tmpdir), + video_sample_distance=2, + rerun=True, + video_start_time="2025_03_14_07_00_30_000", + geotag_source=["gpx"], + geotag_source_path=Path("/data/track.gpx"), + ) + + expected = datetime.datetime( + 2025, 3, 14, 7, 0, 30, tzinfo=datetime.timezone.utc + ).timestamp() + assert seen[0].video_sync == VideoSyncOption(start_time=expected) + + def test_sample_video_passes_the_interpolation_options_on( + self, tmpdir, setup_mock, monkeypatch + ): + """video_process registers them for its geotagging stage.""" + seen: list = [] + + def fake_process(paths, options): + seen.extend(options) + return [ + VideoMetadata( + filename=paths[0], + filesize=1, + filetype=FileType.VIDEO, + points=_make_gps_points(4, time_step=1.0), + ) + ] + + monkeypatch.setattr(sample_video.factory, "process", fake_process) + monkeypatch.setattr( + ffmpeglib.Probe, "probe_video_with_max_resolution", lambda _self: None + ) + + with pytest.raises(exceptions.MapillaryVideoError): + sample_video.sample_video( + _PWD.joinpath("data/mock_sample_video/videos/hello.mp4"), + Path(tmpdir), + video_sample_distance=2, + rerun=True, + geotag_source=["gpx"], + geotag_source_path=Path("/data/track.gpx"), + interpolation_offset_time=2.5, + interpolation_use_gpx_start_time=True, + ) + + assert seen[0].video_sync == VideoSyncOption( + offset_time=2.5, use_gpx_start_time=True + ) + + def test_without_it_the_start_time_is_left_unknown(self): + """ + The start time probed from the container is creation_time - duration, + which is a whole video length early when creation_time marks the start. + """ + options = sample_video._parse_geotag_options( + ["gpx"], Path("/data/track.gpx"), None, Path("/data/v.mp4"), None + ) + + assert options[0].video_sync is None + + +class TestTrackThatMissesTheVideo: + def test_it_fails_instead_of_writing_an_empty_sample_directory( + self, tmpdir, setup_mock, monkeypatch, caplog + ): + """ + A GPX placed by a wrong --video_start_time selects no frames. The + video counts as failed, and the log says why. + """ + points = _make_gps_points(4, time_step=1.0) + for point in points: + point.time -= 7200 + monkeypatch.setattr( + sample_video.factory, + "process", + lambda paths, options: [ + VideoMetadata( + filename=paths[0], + filesize=1, + filetype=FileType.VIDEO, + points=points, + ) + ], + ) + monkeypatch.setattr( + ffmpeglib.Probe, + "probe_video_with_max_resolution", + lambda _self: {"index": 0}, + ) + monkeypatch.setattr( + mp4_sample_parser.MovieBoxParser, "parse_file", mock.MagicMock() + ) + monkeypatch.setattr( + sample_video, "_sample_video_stream_by_distance", lambda *args: {} + ) + + with pytest.raises(exceptions.MapillaryVideoSampleError): + sample_video.sample_video( + _PWD.joinpath("data/mock_sample_video/videos/hello.mp4"), + Path(tmpdir), + video_sample_distance=2, + rerun=True, + ) + + assert "none of its frames falls within its GPS track" in caplog.text + assert "before the video starts" in caplog.text + assert not Path(tmpdir).joinpath("hello.mp4").exists()