diff --git a/mapillary_tools/camm/camm_builder.py b/mapillary_tools/camm/camm_builder.py index f4207fb7..d8f6a6b6 100644 --- a/mapillary_tools/camm/camm_builder.py +++ b/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, }, } diff --git a/mapillary_tools/camm/camm_parser.py b/mapillary_tools/camm/camm_parser.py index a99da0e7..d3a9bbd1 100644 --- a/mapillary_tools/camm/camm_parser.py +++ b/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) diff --git a/mapillary_tools/geo.py b/mapillary_tools/geo.py index 32a5142d..3371b1cb 100644 --- a/mapillary_tools/geo.py +++ b/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: diff --git a/mapillary_tools/geotag/options.py b/mapillary_tools/geotag/options.py index 7bbcd7c5..91e169b9 100644 --- a/mapillary_tools/geotag/options.py +++ b/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) diff --git a/mapillary_tools/geotag/utils.py b/mapillary_tools/geotag/utils.py index e1959347..2b2badc1 100644 --- a/mapillary_tools/geotag/utils.py +++ b/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, diff --git a/mapillary_tools/geotag/video_extractors/gpx.py b/mapillary_tools/geotag/video_extractors/gpx.py index f517cbca..00722bd1 100644 --- a/mapillary_tools/geotag/video_extractors/gpx.py +++ b/mapillary_tools/geotag/video_extractors/gpx.py @@ -17,7 +17,7 @@ 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 @@ -25,6 +25,11 @@ 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 diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py index 12978718..1e3e7764 100644 --- a/mapillary_tools/sample_video.py +++ b/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) diff --git a/mapillary_tools/serializer/description.py b/mapillary_tools/serializer/description.py index 5540aba7..64b2e9a7 100644 --- a/mapillary_tools/serializer/description.py +++ b/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, diff --git a/mapillary_tools/serializer/gpx.py b/mapillary_tools/serializer/gpx.py index 10f36315..42f5af00 100644 --- a/mapillary_tools/serializer/gpx.py +++ b/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 diff --git a/mapillary_tools/telemetry.py b/mapillary_tools/telemetry.py index 5c581e19..39f0a942 100644 --- a/mapillary_tools/telemetry.py +++ b/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) diff --git a/mapillary_tools/uploader.py b/mapillary_tools/uploader.py index f514229b..e3e7f12c 100644 --- a/mapillary_tools/uploader.py +++ b/mapillary_tools/uploader.py @@ -304,14 +304,16 @@ def prepare_camm_info( elif isinstance(point, telemetry.GPSPoint): # 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: + gps_epoch_time = point.get_gps_epoch_time() + if gps_epoch_time is not None: 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, + # CAMM type 6 stores GPS time, not Unix time + time_gps_epoch=gps_epoch_time, gps_fix_type=point.fix.value if point.fix is not None else (3 if point.alt is not None else 2), diff --git a/schema/image_description_schema.json b/schema/image_description_schema.json index 5d8edb72..5a4c7e33 100644 --- a/schema/image_description_schema.json +++ b/schema/image_description_schema.json @@ -40,7 +40,7 @@ "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" } ] } diff --git a/tests/unit/test_camm_parser.py b/tests/unit/test_camm_parser.py index 23f274a3..e6ba09be 100644 --- a/tests/unit/test_camm_parser.py +++ b/tests/unit/test_camm_parser.py @@ -525,7 +525,11 @@ def test_prepare_camm_info_gpspoint_with_epoch_time(): assert converted.lon == original.lon assert converted.alt == original.alt assert converted.time == original.time - assert converted.time_gps_epoch == original.epoch_time + # epoch_time is Unix time, time_gps_epoch is GPS time + assert converted.time_gps_epoch == telemetry.unix_to_gps_epoch( + original.epoch_time + ) + assert converted.get_unix_time() == original.epoch_time # Verify fix type was correctly converted from GPSFix enum assert camm_info.gps[0].gps_fix_type == 3 # FIX_3D.value @@ -672,8 +676,10 @@ def test_prepare_camm_info_mixed_point_types(): # 2 points in gps (CAMMGPSPoint + converted GPSPoint) assert camm_info.gps is not None assert len(camm_info.gps) == 2 + # gps[0] was already a CAMMGPSPoint, so its GPS time is passed through assert camm_info.gps[0].time_gps_epoch == 1706000000.0 - assert camm_info.gps[1].time_gps_epoch == 1706000001.0 + # gps[1] was converted from a GPSPoint, whose epoch_time is Unix time + assert camm_info.gps[1].time_gps_epoch == telemetry.unix_to_gps_epoch(1706000001.0) # 2 points in mini_gps (GPSPoint without epoch + geo.Point) assert camm_info.mini_gps is not None @@ -718,6 +724,79 @@ def test_prepare_camm_info_gpspoint_roundtrip(): for original, decoded in zip(points, x.points): assert isinstance(decoded, telemetry.CAMMGPSPoint) decoded_camm = T.cast(telemetry.CAMMGPSPoint, decoded) - assert abs(original.epoch_time - decoded_camm.time_gps_epoch) < 10e-6 + # The wall clock timestamp survives the Unix -> GPS -> Unix round trip + assert abs(original.epoch_time - decoded_camm.get_unix_time()) < 10e-6 assert abs(original.lat - decoded_camm.lat) < 10e-6 assert abs(original.lon - decoded_camm.lon) < 10e-6 + + +def _extract_camm_info_from_points( + points: T.Sequence[geo.Point], +) -> camm_parser.CAMMInfo: + """Build an in-memory CAMM mp4 out of points and parse it back.""" + movie_timescale = 1_000_000 + + mvhd: cparser.BoxDict = { + "type": b"mvhd", + "data": { + "creation_time": 1, + "modification_time": 2, + "timescale": movie_timescale, + "duration": int(36000 * movie_timescale), + }, + } + empty_mp4: T.List[cparser.BoxDict] = [ + {"type": b"ftyp", "data": b"test"}, + {"type": b"moov", "data": [mvhd]}, + ] + src = cparser.MP4WithoutSTBLBuilderConstruct.build_boxlist(empty_mp4) + + metadata = types.VideoMetadata( + Path(""), filetype=types.FileType.CAMM, points=list(points) + ) + input_camm_info = uploader.VideoUploader.prepare_camm_info(metadata) + target_fp = simple_mp4_builder.transform_mp4( + io.BytesIO(src), camm_builder.camm_sample_generator2(input_camm_info) + ) + + camm_info = camm_parser.extract_camm_info(T.cast(T.BinaryIO, target_fp)) + assert camm_info is not None + return camm_info + + +def test_extract_camm_info_routes_gps_points_to_gps(): + """CAMMGPSPoint is a subclass of geo.Point, so type 6 must be tested first + or every GPS point silently lands in mini_gps (type 5).""" + camm_info = _extract_camm_info_from_points( + [ + telemetry.CAMMGPSPoint( + time=0.0, + lat=37.7749, + lon=-122.4194, + alt=10.0, + angle=None, + time_gps_epoch=1470558405.0, + gps_fix_type=3, + horizontal_accuracy=0.0, + vertical_accuracy=0.0, + velocity_east=0.0, + velocity_north=0.0, + velocity_up=0.0, + speed_accuracy=0.0, + ) + ] + ) + assert camm_info.gps is not None + assert len(camm_info.gps) == 1 + assert isinstance(camm_info.gps[0], telemetry.CAMMGPSPoint) + assert not camm_info.mini_gps + + +def test_extract_camm_info_routes_plain_points_to_mini_gps(): + camm_info = _extract_camm_info_from_points( + [geo.Point(time=0.0, lat=37.7749, lon=-122.4194, alt=10.0, angle=None)] + ) + assert not camm_info.gps + assert camm_info.mini_gps is not None + assert len(camm_info.mini_gps) == 1 + assert type(camm_info.mini_gps[0]) is geo.Point diff --git a/tests/unit/test_description.py b/tests/unit/test_description.py index 4dd5c5c4..7210b901 100644 --- a/tests/unit/test_description.py +++ b/tests/unit/test_description.py @@ -167,7 +167,8 @@ def test_encode_camm_gps_point(): encoded = PointEncoder.encode(p) assert len(encoded) == 6 assert encoded[0] == 2000 - assert encoded[5] == 1700000001.0 + # The description carries Unix time, the point carries GPS time + assert encoded[5] == telemetry.gps_epoch_to_unix(1700000001.0) def test_decode_camm_gps_point(): @@ -179,7 +180,9 @@ def test_decode_camm_gps_point(): assert p.lon == -122.4194 assert p.alt == 15.0 assert p.angle == 180.0 - assert p.time_gps_epoch == 1700000001.0 + # entry[5] is Unix time, CAMMGPSPoint.time_gps_epoch is GPS time + assert p.time_gps_epoch == telemetry.unix_to_gps_epoch(1700000001.0) + assert p.get_unix_time() == 1700000001.0 assert p.gps_fix_type == 3 # alt is not None diff --git a/tests/unit/test_gps_epoch.py b/tests/unit/test_gps_epoch.py new file mode 100644 index 00000000..5888997f --- /dev/null +++ b/tests/unit/test_gps_epoch.py @@ -0,0 +1,171 @@ +# Copyright (c) Meta Platforms, Inc. and affiliates. +# +# This source code is licensed under the BSD license found in the +# LICENSE file in the root directory of this source tree. + +""" +Regression tests for mixing up the GPS epoch (1980-01-06) with the Unix epoch +(1970-01-01). + +CAMM stores GPS time (``CAMMGPSPoint.time_gps_epoch``) while GPX, GPMF and +BlackVue store Unix time (``GPSPoint.epoch_time``). Subtracting one from the +other yields ~315,964,800s, which used to overflow the 32-bit +``segment_duration`` of a version 0 ``elst`` and abort the upload. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from mapillary_tools import geo, telemetry +from mapillary_tools.camm import camm_builder +from mapillary_tools.geotag.options import SourceOption, SourceType +from mapillary_tools.geotag.video_extractors.gpx import GPXVideoExtractor +from mapillary_tools.mp4 import construct_mp4_parser as cparser + + +# Seconds between the two epochs, i.e. the size of the bug +GPS_UNIX_DELTA = 315964800 + +# 2026-08-12T08:26:27Z, taken from a PanoX V2 capture +A_GPS_TIME = 1470558405.9798455 +A_UNIX_TIME = 1786523187.9798455 + + +def _camm_point(time: float, time_gps_epoch: float) -> telemetry.CAMMGPSPoint: + return telemetry.CAMMGPSPoint( + time=time, + lat=37.8436443, + lon=14.9886571, + alt=1202.345, + angle=None, + time_gps_epoch=time_gps_epoch, + gps_fix_type=3, + horizontal_accuracy=0.0, + vertical_accuracy=0.0, + velocity_east=0.0, + velocity_north=0.0, + velocity_up=0.0, + speed_accuracy=0.0, + ) + + +def _gps_point(time: float, epoch_time: float | None) -> telemetry.GPSPoint: + return telemetry.GPSPoint( + time=time, + lat=37.8436443, + lon=14.9886571, + alt=1202.345, + angle=None, + epoch_time=epoch_time, + fix=telemetry.GPSFix.FIX_3D, + precision=None, + ground_speed=None, + ) + + +class TestEpochConversion: + def test_known_instant(self): + assert telemetry.gps_epoch_to_unix(A_GPS_TIME) == A_UNIX_TIME + + def test_round_trip(self): + for unix_time in [0.0, 1e9, A_UNIX_TIME, 2e9]: + assert telemetry.unix_to_gps_epoch( + telemetry.gps_epoch_to_unix(unix_time) + ) == pytest.approx(unix_time) + + def test_leap_seconds_accumulate(self): + # No leap seconds had accumulated at the GPS epoch itself + assert telemetry.gps_epoch_to_unix(0) == GPS_UNIX_DELTA + # 18 by 2026, so the naive +315964800 conversion is 18s too late + assert ( + telemetry.gps_epoch_to_unix(A_GPS_TIME) == A_GPS_TIME + GPS_UNIX_DELTA - 18 + ) + + +class TestPointAccessors: + def test_camm_point_exposes_both_epochs(self): + p = _camm_point(time=0.0, time_gps_epoch=A_GPS_TIME) + assert p.get_gps_epoch_time() == A_GPS_TIME + assert p.get_unix_time() == A_UNIX_TIME + + def test_gps_point_exposes_both_epochs(self): + p = _gps_point(time=0.0, epoch_time=A_UNIX_TIME) + assert p.get_unix_time() == A_UNIX_TIME + assert p.get_gps_epoch_time() == A_GPS_TIME + + def test_invalid_timestamps_are_ignored(self): + assert _camm_point(time=0.0, time_gps_epoch=0.0).get_unix_time() is None + assert _gps_point(time=0.0, epoch_time=None).get_unix_time() is None + assert _gps_point(time=0.0, epoch_time=0.0).get_unix_time() is None + # A plain Point carries no absolute timestamp at all + assert ( + geo.Point(time=1.0, lat=0, lon=0, alt=None, angle=None).get_unix_time() + is None + ) + + +class TestGPXOffset: + """A GPX recorded alongside the video must sync to ~0, not to ~10 years.""" + + def test_camm_video_syncs_to_zero(self): + # Same instant, expressed in each container's native epoch + gpx_points = [_camm_point(time=A_UNIX_TIME, time_gps_epoch=A_GPS_TIME)] + video_points = [_camm_point(time=0.0, time_gps_epoch=A_GPS_TIME)] + assert GPXVideoExtractor._gpx_offset(gpx_points, video_points) == 0.0 + + def test_gopro_video_syncs_to_zero(self): + gpx_points = [_camm_point(time=A_UNIX_TIME, time_gps_epoch=A_GPS_TIME)] + video_points = [_gps_point(time=0.0, epoch_time=A_UNIX_TIME)] + assert GPXVideoExtractor._gpx_offset(gpx_points, video_points) == 0.0 + + def test_real_offset_is_preserved(self): + gpx_points = [ + _camm_point(time=A_UNIX_TIME + 30, time_gps_epoch=A_GPS_TIME + 30) + ] + video_points = [_camm_point(time=0.0, time_gps_epoch=A_GPS_TIME)] + assert GPXVideoExtractor._gpx_offset(gpx_points, video_points) == 30.0 + + def test_missing_video_timestamp_yields_no_offset(self): + gpx_points = [_camm_point(time=A_UNIX_TIME, time_gps_epoch=A_GPS_TIME)] + video_points = [_camm_point(time=0.0, time_gps_epoch=0.0)] + assert GPXVideoExtractor._gpx_offset(gpx_points, video_points) == 0.0 + + +class TestEditListOverflow: + """An oversized initial gap must not abort the upload.""" + + def test_small_offset_stays_version_0(self): + points = [geo.Point(time=1.5, lat=0, lon=0, alt=None, angle=None)] + elst = camm_builder._create_edit_list_from_points([points], 1000, 1000) + assert elst["data"]["version"] == 0 + assert elst["data"]["entries"][0]["segment_duration"] == 1500 + + def test_oversized_offset_falls_back_to_version_1(self): + # The exact shape of the reported crash: a whole GPS epoch of offset + points = [geo.Point(time=GPS_UNIX_DELTA, lat=0, lon=0, alt=None, angle=None)] + elst = camm_builder._create_edit_list_from_points([points], 1000, 1000) + assert elst["data"]["version"] == 1 + assert elst["data"]["entries"][0]["segment_duration"] == GPS_UNIX_DELTA * 1000 + # Must serialize rather than raise construct.core.FormatFieldError + assert cparser.EditBox.build(elst["data"]) + + +class TestSourceOption: + def test_explicit_source_path_is_not_dropped(self): + opt = SourceOption.from_dict( + { + "source": "gpx", + "pattern": "%g.gpx", + "source_path": "/tmp/explicit.gpx", + } + ) + assert opt.source is SourceType.GPX + assert opt.source_path is not None + assert opt.source_path.source_path == Path("/tmp/explicit.gpx") + # source_path wins over pattern when resolving + assert opt.source_path.resolve(Path("/data/video.mp4")) == Path( + "/tmp/explicit.gpx" + ) diff --git a/tests/unit/test_gpx_serializer.py b/tests/unit/test_gpx_serializer.py index 3a52cad5..037b729e 100644 --- a/tests/unit/test_gpx_serializer.py +++ b/tests/unit/test_gpx_serializer.py @@ -10,7 +10,12 @@ from mapillary_tools.geo import Point from mapillary_tools.serializer.gpx import GPXSerializer -from mapillary_tools.telemetry import CAMMGPSPoint, GPSFix, GPSPoint +from mapillary_tools.telemetry import ( + CAMMGPSPoint, + gps_epoch_to_unix, + GPSFix, + GPSPoint, +) from mapillary_tools.types import ErrorMetadata, FileType, ImageMetadata, VideoMetadata @@ -151,7 +156,7 @@ def test_image_metadata_point_has_name(self): gpx_pt = GPXSerializer.as_gpx_point(img) assert gpx_pt.name == "photo.jpg" - def test_camm_gps_point_uses_gps_epoch_time(self): + def test_camm_gps_point_uses_gps_timestamp(self): p = CAMMGPSPoint( time=5.0, lat=48.0, @@ -168,9 +173,10 @@ def test_camm_gps_point_uses_gps_epoch_time(self): speed_accuracy=0.5, ) gpx_pt = GPXSerializer.as_gpx_point(p) - # time should be based on time_gps_epoch, not the video time (5.0) + # time should be based on time_gps_epoch, not the video time (5.0), + # and converted from GPS time to UTC assert gpx_pt.time is not None - assert gpx_pt.time.timestamp() == 1700000000.0 + assert gpx_pt.time.timestamp() == gps_epoch_to_unix(1700000000.0) def test_gps_point_with_epoch_time(self): p = GPSPoint( diff --git a/tests/unit/test_parse_gpx.py b/tests/unit/test_parse_gpx.py index 17ca9770..38944613 100644 --- a/tests/unit/test_parse_gpx.py +++ b/tests/unit/test_parse_gpx.py @@ -23,7 +23,10 @@ def test_parse_gpx_creates_camm_gps_points(): for point in track: assert isinstance(point, telemetry.CAMMGPSPoint) - assert point.time_gps_epoch == point.time + # GPX timestamps are UTC; time_gps_epoch is GPS time, so the two differ + # by the GPS epoch offset plus leap seconds + assert point.time_gps_epoch == telemetry.unix_to_gps_epoch(point.time) + assert point.get_unix_time() == point.time assert point.gps_fix_type == 3 # all points have assert point.horizontal_accuracy == 0.0 assert point.vertical_accuracy == 0.0