From 25e74ce9070b883a2e012125b6902d7b4105d2a5 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Mon, 7 Sep 2026 16:45:11 +0200 Subject: [PATCH 1/2] Fix GPS epoch vs Unix epoch mix-up when geotagging CAMM videos CAMM stores GPS time in CAMMGPSPoint.time_gps_epoch (seconds since 1980-01-06, per the CAMM spec), while GPX, GPMF and BlackVue all store Unix time. Those two were being compared and rendered interchangeably, which is a ~315,964,800s (10 year) error. The visible symptom: geotagging a CAMM video from an external GPX made GPXVideoExtractor._gpx_offset() subtract a GPS time from a Unix time, so every GPX point was rebased ~10 years after the video start. At upload, camm_builder wrote that as the segment_duration of a version 0 elst, which is Int32sb, and the upload aborted with construct.core.FormatFieldError: Error in path (building) -> ... -> segment_duration struct '>l' error during building, given value 315964800000 There is also a silent variant with no crash: sampled frames, exported GPX and the description file all rendered a CAMM GPS timestamp directly as Unix time, dating imagery ~10 years too early. Changes: - telemetry: add gps_epoch_to_unix()/unix_to_gps_epoch(), including leap seconds (GPS time does not count them, so a naive +315964800 is currently 18s off -- ~250m of error at highway speed). - Point.get_unix_time() is now the canonical wall clock accessor; get_gps_epoch_time() is kept for the CAMM serialization boundary. Both are honest about their epoch for GPSPoint and CAMMGPSPoint. - parse_gpx() and uploader.prepare_camm_info() now convert to GPS time when populating time_gps_epoch, instead of storing Unix time in it. - _gpx_offset() compares Unix times via get_unix_time(), which also ignores zero/invalid timestamps rather than treating them as 1970. - sample_video, the GPX serializer and the description serializer render get_unix_time(). MAPGPSTrack[5] is now consistently Unix time; the schema said "GPS epoch time" but every consumer read it as Unix. - camm_builder falls back to a version 1 (64-bit) elst instead of overflowing, so an oversized offset can never abort an upload again. - warn when a GPX has to be shifted more than a day to sync. Two unrelated bugs found while tracing this: - camm_parser filed every CAMM type 6 GPS point into CAMMInfo.mini_gps, leaving CAMMInfo.gps always empty, because CAMMGPSPoint subclasses geo.Point and the isinstance checks were in the wrong order. - SourceOption.from_dict() assigned to a misspelled `.sourthe_path`, silently dropping an explicit source_path passed alongside a pattern. --- mapillary_tools/camm/camm_builder.py | 16 ++ mapillary_tools/camm/camm_parser.py | 8 +- mapillary_tools/geo.py | 26 ++- mapillary_tools/geotag/options.py | 2 +- mapillary_tools/geotag/utils.py | 3 +- .../geotag/video_extractors/gpx.py | 35 ++-- mapillary_tools/sample_video.py | 8 +- mapillary_tools/serializer/description.py | 8 +- mapillary_tools/serializer/gpx.py | 12 +- mapillary_tools/telemetry.py | 92 +++++++++- mapillary_tools/uploader.py | 6 +- schema/image_description_schema.json | 2 +- tests/unit/test_camm_parser.py | 85 ++++++++- tests/unit/test_description.py | 7 +- tests/unit/test_gps_epoch.py | 171 ++++++++++++++++++ tests/unit/test_gpx_serializer.py | 14 +- tests/unit/test_parse_gpx.py | 5 +- 17 files changed, 445 insertions(+), 55 deletions(-) create mode 100644 tests/unit/test_gps_epoch.py 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 From 2e00e71697239265ccacd332d80c1f05d8562fcb Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Wed, 9 Sep 2026 15:24:01 +0200 Subject: [PATCH 2/2] Normalize CAMM GPS timestamps at the parse boundary, not everywhere Follow-up to review of the previous commit. Treating time_gps_epoch as uniformly GPS time was wrong, and broke two things. What producers actually write in CAMM type 6 time_gps_epoch, measured across 608 videos in the device corpus: Labpano Pilot One / Pilot Era / PanoX V2 GPS time 14 files Insta360 Pro Unix time 1 file mapillary_tools itself Unix time GPMF / NMEA sources (converted on write) Unix time Defect 1, read side: converting unconditionally put Insta360 Pro at 2030 and CAMM written by mapillary_tools at 2033. Defect 2, write side: prepare_camm_info() converted GoPro/BlackVue/NMEA timestamps to GPS time on the way out, so the uploaded artifact carried capture times ~10 years in the past (a 2022-06-17 GoPro recording came back as 2012-06-12). Read and write were inverses of each other, so the round trip looked fine while being incompatible with every released version. This is the higher severity of the two: it ships wrong data, not just a red test. Instead: - CAMMGPSPoint.time_gps_epoch is renamed to epoch_time and now always holds Unix time, the same meaning GPSPoint.epoch_time already had. The old name described the CAMM box field, not the value in memory, which is what made both defects easy to write. - The conversion happens exactly once, in camm_parser, keyed on the camera make, right after the samples are parsed. Nowhere else. - The serializer writes epoch_time straight through, so on-disk stays Unix time exactly as every released version writes it. No format change, no migration needed, files stay readable both ways. - get_gps_epoch_time() and unix_to_gps_epoch() are gone; nothing needed GPS time once the boundary was fixed. gps_epoch_to_unix() now has a single caller. A camera writing GPS time that is not on the make list would be silently wrong, so parsing also warns when the first GPS timestamp sits almost exactly one GPS epoch from the container creation_time. It checks for that specific distance rather than general implausibility because some cameras write a meaningless creation_time -- a GoPro HERO7 recorded in 2022 reports 2016 -- which a generic bound would flag constantly. --- mapillary_tools/camm/camm_parser.py | 67 ++++++++++++++- mapillary_tools/geo.py | 13 +-- mapillary_tools/geotag/utils.py | 3 +- mapillary_tools/serializer/description.py | 2 +- mapillary_tools/telemetry.py | 50 ++++------- mapillary_tools/uploader.py | 7 +- tests/unit/test_camm_parser.py | 100 +++++++++++++++++----- tests/unit/test_description.py | 15 ++-- tests/unit/test_geo.py | 24 +++--- tests/unit/test_gps_epoch.py | 91 ++++++++++++-------- tests/unit/test_gpx_serializer.py | 14 +-- tests/unit/test_parse_gpx.py | 5 +- 12 files changed, 249 insertions(+), 142 deletions(-) diff --git a/mapillary_tools/camm/camm_parser.py b/mapillary_tools/camm/camm_parser.py index d3a9bbd1..b5f42049 100644 --- a/mapillary_tools/camm/camm_parser.py +++ b/mapillary_tools/camm/camm_parser.py @@ -123,11 +123,69 @@ def extract_camm_info(fp: T.BinaryIO, telemetry_only: bool = False) -> CAMMInfo 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() @@ -227,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, @@ -245,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, diff --git a/mapillary_tools/geo.py b/mapillary_tools/geo.py index 3371b1cb..d070c82a 100644 --- a/mapillary_tools/geo.py +++ b/mapillary_tools/geo.py @@ -37,19 +37,10 @@ class Point: alt: float | None angle: float | None - def get_gps_epoch_time(self) -> float | None: - """ - 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. + 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 diff --git a/mapillary_tools/geotag/utils.py b/mapillary_tools/geotag/utils.py index 2b2badc1..3c1e2410 100644 --- a/mapillary_tools/geotag/utils.py +++ b/mapillary_tools/geotag/utils.py @@ -37,8 +37,7 @@ def parse_gpx(gpx_file: Path) -> list[Track]: lon=point.longitude, alt=point.elevation, angle=None, - # GPX timestamps are UTC; time_gps_epoch is GPS time - time_gps_epoch=telemetry.unix_to_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, diff --git a/mapillary_tools/serializer/description.py b/mapillary_tools/serializer/description.py index 64b2e9a7..62597b0c 100644 --- a/mapillary_tools/serializer/description.py +++ b/mapillary_tools/serializer/description.py @@ -538,7 +538,7 @@ def decode(cls, entry: T.Sequence[T.Any]) -> geo.Point: lon=lon, alt=alt, angle=angle, - time_gps_epoch=telemetry.unix_to_gps_epoch(unix_time), + epoch_time=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/telemetry.py b/mapillary_tools/telemetry.py index 39f0a942..d6084a05 100644 --- a/mapillary_tools/telemetry.py +++ b/mapillary_tools/telemetry.py @@ -64,6 +64,8 @@ def gps_epoch_to_unix(gps_epoch_time: float) -> float: """ Convert seconds since the GPS epoch (GPS time) to Unix time (UTC). + Only called at the parse boundary, for producers known to record GPS time. + >>> gps_epoch_to_unix(1470558405.9798455) 1786523187.9798455 """ @@ -73,16 +75,6 @@ def gps_epoch_to_unix(gps_epoch_time: float) -> float: 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 @@ -116,13 +108,6 @@ def get_unix_time(self) -> float | None: 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) @@ -171,9 +156,15 @@ 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 + # Unix time (UTC), same meaning as GPSPoint.epoch_time. + # + # The corresponding CAMM box field is named time_gps_epoch, but what + # producers actually store there varies: Labpano cameras record GPS time, + # while Insta360 and mapillary_tools itself record Unix time. Whatever the + # producer wrote is normalized to Unix time once, when the CAMM track is + # parsed (see camm_parser.extract_camm_info), so that everything + # downstream can rely on a single meaning. + epoch_time: float gps_fix_type: int horizontal_accuracy: float vertical_accuracy: float @@ -182,18 +173,11 @@ class CAMMGPSPoint(TimestampedMeasurement, Point): velocity_up: float speed_accuracy: float - def get_gps_epoch_time(self) -> float | None: - """Return the GPS epoch time if valid, otherwise None.""" - if self.time_gps_epoch > 0: - 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) + if self.epoch_time > 0: + return self.epoch_time + return None def interpolate_with(self, other: Point, t: float) -> Point: """Create a new interpolated CAMMGPSPoint using this and other point at time t.""" @@ -203,9 +187,7 @@ def interpolate_with(self, other: Point, t: float) -> Point: # Interpolate all CAMM-specific fields weight = self._calculate_weight_for_interpolation(other, t) - time_gps_epoch = ( - self.time_gps_epoch + (other.time_gps_epoch - self.time_gps_epoch) * weight - ) + epoch_time = self.epoch_time + (other.epoch_time - self.epoch_time) * weight horizontal_accuracy = ( self.horizontal_accuracy + (other.horizontal_accuracy - self.horizontal_accuracy) * weight @@ -231,7 +213,7 @@ def interpolate_with(self, other: Point, t: float) -> Point: lon=base.lon, alt=base.alt, angle=base.angle, - time_gps_epoch=time_gps_epoch, + epoch_time=epoch_time, gps_fix_type=self.gps_fix_type, # Use start point's fix type horizontal_accuracy=horizontal_accuracy, vertical_accuracy=vertical_accuracy, diff --git a/mapillary_tools/uploader.py b/mapillary_tools/uploader.py index e3e7f12c..17bd34c3 100644 --- a/mapillary_tools/uploader.py +++ b/mapillary_tools/uploader.py @@ -304,16 +304,15 @@ 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 - gps_epoch_time = point.get_gps_epoch_time() - if gps_epoch_time is not None: + unix_time = point.get_unix_time() + if unix_time is not None: camm_point = telemetry.CAMMGPSPoint( time=point.time, lat=point.lat, lon=point.lon, alt=point.alt, angle=point.angle, - # CAMM type 6 stores GPS time, not Unix time - time_gps_epoch=gps_epoch_time, + epoch_time=unix_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/tests/unit/test_camm_parser.py b/tests/unit/test_camm_parser.py index e6ba09be..112a1df5 100644 --- a/tests/unit/test_camm_parser.py +++ b/tests/unit/test_camm_parser.py @@ -134,7 +134,7 @@ def test_build_and_parse_camm_gps_points(): lon=0.2, alt=None, angle=None, - time_gps_epoch=1.1, + epoch_time=1.1, gps_fix_type=1, horizontal_accuracy=3.3, vertical_accuracy=4.4, @@ -149,7 +149,7 @@ def test_build_and_parse_camm_gps_points(): lon=0.2, alt=None, angle=None, - time_gps_epoch=1.2, + epoch_time=1.2, gps_fix_type=1, horizontal_accuracy=3.3, vertical_accuracy=4.4, @@ -164,7 +164,7 @@ def test_build_and_parse_camm_gps_points(): lon=0.21, alt=None, angle=None, - time_gps_epoch=1.3, + epoch_time=1.3, gps_fix_type=1, horizontal_accuracy=3.3, vertical_accuracy=4.4, @@ -188,7 +188,7 @@ def test_build_and_parse_camm_gps_points(): lon=0.2, alt=-1, angle=None, - time_gps_epoch=1.2, + epoch_time=1.2, gps_fix_type=1, horizontal_accuracy=3.3, vertical_accuracy=4.4, @@ -203,7 +203,7 @@ def test_build_and_parse_camm_gps_points(): lon=0.21, alt=-1, angle=None, - time_gps_epoch=1.3, + epoch_time=1.3, gps_fix_type=1, horizontal_accuracy=3.3, vertical_accuracy=4.4, @@ -423,7 +423,7 @@ def test_build_and_parse_gpx_sourced_camm_gps_points(): lon=-122.4194, alt=10.0, angle=None, - time_gps_epoch=1706000000.0, + epoch_time=1706000000.0, gps_fix_type=3, horizontal_accuracy=0.0, vertical_accuracy=0.0, @@ -438,7 +438,7 @@ def test_build_and_parse_gpx_sourced_camm_gps_points(): lon=-122.4195, alt=11.0, angle=None, - time_gps_epoch=1706000001.0, + epoch_time=1706000001.0, gps_fix_type=3, horizontal_accuracy=0.0, vertical_accuracy=0.0, @@ -453,7 +453,7 @@ def test_build_and_parse_gpx_sourced_camm_gps_points(): lon=-122.4196, alt=12.0, angle=None, - time_gps_epoch=1706000002.0, + epoch_time=1706000002.0, gps_fix_type=3, horizontal_accuracy=0.0, vertical_accuracy=0.0, @@ -469,12 +469,12 @@ def test_build_and_parse_gpx_sourced_camm_gps_points(): points=points, ) x = encode_decode_empty_camm_mp4(metadata) - # Verify points round-trip with time_gps_epoch preserved + # Verify points round-trip with epoch_time preserved assert len(x.points) == 3 for original, decoded in zip(points, x.points): assert isinstance(decoded, telemetry.CAMMGPSPoint) decoded_camm = T.cast(telemetry.CAMMGPSPoint, decoded) - assert abs(original.time_gps_epoch - decoded_camm.time_gps_epoch) < 10e-6 + assert abs(original.epoch_time - decoded_camm.epoch_time) < 10e-6 assert abs(original.time - decoded_camm.time) < 10e-6 assert abs(original.lat - decoded_camm.lat) < 10e-6 assert abs(original.lon - decoded_camm.lon) < 10e-6 @@ -525,10 +525,8 @@ def test_prepare_camm_info_gpspoint_with_epoch_time(): assert converted.lon == original.lon assert converted.alt == original.alt assert converted.time == original.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 - ) + # Both point types carry Unix time, so the value passes through + assert converted.epoch_time == original.epoch_time assert converted.get_unix_time() == original.epoch_time # Verify fix type was correctly converted from GPSFix enum @@ -624,7 +622,7 @@ def test_prepare_camm_info_mixed_point_types(): lon=-122.4194, alt=10.0, angle=None, - time_gps_epoch=1706000000.0, + epoch_time=1706000000.0, gps_fix_type=3, horizontal_accuracy=1.0, vertical_accuracy=2.0, @@ -676,10 +674,9 @@ 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 - # 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) + # Both were already Unix time, so both pass through unchanged + assert camm_info.gps[0].epoch_time == 1706000000.0 + assert camm_info.gps[1].epoch_time == 1706000001.0 # 2 points in mini_gps (GPSPoint without epoch + geo.Point) assert camm_info.mini_gps is not None @@ -775,7 +772,7 @@ def test_extract_camm_info_routes_gps_points_to_gps(): lon=-122.4194, alt=10.0, angle=None, - time_gps_epoch=1470558405.0, + epoch_time=1470558405.0, gps_fix_type=3, horizontal_accuracy=0.0, vertical_accuracy=0.0, @@ -800,3 +797,66 @@ def test_extract_camm_info_routes_plain_points_to_mini_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 + + +def test_camm_gps_timestamps_round_trip_as_unix(): + """process -> build CAMM -> re-read must return the input timestamps. + + mapillary_tools has always written Unix time into the CAMM type 6 + time_gps_epoch field, and released versions read it back as Unix time. + Writing anything else would make our output unreadable by them, so the + serializer must not convert. + """ + unix_times = [1655503450.5, 1655503451.5] + points = [ + telemetry.CAMMGPSPoint( + time=float(idx), + lat=37.7749 + idx * 1e-4, + lon=-122.4194, + alt=10.0, + angle=None, + epoch_time=unix_time, + 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, + ) + for idx, unix_time in enumerate(unix_times) + ] + metadata = types.VideoMetadata( + Path(""), filetype=types.FileType.CAMM, points=points + ) + + decoded = encode_decode_empty_camm_mp4(metadata).points + + assert [T.cast(telemetry.CAMMGPSPoint, p).epoch_time for p in decoded] == unix_times + assert [p.get_unix_time() for p in decoded] == unix_times + + +def test_gpspoint_timestamps_round_trip_as_unix(): + """The same, for GoPro/BlackVue/NMEA sources converted on the way out.""" + unix_time = 1655503450.5 + metadata = types.VideoMetadata( + Path(""), + filetype=types.FileType.GOPRO, + points=[ + telemetry.GPSPoint( + time=0.0, + lat=37.7749, + lon=-122.4194, + alt=10.0, + angle=None, + epoch_time=unix_time, + fix=telemetry.GPSFix.FIX_3D, + precision=None, + ground_speed=None, + ) + ], + ) + + decoded = encode_decode_empty_camm_mp4(metadata).points + + assert T.cast(telemetry.CAMMGPSPoint, decoded[0]).epoch_time == unix_time diff --git a/tests/unit/test_description.py b/tests/unit/test_description.py index 7210b901..adf95b6e 100644 --- a/tests/unit/test_description.py +++ b/tests/unit/test_description.py @@ -155,7 +155,7 @@ def test_encode_camm_gps_point(): lon=-122.4194, alt=15.0, angle=180.0, - time_gps_epoch=1700000001.0, + epoch_time=1700000001.0, gps_fix_type=3, horizontal_accuracy=1.5, vertical_accuracy=2.0, @@ -168,7 +168,8 @@ def test_encode_camm_gps_point(): assert len(encoded) == 6 assert encoded[0] == 2000 # The description carries Unix time, the point carries GPS time - assert encoded[5] == telemetry.gps_epoch_to_unix(1700000001.0) + # Both the point and the description carry Unix time -- no conversion + assert encoded[5] == 1700000001.0 def test_decode_camm_gps_point(): @@ -180,8 +181,8 @@ def test_decode_camm_gps_point(): assert p.lon == -122.4194 assert p.alt == 15.0 assert p.angle == 180.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) + # entry[5] and CAMMGPSPoint.epoch_time are both Unix time + assert p.epoch_time == 1700000001.0 assert p.get_unix_time() == 1700000001.0 assert p.gps_fix_type == 3 # alt is not None @@ -206,7 +207,7 @@ def test_encode_decode_roundtrip_camm_gps_point(): lon=-0.1278, alt=20.0, angle=45.0, - time_gps_epoch=1700000500.0, + epoch_time=1700000500.0, gps_fix_type=3, horizontal_accuracy=1.0, vertical_accuracy=1.0, @@ -218,7 +219,7 @@ def test_encode_decode_roundtrip_camm_gps_point(): encoded = PointEncoder.encode(original) decoded = PointEncoder.decode(encoded) assert isinstance(decoded, telemetry.CAMMGPSPoint) - assert decoded.time_gps_epoch == original.time_gps_epoch + assert decoded.epoch_time == original.epoch_time def test_decode_6_element_with_none_gps_epoch(): @@ -242,6 +243,6 @@ def test_encode_gps_point_without_epoch(): ground_speed=None, ) encoded = PointEncoder.encode(p) - # get_gps_epoch_time() returns None, so 6th element is None + # get_unix_time() returns None, so 6th element is None assert len(encoded) == 6 assert encoded[5] is None diff --git a/tests/unit/test_geo.py b/tests/unit/test_geo.py index bd3a785e..e381be2a 100644 --- a/tests/unit/test_geo.py +++ b/tests/unit/test_geo.py @@ -877,7 +877,7 @@ def test_avg_speed_with_gps_points_zero_epoch_time_fallback(self): self.assertAlmostEqual(speed, 11.1, delta=0.5) def test_avg_speed_with_camm_gps_points(self): - """Test avg_speed with CAMMGPSPoint using time_gps_epoch field.""" + """Test avg_speed with CAMMGPSPoint using epoch_time field.""" # Video time is 0-10 seconds, but GPS epoch time spans 50 seconds points = [ @@ -887,7 +887,7 @@ def test_avg_speed_with_camm_gps_points(self): lon=0.0, alt=0.0, angle=None, - time_gps_epoch=2000.0, # GPS epoch time + epoch_time=2000.0, # GPS epoch time gps_fix_type=3, horizontal_accuracy=1.0, vertical_accuracy=1.0, @@ -902,7 +902,7 @@ def test_avg_speed_with_camm_gps_points(self): lon=0.0, alt=0.0, angle=None, - time_gps_epoch=2050.0, # GPS epoch time (50 sec elapsed) + epoch_time=2050.0, # GPS epoch time (50 sec elapsed) gps_fix_type=3, horizontal_accuracy=1.0, vertical_accuracy=1.0, @@ -919,7 +919,7 @@ def test_avg_speed_with_camm_gps_points(self): self.assertAlmostEqual(speed, 11.1, delta=0.5) def test_avg_speed_with_camm_gps_points_zero_epoch_fallback(self): - """Test avg_speed with CAMMGPSPoint falls back when time_gps_epoch is 0.""" + """Test avg_speed with CAMMGPSPoint falls back when epoch_time is 0.""" points = [ CAMMGPSPoint( @@ -928,7 +928,7 @@ def test_avg_speed_with_camm_gps_points_zero_epoch_fallback(self): lon=0.0, alt=0.0, angle=None, - time_gps_epoch=0.0, # Zero triggers fallback + epoch_time=0.0, # Zero triggers fallback gps_fix_type=3, horizontal_accuracy=1.0, vertical_accuracy=1.0, @@ -943,7 +943,7 @@ def test_avg_speed_with_camm_gps_points_zero_epoch_fallback(self): lon=0.0, alt=0.0, angle=None, - time_gps_epoch=0.0, # Zero triggers fallback + epoch_time=0.0, # Zero triggers fallback gps_fix_type=3, horizontal_accuracy=1.0, vertical_accuracy=1.0, @@ -1073,7 +1073,7 @@ def test_interpolate_camm_gps_points_returns_camm_gps_point(self): lon=0.0, alt=100.0, angle=0.0, - time_gps_epoch=2000.0, + epoch_time=2000.0, gps_fix_type=3, horizontal_accuracy=1.0, vertical_accuracy=2.0, @@ -1088,7 +1088,7 @@ def test_interpolate_camm_gps_points_returns_camm_gps_point(self): lon=1.0, alt=200.0, angle=45.0, - time_gps_epoch=2010.0, + epoch_time=2010.0, gps_fix_type=3, horizontal_accuracy=3.0, vertical_accuracy=4.0, @@ -1111,7 +1111,7 @@ def test_interpolate_camm_gps_points_returns_camm_gps_point(self): self.assertAlmostEqual(result.alt, 150.0) # Check CAMMGPSPoint-specific fields are interpolated - self.assertAlmostEqual(result.time_gps_epoch, 2005.0) + self.assertAlmostEqual(result.epoch_time, 2005.0) self.assertEqual(result.gps_fix_type, 3) # Taken from start point self.assertAlmostEqual(result.horizontal_accuracy, 2.0) self.assertAlmostEqual(result.vertical_accuracy, 3.0) @@ -1180,7 +1180,7 @@ def test_interpolator_preserves_camm_gps_point_type(self): lon=0.0, alt=100.0, angle=0.0, - time_gps_epoch=2000.0, + epoch_time=2000.0, gps_fix_type=3, horizontal_accuracy=1.0, vertical_accuracy=2.0, @@ -1195,7 +1195,7 @@ def test_interpolator_preserves_camm_gps_point_type(self): lon=1.0, alt=200.0, angle=45.0, - time_gps_epoch=2010.0, + epoch_time=2010.0, gps_fix_type=3, horizontal_accuracy=3.0, vertical_accuracy=4.0, @@ -1210,5 +1210,5 @@ def test_interpolator_preserves_camm_gps_point_type(self): result = interpolator.interpolate(5.0) self.assertIsInstance(result, CAMMGPSPoint) - self.assertAlmostEqual(result.time_gps_epoch, 2005.0) + self.assertAlmostEqual(result.epoch_time, 2005.0) self.assertAlmostEqual(result.velocity_east, 15.0) diff --git a/tests/unit/test_gps_epoch.py b/tests/unit/test_gps_epoch.py index 5888997f..87627b70 100644 --- a/tests/unit/test_gps_epoch.py +++ b/tests/unit/test_gps_epoch.py @@ -4,13 +4,17 @@ # 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). +Regression tests for the epoch of CAMM GPS timestamps. -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. +The CAMM box field is called ``time_gps_epoch``, but producers disagree about +what goes in it: Labpano cameras write GPS time (seconds since 1980-01-06), +while Insta360 and mapillary_tools itself write Unix time. Confusing the two +is a ~315,964,800s (10 year) error. + +The invariant these tests protect: ``CAMMGPSPoint.epoch_time`` is *always* +Unix time in memory. The conversion happens exactly once, when the CAMM track +is parsed, and never again -- in particular not in the serializer, which must +keep writing Unix time so files stay readable by released versions. """ from __future__ import annotations @@ -20,7 +24,7 @@ import pytest from mapillary_tools import geo, telemetry -from mapillary_tools.camm import camm_builder +from mapillary_tools.camm import camm_builder, camm_parser 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 @@ -34,14 +38,14 @@ A_UNIX_TIME = 1786523187.9798455 -def _camm_point(time: float, time_gps_epoch: float) -> telemetry.CAMMGPSPoint: +def _camm_point(time: float, epoch_time: 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, + epoch_time=epoch_time, gps_fix_type=3, horizontal_accuracy=0.0, vertical_accuracy=0.0, @@ -70,34 +74,52 @@ 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 + # 18 by 2026, so the naive +315964800 conversion lands 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 +class TestParseBoundaryNormalization: + """The one place an epoch conversion is allowed to happen.""" + + def test_gps_epoch_producer_is_converted(self): + points = [_camm_point(time=0.0, epoch_time=A_GPS_TIME)] + camm_parser._normalize_gps_epochs(points, "Labpano") + assert points[0].epoch_time == A_UNIX_TIME + assert points[0].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 + @pytest.mark.parametrize("make", ["Insta360", "GoPro", "", "Unknown"]) + def test_unix_producers_are_left_alone(self, make: str): + """Converting these would push them ~10 years into the future.""" + points = [_camm_point(time=0.0, epoch_time=A_UNIX_TIME)] + camm_parser._normalize_gps_epochs(points, make) + assert points[0].epoch_time == A_UNIX_TIME + + def test_make_match_is_case_insensitive(self): + for make in ["labpano", "LABPANO", " Labpano "]: + points = [_camm_point(time=0.0, epoch_time=A_GPS_TIME)] + camm_parser._normalize_gps_epochs(points, make) + assert points[0].epoch_time == A_UNIX_TIME, make + + def test_invalid_timestamps_are_not_converted(self): + """A zero timestamp must stay zero, not become 1980.""" + points = [_camm_point(time=0.0, epoch_time=0.0)] + camm_parser._normalize_gps_epochs(points, "Labpano") + assert points[0].epoch_time == 0.0 + assert points[0].get_unix_time() is None + + +class TestPointAccessors: + def test_both_point_types_report_unix_time(self): + assert _camm_point(0.0, A_UNIX_TIME).get_unix_time() == A_UNIX_TIME + assert _gps_point(0.0, A_UNIX_TIME).get_unix_time() == A_UNIX_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 _camm_point(time=0.0, epoch_time=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 @@ -111,26 +133,23 @@ 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)] + gpx_points = [_camm_point(time=A_UNIX_TIME, epoch_time=A_UNIX_TIME)] + video_points = [_camm_point(time=0.0, epoch_time=A_UNIX_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)] + gpx_points = [_camm_point(time=A_UNIX_TIME, epoch_time=A_UNIX_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)] + gpx_points = [_camm_point(time=A_UNIX_TIME + 30, epoch_time=A_UNIX_TIME + 30)] + video_points = [_camm_point(time=0.0, epoch_time=A_UNIX_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)] + gpx_points = [_camm_point(time=A_UNIX_TIME, epoch_time=A_UNIX_TIME)] + video_points = [_camm_point(time=0.0, epoch_time=0.0)] assert GPXVideoExtractor._gpx_offset(gpx_points, video_points) == 0.0 diff --git a/tests/unit/test_gpx_serializer.py b/tests/unit/test_gpx_serializer.py index 037b729e..f1d61b74 100644 --- a/tests/unit/test_gpx_serializer.py +++ b/tests/unit/test_gpx_serializer.py @@ -10,12 +10,7 @@ from mapillary_tools.geo import Point from mapillary_tools.serializer.gpx import GPXSerializer -from mapillary_tools.telemetry import ( - CAMMGPSPoint, - gps_epoch_to_unix, - GPSFix, - GPSPoint, -) +from mapillary_tools.telemetry import CAMMGPSPoint, GPSFix, GPSPoint from mapillary_tools.types import ErrorMetadata, FileType, ImageMetadata, VideoMetadata @@ -163,7 +158,7 @@ def test_camm_gps_point_uses_gps_timestamp(self): lon=11.0, alt=100.0, angle=0.0, - time_gps_epoch=1700000000.0, + epoch_time=1700000000.0, gps_fix_type=3, horizontal_accuracy=1.0, vertical_accuracy=1.0, @@ -173,10 +168,9 @@ def test_camm_gps_point_uses_gps_timestamp(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), - # and converted from GPS time to UTC + # time should be based on epoch_time, not the video time (5.0) assert gpx_pt.time is not None - assert gpx_pt.time.timestamp() == gps_epoch_to_unix(1700000000.0) + assert gpx_pt.time.timestamp() == 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 38944613..411addd8 100644 --- a/tests/unit/test_parse_gpx.py +++ b/tests/unit/test_parse_gpx.py @@ -23,9 +23,8 @@ def test_parse_gpx_creates_camm_gps_points(): for point in track: assert isinstance(point, telemetry.CAMMGPSPoint) - # 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) + # GPX timestamps are UTC, and so is epoch_time + assert point.epoch_time == point.time assert point.get_unix_time() == point.time assert point.gps_fix_type == 3 # all points have assert point.horizontal_accuracy == 0.0