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..b5f42049 100644 --- a/mapillary_tools/camm/camm_parser.py +++ b/mapillary_tools/camm/camm_parser.py @@ -116,16 +116,76 @@ def extract_camm_info(fp: T.BinaryIO, telemetry_only: bool = False) -> CAMMInfo gps: list[telemetry.CAMMGPSPoint] = [] for measurement in measurements: - if isinstance(measurement, geo.Point): - mini_gps.append(measurement) - elif isinstance(measurement, telemetry.CAMMGPSPoint): + # NOTE: CAMMGPSPoint is a subclass of geo.Point, so it has + # to be tested first or every GPS point ends up in mini_gps + if isinstance(measurement, telemetry.CAMMGPSPoint): gps.append(measurement) + elif isinstance(measurement, geo.Point): + mini_gps.append(measurement) + + _normalize_gps_epochs(gps, make, moov) return CAMMInfo(mini_gps=mini_gps, gps=gps, make=make, model=model) return None +# Makes whose CAMM type 6 samples record GPS time (seconds since 1980-01-06), +# as the CAMM spec describes. Everything else -- Insta360, and the CAMM tracks +# mapillary_tools writes itself -- records Unix time in the same field, so +# converting unconditionally would push those ~10 years into the future. +_GPS_EPOCH_MAKES = frozenset(["labpano"]) + +# Seconds between the mp4 epoch (1904-01-01) and the Unix epoch. +_MP4_EPOCH_UNIX_OFFSET = 2082844800 + +# Tolerance for recognizing a gap as "off by exactly one GPS epoch". Checking +# for that specific distance rather than for general implausibility matters: +# some cameras write a meaningless mvhd creation_time (a GoPro HERO7 recorded +# in 2022 reports 2016), so a generic bound would fire constantly. +_GPS_EPOCH_GAP_TOLERANCE = 30 * 24 * 3600 + + +def _normalize_gps_epochs( + gps: list[telemetry.CAMMGPSPoint], make: str, moov: MovieBoxParser | None = None +) -> None: + """ + Rewrite CAMMGPSPoint.epoch_time in place so it is Unix time regardless of + which epoch the producer used. + + This is the only place CAMM GPS timestamps change epoch. Everything + downstream, including the serializer, treats them as Unix time. + """ + if not gps: + return + + if make.strip().lower() in _GPS_EPOCH_MAKES: + for point in gps: + if point.epoch_time > 0: + point.epoch_time = telemetry.gps_epoch_to_unix(point.epoch_time) + + first = next((p.epoch_time for p in gps if p.epoch_time > 0), None) + if first is None or moov is None: + return + + try: + creation_time = moov.extract_mvhd_boxdata().get("creation_time", 0) + except Exception: + return + + if not creation_time: + return + + gap = abs(first - (creation_time - _MP4_EPOCH_UNIX_OFFSET)) + if abs(gap - telemetry.GPS_EPOCH_UNIX_OFFSET) < _GPS_EPOCH_GAP_TOLERANCE: + LOG.warning( + "CAMM GPS timestamps are one GPS epoch away from the creation time " + "of the video. The camera (make %r) may record GPS time where Unix " + "time is expected, or the reverse; please report this video", + make, + ) + + def extract_camera_make_and_model(fp: T.BinaryIO) -> tuple[str, str]: moov = MovieBoxParser.parse_stream(fp) udta_boxdata = moov.extract_udta_boxdata() @@ -225,7 +285,9 @@ def deserialize(cls, sample: Sample, data: T.Any) -> telemetry.CAMMGPSPoint: lon=data.longitude, alt=data.altitude, angle=None, - time_gps_epoch=data.time_gps_epoch, + # Raw, still in whatever epoch the producer used. Normalized to + # Unix time by _normalize_gps_epochs() once the make is known. + epoch_time=data.time_gps_epoch, gps_fix_type=data.gps_fix_type, horizontal_accuracy=data.horizontal_accuracy, vertical_accuracy=data.vertical_accuracy, @@ -243,7 +305,10 @@ def serialize(cls, data: telemetry.CAMMGPSPoint) -> bytes: { "type": cls.serialized_camm_type.value, "data": { - "time_gps_epoch": data.time_gps_epoch, + # Written as Unix time, which is what every released + # version of mapillary_tools has written and what readers + # of our output expect. Do not convert here. + "time_gps_epoch": data.epoch_time, "gps_fix_type": data.gps_fix_type, "latitude": data.lat, "longitude": data.lon, diff --git a/mapillary_tools/geo.py b/mapillary_tools/geo.py index 32a5142d..d070c82a 100644 --- a/mapillary_tools/geo.py +++ b/mapillary_tools/geo.py @@ -37,9 +37,10 @@ class Point: alt: float | None angle: float | None - def get_gps_epoch_time(self) -> float | None: + def get_unix_time(self) -> float | None: """ - Return the GPS epoch time for this point. + Return the absolute time of this point in Unix time (seconds since + 1970-01-01, UTC), or None if the point carries no absolute timestamp. Base Point class returns None, subclasses can override. """ return None @@ -100,7 +101,7 @@ def gps_distance(latlon_1: tuple[float, float], latlon_2: tuple[float, float]) - def avg_speed(sequence: T.Sequence[PointLike]) -> float: """ Calculate average speed over a sequence of points. - Uses GPS epoch time when available (via get_gps_epoch_time()), + Uses Unix time when available (via get_unix_time()), otherwise falls back to the time field. Returns 0.0 for empty or single-element sequences. Returns NaN if time difference is zero (undefined speed). @@ -116,14 +117,14 @@ def avg_speed(sequence: T.Sequence[PointLike]) -> float: first = sequence[0] last = sequence[-1] - # Try to use GPS epoch time if available (via polymorphic method) - first_gps_time = first.get_gps_epoch_time() - last_gps_time = last.get_gps_epoch_time() + # Try to use Unix time if available (via polymorphic method) + first_unix_time = first.get_unix_time() + last_unix_time = last.get_unix_time() - if first_gps_time is not None and last_gps_time is not None: - time_diff = last_gps_time - first_gps_time + if first_unix_time is not None and last_unix_time is not None: + time_diff = last_unix_time - first_unix_time else: - # Fall back to time field if GPS epoch time not available + # Fall back to time field if Unix time not available time_diff = last.time - first.time if time_diff == 0.0: 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..3c1e2410 100644 --- a/mapillary_tools/geotag/utils.py +++ b/mapillary_tools/geotag/utils.py @@ -37,7 +37,7 @@ def parse_gpx(gpx_file: Path) -> list[Track]: lon=point.longitude, alt=point.elevation, angle=None, - time_gps_epoch=unix_time, + epoch_time=unix_time, gps_fix_type=3 if point.elevation is not None else 2, horizontal_accuracy=0.0, vertical_accuracy=0.0, 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..62597b0c 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, + 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/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..d6084a05 100644 --- a/mapillary_tools/telemetry.py +++ b/mapillary_tools/telemetry.py @@ -6,12 +6,75 @@ # 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). + + Only called at the parse boundary, for producers known to record GPS time. + + >>> 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) + + @unique class GPSFix(Enum): NO_FIX = 0 @@ -33,13 +96,14 @@ 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 @@ -92,7 +156,15 @@ def interpolate_with(self, other: Point, t: float) -> Point: @dataclasses.dataclass class CAMMGPSPoint(TimestampedMeasurement, Point): - 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 @@ -101,10 +173,10 @@ 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 + def get_unix_time(self) -> float | None: + """Return the Unix time if valid, otherwise None.""" + if self.epoch_time > 0: + return self.epoch_time return None def interpolate_with(self, other: Point, t: float) -> Point: @@ -115,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 @@ -143,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 f514229b..17bd34c3 100644 --- a/mapillary_tools/uploader.py +++ b/mapillary_tools/uploader.py @@ -304,14 +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 - if point.epoch_time is not None and point.epoch_time > 0: + 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, - time_gps_epoch=point.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/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..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,7 +525,9 @@ 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 + # 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 assert camm_info.gps[0].gps_fix_type == 3 # FIX_3D.value @@ -620,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, @@ -672,8 +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 - assert camm_info.gps[0].time_gps_epoch == 1706000000.0 - assert camm_info.gps[1].time_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 @@ -718,6 +721,142 @@ 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, + epoch_time=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 + + +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 4dd5c5c4..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, @@ -167,6 +167,8 @@ def test_encode_camm_gps_point(): encoded = PointEncoder.encode(p) assert len(encoded) == 6 assert encoded[0] == 2000 + # The description carries Unix time, the point carries GPS time + # Both the point and the description carry Unix time -- no conversion assert encoded[5] == 1700000001.0 @@ -179,7 +181,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] 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 @@ -203,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, @@ -215,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(): @@ -239,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 new file mode 100644 index 00000000..87627b70 --- /dev/null +++ b/tests/unit/test_gps_epoch.py @@ -0,0 +1,190 @@ +# 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 the epoch of CAMM GPS timestamps. + +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 + +from pathlib import Path + +import pytest + +from mapillary_tools import geo, telemetry +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 + + +# 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, epoch_time: float) -> telemetry.CAMMGPSPoint: + return telemetry.CAMMGPSPoint( + time=time, + lat=37.8436443, + lon=14.9886571, + alt=1202.345, + angle=None, + epoch_time=epoch_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, + ) + + +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_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 lands 18s too late + assert ( + telemetry.gps_epoch_to_unix(A_GPS_TIME) == A_GPS_TIME + GPS_UNIX_DELTA - 18 + ) + + +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 + + @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, 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 + 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): + 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, 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, 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, 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 + + +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..f1d61b74 100644 --- a/tests/unit/test_gpx_serializer.py +++ b/tests/unit/test_gpx_serializer.py @@ -151,14 +151,14 @@ 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, 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, @@ -168,7 +168,7 @@ 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 epoch_time, not the video time (5.0) assert gpx_pt.time is not None assert gpx_pt.time.timestamp() == 1700000000.0 diff --git a/tests/unit/test_parse_gpx.py b/tests/unit/test_parse_gpx.py index 17ca9770..411addd8 100644 --- a/tests/unit/test_parse_gpx.py +++ b/tests/unit/test_parse_gpx.py @@ -23,7 +23,9 @@ 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, 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 assert point.vertical_accuracy == 0.0