diff --git a/README.md b/README.md index 6a439630..10c3fd64 100644 --- a/README.md +++ b/README.md @@ -284,7 +284,11 @@ It is used to locate the images along the GPS tracks. mapillary_tools process MY_IMAGE_DIR --geotag_source "gpx" --geotag_source_path MY_EXTERNAL_GPS.gpx ``` -To geotag videos with a GPX file, video start time (video creation time minus video duration) is required to locate the sample images along the GPS tracks. +To geotag videos with a GPX file, video start time is required to locate the sample images along the GPS tracks. +It is read from the video's own GPS track when it has one, and otherwise from the video creation time. +Cameras disagree on whether the creation time marks the start or the end of the recording (most dashcams write the end), +so mapillary_tools uses the camera model or a date and time in the file name to tell which, and assumes the start when neither says. +Use `--video_start_time` to override it, in UTC, when the sample images end up one video duration off along the track. ```sh # Geotagging with GPX works with interval-based sampling only, diff --git a/mapillary_tools/blackvue_parser.py b/mapillary_tools/blackvue_parser.py index 625c95dc..cc089da1 100644 --- a/mapillary_tools/blackvue_parser.py +++ b/mapillary_tools/blackvue_parser.py @@ -51,15 +51,19 @@ def extract_blackvue_info(fp: T.BinaryIO) -> BlackVueInfo | None: if gps_data is None: return None - points = _parse_gps_box(gps_data) + points, recording_start_time = _parse_gps_box_with_start_time(gps_data) points.sort(key=lambda p: p.time) if points: - # Convert the time field to relative time to the first point + # Convert the time field to the video time, i.e. relative to the start + # of the recording. That is the first NMEA line the camera logged, not + # the first valid fix: until the receiver gets a fix, which can take + # minutes after a cold start, the camera logs lines without positions. # epoch_time stays as the original time in seconds - first_point_time = points[0].time + assert recording_start_time is not None for p in points: - p.time = p.time - first_point_time + # Rounding needed to avoid floating point precision issues + p.time = round(p.time - recording_start_time, 3) # Camera model try: @@ -76,6 +80,22 @@ def extract_blackvue_info(fp: T.BinaryIO) -> BlackVueInfo | None: return BlackVueInfo(model=model, gps=points) +def is_blackvue(fp: T.BinaryIO) -> bool: + """ + Tell whether a video was recorded by a BlackVue dashcam, which writes its + GPS log and its camera model into boxes nested in a top-level free box, + whether or not it ever got a GPS fix + """ + for path in [[b"free", b"gps "], [b"free", b"cprt"]]: + fp.seek(0) + try: + if sparser.parse_mp4_data_first(fp, path) is not None: + return True + except sparser.ParsingError: + pass + return False + + def _extract_camera_model_from_cprt(cprt_bytes: bytes) -> str: """ >>> _extract_camera_model_from_cprt(b' {"model":"DR900X Plus","ver":0.918,"lang":"English","direct":1,"psn":"","temp":34,"GPS":1}') @@ -254,6 +274,29 @@ def _parse_gps_box(gps_data: bytes) -> list[telemetry.GPSPoint]: >>> list(_parse_gps_box(b"[1623057074211]$GPVTG,,T,,M,0.078,N,0.144,K,D*28[1623057075215]")) [] """ + points, _ = _parse_gps_box_with_start_time(gps_data) + return points + + +def _parse_gps_box_with_start_time( + gps_data: bytes, +) -> tuple[list[telemetry.GPSPoint], float | None]: + """ + Parse the GPS points, and the time of the first NMEA line in the same + corrected clock, which is when the recording started + + >>> _parse_gps_box_with_start_time(b"[1623057074211]$GPGGA,202530.00,5109.0262,N,11401.8407,W,5,40,0.5,1097.36,M,-17.00,M,18,TSTR*61")[1] + 1623097530.0 + >>> points, start_time = _parse_gps_box_with_start_time(b''' + ... [1623057072211]$GPGGA,,,,,,0,00,99.99,,,,,,*48 + ... [1623057073211]$GPRMC,,V,,,,,,,,,,N*53 + ... [1623057074211]$GPGGA,202530.00,5109.0262,N,11401.8407,W,5,40,0.5,1097.36,M,-17.00,M,18,TSTR*61 + ... ''') + >>> len(points), points[0].time - start_time + (1, 2.0) + >>> _parse_gps_box_with_start_time(b"") + ([], None) + """ parsed_lines: list[tuple[float, pynmea2.NMEASentence]] = [] # First pass: collect parsed_lines @@ -264,6 +307,13 @@ def _parse_gps_box(gps_data: bytes) -> list[telemetry.GPSPoint]: timezone_offset = _detect_timezone_offset(parsed_lines) + if parsed_lines: + start_time: float | None = round( + min(epoch_sec for epoch_sec, _ in parsed_lines) + timezone_offset, 3 + ) + else: + start_time = None + points_by_sentence_type: dict[str, list[telemetry.GPSPoint]] = {} # Second pass: apply offset to all GPS points @@ -298,12 +348,12 @@ def _parse_gps_box(gps_data: bytes) -> list[telemetry.GPSPoint]: # This is the extraction order in exiftool if "RMC" in points_by_sentence_type: - return points_by_sentence_type["RMC"] + return points_by_sentence_type["RMC"], start_time if "GGA" in points_by_sentence_type: - return points_by_sentence_type["GGA"] + return points_by_sentence_type["GGA"], start_time if "GLL" in points_by_sentence_type: - return points_by_sentence_type["GLL"] + return points_by_sentence_type["GLL"], start_time - return [] + return [], start_time diff --git a/mapillary_tools/ffmpeg.py b/mapillary_tools/ffmpeg.py index 0264e2b2..c75e2abf 100644 --- a/mapillary_tools/ffmpeg.py +++ b/mapillary_tools/ffmpeg.py @@ -9,6 +9,7 @@ import datetime import json import logging +import math import os import re import subprocess @@ -55,8 +56,13 @@ class Stream(T.TypedDict): nb_frames: str -class ProbeOutput(T.TypedDict): +class Format(T.TypedDict, total=False): + tags: dict[str, str] + + +class ProbeOutput(T.TypedDict, total=False): streams: list[Stream] + format: Format class FFmpegNotFoundError(Exception): @@ -605,40 +611,63 @@ def __init__(self, probe_output: ProbeOutput) -> None: """ self.probe_output = probe_output - def probe_video_start_time(self) -> datetime.datetime | None: + def probe_video_creation_time(self) -> datetime.datetime | None: """ - Determine the start time of the video by analyzing stream metadata. + Read the creation time the camera stamped into the stream metadata. - Searches for creation time and duration information in video streams first, - then falls back to other stream types. Calculates start time as: - creation_time - duration + Searches video streams first, then falls back to other stream types. + Whether the creation time marks the start or the end of the recording + depends on the camera (see sample_video._creation_time_to_start_time). Returns: - Video start time as datetime object, or None if cannot be determined + Creation time as datetime object, or None if cannot be determined Note: Prioritizes video streams with highest resolution when multiple exist. """ - streams = self.probe_output.get("streams", []) + for stream in self._iterate_streams_by_priority(): + creation_time = self.extract_stream_creation_time(stream) + if creation_time is not None: + return creation_time + + return None + + def probe_video_duration(self) -> float | None: + """ + Read the duration of the video in seconds from the stream metadata. + + Searches the streams in the same order as probe_video_creation_time. + + Returns: + Duration in seconds, or None if cannot be determined + """ + for stream in self._iterate_streams_by_priority(): + duration = self.extract_stream_duration(stream) + if duration is not None: + return duration - # Search start time from video streams + return None + + def probe_format_tag(self, key: str) -> str | None: + """ + Read a tag of the container, such as "make" or "model". + + Returns: + The tag value, or None if the container does not have the tag + """ + return self.probe_output.get("format", {}).get("tags", {}).get(key) + + def _iterate_streams_by_priority(self) -> T.Generator[Stream, None, None]: + # Video streams by resolution, from the highest, then the other streams video_streams = self.probe_video_streams() video_streams.sort( key=lambda s: s.get("width", 0) * s.get("height", 0), reverse=True ) - for stream in video_streams: - start_time = self.extract_stream_start_time(stream) - if start_time is not None: - return start_time + yield from video_streams - # Search start time from the other streams - for stream in streams: + for stream in self.probe_output.get("streams", []): if stream.get("codec_type") != "video": - start_time = self.extract_stream_start_time(stream) - if start_time is not None: - return start_time - - return None + yield stream def probe_video_streams(self) -> list[Stream]: """ @@ -671,36 +700,54 @@ def probe_video_with_max_resolution(self) -> Stream | None: return video_streams[0] @classmethod - def extract_stream_start_time(cls, stream: Stream) -> datetime.datetime | None: + def extract_stream_creation_time(cls, stream: Stream) -> datetime.datetime | None: """ - Calculate the start time of a specific stream. - - Determines start time by subtracting stream duration from creation time: - start_time = creation_time - duration + Read the creation time of a specific stream. Args: - stream: Stream dictionary containing metadata including tags and duration + stream: Stream dictionary containing metadata including tags Returns: - Stream start time as datetime object, or None if required metadata is missing + Creation time as datetime object, or None if it is missing or malformed Note: Handles multiple datetime formats including ISO format and custom patterns. """ - duration_str = stream.get("duration") - LOG.debug("Extracted video duration: %s", duration_str) - if duration_str is None: - return None - duration = float(duration_str) - creation_time_str = stream.get("tags", {}).get("creation_time") LOG.debug("Extracted video creation time: %s", creation_time_str) if creation_time_str is None: return None try: - creation_time = datetime.datetime.fromisoformat(creation_time_str) + return datetime.datetime.fromisoformat(creation_time_str) except ValueError: - creation_time = datetime.datetime.strptime( + pass + try: + return datetime.datetime.strptime( creation_time_str, "%Y-%m-%dT%H:%M:%S.%f%z" ) - return creation_time - datetime.timedelta(seconds=duration) + except ValueError: + LOG.warning("Ignoring malformed video creation time: %s", creation_time_str) + return None + + @classmethod + def extract_stream_duration(cls, stream: Stream) -> float | None: + """ + Read the duration of a specific stream in seconds. + + Args: + stream: Stream dictionary containing metadata + + Returns: + Duration in seconds, or None if it is missing or malformed + """ + duration_str = stream.get("duration") + LOG.debug("Extracted video duration: %s", duration_str) + if duration_str is None: + return None + try: + duration = float(duration_str) + except ValueError: + return None + if not math.isfinite(duration) or duration < 0: + return None + return duration diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py index 1e3e7764..52776262 100644 --- a/mapillary_tools/sample_video.py +++ b/mapillary_tools/sample_video.py @@ -8,17 +8,28 @@ import datetime import logging import os +import re import shutil +import statistics import time import typing as T from contextlib import contextmanager from pathlib import Path -from . import constants, exceptions, ffmpeg as ffmpeglib, geo, types, utils +from . import ( + blackvue_parser, + constants, + exceptions, + ffmpeg as ffmpeglib, + geo, + types, + utils, +) from .exif_write import ExifEdit from .geotag import geotag_videos_from_video +from .geotag.video_extractors.native import NativeVideoExtractor from .mp4 import mp4_sample_parser -from .serializer.description import parse_capture_time +from .serializer.description import build_capture_time, parse_capture_time LOG = logging.getLogger(__name__) @@ -179,6 +190,225 @@ def wip_sample_dir(sample_dir: Path) -> Path: ) +# GPS clocks that report a time before this have never been set: GoPro writes +# 2000-01-01 until it gets its first fix, for example +_MIN_PLAUSIBLE_START_TIME = datetime.datetime( + 2010, 1, 1, tzinfo=datetime.timezone.utc +).timestamp() + +# Leave room for the clock of the machine running this to be behind +_MAX_FUTURE_START_TIME_SECONDS = 24 * 3600 + +# How many GPS timestamps to take the median of, so that a single bad one cannot +# shift the start time: the Labpano PanoX V2 can record a stale first fix, +# seconds older than the rest. Only the first few are used because in timelapses +# the video clock runs slower than the GPS clock, so the two drift apart +_GPS_CLOCK_SAMPLES = 5 + +# Cameras known to stamp the creation time at the end of the recording, as +# lowercase (make, model) from the container tags. BlackVue is not listed +# because it does not write those tags: blackvue_parser.is_blackvue detects it +_END_STAMPING_CAMERAS = { + ("ricoh", "ricoh theta x"), + ("labpano", "panox v2"), +} + +# A date and time in a file name, for example 20230512_101530 (BlackVue, +# Insta360) or 2023_0512_101530 (Viofo), in the camera's local time +_FILENAME_TIME_RE = re.compile( + r"(? datetime.datetime | None: + """ + Map the absolute GPS timestamps at the start of a track back to the video's time 0. + + Point times are relative to the start of the video, so subtracting one from + its own absolute timestamp gives the wall clock at which the video started. + Timestamps outside the plausible range are skipped. + """ + max_start_time = time.time() + _MAX_FUTURE_START_TIME_SECONDS + + start_times: list[float] = [] + for point in points: + unix_time = point.get_unix_time() + if unix_time is None: + continue + start_time = unix_time - point.time + # Written as a negated range check so that NaN is skipped too + if not (_MIN_PLAUSIBLE_START_TIME <= start_time <= max_start_time): + continue + start_times.append(start_time) + if len(start_times) >= _GPS_CLOCK_SAMPLES: + break + + if not start_times: + return None + + return datetime.datetime.fromtimestamp( + statistics.median(start_times), tz=datetime.timezone.utc + ) + + +def _telemetry_start_time(video_path: Path) -> datetime.datetime | None: + try: + video_metadata = NativeVideoExtractor(video_path).extract() + except exceptions.MapillaryDescriptionError as ex: + LOG.debug("No video telemetry to read the start time from: %s", ex) + return None + + return _gps_clock_start_time(video_metadata.points) + + +def _parse_filename_time(video_path: Path) -> datetime.datetime | None: + for match in _FILENAME_TIME_RE.finditer(video_path.stem): + year, month, day, hour, minute, second = (int(g) for g in match.groups()) + try: + return datetime.datetime( + year, month, day, hour, minute, second, tzinfo=datetime.timezone.utc + ) + except ValueError: + continue + + return None + + +def _as_utc(dt: datetime.datetime) -> datetime.datetime: + if dt.tzinfo is None: + return dt.replace(tzinfo=datetime.timezone.utc) + return dt + + +def _is_same_time_in_another_time_zone( + a: datetime.datetime, b: datetime.datetime +) -> bool: + """ + Tell whether two times could be the same moment, one of them possibly in + local time labelled as UTC + """ + delta = abs((_as_utc(a) - _as_utc(b)).total_seconds()) + if _MAX_UTC_OFFSET_SECONDS + _FILENAME_TIME_TOLERANCE_SECONDS < delta: + return False + remainder = delta % _UTC_OFFSET_GRANULARITY_SECONDS + return ( + min(remainder, _UTC_OFFSET_GRANULARITY_SECONDS - remainder) + <= _FILENAME_TIME_TOLERANCE_SECONDS + ) + + +def _is_end_stamping_camera(video_path: Path, probe: ffmpeglib.Probe) -> bool: + make = probe.probe_format_tag("make") + model = probe.probe_format_tag("model") + if make is not None and model is not None: + if (make.strip().lower(), model.strip().lower()) in _END_STAMPING_CAMERAS: + return True + + try: + with video_path.open("rb") as fp: + return blackvue_parser.is_blackvue(fp) + except Exception as ex: + LOG.debug("Unable to tell whether %s is a BlackVue video: %s", video_path, ex) + return False + + +def _creation_time_to_start_time( + video_path: Path, probe: ffmpeglib.Probe +) -> datetime.datetime | None: + """ + Determine the wall clock time at which a video started recording from the + creation time in its metadata. + + Cameras disagree on what the creation time marks. Many stamp the end of the + recording: BlackVue, Viofo, Vantrue and most other dashcams, the Ricoh Theta + X and the Labpano PanoX V2. Others stamp the start: Sony, Insta360 and the + dashcam in the bug report that motivated reading this at all. Nothing in the + metadata says which, so look for evidence: a camera known to stamp the end, + or a time in the file name that matches only one of the two. Without any, + assume the start and warn about the alternative. + """ + creation_time = probe.probe_video_creation_time() + if creation_time is None: + return None + + duration = probe.probe_video_duration() + if duration is None: + LOG.warning( + "Unable to read the duration of %s, so assuming its creation time %s marks the start of the recording", + video_path.name, + creation_time, + ) + return creation_time + + start_time_if_end_stamped = creation_time - datetime.timedelta(seconds=duration) + + if _is_end_stamping_camera(video_path, probe): + return start_time_if_end_stamped + + filename_time = _parse_filename_time(video_path) + if filename_time is not None: + matches_start = _is_same_time_in_another_time_zone(filename_time, creation_time) + matches_end = _is_same_time_in_another_time_zone( + filename_time, start_time_if_end_stamped + ) + if matches_end and not matches_start: + return start_time_if_end_stamped + if matches_start and not matches_end: + return creation_time + + LOG.warning( + "Assuming the creation time %s of %s marks the start of the recording. " + "If the camera stamps the end instead, as most dashcams do, the recording started %.1f seconds earlier: " + "specify --video_start_time %s to use that", + creation_time, + video_path.name, + duration, + build_capture_time(start_time_if_end_stamped), + ) + return creation_time + + +def _extract_video_start_time( + video_path: Path, probe: ffmpeglib.Probe +) -> datetime.datetime | None: + """ + Determine the wall clock time at which a video started recording. + + A video's own telemetry is the better clock: it is absolute UTC, so it is + immune both to cameras that stamp the container's creation time at the end + of the recording and to cameras that stamp it in local time (GoPro). Fall + 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 + + if start_time is not None: + return start_time + + return _creation_time_to_start_time(video_path, probe) + + def _sample_single_video_by_interval( video_path: Path, sample_dir: Path, @@ -189,9 +419,9 @@ def _sample_single_video_by_interval( ffmpeg = ffmpeglib.FFMPEG(constants.FFMPEG_PATH, constants.FFPROBE_PATH) if start_time is None: - start_time = ffmpeglib.Probe( - ffmpeg.probe_format_and_streams(video_path) - ).probe_video_start_time() + start_time = _extract_video_start_time( + video_path, ffmpeglib.Probe(ffmpeg.probe_format_and_streams(video_path)) + ) if start_time is None: raise exceptions.MapillaryVideoError( f"Unable to extract video start time from {video_path}" @@ -286,13 +516,6 @@ def _sample_single_video_by_distance( probe = ffmpeglib.Probe(ffmpeg.probe_format_and_streams(video_path)) - if start_time is None: - start_time = probe.probe_video_start_time() - if start_time is None: - raise exceptions.MapillaryVideoError( - f"Unable to extract video start time from {video_path}" - ) - LOG.info("Extracting video metdata") video_metadatas = geotag_videos_from_video.GeotagVideosFromVideo().to_description( @@ -321,6 +544,20 @@ def _sample_single_video_by_distance( ) sorted_sample_indices = sorted(sample_points_by_frame_idx.keys()) + # Frames at points with an absolute timestamp are timestamped from it + # below, so only the others need the start time + if start_time is None and any( + interp.get_unix_time() is None + for _, interp in sample_points_by_frame_idx.values() + ): + start_time = _gps_clock_start_time(video_metadata.points) + if start_time is None: + start_time = _creation_time_to_start_time(video_path, probe) + if start_time is None: + raise exceptions.MapillaryVideoError( + f"Unable to extract video start time from {video_path}" + ) + with wip_dir_context(wip_sample_dir(sample_dir), sample_dir) as wip_dir: ffmpeg.extract_specified_frames( video_path, @@ -361,6 +598,7 @@ def _sample_single_video_by_distance( gps_unix_time, tz=datetime.timezone.utc ) else: + assert start_time is not None timestamp = start_time + datetime.timedelta(seconds=interp.time) exif_edit = ExifEdit(sample_paths[0]) exif_edit.add_date_time_original(timestamp) diff --git a/tests/integration/test_gopro.py b/tests/integration/test_gopro.py index 199b780d..32cba706 100644 --- a/tests/integration/test_gopro.py +++ b/tests/integration/test_gopro.py @@ -31,12 +31,16 @@ "MAPILLARY_TOOLS_GOPRO_GPS_PRECISION": "10000000", "MAPILLARY_TOOLS_MAX_CAPTURE_SPEED_KMH": "2000000", # km/h } +# The capture times come from the GPMF GPS clock (the median of the first fixes +# puts the start at 2019-11-18T23:42:08.539Z), not from the container's creation +# time. This camera writes the creation time in local time, so reading the start +# time from there would timestamp every sample 8 hours off. EXPECTED_DESCS: T.List[T.Any] = [ { "filename": "hero8.mp4/hero8_v_000001.jpg", "filetype": "image", "MAPAltitude": 9540.24, - "MAPCaptureTime": "2019_11_18_15_41_12_354", + "MAPCaptureTime": "2019_11_18_23_42_08_539", "MAPCompassHeading": { "TrueHeading": 123.93587938690177, "MagneticHeading": 123.93587938690177, @@ -51,7 +55,7 @@ "filename": "hero8.mp4/hero8_v_000002.jpg", "filetype": "image", "MAPAltitude": 7112.573717404068, - "MAPCaptureTime": "2019_11_18_15_41_14_354", + "MAPCaptureTime": "2019_11_18_23_42_10_539", "MAPCompassHeading": { "TrueHeading": 140.8665026186285, "MagneticHeading": 140.8665026186285, @@ -66,7 +70,7 @@ "filename": "hero8.mp4/hero8_v_000003.jpg", "filetype": "image", "MAPAltitude": 7463.642846094319, - "MAPCaptureTime": "2019_11_18_15_41_16_354", + "MAPCaptureTime": "2019_11_18_23_42_12_539", "MAPCompassHeading": { "TrueHeading": 138.44255851085705, "MagneticHeading": 138.44255851085705, @@ -81,7 +85,7 @@ "filename": "hero8.mp4/hero8_v_000004.jpg", "filetype": "image", "MAPAltitude": 6909.8168472111465, - "MAPCaptureTime": "2019_11_18_15_41_18_354", + "MAPCaptureTime": "2019_11_18_23_42_14_539", "MAPCompassHeading": { "TrueHeading": 142.23462669862568, "MagneticHeading": 142.23462669862568, @@ -96,7 +100,7 @@ "filename": "hero8.mp4/hero8_v_000005.jpg", "filetype": "image", "MAPAltitude": 7212.594480737465, - "MAPCaptureTime": "2019_11_18_15_41_20_354", + "MAPCaptureTime": "2019_11_18_23_42_16_539", "MAPCompassHeading": { "TrueHeading": 164.70819093235514, "MagneticHeading": 164.70819093235514, @@ -111,7 +115,7 @@ "filename": "hero8.mp4/hero8_v_000006.jpg", "filetype": "image", "MAPAltitude": 7274.361994963208, - "MAPCaptureTime": "2019_11_18_15_41_22_354", + "MAPCaptureTime": "2019_11_18_23_42_18_539", "MAPCompassHeading": { "TrueHeading": 139.71549328876722, "MagneticHeading": 139.71549328876722, diff --git a/tests/unit/test_blackvue_parser.py b/tests/unit/test_blackvue_parser.py index ec2956a0..1d767221 100644 --- a/tests/unit/test_blackvue_parser.py +++ b/tests/unit/test_blackvue_parser.py @@ -46,10 +46,11 @@ def test_parse_points(): box = {"type": b"free", "data": [{"type": b"gps ", "data": gps_data}]} data = cparser.Box32ConstructBuilder({b"free": {}}).Box.build(box) info = blackvue_parser.extract_blackvue_info(io.BytesIO(data)) + # Times are relative to the earliest line (1623057129253), not the first RMC assert info == blackvue_parser.BlackVueInfo( gps=[ telemetry.GPSPoint( - time=0.0, + time=0.003, lat=38.88615816666667, lon=-76.992434, alt=None, @@ -60,7 +61,7 @@ def test_parse_points(): ground_speed=None, ), telemetry.GPSPoint( - time=3.0, + time=3.003, lat=38.88615816666667, lon=-76.992434, alt=None, @@ -76,6 +77,50 @@ def test_parse_points(): ) +def test_parse_points_relative_to_recording_start(): + # After a cold start the receiver can take a minute to get its first fix. + # Until then the camera logs lines without a position, so the first fix + # here is 40 seconds into the video, not at its start + gps_data = b""" +[1623057089253]$GPRMC,,V,,,,,,,,,,N*53 + +[1623057109253]$GPGGA,,,,,,0,00,99.99,,,,,,*48 + +[1623057129256]$GPRMC,201205.00,A,3853.16949,N,07659.54604,W,5.849,284.43,070621,,,D*76 + +[1623057132256]$GPRMC,201208.00,A,3853.16949,N,07659.54604,W,5.849,284.43,070621,,,D*7B + """ + + box = {"type": b"free", "data": [{"type": b"gps ", "data": gps_data}]} + data = cparser.Box32ConstructBuilder({b"free": {}}).Box.build(box) + info = blackvue_parser.extract_blackvue_info(io.BytesIO(data)) + assert info is not None + assert [p.time for p in info.gps] == [40.003, 43.003] + assert [p.epoch_time for p in info.gps] == [1623096725, 1623096728] + + +def _build_free_box(child_type: bytes, child_data: bytes) -> bytes: + box = {"type": b"free", "data": [{"type": child_type, "data": child_data}]} + return cparser.Box32ConstructBuilder({b"free": {}}).Box.build(box) + + +def test_is_blackvue(): + # Without a fix the GPS log has no positions, but it is still there + no_fix = _build_free_box(b"gps ", b"[1623057089253]$GPRMC,,V,,,,,,,,,,N*53") + assert blackvue_parser.is_blackvue(io.BytesIO(no_fix)) + + cprt = _build_free_box(b"cprt", b" Pittasoft Co., Ltd.;;DR900S-1CH;") + assert blackvue_parser.is_blackvue(io.BytesIO(cprt)) + + ftyp = cparser.Box32ConstructBuilder({}).Box.build( + {"type": b"ftyp", "data": b"isom\x00\x00\x02\x00isomiso2avc1mp41"} + ) + assert not blackvue_parser.is_blackvue(io.BytesIO(ftyp)) + assert not blackvue_parser.is_blackvue(io.BytesIO(_build_free_box(b"abcd", b""))) + assert not blackvue_parser.is_blackvue(io.BytesIO(b"")) + assert not blackvue_parser.is_blackvue(io.BytesIO(b"\xff" * 64)) + + def test_gpspoint_gga(): gps_data = b"[1623057074211]$GPGGA,202530.25,5109.0262,N,11401.8407,W,5,40,0.5,1097.36,M,-17.00,M,18,TSTR*66" points = blackvue_parser._parse_gps_box(gps_data) diff --git a/tests/unit/test_ffmpeg.py b/tests/unit/test_ffmpeg.py index c79fc8ce..78098017 100644 --- a/tests/unit/test_ffmpeg.py +++ b/tests/unit/test_ffmpeg.py @@ -139,8 +139,8 @@ def test_probe_format_and_streams_ok(setup_data: py.path.local): probe_output = ff.probe_format_and_streams(video_path) probe = ffmpeg.Probe(probe_output) - start_time = probe.probe_video_start_time() - assert start_time is None + creation_time = probe.probe_video_creation_time() + assert creation_time is None max_stream = probe.probe_video_with_max_resolution() assert max_stream is not None assert max_stream["index"] == 0 @@ -156,9 +156,9 @@ def test_probe_format_and_streams_gopro_ok(setup_data: py.path.local): probe_output = ff.probe_format_and_streams(video_path) probe = ffmpeg.Probe(probe_output) - start_time = probe.probe_video_start_time() - assert start_time is not None - assert datetime.datetime.isoformat(start_time) == "2019-11-18T15:41:12.354033+00:00" + creation_time = probe.probe_video_creation_time() + assert creation_time is not None + assert datetime.datetime.isoformat(creation_time) == "2019-11-18T15:41:25+00:00" max_stream = probe.probe_video_with_max_resolution() assert max_stream is not None assert max_stream["index"] == 0 @@ -220,51 +220,98 @@ def test_ffprobe_not_exists(): assert False, "RuntimeError not raised" +def _probe_with_video_stream(creation_time, duration) -> ffmpeg.Probe: + return ffmpeg.Probe( + { + "streams": [ + { + "index": 0, + "codec_type": "video", + "codec_tag_string": "avc1", + "width": 2880, + "height": 1620, + "coded_width": 2880, + "coded_height": 1620, + "duration": duration, + "tags": { + "creation_time": creation_time, + "language": "und", + "handler_name": "Core Media Video", + "vendor_id": "[0][0][0][0]", + "encoder": "H.264", + }, + } + ] + } + ) + + def test_probe(): def test_creation_time(expected, probe_creation_time, probe_duration): - probe = ffmpeg.Probe( - { - "streams": [ - { - "index": 0, - "codec_type": "video", - "codec_tag_string": "avc1", - "width": 2880, - "height": 1620, - "coded_width": 2880, - "coded_height": 1620, - "duration": probe_duration, - "tags": { - "creation_time": probe_creation_time, - "language": "und", - "handler_name": "Core Media Video", - "vendor_id": "[0][0][0][0]", - "encoder": "H.264", - }, - } - ] - } - ) - creation_time = probe.probe_video_start_time() - assert expected == creation_time + probe = _probe_with_video_stream(probe_creation_time, probe_duration) + assert expected == probe.probe_video_creation_time() + assert float(probe_duration) == probe.probe_video_duration() + # Whether the creation time marks the start or the end of the recording is + # up to the caller, so the duration is not subtracted from it here test_creation_time( - datetime.datetime(2023, 3, 7, 1, 35, 29, 190123, tzinfo=datetime.timezone.utc), + datetime.datetime(2023, 3, 7, 1, 35, 34, 123456, tzinfo=datetime.timezone.utc), "2023-03-07T01:35:34.123456Z", "4.933333", ) test_creation_time( - datetime.datetime(2023, 3, 7, 1, 35, 29, 66667, tzinfo=datetime.timezone.utc), + datetime.datetime(2023, 3, 7, 1, 35, 34, tzinfo=datetime.timezone.utc), "2023-03-07T01:35:34.000000Z", "4.933333", ) test_creation_time( - datetime.datetime(2023, 3, 7, 1, 35, 29, 66667), + datetime.datetime(2023, 3, 7, 1, 35, 34), "2023-03-07 01:35:34", "4.933333", ) +def test_probe_malformed_metadata(): + # Treated as missing rather than raised, so that sampling can fall back + probe = _probe_with_video_stream("not a time", "N/A") + assert probe.probe_video_creation_time() is None + assert probe.probe_video_duration() is None + + for duration in ["nan", "inf", "-1"]: + probe = _probe_with_video_stream("2023-03-07T01:35:34.000000Z", duration) + assert probe.probe_video_duration() is None + + +def test_probe_falls_back_to_other_streams(): + probe = ffmpeg.Probe( + { + "streams": [ + {"index": 0, "codec_type": "video", "width": 1920, "height": 1080}, + { + "index": 1, + "codec_type": "audio", + "duration": "5.0", + "tags": {"creation_time": "2023-03-07T01:35:34.000000Z"}, + }, + ] + } + ) + assert probe.probe_video_creation_time() == datetime.datetime( + 2023, 3, 7, 1, 35, 34, tzinfo=datetime.timezone.utc + ) + assert probe.probe_video_duration() == 5.0 + + +def test_probe_format_tag(): + probe = ffmpeg.Probe( + {"streams": [], "format": {"tags": {"make": "RICOH", "model": "RICOH THETA X"}}} + ) + assert probe.probe_format_tag("make") == "RICOH" + assert probe.probe_format_tag("model") == "RICOH THETA X" + assert probe.probe_format_tag("firmware") is None + assert ffmpeg.Probe({"streams": []}).probe_format_tag("make") is None + + def _ffmpeg_with_version(version): ff = ffmpeg.FFMPEG() ff._version_probed = True diff --git a/tests/unit/test_sample_video.py b/tests/unit/test_sample_video.py index 0743eeb4..271929ec 100644 --- a/tests/unit/test_sample_video.py +++ b/tests/unit/test_sample_video.py @@ -7,8 +7,10 @@ import datetime import json +import logging import os import shutil +import time import typing as T from pathlib import Path from unittest import mock @@ -21,13 +23,20 @@ ffmpeg as ffmpeglib, geo, sample_video, + telemetry, ) -from mapillary_tools.mp4 import mp4_sample_parser +from mapillary_tools.mp4 import construct_mp4_parser as cparser, mp4_sample_parser from mapillary_tools.serializer import description from mapillary_tools.types import FileType, VideoMetadata _PWD = Path(os.path.dirname(os.path.abspath(__file__))) +# The creation time of the hello.mp4 probe fixture, which is where videos +# without their own GPS clock get their start time from +PROBE_START_TIME = datetime.datetime( + 2021, 8, 10, 14, 38, 6, tzinfo=datetime.timezone.utc +) + # --------------------------------------------------------------------------- # Interval-based sampling tests (using MOCK_FFMPEG) @@ -85,8 +94,7 @@ def test_sample_video(tmpdir: py.path.local, setup_mock): rerun=True, ) samples = sample_dir.join("hello.mp4").listdir() - video_start_time = description.parse_capture_time("2021_08_10_14_37_05_023") - _validate_interval([Path(s) for s in samples], video_start_time) + _validate_interval([Path(s) for s in samples], PROBE_START_TIME) def test_sample_single_video(tmpdir: py.path.local, setup_mock): @@ -101,8 +109,7 @@ def test_sample_single_video(tmpdir: py.path.local, setup_mock): rerun=True, ) samples = sample_dir.join("hello.mp4").listdir() - video_start_time = description.parse_capture_time("2021_08_10_14_37_05_023") - _validate_interval([Path(s) for s in samples], video_start_time) + _validate_interval([Path(s) for s in samples], PROBE_START_TIME) def test_sample_video_with_start_time(tmpdir: py.path.local, setup_mock): @@ -123,6 +130,330 @@ def test_sample_video_with_start_time(tmpdir: py.path.local, setup_mock): _validate_interval([Path(s) for s in samples], video_start_time) +def test_sample_video_from_gps_clock(tmpdir: py.path.local, setup_mock, monkeypatch): + """A video's own GPS clock wins over the container's creation time.""" + root = _PWD.joinpath("data/mock_sample_video") + video_dir = root.joinpath("videos") + sample_dir = tmpdir.mkdir("sampled_video_frames") + + # A camera that stamps the creation time at the end of the recording, or in + # local time, still has a correct absolute clock in its telemetry + gps_start_time = datetime.datetime( + 2021, 8, 10, 6, 38, 6, tzinfo=datetime.timezone.utc + ) + points = [ + telemetry.GPSPoint( + time=float(i), + lat=40.0 + i * 0.001, + lon=-74.0, + alt=None, + angle=None, + epoch_time=gps_start_time.timestamp() + i, + fix=None, + precision=None, + ground_speed=None, + ) + for i in range(3) + ] + monkeypatch.setattr( + sample_video, + "NativeVideoExtractor", + lambda video_path: mock.Mock( + extract=lambda: VideoMetadata( + filename=video_path, + filetype=FileType.BLACKVUE, + points=T.cast(T.List[geo.Point], points), + ) + ), + ) + + sample_video.sample_video( + video_dir, + Path(sample_dir), + video_sample_distance=-1, + video_sample_interval=2, + rerun=True, + ) + + samples = sample_dir.join("hello.mp4").listdir() + _validate_interval([Path(s) for s in samples], gps_start_time) + + +def test_sample_video_when_telemetry_fails( + tmpdir: py.path.local, setup_mock, monkeypatch, caplog +): + """A telemetry parser bug must not fail sampling, which did not need it.""" + root = _PWD.joinpath("data/mock_sample_video") + video_dir = root.joinpath("videos") + sample_dir = tmpdir.mkdir("sampled_video_frames") + + def raise_type_error(): + raise TypeError("'NoneType' object is not subscriptable") + + monkeypatch.setattr( + sample_video, + "NativeVideoExtractor", + lambda video_path: mock.Mock(extract=raise_type_error), + ) + + with caplog.at_level(logging.WARNING, logger=sample_video.LOG.name): + sample_video.sample_video( + video_dir, + Path(sample_dir), + video_sample_distance=-1, + video_sample_interval=2, + rerun=True, + ) + + 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 + ) + + +class TestGPSClockStartTime: + """Tests for _gps_clock_start_time.""" + + @staticmethod + def _gps_point(time: float, epoch_time: float | None) -> telemetry.GPSPoint: + return telemetry.GPSPoint( + time=time, + lat=40.0, + lon=-74.0, + alt=None, + angle=None, + epoch_time=epoch_time, + fix=None, + precision=None, + ground_speed=None, + ) + + def test_maps_first_timestamp_back_to_video_start(self) -> None: + # The first point is 2.5s into the video, so the video started 2.5s + # before that point was recorded + points = [self._gps_point(2.5, 1628599086.0)] + assert sample_video._gps_clock_start_time(points) == datetime.datetime( + 2021, 8, 10, 12, 38, 3, 500000, tzinfo=datetime.timezone.utc + ) + + def test_skips_points_without_a_timestamp(self) -> None: + points = [self._gps_point(0.0, None), self._gps_point(1.0, 1628599086.0)] + assert sample_video._gps_clock_start_time(points) == datetime.datetime( + 2021, 8, 10, 12, 38, 5, tzinfo=datetime.timezone.utc + ) + + def test_skips_unset_clock(self) -> None: + # GoPro reports 2000-01-01 until it gets its first fix + points = [ + self._gps_point(0.0, 946684800.0), + self._gps_point(1.0, 1628599086.0), + ] + assert sample_video._gps_clock_start_time(points) == datetime.datetime( + 2021, 8, 10, 12, 38, 5, tzinfo=datetime.timezone.utc + ) + + def test_skips_future_timestamps(self) -> None: + points = [ + self._gps_point(0.0, time.time() + 7 * 24 * 3600), + self._gps_point(1.0, 1628599086.0), + ] + assert sample_video._gps_clock_start_time(points) == datetime.datetime( + 2021, 8, 10, 12, 38, 5, tzinfo=datetime.timezone.utc + ) + + def test_skips_non_finite_timestamps(self) -> None: + points = [ + self._gps_point(float("nan"), 1628599086.0), + self._gps_point(0.0, float("inf")), + self._gps_point(0.0, 1e300), + self._gps_point(1.0, 1628599086.0), + ] + assert sample_video._gps_clock_start_time(points) == datetime.datetime( + 2021, 8, 10, 12, 38, 5, tzinfo=datetime.timezone.utc + ) + + def test_only_implausible_timestamps(self) -> None: + points = [self._gps_point(0.0, 946684800.0), self._gps_point(1.0, 1e300)] + assert sample_video._gps_clock_start_time(points) is None + + def test_median_ignores_a_bad_timestamp(self) -> None: + epoch_times = [1628599086.0 + i for i in range(5)] + epoch_times[0] += 30 + points = [self._gps_point(float(i), t) for i, t in enumerate(epoch_times)] + assert sample_video._gps_clock_start_time(points) == datetime.datetime( + 2021, 8, 10, 12, 38, 6, tzinfo=datetime.timezone.utc + ) + + def test_uses_the_start_of_a_timelapse(self) -> None: + # Each second of this timelapse spans 10 seconds of GPS time, so the + # offset between the two clocks grows by 9 seconds per point. Reading it + # at the start of the track keeps the error to a couple of points + # rather than half the track + points = [self._gps_point(float(i), 1628599086.0 + 10 * i) for i in range(1000)] + assert sample_video._gps_clock_start_time(points) == datetime.datetime( + 2021, 8, 10, 12, 38, 24, tzinfo=datetime.timezone.utc + ) + + def test_no_absolute_timestamps(self) -> None: + assert sample_video._gps_clock_start_time(_make_gps_points(3)) is None + + def test_no_points(self) -> None: + assert sample_video._gps_clock_start_time([]) is None + + +class TestCreationTimeToStartTime: + """Tests for _creation_time_to_start_time.""" + + # A 60-second video, so the start is a minute before this if the camera + # stamped the end + CREATION_TIME = datetime.datetime( + 2023, 3, 7, 1, 36, 34, tzinfo=datetime.timezone.utc + ) + START_TIME_IF_END_STAMPED = datetime.datetime( + 2023, 3, 7, 1, 35, 34, tzinfo=datetime.timezone.utc + ) + + @staticmethod + def _probe( + creation_time: str | None = "2023-03-07T01:36:34.000000Z", + duration: str | None = "60.0", + format_tags: dict[str, str] | None = None, + ) -> ffmpeglib.Probe: + stream: dict[str, T.Any] = { + "index": 0, + "codec_type": "video", + "width": 1920, + "height": 1080, + "tags": {}, + } + if creation_time is not None: + stream["tags"]["creation_time"] = creation_time + if duration is not None: + stream["duration"] = duration + return ffmpeglib.Probe( + T.cast( + ffmpeglib.ProbeOutput, + {"streams": [stream], "format": {"tags": format_tags or {}}}, + ) + ) + + @staticmethod + def _video(tmp_path: Path, name: str, data: bytes = b"") -> Path: + video_path = tmp_path / name + video_path.write_bytes(data) + return video_path + + def test_assumes_start_without_evidence(self, tmp_path: Path, caplog) -> None: + video_path = self._video(tmp_path, "clip.mp4") + with caplog.at_level(logging.WARNING, logger=sample_video.LOG.name): + start_time = sample_video._creation_time_to_start_time( + video_path, self._probe() + ) + assert start_time == self.CREATION_TIME + # Tells the user how to override it if the camera stamps the end + assert "--video_start_time 2023_03_07_01_35_34_000" in caplog.text + + def test_known_end_stamping_camera(self, tmp_path: Path) -> None: + video_path = self._video(tmp_path, "R0020627.MP4") + probe = self._probe(format_tags={"make": "RICOH", "model": "RICOH THETA X"}) + assert ( + sample_video._creation_time_to_start_time(video_path, probe) + == self.START_TIME_IF_END_STAMPED + ) + + def test_other_models_of_the_same_make(self, tmp_path: Path) -> None: + video_path = self._video(tmp_path, "R0010001.MP4") + probe = self._probe(format_tags={"make": "RICOH", "model": "RICOH THETA Z1"}) + assert ( + sample_video._creation_time_to_start_time(video_path, probe) + == self.CREATION_TIME + ) + + def test_blackvue_without_gps_fix(self, tmp_path: Path) -> None: + # BlackVue stamps the end. Without a fix there is no GPS clock to read + # the start from, but its GPS box is still there to identify it + box = { + "type": b"free", + "data": [ + {"type": b"gps ", "data": b"[1678152934000]$GPRMC,,V,,,,,,,,,,N*53"} + ], + } + data = cparser.Box32ConstructBuilder({b"free": {}}).Box.build(box) + video_path = self._video(tmp_path, "clip.mp4", data) + assert ( + sample_video._creation_time_to_start_time(video_path, self._probe()) + == self.START_TIME_IF_END_STAMPED + ) + + def test_file_name_matches_end(self, tmp_path: Path) -> None: + # Viofo names files by the start in local time (UTC+9 here), and stamps + # the creation time at the end. The two clocks can be seconds apart + for name in ["2023_0307_103534_0001F.MP4", "2023_0307_103544_0001F.MP4"]: + video_path = self._video(tmp_path, name) + assert ( + sample_video._creation_time_to_start_time(video_path, self._probe()) + == self.START_TIME_IF_END_STAMPED + ) + + def test_file_name_matches_end_with_naive_creation_time( + self, tmp_path: Path + ) -> None: + video_path = self._video(tmp_path, "2023_0307_103534_0001F.MP4") + probe = self._probe(creation_time="2023-03-07 01:36:34") + assert sample_video._creation_time_to_start_time( + video_path, probe + ) == datetime.datetime(2023, 3, 7, 1, 35, 34) + + def test_file_name_matches_start(self, tmp_path: Path) -> None: + # Insta360 names files a few seconds after it stamps the creation time + for name in ["VID_20230307_103634_00_001.mp4", "VID_20230307_103644.mp4"]: + video_path = self._video(tmp_path, name) + assert ( + sample_video._creation_time_to_start_time(video_path, self._probe()) + == self.CREATION_TIME + ) + + def test_file_name_matches_both(self, tmp_path: Path, caplog) -> None: + # A 15-minute video starts and ends at times that are both a whole + # time zone away from the file name + video_path = self._video(tmp_path, "20230307_103634.mp4") + with caplog.at_level(logging.WARNING, logger=sample_video.LOG.name): + start_time = sample_video._creation_time_to_start_time( + video_path, self._probe(duration="900.0") + ) + assert start_time == self.CREATION_TIME + assert "--video_start_time" in caplog.text + + def test_file_name_matches_neither(self, tmp_path: Path) -> None: + for name in ["20230307_104004.mp4", "20230309_103634.mp4"]: + video_path = self._video(tmp_path, name) + assert ( + sample_video._creation_time_to_start_time(video_path, self._probe()) + == self.CREATION_TIME + ) + + def test_file_name_with_invalid_date(self, tmp_path: Path) -> None: + video_path = self._video(tmp_path, "20231399_103534.mp4") + assert ( + sample_video._creation_time_to_start_time(video_path, self._probe()) + == self.CREATION_TIME + ) + + def test_no_creation_time(self, tmp_path: Path) -> None: + video_path = self._video(tmp_path, "clip.mp4") + probe = self._probe(creation_time=None) + assert sample_video._creation_time_to_start_time(video_path, probe) is None + + def test_no_duration(self, tmp_path: Path) -> None: + video_path = self._video(tmp_path, "2023_0307_103534_0001F.MP4") + probe = self._probe(duration=None) + assert ( + sample_video._creation_time_to_start_time(video_path, probe) + == self.CREATION_TIME + ) + + # --------------------------------------------------------------------------- # Helpers for distance-based sampling tests # --------------------------------------------------------------------------- @@ -130,12 +461,6 @@ def test_sample_video_with_start_time(tmpdir: py.path.local, setup_mock): MOCK_PROBE_JSON = _PWD / "data" / "mock_sample_video" / "videos" / "hello.mp4" TEST_EXIF_JPG = _PWD / "data" / "test_exif.jpg" -# Start time derived from the hello.mp4 probe fixture: -# creation_time "2021-08-10T14:38:06.000000Z" - duration "60.977000" -PROBE_START_TIME = datetime.datetime( - 2021, 8, 10, 14, 36, 55, 23000, tzinfo=datetime.timezone.utc -) - def _load_probe_output() -> ffmpeglib.ProbeOutput: with open(MOCK_PROBE_JSON) as fp: @@ -417,10 +742,13 @@ def _setup_mocks( tmp_path: Path, video_path: Path, num_gps_points: int = 10, + gps_points: T.Sequence[geo.Point] | None = None, ) -> dict[str, T.Any]: """Set up all the mocks needed for _sample_single_video_by_distance.""" probe_output = _load_probe_output() - gps_points = _make_gps_points(num_gps_points, time_step=1.0) + if gps_points is None: + gps_points = _make_gps_points(num_gps_points, time_step=1.0) + num_gps_points = len(gps_points) video_metadata = VideoMetadata( filename=video_path, @@ -608,3 +936,81 @@ def test_exif_lat_lon_written(self, tmp_path: Path) -> None: # First GPS point is at (40.0, -74.0) assert abs(lat - 40.0) < 0.01 assert abs(lon - (-74.0)) < 0.01 + + def test_timestamps_from_creation_time(self, tmp_path: Path) -> None: + """Without a GPS clock, frames are timestamped from the creation time.""" + video_dir = tmp_path / "videos" + video_dir.mkdir() + video_file = video_dir / "test.mp4" + video_file.touch() + output_dir = tmp_path / "output" + + mocks = self._setup_mocks(tmp_path, video_file) + + with ( + mocks["patches"]["ffmpeg_cls"], + mocks["patches"]["geotag_cls"], + mocks["patches"]["moov_parse"], + ): + sample_video.sample_video( + video_import_path=video_file, + import_path=output_dir, + video_sample_distance=0.0, + ) + + frames = sorted((output_dir / "test.mp4").glob("*.jpg")) + assert len(frames) == 10 + for idx, frame in enumerate(frames): + assert exif_read.ExifRead(frame).extract_capture_time() == ( + PROBE_START_TIME + datetime.timedelta(seconds=idx) + ) + + def test_timestamps_from_gps_clock(self, tmp_path: Path) -> None: + """Frames at points with their own timestamp do not need the creation time.""" + video_dir = tmp_path / "videos" + video_dir.mkdir() + video_file = video_dir / "test.mp4" + video_file.touch() + output_dir = tmp_path / "output" + + gps_start_time = datetime.datetime( + 2021, 8, 10, 6, 38, 6, tzinfo=datetime.timezone.utc + ) + gps_points = [ + telemetry.GPSPoint( + time=p.time, + lat=p.lat, + lon=p.lon, + alt=p.alt, + angle=p.angle, + epoch_time=gps_start_time.timestamp() + p.time, + fix=None, + precision=None, + ground_speed=None, + ) + for p in _make_gps_points(10, time_step=1.0) + ] + mocks = self._setup_mocks(tmp_path, video_file, gps_points=gps_points) + + with ( + mocks["patches"]["ffmpeg_cls"], + mocks["patches"]["geotag_cls"], + mocks["patches"]["moov_parse"], + mock.patch.object( + sample_video, + "_creation_time_to_start_time", + side_effect=AssertionError("creation time should not be read"), + ), + ): + sample_video.sample_video( + video_import_path=video_file, + import_path=output_dir, + video_sample_distance=0.0, + ) + + frames = sorted((output_dir / "test.mp4").glob("*.jpg")) + assert len(frames) == 10 + for idx, frame in enumerate(frames): + assert exif_read.ExifRead(frame).extract_capture_time() == ( + gps_start_time + datetime.timedelta(seconds=idx) + )