Fix GPS epoch vs Unix epoch mix-up when geotagging CAMM videos

mapillary_tools/camm/camm_builder.py

@@ -28,6 +28,10 @@ def _build_camm_sample(measurement: camm_parser.TelemetryMeasurement) -> bytes:
     raise ValueError(f"Unsupported measurement type {type(measurement)}")
 
 
+INT32_MIN = -(2**31)
+INT32_MAX = 2**31 - 1
+
+
 def _create_edit_list_from_points(
     tracks: T.Sequence[T.Sequence[geo.Point]],
     movie_timescale: int,
@@ -68,9 +72,21 @@ def _create_edit_list_from_points(
                 }
             )
 
+    # A version 0 elst stores these as 32-bit signed integers. Fall back to
+    # version 1 (64-bit) rather than letting the build fail on overflow.
+    version = 0
+    for entry in entries:
+        if not (
+            INT32_MIN <= entry["segment_duration"] <= INT32_MAX
+            and INT32_MIN <= entry["media_time"] <= INT32_MAX
+        ):
+            version = 1
+            break
+
     return {
         "type": b"elst",
         "data": {
+            "version": version,
             "entries": entries,
         },
     }

mapillary_tools/camm/camm_parser.py

@@ -116,10 +116,12 @@ def extract_camm_info(fp: T.BinaryIO, telemetry_only: bool = False) -> CAMMInfo
                 gps: list[telemetry.CAMMGPSPoint] = []
 
                 for measurement in measurements:
-                    if isinstance(measurement, geo.Point):
-                        mini_gps.append(measurement)
-                    elif isinstance(measurement, telemetry.CAMMGPSPoint):
+                    # NOTE: CAMMGPSPoint is a subclass of geo.Point, so it has
+                    # to be tested first or every GPS point ends up in mini_gps
+                    if isinstance(measurement, telemetry.CAMMGPSPoint):
                         gps.append(measurement)
+                    elif isinstance(measurement, geo.Point):
+                        mini_gps.append(measurement)
 
                 return CAMMInfo(mini_gps=mini_gps, gps=gps, make=make, model=model)
 

mapillary_tools/geo.py

@@ -39,7 +39,17 @@ class Point:
 
     def get_gps_epoch_time(self) -> float | None:
         """
-        Return the GPS epoch time for this point.
+        Return the time of this point in seconds since the GPS epoch
+        (1980-01-06), i.e. GPS time.
+        Base Point class returns None, subclasses can override.
+        """
+        return None
+
+    def get_unix_time(self) -> float | None:
+        """
+        Return the time of this point in Unix time (seconds since 1970-01-01,
+        UTC). This is the canonical wall clock accessor -- prefer it over
+        get_gps_epoch_time() everywhere except when serializing CAMM.
         Base Point class returns None, subclasses can override.
         """
         return None
@@ -100,7 +110,7 @@ def gps_distance(latlon_1: tuple[float, float], latlon_2: tuple[float, float]) -
 def avg_speed(sequence: T.Sequence[PointLike]) -> float:
     """
     Calculate average speed over a sequence of points.
-    Uses GPS epoch time when available (via get_gps_epoch_time()),
+    Uses Unix time when available (via get_unix_time()),
     otherwise falls back to the time field.
     Returns 0.0 for empty or single-element sequences.
     Returns NaN if time difference is zero (undefined speed).
@@ -116,14 +126,14 @@ def avg_speed(sequence: T.Sequence[PointLike]) -> float:
     first = sequence[0]
     last = sequence[-1]
 
-    # Try to use GPS epoch time if available (via polymorphic method)
-    first_gps_time = first.get_gps_epoch_time()
-    last_gps_time = last.get_gps_epoch_time()
+    # Try to use Unix time if available (via polymorphic method)
+    first_unix_time = first.get_unix_time()
+    last_unix_time = last.get_unix_time()
 
-    if first_gps_time is not None and last_gps_time is not None:
-        time_diff = last_gps_time - first_gps_time
+    if first_unix_time is not None and last_unix_time is not None:
+        time_diff = last_unix_time - first_unix_time
     else:
-        # Fall back to time field if GPS epoch time not available
+        # Fall back to time field if Unix time not available
         time_diff = last.time - first.time
 
     if time_diff == 0.0:

mapillary_tools/geotag/options.py

@@ -67,7 +67,7 @@ def from_dict(cls, data: dict[str, T.Any]) -> SourceOption:
             elif k == "source_path":
                 kwargs.setdefault(
                     "source_path", SourcePathOption(source_path=Path(v))
-                ).sourthe_path = Path(v)
+                ).source_path = Path(v)
             elif k == "pattern":
                 kwargs.setdefault(
                     "source_path", SourcePathOption(pattern=v)

mapillary_tools/geotag/utils.py

@@ -37,7 +37,8 @@ def parse_gpx(gpx_file: Path) -> list[Track]:
                             lon=point.longitude,
                             alt=point.elevation,
                             angle=None,
-                            time_gps_epoch=unix_time,
+                            # GPX timestamps are UTC; time_gps_epoch is GPS time
+                            time_gps_epoch=telemetry.unix_to_gps_epoch(unix_time),
                             gps_fix_type=3 if point.elevation is not None else 2,
                             horizontal_accuracy=0.0,
                             vertical_accuracy=0.0,

mapillary_tools/geotag/video_extractors/gpx.py

@@ -17,14 +17,19 @@
 else:
     from typing_extensions import override
 
-from ... import exceptions, geo, telemetry, types, utils
+from ... import exceptions, geo, types, utils
 from ..utils import parse_gpx
 from .base import BaseVideoExtractor
 from .native import NativeVideoExtractor
 
 
 LOG = logging.getLogger(__name__)
 
+# A GPX track and the video it is synced against should overlap in time. Warn
+# above a day, which no legitimate pairing needs and an epoch mix-up exceeds by
+# orders of magnitude.
+_IMPLAUSIBLE_OFFSET_SECONDS = 24 * 3600
+
 
 class SyncMode(enum.Enum):
     # Sync by video GPS timestamps if found, otherwise rebase
@@ -73,6 +78,15 @@ def extract(self) -> types.VideoMetadata:
             self._rebase_times(gpx_points)
         else:
             offset = self._gpx_offset(gpx_points, native_video_metadata.points)
+            if abs(offset) > _IMPLAUSIBLE_OFFSET_SECONDS:
+                LOG.warning(
+                    "Syncing %s against %s requires an offset of %.0f seconds (%.1f days). "
+                    "The GPX file probably does not belong to this video",
+                    self.video_path,
+                    self.gpx_path,
+                    offset,
+                    offset / 86400,
+                )
             self._rebase_times(gpx_points, offset=offset)
 
         return dataclasses.replace(native_video_metadata, points=gpx_points)
@@ -107,16 +121,13 @@ def _gpx_offset(
         if not gpx_points or not video_gps_points:
             return offset
 
-        gps_epoch_time: float | None = None
-        gps_point = video_gps_points[0]
-        if isinstance(gps_point, telemetry.GPSPoint):
-            if gps_point.epoch_time is not None:
-                gps_epoch_time = gps_point.epoch_time
-        elif isinstance(gps_point, telemetry.CAMMGPSPoint):
-            if gps_point.time_gps_epoch is not None:
-                gps_epoch_time = gps_point.time_gps_epoch
-
-        if gps_epoch_time is not None:
-            offset = gpx_points[0].time - gps_epoch_time
+        # Both sides must be Unix time here. Video GPS timestamps are stored in
+        # whatever epoch their container uses (CAMM records GPS time, GoPro
+        # records Unix time), so go through get_unix_time() rather than reading
+        # the raw attributes -- that also skips zero/invalid timestamps.
+        video_unix_time = video_gps_points[0].get_unix_time()
+
+        if video_unix_time is not None:
+            offset = gpx_points[0].time - video_unix_time
 
         return offset

mapillary_tools/sample_video.py

@@ -354,11 +354,11 @@ def _sample_single_video_by_distance(
                 f"interpolated time {interp.time} should match the video sample time {video_sample.exact_composition_time}"
             )
 
-            # Try to use GPS epoch time if available (for timelapse videos)
-            gps_epoch_time = interp.get_gps_epoch_time()
-            if gps_epoch_time is not None:
+            # Try to use the GPS timestamp if available (for timelapse videos)
+            gps_unix_time = interp.get_unix_time()
+            if gps_unix_time is not None:
                 timestamp = datetime.datetime.fromtimestamp(
-                    gps_epoch_time, tz=datetime.timezone.utc
+                    gps_unix_time, tz=datetime.timezone.utc
                 )
             else:
                 timestamp = start_time + datetime.timedelta(seconds=interp.time)

mapillary_tools/serializer/description.py

@@ -196,7 +196,7 @@ class ErrorDescription(TypedDict, total=False):
                     },
                     {
                         "type": ["number", "null"],
-                        "description": "GPS epoch time of the track point, in seconds. If present, used as the authoritative timestamp",
+                        "description": "Unix time (UTC) of the track point, in seconds. If present, used as the authoritative timestamp",
                     },
                 ],
             },
@@ -517,14 +517,14 @@ def encode(cls, p: geo.Point) -> T.Sequence[float | int | None]:
             round(p.lat, _COORDINATES_PRECISION),
             round(p.alt, _ALTITUDE_PRECISION) if p.alt is not None else None,
             round(p.angle, _ANGLE_PRECISION) if p.angle is not None else None,
-            p.get_gps_epoch_time(),
+            p.get_unix_time(),
         ]
         return entry
 
     @classmethod
     def decode(cls, entry: T.Sequence[T.Any]) -> geo.Point:
         if len(entry) >= 6 and entry[5] is not None:
-            time_ms, lon, lat, alt, angle, time_gps_epoch = (
+            time_ms, lon, lat, alt, angle, unix_time = (
                 entry[0],
                 entry[1],
                 entry[2],
@@ -538,7 +538,7 @@ def decode(cls, entry: T.Sequence[T.Any]) -> geo.Point:
                 lon=lon,
                 alt=alt,
                 angle=angle,
-                time_gps_epoch=time_gps_epoch,
+                time_gps_epoch=telemetry.unix_to_gps_epoch(unix_time),
                 gps_fix_type=3 if alt is not None else 2,
                 horizontal_accuracy=0.0,
                 vertical_accuracy=0.0,

mapillary_tools/serializer/gpx.py

@@ -77,14 +77,12 @@ def as_gpx_point(cls, point: geo.Point) -> gpxpy.gpx.GPXTrackPoint:
 
         if isinstance(point, types.ImageMetadata):
             gpx_point.name = point.filename.name
-        elif isinstance(point, CAMMGPSPoint):
-            gpx_point.time = datetime.datetime.fromtimestamp(
-                point.time_gps_epoch, datetime.timezone.utc
-            )
-        elif isinstance(point, GPSPoint):
-            if point.epoch_time is not None:
+        elif isinstance(point, (CAMMGPSPoint, GPSPoint)):
+            # GPX timestamps are UTC, so normalize whatever epoch the point uses
+            unix_time = point.get_unix_time()
+            if unix_time is not None:
                 gpx_point.time = datetime.datetime.fromtimestamp(
-                    point.epoch_time, datetime.timezone.utc
+                    unix_time, datetime.timezone.utc
                 )
 
         return gpx_point

mapillary_tools/telemetry.py

@@ -6,12 +6,83 @@
 # pyre-ignore-all-errors[16]
 from __future__ import annotations
 
+import bisect
+import calendar
 import dataclasses
 from enum import Enum, unique
 
 from .geo import Point
 
 
+# Seconds between the Unix epoch (1970-01-01) and the GPS epoch (1980-01-06).
+GPS_EPOCH_UNIX_OFFSET = 315964800
+
+# UTC dates on which a leap second took effect since the GPS epoch. GPS time is
+# a continuous scale that does not count leap seconds, so converting it to UTC
+# requires subtracting however many have accumulated. There has been no leap
+# second since 2017-01-01 (GPS - UTC = 18s); append here if one is announced.
+_LEAP_SECOND_UTC_DATES: tuple[tuple[int, int, int], ...] = (
+    (1981, 7, 1),
+    (1982, 7, 1),
+    (1983, 7, 1),
+    (1985, 7, 1),
+    (1988, 1, 1),
+    (1990, 1, 1),
+    (1991, 1, 1),
+    (1992, 7, 1),
+    (1993, 7, 1),
+    (1994, 7, 1),
+    (1996, 1, 1),
+    (1997, 7, 1),
+    (1999, 1, 1),
+    (2006, 1, 1),
+    (2009, 1, 1),
+    (2012, 7, 1),
+    (2015, 7, 1),
+    (2017, 1, 1),
+)
+
+_LEAP_SECOND_UNIX_TIMES: tuple[int, ...] = tuple(
+    calendar.timegm((year, month, day, 0, 0, 0))
+    for year, month, day in _LEAP_SECOND_UTC_DATES
+)
+
+
+def _gps_utc_offset_at(unix_time: float) -> int:
+    """
+    Number of leap seconds GPS time is ahead of UTC at the given Unix time.
+
+    >>> _gps_utc_offset_at(0)  # before the GPS epoch
+    0
+    >>> _gps_utc_offset_at(1786523187)  # 2026
+    18
+    """
+    return bisect.bisect_right(_LEAP_SECOND_UNIX_TIMES, unix_time)
+
+
+def gps_epoch_to_unix(gps_epoch_time: float) -> float:
+    """
+    Convert seconds since the GPS epoch (GPS time) to Unix time (UTC).
+
+    >>> gps_epoch_to_unix(1470558405.9798455)
+    1786523187.9798455
+    """
+    # The leap-second lookup is done on the uncorrected value. That is only
+    # ambiguous for instants within ~18s of a leap-second boundary.
+    approx_unix_time = gps_epoch_time + GPS_EPOCH_UNIX_OFFSET
+    return approx_unix_time - _gps_utc_offset_at(approx_unix_time)
+
+
+def unix_to_gps_epoch(unix_time: float) -> float:
+    """
+    Convert Unix time (UTC) to seconds since the GPS epoch (GPS time).
+
+    >>> unix_to_gps_epoch(1786523187.9798455)
+    1470558405.9798455
+    """
+    return unix_time - GPS_EPOCH_UNIX_OFFSET + _gps_utc_offset_at(unix_time)
+
+
 @unique
 class GPSFix(Enum):
     NO_FIX = 0
@@ -33,17 +104,25 @@ class TimestampedMeasurement:
 
 @dataclasses.dataclass
 class GPSPoint(TimestampedMeasurement, Point):
+    # Unix time (UTC), NOT seconds since the GPS epoch
     epoch_time: float | None
     fix: GPSFix | None
     precision: float | None
     ground_speed: float | None
 
-    def get_gps_epoch_time(self) -> float | None:
-        """Return the GPS epoch time if valid, otherwise None."""
+    def get_unix_time(self) -> float | None:
+        """Return the Unix time if valid, otherwise None."""
         if self.epoch_time is not None and self.epoch_time > 0:
             return self.epoch_time
         return None
 
+    def get_gps_epoch_time(self) -> float | None:
+        """Return the GPS epoch time if valid, otherwise None."""
+        unix_time = self.get_unix_time()
+        if unix_time is None:
+            return None
+        return unix_to_gps_epoch(unix_time)
+
     def interpolate_with(self, other: Point, t: float) -> Point:
         """Create a new interpolated GPSPoint using this and other point at time t."""
         base = super().interpolate_with(other, t)
@@ -92,6 +171,8 @@ def interpolate_with(self, other: Point, t: float) -> Point:
 
 @dataclasses.dataclass
 class CAMMGPSPoint(TimestampedMeasurement, Point):
+    # Seconds since the GPS epoch (GPS time), as defined by the CAMM spec.
+    # NOT Unix time -- use get_unix_time() to get a wall clock timestamp.
     time_gps_epoch: float
     gps_fix_type: int
     horizontal_accuracy: float
@@ -107,6 +188,13 @@ def get_gps_epoch_time(self) -> float | None:
             return self.time_gps_epoch
         return None
 
+    def get_unix_time(self) -> float | None:
+        """Return the Unix time if valid, otherwise None."""
+        gps_epoch_time = self.get_gps_epoch_time()
+        if gps_epoch_time is None:
+            return None
+        return gps_epoch_to_unix(gps_epoch_time)
+
     def interpolate_with(self, other: Point, t: float) -> Point:
         """Create a new interpolated CAMMGPSPoint using this and other point at time t."""
         base = super().interpolate_with(other, t)