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,16 +116,76 @@ 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)
+
+                _normalize_gps_epochs(gps, make, moov)
 
                 return CAMMInfo(mini_gps=mini_gps, gps=gps, make=make, model=model)
 
     return None
 
 
+# Makes whose CAMM type 6 samples record GPS time (seconds since 1980-01-06),
+# as the CAMM spec describes. Everything else -- Insta360, and the CAMM tracks
+# mapillary_tools writes itself -- records Unix time in the same field, so
+# converting unconditionally would push those ~10 years into the future.
+_GPS_EPOCH_MAKES = frozenset(["labpano"])
+
+# Seconds between the mp4 epoch (1904-01-01) and the Unix epoch.
+_MP4_EPOCH_UNIX_OFFSET = 2082844800
+
+# Tolerance for recognizing a gap as "off by exactly one GPS epoch". Checking
+# for that specific distance rather than for general implausibility matters:
+# some cameras write a meaningless mvhd creation_time (a GoPro HERO7 recorded
+# in 2022 reports 2016), so a generic bound would fire constantly.
+_GPS_EPOCH_GAP_TOLERANCE = 30 * 24 * 3600
+
+
+def _normalize_gps_epochs(
+    gps: list[telemetry.CAMMGPSPoint], make: str, moov: MovieBoxParser | None = None
+) -> None:
+    """
+    Rewrite CAMMGPSPoint.epoch_time in place so it is Unix time regardless of
+    which epoch the producer used.
+
+    This is the only place CAMM GPS timestamps change epoch. Everything
+    downstream, including the serializer, treats them as Unix time.
+    """
+    if not gps:
+        return
+
+    if make.strip().lower() in _GPS_EPOCH_MAKES:
+        for point in gps:
+            if point.epoch_time > 0:
+                point.epoch_time = telemetry.gps_epoch_to_unix(point.epoch_time)
+
+    first = next((p.epoch_time for p in gps if p.epoch_time > 0), None)
+    if first is None or moov is None:
+        return
+
+    try:
+        creation_time = moov.extract_mvhd_boxdata().get("creation_time", 0)
+    except Exception:
+        return
+
+    if not creation_time:
+        return
+
+    gap = abs(first - (creation_time - _MP4_EPOCH_UNIX_OFFSET))
+    if abs(gap - telemetry.GPS_EPOCH_UNIX_OFFSET) < _GPS_EPOCH_GAP_TOLERANCE:
+        LOG.warning(
+            "CAMM GPS timestamps are one GPS epoch away from the creation time "
+            "of the video. The camera (make %r) may record GPS time where Unix "
+            "time is expected, or the reverse; please report this video",
+            make,
+        )
+
+
 def extract_camera_make_and_model(fp: T.BinaryIO) -> tuple[str, str]:
     moov = MovieBoxParser.parse_stream(fp)
     udta_boxdata = moov.extract_udta_boxdata()
@@ -225,7 +285,9 @@ def deserialize(cls, sample: Sample, data: T.Any) -> telemetry.CAMMGPSPoint:
             lon=data.longitude,
             alt=data.altitude,
             angle=None,
-            time_gps_epoch=data.time_gps_epoch,
+            # Raw, still in whatever epoch the producer used. Normalized to
+            # Unix time by _normalize_gps_epochs() once the make is known.
+            epoch_time=data.time_gps_epoch,
             gps_fix_type=data.gps_fix_type,
             horizontal_accuracy=data.horizontal_accuracy,
             vertical_accuracy=data.vertical_accuracy,
@@ -243,7 +305,10 @@ def serialize(cls, data: telemetry.CAMMGPSPoint) -> bytes:
             {
                 "type": cls.serialized_camm_type.value,
                 "data": {
-                    "time_gps_epoch": data.time_gps_epoch,
+                    # Written as Unix time, which is what every released
+                    # version of mapillary_tools has written and what readers
+                    # of our output expect. Do not convert here.
+                    "time_gps_epoch": data.epoch_time,
                     "gps_fix_type": data.gps_fix_type,
                     "latitude": data.lat,
                     "longitude": data.lon,

mapillary_tools/geo.py

@@ -37,9 +37,10 @@ class Point:
     alt: float | None
     angle: float | None
 
-    def get_gps_epoch_time(self) -> float | None:
+    def get_unix_time(self) -> float | None:
         """
-        Return the GPS epoch time for this point.
+        Return the absolute time of this point in Unix time (seconds since
+        1970-01-01, UTC), or None if the point carries no absolute timestamp.
         Base Point class returns None, subclasses can override.
         """
         return None
@@ -100,7 +101,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 +117,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,7 @@ def parse_gpx(gpx_file: Path) -> list[Track]:
                             lon=point.longitude,
                             alt=point.elevation,
                             angle=None,
-                            time_gps_epoch=unix_time,
+                            epoch_time=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,
+                epoch_time=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