diff --git a/mapillary_tools/geotag/geotag_images_from_video.py b/mapillary_tools/geotag/geotag_images_from_video.py index c20d9f1a3..0adbd7fd9 100644 --- a/mapillary_tools/geotag/geotag_images_from_video.py +++ b/mapillary_tools/geotag/geotag_images_from_video.py @@ -5,6 +5,8 @@ from __future__ import annotations +import bisect +import dataclasses import logging import sys import typing as T @@ -15,15 +17,78 @@ else: from typing_extensions import override -from .. import types, utils +from .. import geo, types, utils from .base import GeotagImagesFromGeneric from .geotag_images_from_gpx import GeotagImagesFromGPX from .geotag_videos_from_video import GeotagVideosFromVideo +from .video_extractors.gpx import GPXVideoExtractor LOG = logging.getLogger(__name__) +@dataclasses.dataclass(frozen=True) +class GPSTimeline: + """ + The Unix time of the video's GPS clock along the video: interpolated + between its GPS timestamps, and extrapolated beyond them at the clock's + rate. A timelapse captured its frames at those times, which one rate from + one of them does not follow: the rate drifts from them over the video, and + the first timestamp of a GoPro MAX timelapse can be seconds early. + + >>> timeline = GPSTimeline((0.0, 1.0, 2.0), (100.0, 104.0, 114.0), rate=10.0) + >>> timeline.to_unix_time(0.5) + 102.0 + >>> timeline.to_unix_time(3.0) + 124.0 + """ + + video_times: tuple[float, ...] + unix_times: tuple[float, ...] + # GPS seconds per video second beyond the timestamps + rate: float + + @classmethod + def read(cls, points: T.Sequence[geo.Point], rate: float) -> GPSTimeline | None: + """ + Read the timeline from the GPS timestamps of the points, or None if they + have none + """ + pairs: list[tuple[float, float]] = [] + for p in points: + unix_time = p.get_unix_time() + if unix_time is None: + continue + if pairs and unix_time == pairs[-1][1]: + # GoPro GPS5 stamps a whole block (~1s) of points with one + # timestamp. Keep the first point of the block + continue + pairs.append((p.time, unix_time)) + + if not pairs: + return None + + return cls( + video_times=tuple(v for v, _ in pairs), + unix_times=tuple(u for _, u in pairs), + rate=rate, + ) + + def to_unix_time(self, video_time: float) -> float: + idx = bisect.bisect_right(self.video_times, video_time) + if 0 < idx < len(self.video_times): + v0, v1 = self.video_times[idx - 1], self.video_times[idx] + u0, u1 = self.unix_times[idx - 1], self.unix_times[idx] + return u0 + (u1 - u0) * (video_time - v0) / (v1 - v0) + + # Before the first timestamp or after the last + anchor = 0 if idx == 0 else -1 + return ( + self.unix_times[anchor] + + (video_time - self.video_times[anchor]) * self.rate + ) + + class GeotagImagesFromVideo(GeotagImagesFromGeneric): def __init__( self, @@ -74,7 +139,7 @@ def to_description( ) geotag = GeotagImagesFromGPX( - video_metadata.points, + self._on_capture_timeline(video_metadata.points), use_gpx_start_time=False, use_image_start_time=True, offset_time=self.offset_time, @@ -98,6 +163,28 @@ def to_description( return final_image_metadatas + @classmethod + def _on_capture_timeline( + cls, points: T.Sequence[geo.Point] + ) -> T.Sequence[geo.Point]: + """ + Time the video's GPS points in GPS seconds from the start of the video, + like the samples, which are stamped with their capture times. A + timelapse packs several into each second of video, as many as its GPS + timestamps tell. + """ + clock = GPXVideoExtractor._read_video_clock(points) + if clock is None or clock.rate == 1.0: + return points + + timeline = GPSTimeline.read(points, clock.rate) + assert timeline is not None, "a GPS clock is read from GPS timestamps" + start = timeline.to_unix_time(0.0) + return [ + dataclasses.replace(p, time=timeline.to_unix_time(p.time) - start) + for p in points + ] + class GeotagImageSamplesFromVideo(GeotagImagesFromGeneric): def __init__( diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py index e945d2526..d555f6844 100644 --- a/mapillary_tools/sample_video.py +++ b/mapillary_tools/sample_video.py @@ -29,7 +29,9 @@ ) from .exif_write import ExifEdit from .geotag import factory +from .geotag.geotag_images_from_video import GPSTimeline from .geotag.options import SourceOption, SourceType, VideoSyncOption +from .geotag.video_extractors.gpx import GPXVideoExtractor from .geotag.video_extractors.native import NativeVideoExtractor from .mp4 import mp4_sample_parser from .serializer.description import build_capture_time, parse_capture_time @@ -412,14 +414,44 @@ def _gps_clock_start_time( ) -def _telemetry_start_time(video_path: Path) -> datetime.datetime | None: +def _read_telemetry(video_path: Path) -> T.Sequence[geo.Point]: + """ + Read the GPS points of the video's own telemetry, or none if it has none + """ try: - video_metadata = NativeVideoExtractor(video_path).extract() + return NativeVideoExtractor(video_path).extract().points except exceptions.MapillaryDescriptionError as ex: - LOG.debug("No video telemetry to read the start time from: %s", ex) + LOG.debug("No video telemetry to read the clock from: %s", ex) + except Exception as ex: + # Telemetry is only one way to tell when the frames were captured, so + # a video whose telemetry fails to parse must still be sampled + LOG.warning( + "Unable to read the telemetry of %s: %s", + video_path.name, + ex, + exc_info=LOG.isEnabledFor(logging.DEBUG), + ) + return [] + + +def _telemetry_clock_rate( + video_path: Path, telemetry_points: T.Sequence[geo.Point] +) -> float | None: + """ + GPS seconds per video second of the video's own GPS clock: more than 1 if + the video is a timelapse, or None if it has no GPS clock + """ + clock = GPXVideoExtractor._read_video_clock(telemetry_points) + if clock is None: return None - return _gps_clock_start_time(video_metadata.points) + if clock.rate != 1.0: + LOG.info( + "The GPS clock in %s runs %.1f times faster than the video (timelapse)", + video_path.name, + clock.rate, + ) + return clock.rate def _parse_filename_time(video_path: Path) -> datetime.datetime | None: @@ -530,7 +562,9 @@ def _creation_time_to_start_time( def _extract_video_start_time( - video_path: Path, probe: ffmpeglib.Probe + video_path: Path, + probe: ffmpeglib.Probe, + telemetry_points: T.Sequence[geo.Point], ) -> datetime.datetime | None: """ Determine the wall clock time at which a video started recording. @@ -541,19 +575,7 @@ def _extract_video_start_time( back to the creation time when there is no telemetry to sync against, which is the case for the plain MP4s that get geotagged from a GPX. """ - try: - start_time = _telemetry_start_time(video_path) - except Exception as ex: - # Telemetry is only one way to find the start time, so a video whose - # telemetry fails to parse must still be sampled - LOG.warning( - "Unable to read the start time of %s from its telemetry: %s", - video_path.name, - ex, - exc_info=LOG.isEnabledFor(logging.DEBUG), - ) - start_time = None - + start_time = _gps_clock_start_time(telemetry_points) if start_time is not None: return start_time @@ -569,10 +591,41 @@ def _sample_single_video_by_interval( ) -> None: ffmpeg = ffmpeglib.FFMPEG(constants.FFMPEG_PATH, constants.FFPROBE_PATH) + telemetry_points = _read_telemetry(video_path) + + # Frames are stamped with their capture times, which a timelapse packs + # several seconds of into each second of video: as the video's own GPS + # timestamps tell, and for a video without a GPS clock, at + # --video_duration_ratio + clock_rate = _telemetry_clock_rate(video_path, telemetry_points) + timeline: GPSTimeline | None = None + if clock_rate is None: + rate = duration_ratio + else: + rate = clock_rate + if duration_ratio != 1.0: + LOG.warning( + "Ignoring --video_duration_ratio %g for %s, whose own GPS clock runs at %.3gx", + duration_ratio, + video_path.name, + clock_rate, + ) + if clock_rate != 1.0: + # As distance samples are stamped, and as the geotagging stage + # times the video's track to place the samples along it + timeline = GPSTimeline.read(telemetry_points, clock_rate) + if start_time is None: - start_time = _extract_video_start_time( - video_path, ffmpeglib.Probe(ffmpeg.probe_format_and_streams(video_path)) - ) + if timeline is not None: + start_time = datetime.datetime.fromtimestamp( + timeline.to_unix_time(0.0), tz=datetime.timezone.utc + ) + else: + start_time = _extract_video_start_time( + video_path, + ffmpeglib.Probe(ffmpeg.probe_format_and_streams(video_path)), + telemetry_points, + ) if start_time is None: raise exceptions.MapillaryVideoError( f"Unable to extract video start time from {video_path}" @@ -586,7 +639,11 @@ def _sample_single_video_by_interval( if sample_paths[0] is None: continue # extract_frames() produces 1-based frame indices so we need to subtract 1 here - seconds = (frame_idx_1based - 1) * sample_interval * duration_ratio + video_time = (frame_idx_1based - 1) * sample_interval + if timeline is None: + seconds = video_time * rate + else: + seconds = timeline.to_unix_time(video_time) - timeline.to_unix_time(0.0) timestamp = start_time + datetime.timedelta(seconds=seconds) exif_edit = ExifEdit(sample_paths[0]) exif_edit.add_date_time_original(timestamp) diff --git a/tests/unit/test_geotag_images_from_video.py b/tests/unit/test_geotag_images_from_video.py new file mode 100644 index 000000000..9664ed464 --- /dev/null +++ b/tests/unit/test_geotag_images_from_video.py @@ -0,0 +1,126 @@ +# Copyright (c) Meta Platforms, Inc. and affiliates. +# +# This source code is licensed under the BSD license found in the +# LICENSE file in the root directory of this source tree. + +from __future__ import annotations + +import datetime +import os +import shutil +import typing as T +from pathlib import Path + +import pytest +from mapillary_tools import geo, telemetry, types +from mapillary_tools.exif_write import ExifEdit +from mapillary_tools.geotag.geotag_images_from_video import GeotagImagesFromVideo + +_PWD = Path(os.path.dirname(os.path.abspath(__file__))) + +START_TIME = datetime.datetime(2021, 8, 10, 6, 38, 6, tzinfo=datetime.timezone.utc) + + +def _video_metadata( + video_path: Path, gps_seconds: T.Sequence[float] +) -> types.VideoMetadata: + # One point per second of video, at gps_seconds[i] from START_TIME + points = [ + telemetry.GPSPoint( + time=float(i), + lat=40.0 + i * 0.0001, + lon=-74.0, + alt=None, + angle=None, + epoch_time=START_TIME.timestamp() + seconds, + fix=None, + precision=None, + ground_speed=None, + ) + for i, seconds in enumerate(gps_seconds) + ] + return types.VideoMetadata( + filename=video_path, + filetype=types.FileType.GOPRO, + points=T.cast(T.List[geo.Point], points), + ) + + +def _sample(sample_dir: Path, idx: int, capture_time: datetime.datetime) -> Path: + sample_path = sample_dir / f"hello_v_{idx:06d}.jpg" + shutil.copyfile(_PWD / "data/test_exif.jpg", sample_path) + exif_edit = ExifEdit(sample_path) + exif_edit.add_date_time_original(capture_time) + exif_edit.add_gps_datetime(capture_time) + exif_edit.write() + return sample_path + + +@pytest.mark.parametrize("rate", [1.0, 10.0]) +def test_samples_are_placed_by_their_capture_times(tmp_path: Path, rate: float): + video_path = tmp_path / "hello.mp4" + sample_dir = tmp_path / "samples" / "hello.mp4" + sample_dir.mkdir(parents=True) + + # A sample every 2 seconds of video, stamped with its capture time, which + # a timelapse spaces **rate** times as far apart + samples = [ + _sample( + sample_dir, k + 1, START_TIME + datetime.timedelta(seconds=2 * k * rate) + ) + for k in range(4) + ] + + metadatas = GeotagImagesFromVideo( + [_video_metadata(video_path, [rate * i for i in range(10)])], num_processes=0 + ).to_description(samples) + + lat_by_name = {} + for metadata in metadatas: + assert isinstance(metadata, types.ImageMetadata), metadata + lat_by_name[metadata.filename.name] = metadata.lat + assert lat_by_name == { + sample.name: pytest.approx(40.0 + 2 * k * 0.0001) + for k, sample in enumerate(samples) + } + + +def test_timelapse_samples_are_placed_by_the_gps_timestamps(tmp_path: Path): + """ + Distance samples are stamped with the GPS time of their video time. A + GoPro MAX timelapse can stamp its first point seconds early, and its clock + keeps no one rate, so timed at one rate the track placed the samples off, + and the last ones past its end. + """ + video_path = tmp_path / "hello.mp4" + sample_dir = tmp_path / "samples" / "hello.mp4" + sample_dir.mkdir(parents=True) + + # 10x, but the first timestamp 3 seconds early and the clock speeding up + gps_seconds = [0.0] + [3 + 10 * i + 0.1 * i * i for i in range(1, 10)] + + def stamped(video_time: float) -> datetime.datetime: + i = int(video_time) + seconds = gps_seconds[i] + if i + 1 < len(gps_seconds): + seconds += (gps_seconds[i + 1] - gps_seconds[i]) * (video_time - i) + return START_TIME + datetime.timedelta(seconds=seconds) + + video_times = [0.0, 0.5, 2.0, 4.5, 8.0, 9.0] + samples = [ + _sample(sample_dir, k + 1, stamped(video_time)) + for k, video_time in enumerate(video_times) + ] + + metadatas = GeotagImagesFromVideo( + [_video_metadata(video_path, gps_seconds)], num_processes=0 + ).to_description(samples) + + lat_by_name = {} + for metadata in metadatas: + assert isinstance(metadata, types.ImageMetadata), metadata + lat_by_name[metadata.filename.name] = metadata.lat + assert lat_by_name == { + sample.name: pytest.approx(40.0 + video_time * 0.0001) + for sample, video_time in zip(samples, video_times) + } diff --git a/tests/unit/test_sample_video.py b/tests/unit/test_sample_video.py index cf4760b73..fd646f3e9 100644 --- a/tests/unit/test_sample_video.py +++ b/tests/unit/test_sample_video.py @@ -74,12 +74,16 @@ def setup_mock(monkeypatch): monkeypatch.setattr(ffmpeglib, "FFMPEG", MOCK_FFMPEG) -def _validate_interval(samples: T.Sequence[Path], video_start_time): +def _validate_interval( + samples: T.Sequence[Path], video_start_time, seconds_per_sample: float = 2 +): assert len(samples), "expect samples but got none" for idx, sample in enumerate(sorted(samples)): assert sample.name == f"hello_v_{idx + 1:06d}.jpg" exif = exif_read.ExifRead(sample) - expected_dt = video_start_time + datetime.timedelta(seconds=2 * idx) + expected_dt = video_start_time + datetime.timedelta( + seconds=seconds_per_sample * idx + ) assert exif.extract_capture_time() == expected_dt @@ -208,10 +212,129 @@ def raise_type_error(): samples = sample_dir.join("hello.mp4").listdir() _validate_interval([Path(s) for s in samples], PROBE_START_TIME) - assert "Unable to read the start time of hello.mp4 from its telemetry" in ( - caplog.text + assert "Unable to read the telemetry of hello.mp4" in caplog.text + + +class TestSampleTimelapseByInterval: + """ + Interval samples of a timelapse are stamped with their capture times, which + its GPS clock spaces 10 times as far apart as the frames + """ + + GPS_START_TIME = datetime.datetime( + 2021, 8, 10, 6, 38, 6, tzinfo=datetime.timezone.utc ) + @classmethod + def _mock_telemetry(cls, monkeypatch, gps_seconds: T.Callable[[int], float]): + # One point per second of video, at gps_seconds(i) from GPS_START_TIME + points = [ + telemetry.GPSPoint( + time=float(i), + lat=40.0 + i * 0.001, + lon=-74.0, + alt=None, + angle=None, + epoch_time=cls.GPS_START_TIME.timestamp() + gps_seconds(i), + fix=None, + precision=None, + ground_speed=None, + ) + for i in range(60) + ] + monkeypatch.setattr( + sample_video, + "NativeVideoExtractor", + lambda video_path: mock.Mock( + extract=lambda: VideoMetadata( + filename=video_path, + filetype=FileType.GOPRO, + points=T.cast(T.List[geo.Point], points), + ) + ), + ) + + @pytest.fixture + def timelapse_telemetry(self, monkeypatch): + # Recorded 10 seconds apart + self._mock_telemetry(monkeypatch, lambda i: 10 * i) + + def _sample(self, tmpdir: py.path.local, **kwargs) -> list[Path]: + root = _PWD.joinpath("data/mock_sample_video") + sample_dir = tmpdir.mkdir("sampled_video_frames") + sample_video.sample_video( + root.joinpath("videos"), + Path(sample_dir), + video_sample_distance=-1, + video_sample_interval=2, + rerun=True, + **kwargs, + ) + return [Path(s) for s in sample_dir.join("hello.mp4").listdir()] + + def test_capture_times_follow_the_gps_clock( + self, tmpdir, setup_mock, timelapse_telemetry, caplog + ): + _validate_interval( + self._sample(tmpdir), self.GPS_START_TIME, seconds_per_sample=20 + ) + assert "--video_duration_ratio" not in caplog.text + + def test_capture_times_follow_the_gps_timestamps( + self, tmpdir, setup_mock, monkeypatch + ): + """ + Not one rate: the GPS clock of a timelapse drifts from it, and distance + samples are stamped from the timestamps. The geotagging stage places + both kinds of samples along a track timed by them + """ + self._mock_telemetry(monkeypatch, lambda i: 10 * i + 0.1 * i * i) + + samples = sorted(self._sample(tmpdir)) + + assert len(samples) > 2 + for idx, sample in enumerate(samples): + video_time = 2 * idx + assert exif_read.ExifRead( + sample + ).extract_capture_time() == self.GPS_START_TIME + datetime.timedelta( + seconds=10 * video_time + 0.1 * video_time * video_time + ) + + def test_from_the_given_start_time(self, tmpdir, setup_mock, timelapse_telemetry): + video_start_time = "2020_08_10_14_37_05_023" + _validate_interval( + self._sample(tmpdir, video_start_time=video_start_time), + description.parse_capture_time(video_start_time), + seconds_per_sample=20, + ) + + def test_the_gps_clock_wins_over_the_duration_ratio( + self, tmpdir, setup_mock, timelapse_telemetry, caplog + ): + """ + The ratio is for the videos without a GPS clock, which a batch of + videos may mix with videos that have one + """ + _validate_interval( + self._sample(tmpdir, video_duration_ratio=5), + self.GPS_START_TIME, + seconds_per_sample=20, + ) + assert ( + "Ignoring --video_duration_ratio 5 for hello.mp4, whose own GPS clock runs at 10x" + in caplog.text + ) + + def test_the_duration_ratio_paces_a_video_without_a_gps_clock( + self, tmpdir, setup_mock + ): + _validate_interval( + self._sample(tmpdir, video_duration_ratio=5), + PROBE_START_TIME, + seconds_per_sample=10, + ) + class TestGPSClockStartTime: """Tests for _gps_clock_start_time."""