Preserve GPS timestamps from GPX tracks in video CAMM output

mapillary_tools/geotag/utils.py

@@ -12,11 +12,15 @@
 
 import gpxpy
 
-from .. import exiftool_read, geo, utils
+from .. import exiftool_read, geo, telemetry, utils
 
 Track = T.List[geo.Point]
 LOG = logging.getLogger(__name__)
 
+# GPS epoch start: January 6, 1980 (Unix timestamp).
+# Any timestamp below this is not a valid GPS time.
+_MIN_GPS_EPOCH_TIME = 315964800.0
+
 
 def parse_gpx(gpx_file: Path) -> list[Track]:
     with gpx_file.open("r") as f:
@@ -29,15 +33,35 @@ def parse_gpx(gpx_file: Path) -> list[Track]:
             tracks.append([])
             for point in segment.points:
                 if point.time is not None:
-                    tracks[-1].append(
-                        geo.Point(
-                            time=geo.as_unix_time(point.time),
-                            lat=point.latitude,
-                            lon=point.longitude,
-                            alt=point.elevation,
-                            angle=None,
+                    unix_time = geo.as_unix_time(point.time)
+                    if unix_time >= _MIN_GPS_EPOCH_TIME:
+                        tracks[-1].append(
+                            telemetry.CAMMGPSPoint(
+                                time=unix_time,
+                                lat=point.latitude,
+                                lon=point.longitude,
+                                alt=point.elevation,
+                                angle=None,
+                                time_gps_epoch=unix_time,
+                                gps_fix_type=3 if point.elevation is not None else 2,
+                                horizontal_accuracy=0.0,
+                                vertical_accuracy=0.0,
+                                velocity_east=0.0,
+                                velocity_north=0.0,
+                                velocity_up=0.0,
+                                speed_accuracy=0.0,
+                            )
+                        )
+                    else:
+                        tracks[-1].append(
+                            geo.Point(
+                                time=unix_time,
+                                lat=point.latitude,
+                                lon=point.longitude,
+                                alt=point.elevation,
+                                angle=None,
+                            )
                         )
-                    )
 
     return tracks
 

mapillary_tools/serializer/description.py

@@ -25,7 +25,7 @@
 
 import jsonschema
 
-from .. import exceptions, geo
+from .. import exceptions, geo, telemetry
 from ..types import (
     BaseSerializer,
     describe_error_metadata,
@@ -196,6 +196,10 @@ class ErrorDescription(TypedDict, total=False):
                         "type": ["number", "null"],
                         "description": "Camera angle of the track point, in degrees. If null, the angle will be interpolated",
                     },
+                    {
+                        "type": ["number", "null"],
+                        "description": "GPS epoch time of the track point, in seconds. If present, used as the authoritative timestamp",
+                    },
                 ],
             },
         },
@@ -509,18 +513,45 @@ def _from_video_desc(cls, desc: VideoDescription) -> VideoMetadata:
 class PointEncoder:
     @classmethod
     def encode(cls, p: geo.Point) -> T.Sequence[float | int | None]:
-        entry = [
+        entry: list[float | int | None] = [
             int(p.time * 1000),
             round(p.lon, _COORDINATES_PRECISION),
             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,
         ]
+        gps_epoch_time = p.get_gps_epoch_time()
+        if gps_epoch_time is not None:
+            entry.append(gps_epoch_time)
         return entry
 
     @classmethod
     def decode(cls, entry: T.Sequence[T.Any]) -> geo.Point:
-        time_ms, lon, lat, alt, angle = entry
+        if len(entry) >= 6 and entry[5] is not None:
+            time_ms, lon, lat, alt, angle, time_gps_epoch = (
+                entry[0],
+                entry[1],
+                entry[2],
+                entry[3],
+                entry[4],
+                entry[5],
+            )
+            return telemetry.CAMMGPSPoint(
+                time=time_ms / 1000,
+                lat=lat,
+                lon=lon,
+                alt=alt,
+                angle=angle,
+                time_gps_epoch=time_gps_epoch,
+                gps_fix_type=3 if alt is not None else 2,
+                horizontal_accuracy=0.0,
+                vertical_accuracy=0.0,
+                velocity_east=0.0,
+                velocity_north=0.0,
+                velocity_up=0.0,
+                speed_accuracy=0.0,
+            )
+        time_ms, lon, lat, alt, angle = entry[0], entry[1], entry[2], entry[3], entry[4]
         return geo.Point(time=time_ms / 1000, lon=lon, lat=lat, alt=alt, angle=angle)
 
 

mapillary_tools/uploader.py

@@ -302,10 +302,33 @@ def prepare_camm_info(
                 camm_info.gps.append(point)
 
             elif isinstance(point, telemetry.GPSPoint):
-                # There is no proper CAMM entry for GoPro GPS
-                if camm_info.mini_gps is None:
-                    camm_info.mini_gps = []
-                camm_info.mini_gps.append(point)
+                # Convert GPSPoint to CAMMGPSPoint if it has a valid epoch_time,
+                # so the GPS timestamp is preserved in the CAMM type 6 entry
+                if point.epoch_time is not None and point.epoch_time > 0:
+                    camm_point = telemetry.CAMMGPSPoint(
+                        time=point.time,
+                        lat=point.lat,
+                        lon=point.lon,
+                        alt=point.alt,
+                        angle=point.angle,
+                        time_gps_epoch=point.epoch_time,
+                        gps_fix_type=point.fix.value
+                        if point.fix is not None
+                        else (3 if point.alt is not None else 2),
+                        horizontal_accuracy=0.0,
+                        vertical_accuracy=0.0,
+                        velocity_east=0.0,
+                        velocity_north=0.0,
+                        velocity_up=0.0,
+                        speed_accuracy=0.0,
+                    )
+                    if camm_info.gps is None:
+                        camm_info.gps = []
+                    camm_info.gps.append(camm_point)
+                else:
+                    if camm_info.mini_gps is None:
+                        camm_info.mini_gps = []
+                    camm_info.mini_gps.append(point)
 
             elif isinstance(point, geo.Point):
                 if camm_info.mini_gps is None:

schema/image_description_schema.json

@@ -34,6 +34,13 @@
                                     "null"
                                 ],
                                 "description": "Camera angle of the track point, in degrees. If null, the angle will be interpolated"
+                            },
+                            {
+                                "type": [
+                                    "number",
+                                    "null"
+                                ],
+                                "description": "GPS epoch time of the track point, in seconds. If present, used as the authoritative timestamp"
                             }
                         ]
                     }