diff --git a/mapillary_tools/camm/camm_parser.py b/mapillary_tools/camm/camm_parser.py
index b5f42049..65934000 100644
--- a/mapillary_tools/camm/camm_parser.py
+++ b/mapillary_tools/camm/camm_parser.py
@@ -10,6 +10,8 @@
import dataclasses
import io
import logging
+import statistics
+import time
import typing as T
from enum import Enum
@@ -131,59 +133,139 @@ def extract_camm_info(fp: T.BinaryIO, telemetry_only: bool = False) -> CAMMInfo
# 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"])
+# as the CAMM spec describes. Everything else -- Insta360, for one -- records
+# Unix time in the same field, so converting unconditionally would push those
+# ~10 years into the future. Matched as a substring of the lowercased make,
+# because firmware reports the same vendor in more than one form.
+_GPS_EPOCH_MAKES = ("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
+# How close the median GPS timestamp of a track has to be to the mvhd
+# creation_time, read either as Unix time or as GPS time, for the creation time
+# to decide the epoch. The two readings are ten years apart, so this can be
+# generous enough to absorb local-time creation times and long recordings
+# without ever being ambiguous.
+_CREATION_TIME_TOLERANCE = 30 * 24 * 3600
+# A timestamp this far from the median of its track cannot belong to it: even
+# a time-lapse spans days, not months
+_STRAY_TOLERANCE = 30 * 24 * 3600
-def _normalize_gps_epochs(
- gps: list[telemetry.CAMMGPSPoint], make: str, moov: MovieBoxParser | None = None
-) -> None:
+
+def make_records_gps_time(make: str) -> bool:
"""
- Rewrite CAMMGPSPoint.epoch_time in place so it is Unix time regardless of
- which epoch the producer used.
+ Whether cameras of this make record GPS time in CAMM type 6 samples.
+
+ >>> make_records_gps_time("Labpano")
+ True
+ >>> make_records_gps_time("Labpano Technology Co.,Ltd")
+ True
+ >>> make_records_gps_time("Insta360")
+ False
+ """
+ normalized = make.strip().lower()
+ return any(m in normalized for m in _GPS_EPOCH_MAKES)
- This is the only place CAMM GPS timestamps change epoch. Everything
- downstream, including the serializer, treats them as Unix time.
+
+def _records_gps_time(
+ epoch_time: float, make: str, creation_time: float | None
+) -> bool:
"""
- if not gps:
- return
+ Decide the epoch of raw CAMM type 6 timestamps, given one representative
+ of them.
+
+ The creation time of the video decides when it can: GPS time reads as one
+ GPS epoch before it, Unix time reads close to it. The make decides only
+ when the creation time settles neither, because it is missing or
+ meaningless (a GoPro HERO7 recorded in 2022 reports 2016). Even then, a
+ timestamp that GPS time would put in the future is Unix time: that is what
+ mapillary_tools writes for a Labpano source without a creation time.
+
+ >>> creation_time = 1705574637 # 2024-01-18T10:43:57Z
+ >>> _records_gps_time(1389609661, "", creation_time) # GPS time
+ True
+ >>> _records_gps_time(1705574443, "Labpano", creation_time) # Unix time
+ False
+ >>> _records_gps_time(1389609661, "Labpano", None)
+ True
+ >>> _records_gps_time(1705574443, "Insta360", None)
+ False
+ """
+ if creation_time:
+ gap = creation_time - epoch_time
+ if abs(gap - telemetry.GPS_EPOCH_UNIX_OFFSET) < _CREATION_TIME_TOLERANCE:
+ return True
+ if abs(gap) < _CREATION_TIME_TOLERANCE:
+ return False
- 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)
+ if not make_records_gps_time(make):
+ return False
- first = next((p.epoch_time for p in gps if p.epoch_time > 0), None)
- if first is None or moov is None:
- return
+ # Read as GPS time, Unix time lands ten years late, so in the future for
+ # any recording less than ten years old
+ return (
+ telemetry.gps_epoch_to_unix(epoch_time) < time.time() + _CREATION_TIME_TOLERANCE
+ )
+
+
+def _extract_creation_time(moov: MovieBoxParser | None) -> float | None:
+ """Return the mvhd creation_time as Unix time, or None if unset."""
+ if moov is None:
+ return None
try:
creation_time = moov.extract_mvhd_boxdata().get("creation_time", 0)
except Exception:
- return
+ return None
if not creation_time:
+ return None
+
+ return creation_time - _MP4_EPOCH_UNIX_OFFSET
+
+
+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, and drop the points whose timestamps are
+ strays.
+
+ This is the only place CAMM GPS timestamps change epoch. Everything
+ downstream, including the serializer, treats them as Unix time.
+ """
+ epoch_times = [p.epoch_time for p in gps if p.epoch_time > 0]
+ if not epoch_times:
return
- gap = abs(first - (creation_time - _MP4_EPOCH_UNIX_OFFSET))
- if abs(gap - telemetry.GPS_EPOCH_UNIX_OFFSET) < _GPS_EPOCH_GAP_TOLERANCE:
+ # Decide by the median rather than by any one sample, so that a stray
+ # timestamp cannot flip the epoch of the whole track
+ median_epoch_time = statistics.median(epoch_times)
+
+ # A stray timestamp would still be wrong by years once converted with the
+ # track, and would corrupt the timestamps interpolated next to it
+ kept = [
+ p
+ for p in gps
+ if p.epoch_time <= 0 or abs(p.epoch_time - median_epoch_time) < _STRAY_TOLERANCE
+ ]
+ if len(kept) < len(gps):
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,
+ "Dropped %d of %d CAMM GPS points whose timestamps are more than 30 days from the rest of the track",
+ len(gps) - len(kept),
+ len(gps),
)
+ gps[:] = kept
+
+ if not _records_gps_time(median_epoch_time, make, _extract_creation_time(moov)):
+ return
+
+ for point in gps:
+ if point.epoch_time > 0:
+ point.epoch_time = telemetry.gps_epoch_to_unix(point.epoch_time)
def extract_camera_make_and_model(fp: T.BinaryIO) -> tuple[str, str]:
@@ -305,9 +387,10 @@ def serialize(cls, data: telemetry.CAMMGPSPoint) -> bytes:
{
"type": cls.serialized_camm_type.value,
"data": {
- # 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.
+ # Written as Unix time, whatever the make. Readers tell it
+ # from the GPS time some cameras record by the creation
+ # time the file carries (see _records_gps_time). Do not
+ # convert here.
"time_gps_epoch": data.epoch_time,
"gps_fix_type": data.gps_fix_type,
"latitude": data.lat,
diff --git a/mapillary_tools/exceptions.py b/mapillary_tools/exceptions.py
index 5474fd1b..af1f34cb 100644
--- a/mapillary_tools/exceptions.py
+++ b/mapillary_tools/exceptions.py
@@ -72,7 +72,9 @@ class MapillaryStationaryVideoError(MapillaryDescriptionError):
pass
-class MapillaryOutsideGPXTrackError(MapillaryDescriptionError):
+# A geotagging error, so that the next geotag source gets its turn: a GPX track
+# that misses the capture time says nothing about the file itself
+class MapillaryOutsideGPXTrackError(MapillaryGeoTaggingError):
def __init__(
self, message: str, image_time: str, gpx_start_time: str, gpx_end_time: str
):
@@ -81,6 +83,14 @@ def __init__(
self.gpx_start_time = gpx_start_time
self.gpx_end_time = gpx_end_time
+ def __reduce__(self):
+ # Pickle with every argument, so that the error survives the trip back
+ # from a worker process (video geotagging raises it in one)
+ return (
+ self.__class__,
+ (self.args[0], self.image_time, self.gpx_start_time, self.gpx_end_time),
+ )
+
class MapillaryDuplicationError(MapillaryDescriptionError):
def __init__(
diff --git a/mapillary_tools/exiftool_read_video.py b/mapillary_tools/exiftool_read_video.py
index 4257abca..a9834c11 100644
--- a/mapillary_tools/exiftool_read_video.py
+++ b/mapillary_tools/exiftool_read_video.py
@@ -11,7 +11,8 @@
import typing as T
import xml.etree.ElementTree as ET
-from . import exif_read, exiftool_read, geo
+from . import exif_read, exiftool_read, geo, telemetry
+from .camm import camm_parser
from .telemetry import GPSFix, GPSPoint
from .utils import sanitize_serial
@@ -51,6 +52,17 @@ def _maybe_float(text: str | None) -> float | None:
return None
+def _exiftool_gps_time_to_unix(exiftool_time: float) -> float:
+ """
+ Convert a CAMM GPS timestamp as ExifTool renders it -- GPS time plus the
+ epoch difference, with no leap-second correction -- to Unix time (UTC).
+
+ >>> _exiftool_gps_time_to_unix(1705574461.6) # 2024-01-18T10:41:01.6
+ 1705574443.6
+ """
+ return telemetry.gps_epoch_to_unix(exiftool_time - telemetry.GPS_EPOCH_UNIX_OFFSET)
+
+
def _index_text_by_tag(elements: T.Iterable[ET.Element]) -> dict[str, list[str]]:
texts_by_tag: dict[str, list[str]] = {}
for element in elements:
@@ -550,9 +562,41 @@ def _extract_gps_track_from_track(self) -> list[GPSPoint]:
gps_precision_tag=f"{track_ns}:GPSHPositioningError",
)
if track:
+ if self._is_camm_track_in_gps_time(track_ns):
+ LOG.debug(
+ f"Correcting the CAMM GPS timestamps in {track_ns} by the leap seconds"
+ )
+ for point in track:
+ if point.epoch_time is not None:
+ point.epoch_time = _exiftool_gps_time_to_unix(
+ point.epoch_time
+ )
return track
return []
+ def _is_camm_track_in_gps_time(self, track_ns: str) -> bool:
+ """
+ Whether the track is CAMM from a camera that records GPS time.
+
+ ExifTool converts those timestamps by the epoch difference alone, so
+ they read 18s (the leap seconds since 1980) ahead of what the native
+ CAMM parser returns for the same video.
+
+ Only camera originals qualify. Cameras put their CAMM track under a
+ meta handler, which ExifTool reports as MetaFormat. The CAMM tracks
+ mapillary_tools writes use a camm handler, which ExifTool reports as
+ OtherFormat, and hold Unix time, which ExifTool reads as is.
+ """
+ meta_format = self._extract_alternative_fields([f"{track_ns}:MetaFormat"], str)
+ if (meta_format or "").strip().lower() != "camm":
+ return False
+
+ make = self.extract_make()
+ if not make:
+ return False
+
+ return camm_parser.make_records_gps_time(make)
+
def _extract_alternative_fields(
self,
fields: T.Sequence[str],
diff --git a/mapillary_tools/geotag/video_extractors/gpx.py b/mapillary_tools/geotag/video_extractors/gpx.py
index 00722bd1..cd8973e0 100644
--- a/mapillary_tools/geotag/video_extractors/gpx.py
+++ b/mapillary_tools/geotag/video_extractors/gpx.py
@@ -6,18 +6,23 @@
from __future__ import annotations
import dataclasses
+import datetime
import enum
import logging
import sys
import typing as T
from pathlib import Path
+import construct as C
+
if sys.version_info >= (3, 12):
from typing import override
else:
from typing_extensions import override
from ... import exceptions, geo, types, utils
+from ...mp4 import construct_mp4_parser as cparser, simple_mp4_parser as sparser
+from ...serializer.description import build_capture_time
from ..utils import parse_gpx
from .base import BaseVideoExtractor
from .native import NativeVideoExtractor
@@ -25,10 +30,16 @@
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
+# When the duration of the video is unknown, a GPX track that misses the
+# video's GPS by more than this cannot belong to it. No camera clock or time
+# zone mistake comes close, while an epoch mix-up exceeds it by orders of
+# magnitude.
+_IMPLAUSIBLE_GAP_SECONDS = 24 * 3600
+
+# How much of the video a GPX track may leave uncovered without a warning. A
+# logger that records whole seconds once a second, started and stopped with
+# the camera, can leave a second at each end.
+_UNCOVERED_TOLERANCE_SECONDS = 2.0
class SyncMode(enum.Enum):
@@ -78,19 +89,101 @@ 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,
+ if gpx_points and native_video_metadata.points:
+ self._check_time_gap(
+ gpx_points,
+ native_video_metadata.points,
offset,
- offset / 86400,
+ self._video_duration(),
)
self._rebase_times(gpx_points, offset=offset)
return dataclasses.replace(native_video_metadata, points=gpx_points)
+ def _video_duration(self) -> float | None:
+ """
+ The duration of the video in seconds, from its mvhd box, or None if it
+ cannot be read.
+ """
+ try:
+ with self.video_path.open("rb") as fp:
+ data = sparser.parse_box_data_first(fp, [b"moov", b"mvhd"])
+ if data is None:
+ return None
+ mvhd = cparser.MovieHeaderBox.parse(data)
+ except (OSError, sparser.ParsingError, C.ConstructError) as ex:
+ LOG.debug("Failed to read the duration of %s: %s", self.video_path, ex)
+ return None
+
+ # All 1s means the duration is unknown
+ if not mvhd.timescale or mvhd.duration in (0, 0xFFFFFFFF, 0xFFFFFFFFFFFFFFFF):
+ return None
+
+ return mvhd.duration / mvhd.timescale
+
+ def _check_time_gap(
+ self,
+ gpx_points: T.Sequence[geo.Point],
+ video_gps_points: T.Sequence[geo.Point],
+ offset: float,
+ video_duration: float | None,
+ ) -> None:
+ """
+ Check the GPX track, once synced by offset, against the video in time.
+
+ A GPX track that misses the video raises: it has no position for any
+ moment of the video. A track that covers only part of the video warns,
+ and one that covers all of it is silent.
+
+ When the video's GPS has no timestamps, the offset is 0 and the track
+ starts at video time 0, so only the end of the video can go uncovered.
+
+ Without the duration of the video, the video is known only up to its
+ last GPS point, and a GPX that starts after it may still overlap the
+ video. Then a gap only warns, unless it is too large for any clock or
+ time zone mistake to explain.
+ """
+ # The Unix time of video time 0, in the convention _rebase_times() uses
+ video_start_time = gpx_points[0].time - offset
+ # From video time 0, since the frames start there even when the video's
+ # own GPS starts later
+ video_end_time = video_start_time + max(p.time for p in video_gps_points)
+ if video_duration is not None:
+ video_end_time = max(video_end_time, video_start_time + video_duration)
+ gpx_first = min(p.time for p in gpx_points)
+ gpx_last = max(p.time for p in gpx_points)
+
+ gpx_track = f"The GPX track in {self.gpx_path} ({_isoformat(gpx_first)} to {_isoformat(gpx_last)})"
+ video = f"the video {self.video_path} ({_isoformat(video_start_time)} to {_isoformat(video_end_time)})"
+
+ gap = max(gpx_first - video_end_time, video_start_time - gpx_last)
+ if gap <= 0:
+ uncovered = max(gpx_first - video_start_time, 0) + max(
+ video_end_time - gpx_last, 0
+ )
+ if uncovered > _UNCOVERED_TOLERANCE_SECONDS:
+ LOG.warning(
+ f"{gpx_track} covers only part of {video}: "
+ f"{_format_duration(uncovered)} of the video fall outside the track"
+ )
+ return
+
+ message = (
+ f"{gpx_track} misses {video} by {_format_duration(gap)}. Check that "
+ "the GPX file belongs to this video, then the camera clock and the "
+ "time zone of the GPX timestamps"
+ )
+
+ if video_duration is not None or gap > _IMPLAUSIBLE_GAP_SECONDS:
+ raise exceptions.MapillaryOutsideGPXTrackError(
+ message,
+ image_time=build_capture_time(video_start_time),
+ gpx_start_time=build_capture_time(gpx_first),
+ gpx_end_time=build_capture_time(gpx_last),
+ )
+
+ LOG.warning(message)
+
@classmethod
def _rebase_times(cls, points: T.Sequence[geo.Point], offset: float = 0.0) -> None:
"""
@@ -121,13 +214,46 @@ def _gpx_offset(
if not gpx_points or not video_gps_points:
return offset
- # 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()
+ # Both sides must be Unix time here. get_unix_time() skips
+ # zero/invalid timestamps, and points that carry none at all, like
+ # CAMM type 5 points, which a track can start with.
+ anchor = next(
+ (p for p in video_gps_points if p.get_unix_time() is not None), None
+ )
- if video_unix_time is not None:
+ if anchor is not None:
+ anchor_unix_time = T.cast(float, anchor.get_unix_time())
+ # The Unix time of video time 0
+ video_unix_time = anchor_unix_time - anchor.time
offset = gpx_points[0].time - video_unix_time
return offset
+
+
+def _isoformat(unix_time: float) -> str:
+ return datetime.datetime.fromtimestamp(
+ unix_time, tz=datetime.timezone.utc
+ ).isoformat()
+
+
+def _format_duration(seconds: float) -> str:
+ """
+ >>> _format_duration(0.089)
+ '0.089 seconds'
+ >>> _format_duration(1)
+ '1 second'
+ >>> _format_duration(13)
+ '13 seconds'
+ >>> _format_duration(90)
+ '1.5 minutes'
+ >>> _format_duration(2 * 3600)
+ '2.0 hours'
+ >>> _format_duration(3 * 86400)
+ '3.0 days'
+ """
+ for unit, size in (("days", 86400), ("hours", 3600), ("minutes", 60)):
+ if seconds >= size:
+ return f"{seconds / size:.1f} {unit}"
+ # Significant digits, so that a gap under a second does not read as 0
+ text = f"{seconds:.3g}"
+ return f"{text} second" if text == "1" else f"{text} seconds"
diff --git a/mapillary_tools/geotag/video_extractors/native.py b/mapillary_tools/geotag/video_extractors/native.py
index a4a329e7..f4a60af3 100644
--- a/mapillary_tools/geotag/video_extractors/native.py
+++ b/mapillary_tools/geotag/video_extractors/native.py
@@ -69,11 +69,19 @@ def extract(self) -> types.VideoMetadata:
if not camm_info.gps and not camm_info.mini_gps:
raise exceptions.MapillaryGPXEmptyError("Empty GPS data found")
+ # A track may mix type 6 and type 5 samples, so use both. No camera is
+ # known to interleave them, and it would cost the absolute timestamps
+ # of sampled frames: interpolating between the two types returns a
+ # plain geo.Point (see CAMMGPSPoint.interpolate_with), so sample_video
+ # falls back to the container start time.
+ points: list[geo.Point] = [*(camm_info.gps or []), *(camm_info.mini_gps or [])]
+ points.sort(key=lambda p: p.time)
+
return types.VideoMetadata(
filename=self.video_path,
filesize=utils.get_file_size(self.video_path),
filetype=types.FileType.CAMM,
- points=T.cast(T.List[geo.Point], camm_info.gps or camm_info.mini_gps),
+ points=points,
make=camm_info.make,
model=camm_info.model,
)
diff --git a/tests/cli/camm_parser.py b/tests/cli/camm_parser.py
index 1acd0155..ff8f7ed8 100644
--- a/tests/cli/camm_parser.py
+++ b/tests/cli/camm_parser.py
@@ -44,7 +44,9 @@ def _convert(path: pathlib.Path):
track.description = "Invalid CAMM video"
return track
- points = T.cast(T.List[geo.Point], camm_info.gps or camm_info.mini_gps)
+ # A track may mix type 6 and type 5 samples, so use both
+ points: T.List[geo.Point] = [*(camm_info.gps or []), *(camm_info.mini_gps or [])]
+ points.sort(key=lambda p: p.time)
track.segments.append(_convert_points_to_gpx_segment(points))
make_model = json.dumps({"make": camm_info.make, "model": camm_info.model})
diff --git a/tests/integration/test_process.py b/tests/integration/test_process.py
index 765fad80..df09b3d7 100644
--- a/tests/integration/test_process.py
+++ b/tests/integration/test_process.py
@@ -9,6 +9,7 @@
import subprocess
from pathlib import Path
+import gpxpy
import py.path
import pytest
@@ -623,11 +624,32 @@ def test_process_video_geotag_source_gpx_not_found(setup_data: py.path.local):
assert descs[0]["error"]["type"] == "MapillaryVideoGPSNotFoundError"
+# The GPS track of gopro_data/max-360mode.mp4 starts at 2019-11-18T23:44:42.59Z
+_GOPRO_MAX_GPS_START = datetime.datetime(
+ 2019, 11, 18, 23, 44, 40, tzinfo=datetime.timezone.utc
+)
+
+
+def _copy_gpx_shifted_to(
+ src: py.path.local, dst: py.path.local, start: datetime.datetime
+) -> None:
+ """Copy a GPX file with its times shifted to begin at start."""
+ with open(src) as fp:
+ gpx = gpxpy.parse(fp)
+ start_time = gpx.get_time_bounds().start_time
+ assert start_time is not None
+ gpx.adjust_time(start - start_time)
+ dst.write(gpx.to_xml())
+
+
def test_process_video_geotag_source_with_gopro_gpx_specified(
setup_data: py.path.local,
):
video_path = setup_data.join("gopro_data").join("max-360mode.mp4")
- gpx_file = setup_data.join("gpx").join("sf_30km_h.gpx")
+ gpx_file = setup_data.join("gpx").join("max-360mode.gpx")
+ _copy_gpx_shifted_to(
+ setup_data.join("gpx").join("sf_30km_h.gpx"), gpx_file, _GOPRO_MAX_GPS_START
+ )
descs = run_process_for_descs(
[
@@ -645,6 +667,33 @@ def test_process_video_geotag_source_with_gopro_gpx_specified(
assert len(descs[0]["MAPGPSTrack"]) > 0
+def test_process_video_geotag_source_with_gpx_outside_video(
+ setup_data: py.path.local,
+):
+ """A GPX file recorded at another time than the video must not sync."""
+ video_path = setup_data.join("gopro_data").join("max-360mode.mp4")
+ # Recorded in 2025, five years after the video
+ gpx_file = setup_data.join("gpx").join("sf_30km_h.gpx")
+
+ descs = run_process_for_descs(
+ [
+ *[
+ "--video_geotag_source",
+ json.dumps({"source": "gpx", "source_path": str(gpx_file)}),
+ ],
+ str(video_path),
+ ]
+ )
+
+ assert len(descs) == 1
+ assert descs[0]["error"]["type"] == "MapillaryOutsideGPXTrackError"
+ assert descs[0]["error"]["vars"] == {
+ "image_time": "2019_11_18_23_44_42_590",
+ "gpx_start_time": "2025_03_14_07_00_00_000",
+ "gpx_end_time": "2025_03_14_07_01_33_624",
+ }
+
+
def test_process_geotag_with_gpx_pattern_not_found(setup_data: py.path.local):
video_path = setup_data.join("gopro_data").join("max-360mode.mp4")
@@ -661,8 +710,11 @@ def test_process_geotag_with_gpx_pattern_not_found(setup_data: py.path.local):
def test_process_geotag_with_gpx_pattern(setup_data: py.path.local):
video_path = setup_data.join("gopro_data").join("max-360mode.mp4")
- gpx_file = setup_data.join("gpx").join("sf_30km_h.gpx")
- gpx_file.copy(setup_data.join("gopro_data").join("max-360mode.gpx"))
+ _copy_gpx_shifted_to(
+ setup_data.join("gpx").join("sf_30km_h.gpx"),
+ setup_data.join("gopro_data").join("max-360mode.gpx"),
+ _GOPRO_MAX_GPS_START,
+ )
descs = run_process_for_descs(
[
diff --git a/tests/unit/test_camm_parser.py b/tests/unit/test_camm_parser.py
index 112a1df5..b687c349 100644
--- a/tests/unit/test_camm_parser.py
+++ b/tests/unit/test_camm_parser.py
@@ -802,10 +802,9 @@ def test_extract_camm_info_routes_plain_points_to_mini_gps():
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.
+ mapillary_tools writes Unix time into the CAMM type 6 time_gps_epoch
+ field whatever the make, as every released version has. Makes that record
+ GPS time are covered in test_gps_epoch.py.
"""
unix_times = [1655503450.5, 1655503451.5]
points = [
diff --git a/tests/unit/test_gps_epoch.py b/tests/unit/test_gps_epoch.py
index 87627b70..8741f5b5 100644
--- a/tests/unit/test_gps_epoch.py
+++ b/tests/unit/test_gps_epoch.py
@@ -4,7 +4,7 @@
# LICENSE file in the root directory of this source tree.
"""
-Regression tests for the epoch of CAMM GPS timestamps.
+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),
@@ -12,22 +12,37 @@
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.
+Unix time in memory. It is converted from GPS time once, when the CAMM track is
+parsed, and never on the way out: mapillary_tools writes Unix time whatever the
+make.
+
+Which epoch a track records is decided by the mvhd creation time of the video
+when that is conclusive, and by the make otherwise. The CAMM tracks
+mapillary_tools writes carry the make and the creation time of their source,
+so for a Labpano video it is the creation time that says Unix time.
"""
from __future__ import annotations
+import datetime
+import io
+import logging
+import pickle
+import shutil
+import typing as T
+import xml.etree.ElementTree as ET
from pathlib import Path
import pytest
-
-from mapillary_tools import geo, telemetry
+from mapillary_tools import exceptions, exiftool_read, geo, telemetry, types, uploader
from mapillary_tools.camm import camm_builder, camm_parser
-from mapillary_tools.geotag.options import SourceOption, SourceType
+from mapillary_tools.exiftool_read_video import ExifToolReadVideo
+from mapillary_tools.exiftool_runner import ExiftoolRunner
+from mapillary_tools.geotag import factory
+from mapillary_tools.geotag.options import SourceOption, SourcePathOption, SourceType
from mapillary_tools.geotag.video_extractors.gpx import GPXVideoExtractor
-from mapillary_tools.mp4 import construct_mp4_parser as cparser
+from mapillary_tools.geotag.video_extractors.native import CAMMVideoExtractor
+from mapillary_tools.mp4 import construct_mp4_parser as cparser, simple_mp4_builder
# Seconds between the two epochs, i.e. the size of the bug
@@ -37,6 +52,24 @@
A_GPS_TIME = 1470558405.9798455
A_UNIX_TIME = 1786523187.9798455
+# Seconds between the mp4 epoch (1904-01-01) and the Unix epoch
+MP4_UNIX_DELTA = 2082844800
+
+# When the tests read tracks: two months after A_UNIX_TIME
+NOW = A_UNIX_TIME + 60 * 24 * 3600
+
+
+class _PinnedClock:
+ @staticmethod
+ def time() -> float:
+ return NOW
+
+
+@pytest.fixture
+def pinned_now(monkeypatch):
+ """Pin the clock the make fallback reads, so the tests do not expire."""
+ monkeypatch.setattr(camm_parser, "time", _PinnedClock)
+
def _camm_point(time: float, epoch_time: float) -> telemetry.CAMMGPSPoint:
return telemetry.CAMMGPSPoint(
@@ -70,6 +103,71 @@ def _gps_point(time: float, epoch_time: float | None) -> telemetry.GPSPoint:
)
+def _write_camm_mp4(
+ points: T.Sequence[geo.Point],
+ make: str,
+ creation_time: float | None,
+ duration: float = 10.0,
+) -> bytes:
+ """
+ Write points as a CAMM track into an empty mp4, the way the uploader does.
+
+ creation_time is the Unix time to put in mvhd, or None to leave it unset.
+ duration is the duration of the video in seconds, or 0 for unknown.
+ """
+ mp4_creation_time = (
+ 0 if creation_time is None else int(creation_time) + MP4_UNIX_DELTA
+ )
+ mvhd: cparser.BoxDict = {
+ "type": b"mvhd",
+ "data": {
+ "creation_time": mp4_creation_time,
+ "modification_time": mp4_creation_time,
+ "timescale": 1000,
+ "duration": int(duration * 1000),
+ },
+ }
+ src = cparser.MP4WithoutSTBLBuilderConstruct.build_boxlist(
+ [
+ {"type": b"ftyp", "data": b"test"},
+ {"type": b"moov", "data": [mvhd]},
+ ]
+ )
+ metadata = types.VideoMetadata(
+ Path(""),
+ filetype=types.FileType.CAMM,
+ points=list(points),
+ make=make,
+ model="PanoX V2",
+ )
+ camm_info = uploader.VideoUploader.prepare_camm_info(metadata)
+ target_fp = simple_mp4_builder.transform_mp4(
+ io.BytesIO(src), camm_builder.camm_sample_generator2(camm_info)
+ )
+ return target_fp.read()
+
+
+def _read_camm(data: bytes) -> camm_parser.CAMMInfo:
+ camm_info = camm_parser.extract_camm_info(io.BytesIO(data))
+ assert camm_info is not None
+ return camm_info
+
+
+def _unix_times(camm_info: camm_parser.CAMMInfo) -> list[float]:
+ return [p.epoch_time for p in camm_info.gps or []]
+
+
+# A two point track at A_UNIX_TIME, as parse_gpx() or the parser would return it
+UNIX_TIMES = [A_UNIX_TIME, A_UNIX_TIME + 1]
+
+
+def _unix_track() -> list[telemetry.CAMMGPSPoint]:
+ return [
+ _camm_point(time=float(idx), epoch_time=epoch_time)
+ for idx, epoch_time in enumerate(UNIX_TIMES)
+ ]
+
+
class TestEpochConversion:
def test_known_instant(self):
assert telemetry.gps_epoch_to_unix(A_GPS_TIME) == A_UNIX_TIME
@@ -112,6 +210,191 @@ def test_invalid_timestamps_are_not_converted(self):
assert points[0].epoch_time == 0.0
assert points[0].get_unix_time() is None
+ @pytest.mark.parametrize(
+ "make", ["Labpano Technology Co.,Ltd", "LABPANO TECHNOLOGY", "Labpano Pilot"]
+ )
+ def test_make_variants_are_matched(self, make: str):
+ """Firmware reports the vendor in more than one form."""
+ 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
+
+
+class TestCreationTimeEvidence:
+ """The creation time decides the epoch whenever it is conclusive."""
+
+ # Labpano stamps the creation time at the end of the recording
+ CREATION_TIME = A_UNIX_TIME + 600
+
+ @pytest.mark.parametrize("make", ["", "Insta360", "Some Future Camera"])
+ def test_gps_time_is_recognized_whatever_the_make(self, make: str):
+ assert camm_parser._records_gps_time(A_GPS_TIME, make, self.CREATION_TIME)
+
+ @pytest.mark.parametrize("make", ["Labpano", "Labpano Technology Co.,Ltd"])
+ def test_unix_time_is_recognized_whatever_the_make(self, make: str):
+ assert not camm_parser._records_gps_time(A_UNIX_TIME, make, self.CREATION_TIME)
+
+ def test_inconclusive_creation_time_falls_back_to_make(self):
+ # A GoPro HERO7 recorded in 2022 reports 2016
+ meaningless = A_UNIX_TIME - 6 * 365 * 24 * 3600
+ assert camm_parser._records_gps_time(A_GPS_TIME, "Labpano", meaningless)
+ assert not camm_parser._records_gps_time(A_UNIX_TIME, "Insta360", meaningless)
+
+ def test_unknown_make_in_gps_time_is_read_from_the_file(self):
+ """The mvhd creation time reaches the decision through the parser."""
+ data = _write_camm_mp4(
+ [_camm_point(time=0.0, epoch_time=A_GPS_TIME)],
+ make="",
+ creation_time=self.CREATION_TIME,
+ )
+ assert _unix_times(_read_camm(data)) == [A_UNIX_TIME]
+
+
+@pytest.mark.usefixtures("pinned_now")
+class TestWithoutCreationTime:
+ """
+ Without a conclusive creation time the make decides, but GPS time must
+ not put a timestamp in the future: Unix time read as GPS time lands ten
+ years late. That keeps our output of a Labpano source without a creation
+ time from reading back as 2036, for recordings less than ten years old
+ when read.
+ """
+
+ @pytest.mark.parametrize("make", ["Labpano", "Labpano Technology Co.,Ltd"])
+ def test_unix_time_is_not_converted(self, make: str):
+ assert not camm_parser._records_gps_time(A_UNIX_TIME, make, None)
+
+ def test_gps_time_is_converted(self):
+ assert camm_parser._records_gps_time(A_GPS_TIME, "Labpano", None)
+
+ def test_labpano_original_is_converted(self):
+ data = _write_camm_mp4(
+ [_camm_point(time=0.0, epoch_time=A_GPS_TIME)], "Labpano", None
+ )
+ assert _unix_times(_read_camm(data)) == [A_UNIX_TIME]
+
+
+class TestStraySamples:
+ """
+ The whole track decides its epoch, not its first timestamp, and a
+ timestamp more than 30 days from the rest of the track is dropped.
+ """
+
+ CREATION_TIME = A_UNIX_TIME + 600
+
+ def _read_back(self, raw_times: list[float]) -> list[float]:
+ points = [
+ _camm_point(time=float(idx), epoch_time=raw_time)
+ for idx, raw_time in enumerate(raw_times)
+ ]
+ # Written as is, so the raw times are what the reader sees
+ data = _write_camm_mp4(points, "Labpano", self.CREATION_TIME)
+ return _unix_times(_read_camm(data))
+
+ def test_stray_unix_time_does_not_flip_a_gps_time_track(self):
+ raw_times = [A_UNIX_TIME] + [A_GPS_TIME + t for t in range(1, 10)]
+ assert self._read_back(raw_times) == [A_UNIX_TIME + t for t in range(1, 10)]
+
+ def test_stray_gps_time_does_not_flip_a_unix_time_track(self):
+ raw_times = [A_GPS_TIME] + [A_UNIX_TIME + t for t in range(1, 10)]
+ assert self._read_back(raw_times) == [A_UNIX_TIME + t for t in range(1, 10)]
+
+ def test_stray_is_dropped_wherever_it_is(self, caplog):
+ """Kept, it would sit ten years off, amid the track it interrupts."""
+ raw_times = [A_UNIX_TIME + t for t in range(10)]
+ raw_times[5] = A_GPS_TIME + 5
+
+ with caplog.at_level(logging.WARNING):
+ unix_times = self._read_back(raw_times)
+
+ assert unix_times == [A_UNIX_TIME + t for t in range(10) if t != 5]
+ [record] = caplog.records
+ assert "Dropped 1 of 10" in record.getMessage()
+
+ def test_points_without_a_timestamp_are_not_strays(self):
+ points = [_camm_point(time=0.0, epoch_time=0.0)] + [
+ _camm_point(time=float(t), epoch_time=A_GPS_TIME + t) for t in range(1, 5)
+ ]
+ camm_parser._normalize_gps_epochs(points, "Labpano")
+ assert [p.epoch_time for p in points] == [0.0] + [
+ A_UNIX_TIME + t for t in range(1, 5)
+ ]
+
+
+class TestWriteRoundTrip:
+ """
+ Whatever mapillary_tools writes must read back to the same Unix times.
+
+ It writes Unix time whatever the make. The source make is copied into the
+ output, so for a Labpano video it is the creation time, copied from the
+ source too, that keeps the reader from converting it one GPS epoch later:
+ 2026 used to become 2036.
+ """
+
+ @pytest.mark.parametrize(
+ "make", ["Labpano", "Labpano Technology Co.,Ltd", "Insta360", ""]
+ )
+ def test_round_trip(self, make: str):
+ data = _write_camm_mp4(_unix_track(), make, A_UNIX_TIME + 600)
+ assert _unix_times(_read_camm(data)) == UNIX_TIMES
+
+ @pytest.mark.usefixtures("pinned_now")
+ @pytest.mark.parametrize(
+ "make", ["Labpano", "Labpano Technology Co.,Ltd", "Insta360", ""]
+ )
+ def test_round_trip_without_creation_time(self, make: str):
+ data = _write_camm_mp4(_unix_track(), make, None)
+ assert _unix_times(_read_camm(data)) == UNIX_TIMES
+
+ def test_reprocessing_output_is_stable(self):
+ """process_and_upload output, processed again, is the same track."""
+ data = _write_camm_mp4(_unix_track(), "Labpano", A_UNIX_TIME + 600)
+ for _ in range(2):
+ camm_info = _read_camm(data)
+ assert _unix_times(camm_info) == UNIX_TIMES
+ data = _write_camm_mp4(
+ camm_info.gps or [], camm_info.make, A_UNIX_TIME + 600
+ )
+
+ @pytest.mark.parametrize("make", ["Labpano", "Insta360", ""])
+ def test_written_as_unix_time_whatever_the_make(self, monkeypatch, make: str):
+ data = _write_camm_mp4(_unix_track(), make, A_UNIX_TIME + 600)
+ monkeypatch.setattr(camm_parser, "_normalize_gps_epochs", lambda *_: None)
+ assert _unix_times(_read_camm(data)) == UNIX_TIMES
+
+
+class TestMixedCAMMTypes:
+ """A track may mix type 6 and type 5 samples; neither may be dropped."""
+
+ def _mixed_track(self) -> list[geo.Point]:
+ return [
+ geo.Point(time=0.0, lat=37.0, lon=14.0, alt=None, angle=None),
+ _camm_point(time=1.0, epoch_time=A_UNIX_TIME + 1),
+ geo.Point(time=2.0, lat=37.0, lon=14.0, alt=None, angle=None),
+ _camm_point(time=3.0, epoch_time=A_UNIX_TIME + 3),
+ ]
+
+ def test_extractor_returns_both_types_in_order(self, tmp_path: Path):
+ video_path = tmp_path / "mixed.mp4"
+ video_path.write_bytes(
+ _write_camm_mp4(self._mixed_track(), "Labpano", A_UNIX_TIME + 600)
+ )
+
+ points = CAMMVideoExtractor(video_path).extract().points
+
+ assert [p.time for p in points] == [0.0, 1.0, 2.0, 3.0]
+ assert [type(p) for p in points] == [
+ geo.Point,
+ telemetry.CAMMGPSPoint,
+ geo.Point,
+ telemetry.CAMMGPSPoint,
+ ]
+
+ def test_gpx_offset_anchors_on_the_first_timestamped_point(self):
+ # The GPX starts 30s before the video does
+ gpx_points = [_camm_point(time=A_UNIX_TIME - 30, epoch_time=A_UNIX_TIME - 30)]
+ assert GPXVideoExtractor._gpx_offset(gpx_points, self._mixed_track()) == -30.0
+
class TestPointAccessors:
def test_both_point_types_report_unix_time(self):
@@ -152,6 +435,454 @@ def test_missing_video_timestamp_yields_no_offset(self):
video_points = [_camm_point(time=0.0, epoch_time=0.0)]
assert GPXVideoExtractor._gpx_offset(gpx_points, video_points) == 0.0
+ def test_video_gps_starting_late_syncs_to_video_time(self):
+ """
+ The offset is to video time 0, not to the first video GPS point, or
+ the GPX track would land early by that point's time.
+ """
+ gpx_points = _gpx_track(A_UNIX_TIME, 20)
+ # The video's GPS starts 5s into the video
+ video_points = [
+ _camm_point(time=5.0 + t, epoch_time=A_UNIX_TIME + 5 + t) for t in range(10)
+ ]
+
+ offset = GPXVideoExtractor._gpx_offset(gpx_points, video_points)
+ GPXVideoExtractor._rebase_times(gpx_points, offset=offset)
+
+ # Recorded at the same instant as the first video GPS point, so it
+ # lands at the same video time
+ assert gpx_points[5].time == 5.0
+
+
+def _gpx_track(start: float, duration: int) -> list[telemetry.CAMMGPSPoint]:
+ return [
+ _camm_point(time=start + t, epoch_time=start + t) for t in range(duration + 1)
+ ]
+
+
+def _write_gpx(path: Path, points: T.Sequence[geo.Point]) -> None:
+ trkpts = "".join(
+ f''
+ for p in points
+ )
+ path.write_text(
+ ''
+ ''
+ f"{trkpts}"
+ )
+
+
+def _isoformat(unix_time: float) -> str:
+ return datetime.datetime.fromtimestamp(unix_time, datetime.timezone.utc).strftime(
+ "%Y-%m-%dT%H:%M:%SZ"
+ )
+
+
+class TestGPXTimeGap:
+ """
+ A GPX track that misses the video must fail, not sync.
+
+ A track that misses the video has no position for any moment of it: an
+ epoch mix-up, a GPX file from another day, or naive GPX timestamps read
+ in the wrong time zone. A track that covers only part of the video warns.
+ """
+
+ # A 10s video track starting at A_UNIX_TIME, in a 12s video
+ VIDEO = [_camm_point(time=float(t), epoch_time=A_UNIX_TIME + t) for t in range(11)]
+ VIDEO_DURATION = 12.0
+
+ # GPS without timestamps, so the GPX can only be aligned to video time 0
+ UNTIMED_VIDEO = [
+ geo.Point(time=float(t), lat=37.0, lon=14.0, alt=None, angle=None)
+ for t in range(11)
+ ]
+ # Timestamps in whole seconds, as in the GPX, so that it syncs to exactly 0
+ WHOLE_SECOND_VIDEO = [
+ _camm_point(time=float(t), epoch_time=int(A_UNIX_TIME) + t) for t in range(11)
+ ]
+
+ def _check(
+ self,
+ gpx_points: list[telemetry.CAMMGPSPoint],
+ video: T.Sequence[geo.Point] = VIDEO,
+ duration: float | None = VIDEO_DURATION,
+ ) -> None:
+ extractor = GPXVideoExtractor(Path("video.mp4"), Path("track.gpx"))
+ offset = extractor._gpx_offset(gpx_points, video)
+ extractor._check_time_gap(gpx_points, video, offset, duration)
+
+ @pytest.mark.parametrize(
+ "start, duration",
+ [
+ # Starts 30s before the video and ends after it
+ (A_UNIX_TIME - 30, 60),
+ # Started a second into the video, and ends with it
+ (A_UNIX_TIME + 1, 11),
+ # A long recording that spans the video
+ (A_UNIX_TIME - 3 * 24 * 3600, 6 * 24 * 3600),
+ ],
+ )
+ def test_covering_gpx_passes_silently(self, caplog, start: float, duration: int):
+ with caplog.at_level(logging.WARNING):
+ self._check(_gpx_track(start, duration))
+ assert not caplog.records
+
+ @pytest.mark.parametrize(
+ "start, duration, uncovered",
+ [
+ # Ends 5s into the video
+ (A_UNIX_TIME - 30, 35, 7),
+ # Starts 5s into the video
+ (A_UNIX_TIME + 5, 60, 5),
+ # Starts after the video's own GPS gives out, before the video ends
+ (A_UNIX_TIME + 11, 20, 11),
+ ],
+ )
+ def test_partial_cover_warns(
+ self, caplog, start: float, duration: int, uncovered: int
+ ):
+ with caplog.at_level(logging.WARNING):
+ self._check(_gpx_track(start, duration))
+ [record] = caplog.records
+ assert "covers only part of" in record.getMessage()
+ assert f"{uncovered} seconds" in record.getMessage()
+ assert "video.mp4" in record.getMessage()
+ assert "track.gpx" in record.getMessage()
+
+ @pytest.mark.parametrize(
+ "start",
+ [
+ # Ended a minute before the video started
+ A_UNIX_TIME - 70,
+ # Starts after the video ends, though within a minute of it
+ A_UNIX_TIME + 13,
+ # Naive GPX timestamps read in a time zone 2h off
+ A_UNIX_TIME + 2 * 3600,
+ # A GPX file from another day
+ A_UNIX_TIME - 3 * 24 * 3600,
+ ],
+ )
+ def test_missing_the_video_raises(self, start: float):
+ with pytest.raises(exceptions.MapillaryOutsideGPXTrackError) as info:
+ self._check(_gpx_track(start, 10))
+ assert "video.mp4" in str(info.value)
+ assert "track.gpx" in str(info.value)
+ assert "belongs to this video" in str(info.value)
+ assert "time zone" in str(info.value)
+
+ @pytest.mark.parametrize(
+ "gap, reported",
+ [
+ # The GPX of the next clip, starting just after the video ends
+ (0.089, "by 0.089 seconds."),
+ (1, "by 1 second."),
+ (70, "by 1.2 minutes."),
+ (2 * 3600, "by 2.0 hours."),
+ (3 * 24 * 3600, "by 3.0 days."),
+ ],
+ )
+ def test_gap_is_reported_in_a_readable_unit(self, gap: float, reported: str):
+ video_end = A_UNIX_TIME + self.VIDEO_DURATION
+ with pytest.raises(exceptions.MapillaryOutsideGPXTrackError) as info:
+ self._check(_gpx_track(video_end + gap, 10))
+ assert reported in str(info.value)
+
+ @pytest.mark.parametrize(
+ "start",
+ [
+ A_UNIX_TIME - 70,
+ A_UNIX_TIME + 2 * 3600,
+ # Starts after the video's own GPS gives out, which the video
+ # itself may not
+ A_UNIX_TIME + 60,
+ ],
+ )
+ def test_without_duration_a_gap_warns(self, caplog, start: float):
+ with caplog.at_level(logging.WARNING):
+ self._check(_gpx_track(start, 10), duration=None)
+ [record] = caplog.records
+ assert "misses" in record.getMessage()
+
+ @pytest.mark.parametrize(
+ "start",
+ [A_UNIX_TIME + 2 * 24 * 3600, A_UNIX_TIME - 3 * 24 * 3600],
+ )
+ def test_without_duration_a_gap_of_days_raises(self, start: float):
+ with pytest.raises(exceptions.MapillaryOutsideGPXTrackError):
+ self._check(_gpx_track(start, 10), duration=None)
+
+ @pytest.mark.parametrize("duration", [VIDEO_DURATION, None])
+ def test_epoch_mixup_raises(self, duration: float | None):
+ """Video timestamps left in GPS time sync ten years off."""
+ video = [
+ _camm_point(time=float(t), epoch_time=A_GPS_TIME + t) for t in range(11)
+ ]
+ gpx_points = _gpx_track(A_UNIX_TIME, 10)
+ offset = GPXVideoExtractor._gpx_offset(gpx_points, video)
+ assert abs(offset - (GPS_UNIX_DELTA - 18)) < 1
+ with pytest.raises(exceptions.MapillaryOutsideGPXTrackError):
+ self._check(gpx_points, video, duration)
+
+ def test_error_survives_a_worker_process(self):
+ """Videos are geotagged in a process pool, so the error gets pickled."""
+ with pytest.raises(exceptions.MapillaryOutsideGPXTrackError) as info:
+ self._check(_gpx_track(A_UNIX_TIME + 2 * 24 * 3600, 10))
+
+ unpickled = pickle.loads(pickle.dumps(info.value))
+
+ assert str(unpickled) == str(info.value)
+ assert vars(unpickled) == vars(info.value)
+
+ def _write_video_and_gpx(
+ self, tmp_path: Path, gpx_start: float, duration: float = VIDEO_DURATION
+ ) -> tuple[Path, Path]:
+ video_path = tmp_path / "labpano.mp4"
+ video_path.write_bytes(
+ _write_camm_mp4(self.VIDEO, "Labpano", A_UNIX_TIME + 600, duration)
+ )
+ gpx_path = tmp_path / "labpano.gpx"
+ _write_gpx(gpx_path, _gpx_track(gpx_start, 20))
+ return video_path, gpx_path
+
+ def test_extract_syncs_labpano_video_to_its_gpx(self, tmp_path: Path, caplog):
+ video_path, gpx_path = self._write_video_and_gpx(tmp_path, A_UNIX_TIME - 5)
+
+ with caplog.at_level(logging.WARNING):
+ points = GPXVideoExtractor(video_path, gpx_path).extract().points
+
+ # The GPX starts ~5s before the video, whose GPS starts at video time 0
+ assert -6 < points[0].time < -4
+ assert not caplog.records
+
+ @pytest.mark.parametrize(
+ "gpx_start",
+ [
+ # From days before
+ A_UNIX_TIME - 3 * 24 * 3600,
+ # In a time zone 2h off
+ A_UNIX_TIME + 2 * 3600,
+ ],
+ )
+ def test_extract_rejects_gpx_that_misses_the_video(
+ self, tmp_path: Path, gpx_start: float
+ ):
+ video_path, gpx_path = self._write_video_and_gpx(tmp_path, gpx_start)
+ with pytest.raises(exceptions.MapillaryOutsideGPXTrackError):
+ GPXVideoExtractor(video_path, gpx_path).extract()
+
+ def test_extract_without_duration_only_warns_of_hours(self, tmp_path: Path, caplog):
+ video_path, gpx_path = self._write_video_and_gpx(
+ tmp_path, A_UNIX_TIME + 2 * 3600, duration=0
+ )
+ with caplog.at_level(logging.WARNING):
+ GPXVideoExtractor(video_path, gpx_path).extract()
+ [record] = caplog.records
+ assert "misses" in record.getMessage()
+
+ def _extract_synced_to_video_time_0(
+ self, tmp_path: Path, video: T.Sequence[geo.Point], gpx_duration: int
+ ) -> list[geo.Point]:
+ video_path = tmp_path / "video.mp4"
+ video_path.write_bytes(
+ _write_camm_mp4(video, "Labpano", A_UNIX_TIME + 600, self.VIDEO_DURATION)
+ )
+ gpx_path = tmp_path / "track.gpx"
+ _write_gpx(gpx_path, _gpx_track(int(A_UNIX_TIME), gpx_duration))
+
+ points = GPXVideoExtractor(video_path, gpx_path).extract().points
+
+ assert points[0].time == 0.0
+ return points
+
+ @pytest.mark.parametrize("video", [UNTIMED_VIDEO, WHOLE_SECOND_VIDEO])
+ def test_extract_warns_of_partial_cover_at_video_time_0(
+ self, tmp_path: Path, caplog, video: T.Sequence[geo.Point]
+ ):
+ """An offset of 0 is a sync like any other, not a reason to skip the check."""
+ with caplog.at_level(logging.WARNING):
+ # Covers the first 6 of the video's 12 seconds
+ self._extract_synced_to_video_time_0(tmp_path, video, 6)
+ [record] = caplog.records
+ assert "6 seconds of the video fall outside the track" in record.getMessage()
+
+ @pytest.mark.parametrize("video", [UNTIMED_VIDEO, WHOLE_SECOND_VIDEO])
+ def test_extract_covering_at_video_time_0_is_silent(
+ self, tmp_path: Path, caplog, video: T.Sequence[geo.Point]
+ ):
+ with caplog.at_level(logging.WARNING):
+ self._extract_synced_to_video_time_0(tmp_path, video, 20)
+ assert not caplog.records
+
+ def test_video_duration_is_read_from_the_file(self, tmp_path: Path):
+ video_path, gpx_path = self._write_video_and_gpx(tmp_path, A_UNIX_TIME)
+ assert GPXVideoExtractor(video_path, gpx_path)._video_duration() == 12.0
+
+ @pytest.mark.parametrize("content", [b"", b"not a real mp4"])
+ def test_unreadable_duration_is_unknown(self, tmp_path: Path, content: bytes):
+ video_path = tmp_path / "video.mp4"
+ video_path.write_bytes(content)
+ extractor = GPXVideoExtractor(video_path, tmp_path / "track.gpx")
+ assert extractor._video_duration() is None
+
+ def test_next_source_gets_its_turn(self, tmp_path: Path):
+ """The GPX misses the video, which says nothing about the video itself."""
+ video_path, gpx_path = self._write_video_and_gpx(
+ tmp_path, A_UNIX_TIME - 3 * 24 * 3600
+ )
+ options = [
+ SourceOption(
+ SourceType.GPX,
+ num_processes=0,
+ source_path=SourcePathOption(source_path=gpx_path),
+ ),
+ SourceOption(SourceType.NATIVE, num_processes=0),
+ ]
+
+ [metadata] = factory.process([video_path], options)
+
+ assert isinstance(metadata, types.VideoMetadata)
+ assert [p.time for p in metadata.points] == [p.time for p in self.VIDEO]
+
+
+class TestImagesOutsideGPXTrack:
+ """
+ Images outside a GPX track fail with the same error, a geotagging error,
+ so they too fall through to the next geotag source.
+ """
+
+ # Captured 2018-06-08T20:24:11Z at 45.5169, -122.5728
+ IMAGE = Path(__file__).parent.parent / "data" / "images" / "DSC00001.JPG"
+
+ def _process(self, tmp_path: Path, sources: list[SourceType]):
+ image_path = tmp_path / self.IMAGE.name
+ shutil.copyfile(self.IMAGE, image_path)
+ # A track recorded years after the image
+ gpx_path = tmp_path / "track.gpx"
+ _write_gpx(gpx_path, _gpx_track(A_UNIX_TIME, 10))
+ options = [
+ SourceOption(
+ source,
+ num_processes=0,
+ source_path=SourcePathOption(source_path=gpx_path),
+ )
+ if source is SourceType.GPX
+ else SourceOption(source, num_processes=0)
+ for source in sources
+ ]
+ [metadata] = factory.process([image_path], options)
+ return metadata
+
+ def test_next_source_gets_its_turn(self, tmp_path: Path):
+ metadata = self._process(tmp_path, [SourceType.GPX, SourceType.EXIF])
+
+ assert isinstance(metadata, types.ImageMetadata)
+ assert (round(metadata.lat, 4), round(metadata.lon, 4)) == (45.5169, -122.5728)
+
+ def test_gpx_as_the_only_source_fails(self, tmp_path: Path):
+ metadata = self._process(tmp_path, [SourceType.GPX])
+
+ assert isinstance(metadata, types.ErrorMetadata)
+ assert isinstance(metadata.error, exceptions.MapillaryOutsideGPXTrackError)
+
+
+def _camm_exiftool_xml(
+ make: str, meta_format: str = "camm", format_tag: str = "MetaFormat"
+) -> ET.ElementTree:
+ """
+ ExifTool XML for a CAMM track, trimmed from a PanoX V2 capture.
+
+ ExifTool reports the format of a camera's CAMM track, under a meta
+ handler, as MetaFormat, and that of the CAMM tracks mapillary_tools
+ writes, under a camm handler, as OtherFormat.
+ """
+ xml = f"""
+
+
+ 2024:01:18 10:43:57
+ {meta_format}
+ 0
+ 0
+ 2024:01:18 10:41:01.600768Z
+ 3
+ 47.36061891
+ 8.52077651
+ 448.905395507812
+ {make}
+ PanoX V2
+
+
+"""
+ root = ET.fromstring(xml)
+ desc = root.find("{http://www.w3.org/1999/02/22-rdf-syntax-ns#}Description")
+ assert desc is not None
+ return ET.ElementTree(desc)
+
+
+class TestExifToolAgreesWithNativeParser:
+ """
+ ExifTool converts CAMM GPS time by the epoch difference alone, so it
+ reads 18s (the leap seconds since 1980) later than the native parser.
+ The native reading is the right one: in the PanoX V2 capture this is taken
+ from, the last GPS sample reads 0.4s before the mvhd creation time, stamped
+ when the file is finalized, where ExifTool's reading of it would be 17.6s
+ after.
+ """
+
+ # 2024-01-18T10:41:01.600768Z, as ExifTool renders it
+ EXIFTOOL_TIME = 1705574461.600768
+ NATIVE_TIME = EXIFTOOL_TIME - 18
+
+ def _first_epoch_time(self, xml: ET.ElementTree) -> float | None:
+ track = ExifToolReadVideo(xml).extract_gps_track()
+ return T.cast(telemetry.GPSPoint, track[0]).epoch_time
+
+ @pytest.mark.parametrize(
+ "make, meta_format",
+ [
+ ("Labpano", "camm"),
+ ("Labpano Technology Co.,Ltd", "camm"),
+ ("Labpano", "CAMM"),
+ ("Labpano", " camm "),
+ ],
+ )
+ def test_gps_time_camm_track_is_leap_corrected(self, make: str, meta_format: str):
+ epoch_time = self._first_epoch_time(_camm_exiftool_xml(make, meta_format))
+ assert epoch_time == pytest.approx(self.NATIVE_TIME, abs=1e-3)
+
+ @pytest.mark.parametrize(
+ "make, meta_format", [("Insta360", "camm"), ("Labpano", "gpmd"), ("", "camm")]
+ )
+ def test_other_tracks_are_left_alone(self, make: str, meta_format: str):
+ epoch_time = self._first_epoch_time(_camm_exiftool_xml(make, meta_format))
+ assert epoch_time == pytest.approx(self.EXIFTOOL_TIME, abs=1e-3)
+
+ def test_camm_tracks_written_by_mapillary_tools_are_left_alone(self):
+ """They hold Unix time, which ExifTool reads as is."""
+ xml = _camm_exiftool_xml("Labpano", format_tag="OtherFormat")
+ epoch_time = self._first_epoch_time(xml)
+ assert epoch_time == pytest.approx(self.EXIFTOOL_TIME, abs=1e-3)
+
+ @pytest.mark.skipif(shutil.which("exiftool") is None, reason="needs ExifTool")
+ def test_real_exiftool_reads_our_output_as_written(self, tmp_path: Path):
+ video_path = tmp_path / "labpano.mp4"
+ video_path.write_bytes(
+ _write_camm_mp4(_unix_track(), "Labpano", A_UNIX_TIME + 600)
+ )
+
+ xml = ExiftoolRunner(T.cast(str, shutil.which("exiftool"))).extract_xml(
+ [video_path]
+ )
+ [rdf] = exiftool_read.index_rdf_description_by_path_from_xml_element(
+ ET.fromstring(xml)
+ ).values()
+ track = ExifToolReadVideo(ET.ElementTree(rdf)).extract_gps_track()
+
+ epoch_times = [T.cast(telemetry.GPSPoint, p).epoch_time for p in track]
+ assert epoch_times == pytest.approx(UNIX_TIMES, abs=1e-3)
+
class TestEditListOverflow:
"""An oversized initial gap must not abort the upload."""