Stamp and geotag the samples of a timelapse by its GPS timestamps

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__(

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)