Preserve GPS timestamps from GPX tracks in video CAMM output

mapillary_tools/exiftool_read_video.py

@@ -202,24 +202,41 @@ def _aggregate_float_values_same_length(
     # aggregate speeds (optional)
     ground_speeds = _aggregate_float_values_same_length(ground_speed_tag)
 
-    # GPS timestamp (optional)
-    epoch_time = None
+    # GPS epoch times (optional)
     if gps_time_tag is not None:
-        gps_time_text = _extract_alternative_fields(texts_by_tag, [gps_time_tag], str)
-        if gps_time_text is not None:
-            dt = exif_read.parse_gps_datetime(gps_time_text)
-            if dt is not None:
-                epoch_time = geo.as_unix_time(dt)
+        gps_epoch_times: list[float | None] = [
+            geo.as_unix_time(dt) if dt is not None else None
+            for dt in (
+                exif_read.parse_gps_datetime(text)
+                for text in _extract_alternative_fields(
+                    texts_by_tag, [gps_time_tag], list
+                )
+                or []
+            )
+        ]
+        if len(gps_epoch_times) != expected_length:
+            LOG.warning(
+                "Found different number of GPS epoch times %d and coordinates %d",
+                len(gps_epoch_times),
+                expected_length,
+            )
+            gps_epoch_times = [None] * expected_length
+    elif time_tag is not None:
+        # Use per-point GPS timestamps as epoch times
+        gps_epoch_times = [t for t in timestamps]
+    else:
+        gps_epoch_times = [None] * expected_length
 
     # build track
     track: list[GPSPoint] = []
-    for timestamp, lon, lat, alt, direction, ground_speed in zip(
+    for timestamp, lon, lat, alt, direction, ground_speed, epoch_time in zip(
         timestamps,
         lons,
         lats,
         alts,
         directions,
         ground_speeds,
+        gps_epoch_times,
     ):
         if timestamp is None or lon is None or lat is None:
             continue
@@ -329,6 +346,7 @@ def _aggregate_gps_track_by_sample_time(
             lon_tag=lon_tag,
             lat_tag=lat_tag,
             alt_tag=alt_tag,
+            gps_time_tag=gps_time_tag,
             direction_tag=direction_tag,
             ground_speed_tag=ground_speed_tag,
         )
@@ -484,6 +502,7 @@ def _extract_gps_track_from_track(self) -> list[GPSPoint]:
                     lon_tag=f"{track_ns}:GPSLongitude",
                     lat_tag=f"{track_ns}:GPSLatitude",
                     alt_tag=f"{track_ns}:GPSAltitude",
+                    gps_time_tag=f"{track_ns}:GPSDateTime",
                     direction_tag=f"{track_ns}:GPSTrack",
                     ground_speed_tag=f"{track_ns}:GPSSpeed",
                     gps_fix_tag=f"{track_ns}:GPSMeasureMode",
@@ -521,5 +540,6 @@ def _extract_gps_track_from_quicktime(
             lon_tag=f"{namespace}:GPSLongitude",
             lat_tag=f"{namespace}:GPSLatitude",
             alt_tag=f"{namespace}:GPSAltitude",
+            gps_time_tag=f"{namespace}:GPSDateTime",
             direction_tag=f"{namespace}:GPSTrack",
         )

mapillary_tools/geotag/utils.py

@@ -12,7 +12,7 @@
 
 import gpxpy
 
-from .. import exiftool_read, geo, utils
+from .. import exiftool_read, geo, telemetry, utils
 
 Track = T.List[geo.Point]
 LOG = logging.getLogger(__name__)
@@ -29,13 +29,22 @@ def parse_gpx(gpx_file: Path) -> list[Track]:
             tracks.append([])
             for point in segment.points:
                 if point.time is not None:
+                    unix_time = geo.as_unix_time(point.time)
                     tracks[-1].append(
-                        geo.Point(
-                            time=geo.as_unix_time(point.time),
+                        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,
                         )
                     )
 

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,43 @@ 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,
+            p.get_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"
                             }
                         ]
                     }