Skip to content

Do not interpolate a video's track across a GPX gap in process #899

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
caglarpir merged 3 commits into from
Oct 24, 2026
Merged

Do not interpolate a video's track across a GPX gap in process #899

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
09ee441
Fail a file named for process that is not a supported image or video
caglarpir Oct 24, 2026
038f9e4
Leave out the interval samples that fall outside their GPS track
caglarpir Oct 24, 2026
433088c
Do not interpolate a video's track across a GPX gap in process
caglarpir Oct 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -240,7 +240,9 @@ description-level processing error are included in `processed_file_count`.
Unsupported entries include their filename, category, stable reason code, and
normalized extension. Hidden/system files and common GPS or metadata sidecars
are ignored during directory discovery, but an explicitly supplied unsupported
file is always reported. The report is written only to the requested local path
file is always reported. Such a file also fails processing, with a
`MapillaryUnsupportedFileError` entry in the image description file, whether
or not the report is requested. The report is written only to the requested local path
and contains absolute source filenames, so treat it as local capture metadata.

## Upload
Expand Down
7 changes: 7 additions & 0 deletions mapillary_tools/exceptions.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,13 @@ class MapillaryGeoTaggingError(MapillaryDescriptionError):
pass


class MapillaryUnsupportedFileError(MapillaryDescriptionError):
"""
A file named for processing is neither an image nor a video that can be
processed
"""


class MapillaryVideoGPSUnusableError(MapillaryDescriptionError):
"""
The video has no GPS track that can be used: none at all, an empty one, or
Expand Down
119 changes: 70 additions & 49 deletions mapillary_tools/geotag/geotag_images_from_gpx.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,75 @@
LOG = logging.getLogger(__name__)


def check_within_track(
time: float,
sorted_points: T.Sequence[geo.Point],
gaps: T.Sequence[tuple[geo.Point, geo.Point]] = (),
) -> None:
"""
Raise MapillaryOutsideGPXTrackError if **time** falls outside the track of
**sorted_points**, with a tolerance of 1 ms: before its start, after its
end, or in one of its **gaps** (see track_gaps())

>>> track = [geo.Point(time=t, lat=0, lon=0, alt=None, angle=None) for t in (10, 20, 9000)]
>>> check_within_track(15.0, track)
>>> check_within_track(9000.0005, track)
>>> check_within_track(9.5, track)
Traceback (most recent call last):
...
mapillary_tools.exceptions.MapillaryOutsideGPXTrackError: The image date time is 0.5 seconds behind the GPX start point
>>> check_within_track(4000.0, track, track_gaps(track, 3600))
Traceback (most recent call last):
...
mapillary_tools.exceptions.MapillaryOutsideGPXTrackError: The GPX track has a gap of 2.5 hours, from 1970-01-01T00:00:20+00:00 at (0, 0) to 1970-01-01T02:30:00+00:00 at (0, 0), which the image time 1970-01-01T01:06:40+00:00 falls in
"""
assert sorted_points, "must have at least one point"

if time < sorted_points[0].time:
delta = sorted_points[0].time - time
gpx_start_time = build_capture_time(sorted_points[0].time)
gpx_end_time = build_capture_time(sorted_points[-1].time)
# with the tolerance of 1ms
if 0.001 < delta:
raise exceptions.MapillaryOutsideGPXTrackError(
f"The image date time is {round(delta, 3)} seconds behind the GPX start point",
image_time=build_capture_time(time),
gpx_start_time=gpx_start_time,
gpx_end_time=gpx_end_time,
)

if sorted_points[-1].time < time:
delta = time - sorted_points[-1].time
gpx_start_time = build_capture_time(sorted_points[0].time)
gpx_end_time = build_capture_time(sorted_points[-1].time)
# with the tolerance of 1ms
if 0.001 < delta:
raise exceptions.MapillaryOutsideGPXTrackError(
f"The image time is {round(delta, 3)} seconds beyond the GPX end point",
image_time=build_capture_time(time),
gpx_start_time=gpx_start_time,
gpx_end_time=gpx_end_time,
)

# The last gap that starts at or before the image
idx = bisect.bisect_right([before.time for before, _ in gaps], time)
if idx:
before, after = gaps[idx - 1]
delta = min(time - before.time, after.time - time)
# with the tolerance of 1ms, as at the ends of the track
if 0.001 < delta:
captured_at = datetime.datetime.fromtimestamp(
time, tz=datetime.timezone.utc
).isoformat()
track_gap = describe_gap(before, after, before.time, after.time)
raise exceptions.MapillaryOutsideGPXTrackError(
f"The GPX track has {track_gap}, which the image time {captured_at} falls in",
image_time=build_capture_time(time),
gpx_start_time=build_capture_time(sorted_points[0].time),
gpx_end_time=build_capture_time(sorted_points[-1].time),
)


class GeotagImagesFromGPX(GeotagImagesFromGeneric):
def __init__(
self,
Expand Down Expand Up @@ -55,55 +124,7 @@ def _interpolate_image_metadata_along(
sorted_points: T.Sequence[geo.Point],
gaps: T.Sequence[tuple[geo.Point, geo.Point]] = (),
) -> types.ImageMetadata:
assert sorted_points, "must have at least one point"

if image_metadata.time < sorted_points[0].time:
delta = sorted_points[0].time - image_metadata.time
gpx_start_time = build_capture_time(sorted_points[0].time)
gpx_end_time = build_capture_time(sorted_points[-1].time)
# with the tolerance of 1ms
if 0.001 < delta:
raise exceptions.MapillaryOutsideGPXTrackError(
f"The image date time is {round(delta, 3)} seconds behind the GPX start point",
image_time=build_capture_time(image_metadata.time),
gpx_start_time=gpx_start_time,
gpx_end_time=gpx_end_time,
)

if sorted_points[-1].time < image_metadata.time:
delta = image_metadata.time - sorted_points[-1].time
gpx_start_time = build_capture_time(sorted_points[0].time)
gpx_end_time = build_capture_time(sorted_points[-1].time)
# with the tolerance of 1ms
if 0.001 < delta:
raise exceptions.MapillaryOutsideGPXTrackError(
f"The image time is {round(delta, 3)} seconds beyond the GPX end point",
image_time=build_capture_time(image_metadata.time),
gpx_start_time=gpx_start_time,
gpx_end_time=gpx_end_time,
)

# The last gap that starts at or before the image
idx = bisect.bisect_right(
[before.time for before, _ in gaps], image_metadata.time
)
if idx:
before, after = gaps[idx - 1]
delta = min(
image_metadata.time - before.time, after.time - image_metadata.time
)
# with the tolerance of 1ms, as at the ends of the track
if 0.001 < delta:
captured_at = datetime.datetime.fromtimestamp(
image_metadata.time, tz=datetime.timezone.utc
).isoformat()
track_gap = describe_gap(before, after, before.time, after.time)
raise exceptions.MapillaryOutsideGPXTrackError(
f"The GPX track has {track_gap}, which the image time {captured_at} falls in",
image_time=build_capture_time(image_metadata.time),
gpx_start_time=build_capture_time(sorted_points[0].time),
gpx_end_time=build_capture_time(sorted_points[-1].time),
)
check_within_track(image_metadata.time, sorted_points, gaps)

interpolated = geo.interpolate(sorted_points, image_metadata.time)

Expand Down
123 changes: 104 additions & 19 deletions mapillary_tools/geotag/video_extractors/gpx.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -204,7 +204,7 @@ def extract(self) -> types.VideoMetadata:
self._apply_offset_time(gpx_points, rate=rate)
self._check_coverage(gpx_points, gpx_unix_times, [], start_clock, rate)
if self.clip_to_video:
gpx_points = self._clip_to_video(gpx_points, start_clock)
gpx_points = self._clip_to_video(gpx_points, start_clock, rate)
return types.VideoMetadata(
filename=self.video_path,
filesize=utils.get_file_size(self.video_path),
Expand Down Expand Up @@ -269,7 +269,7 @@ def extract(self) -> types.VideoMetadata:
rate,
)
if self.clip_to_video:
gpx_points = self._clip_to_video(gpx_points, clock)
gpx_points = self._clip_to_video(gpx_points, clock, rate)

return dataclasses.replace(native_video_metadata, points=gpx_points)

Expand Down Expand Up @@ -313,7 +313,7 @@ def _parse_tracks(self) -> list[list[geo.Point]]:
) from ex

def _clip_to_video(
self, gpx_points: list[geo.Point], clock: VideoClock | None
self, gpx_points: list[geo.Point], clock: VideoClock | None, rate: float
) -> list[geo.Point]:
"""
Clip a GPX track to the video, from video time 0 to where the video
Expand All @@ -330,10 +330,20 @@ def _clip_to_video(
past, so that the first frames keep a position. Its end is left as it
is: it may be that of a timelapse, which runs on past the end of the
video at the pace of the GPS clock rather than that of the video.

A gap of more than constants.MAX_GPX_GAP_SECONDS that video time 0, or
the end of the video at **rate** GPS seconds per second, falls in says
nothing of where the video was: the fixes on its far side are left
out, with no point interpolated across it, as if the track started
(or ended) there. One inside the video fails it (see _check_coverage).
"""
video_duration = _video_duration(self.video_path)
gpx_points = _leave_out_far_side_of_gaps(
gpx_points, constants.MAX_GPX_GAP_SECONDS / rate, video_duration
)
if clock is None:
return _clip_track(gpx_points, None)
return _clip_track(gpx_points, _video_duration(self.video_path))
return _clip_track(gpx_points, video_duration)

def _place_without_video_gps(
self, gpx_points: T.Sequence[geo.Point]
Expand Down Expand Up @@ -446,7 +456,9 @@ def _check_coverage(
warns, and one that covers all of it is silent. The video can fall
between the ends of the track and still miss it, or part of it, in a
gap of more than constants.MAX_GPX_GAP_SECONDS between two of its
points.
points. In process, which uploads the track with the video, a gap
inside the video raises too: sampling leaves out the frames in it,
while the track would be interpolated across it.

Without the duration of the video, the video is known only up to its
last GPS point, and a track that starts after it may still overlap the
Expand Down Expand Up @@ -506,6 +518,66 @@ def _check_coverage(

# In GPS seconds, which is where clock and time zone mistakes are made
gap = min(max(first - video_end, -last) for first, last in parts) * rate
if clock is None:
# Placed at the start of the video, only the offset moved it off
advice = "Check --interpolation_offset_time"
elif offset_time:
advice = (
f"Check --interpolation_offset_time, that the {self.file_format} file belongs to "
f"this video, then the camera clock and the time zone of the {self.file_format} "
"timestamps"
)
else:
advice = (
f"Check that the {self.file_format} file belongs to this video, then the camera "
f"clock and the time zone of the {self.file_format} timestamps"
)

if gap <= 0 and self.clip_to_video:
# process uploads the track with the video, and its frames would be
# placed along it across a gap that the video spans, which says
# nothing of where the video was. Sampling leaves out the frames
# in it (see _clip_to_video for the gaps at its ends)
for before, after in gap_points:
if before.time <= 0 and video_end <= after.time:
# The track touches the video at an instant at one end only
where = f"which {video} falls in{offset_clause}"
if clock is not None or offset_time:
where_advice = advice
elif self.sync_mode is SyncMode.REBASE:
# Said to start with the video
where_advice = f"Trim the {self.file_format} to the video"
else:
# Assumed to start with the video
where_advice = (
f"Trim the {self.file_format} to the video, or sample the video "
"with video_process and --video_start_time"
)
elif 0 <= before.time and after.time <= video_end:
where = (
f"inside {video}{offset_clause}, which says nothing of where "
"the video was in it"
)
where_advice = (
"Sample the video with video_process and --video_geotag_source "
f"{self.file_format.lower()}, which leaves out the frames in the gap"
)
else:
continue
# At the Unix times the points were recorded at
track_gap = describe_gap(
before,
after,
lookup_start + before.time * rate,
lookup_start + after.time * rate,
)
raise exceptions.MapillaryOutsideGPXTrackError(
f"{gpx_track} has {track_gap}, {where}. {where_advice}",
image_time=build_capture_time(lookup_start),
gpx_start_time=build_capture_time(gpx_unix_times[0]),
gpx_end_time=build_capture_time(gpx_unix_times[1]),
)

if gap <= 0:
# In video seconds
uncovered = (
Expand All @@ -523,20 +595,6 @@ def _check_coverage(
)
return

if clock is None:
# Placed at the start of the video, only the offset moved it off
advice = "Check --interpolation_offset_time"
elif offset_time:
advice = (
f"Check --interpolation_offset_time, that the {self.file_format} file belongs to "
f"this video, then the camera clock and the time zone of the {self.file_format} "
"timestamps"
)
else:
advice = (
f"Check that the {self.file_format} file belongs to this video, then the camera "
f"clock and the time zone of the {self.file_format} timestamps"
)
# Between the ends of the track, the video falls in one of its gaps
surrounding_gap = next(
(
Expand Down Expand Up @@ -1027,6 +1085,33 @@ def describe_gap(
)


def _leave_out_far_side_of_gaps(
points: T.Sequence[geo.Point], max_gap: float, end: float | None
) -> list[geo.Point]:
"""
Leave out of a track, sorted by time, the points on the far side of a gap
of more than **max_gap** seconds that time 0, or **end** (None if
unknown), falls in

>>> track = [geo.Point(time=t, lat=t, lon=0, alt=None, angle=None) for t in (-9000, 99, 200, 9000)]
>>> [p.time for p in _leave_out_far_side_of_gaps(track, 3600, 300)]
[99, 200]
>>> [p.time for p in _leave_out_far_side_of_gaps(track, 3600, None)]
[99, 200, 9000]
>>> [p.time for p in _leave_out_far_side_of_gaps(track, 10000, 300)]
[-9000, 99, 200, 9000]
"""
start, stop = 0, len(points)
for idx, (before, after) in enumerate(geo.pairwise(points)):
if after.time - before.time <= max_gap:
continue
if before.time < 0 < after.time:
start = idx + 1
if end is not None and before.time < end < after.time:
stop = idx + 1
return list(points[start:stop])


def _clip_track(
points: T.Sequence[geo.Point], duration: float | None
) -> list[geo.Point]:
Expand Down
Loading
Loading