Skip to content

Read a video's GPS clock as a straight line at the rate it fits #889

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 1 commit into from
Oct 23, 2026
Merged

Read a video's GPS clock as a straight line at the rate it fits #889

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
16 changes: 5 additions & 11 deletions mapillary_tools/geotag/geotag_images_from_video.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
from .base import GeotagImagesFromGeneric
from .geotag_images_from_gpx import GeotagImagesFromGPX
from .geotag_videos_from_video import GeotagVideosFromVideo
from .video_extractors.gpx import GPSTimeline, GPXVideoExtractor
from .video_extractors.gpx import GPXVideoExtractor


LOG = logging.getLogger(__name__)
Expand Down Expand Up @@ -109,21 +109,15 @@ def _on_capture_timeline(
) -> T.Sequence[geo.Point]:
"""
Time the video's GPS points in GPS seconds from the start of the video,
like the samples, which are stamped with their capture times, as its
GPS timestamps tell: a timelapse packs several into each second of
like the samples, which are stamped with their capture times, at the
pace of its GPS clock: a timelapse packs several into each second of
video, and some clocks gain on the video.
"""
clock = GPXVideoExtractor._read_video_clock(points)
if clock is None:
if clock is None or clock.rate == 1.0:
return points

timeline = GPSTimeline.read(points, clock.rate)
assert timeline is not None, "a GPS clock is read from GPS timestamps"
start = timeline.to_unix_time(0.0)
return [
dataclasses.replace(p, time=timeline.to_unix_time(p.time) - start)
for p in points
]
return [dataclasses.replace(p, time=p.time * clock.rate) for p in points]


class GeotagImageSamplesFromVideo(GeotagImagesFromGeneric):
Expand Down
292 changes: 108 additions & 184 deletions mapillary_tools/geotag/video_extractors/gpx.py
View file
Open in desktop

Large diffs are not rendered by default.

78 changes: 30 additions & 48 deletions mapillary_tools/sample_video.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@
SourceType,
VideoSyncOption,
)
from .geotag.video_extractors.gpx import GPSTimeline, GPXVideoExtractor, VideoClock
from .geotag.video_extractors.gpx import GPXVideoExtractor, VideoClock
from .geotag.video_extractors.native import NativeVideoExtractor
from .geotag.video_extractors.nmea import NMEAVideoExtractor
from .mp4 import mp4_sample_parser, simple_mp4_parser as sparser
Expand Down Expand Up @@ -473,24 +473,24 @@ def _read_telemetry(video_path: Path) -> T.Sequence[geo.Point]:
return []


def _telemetry_clock_rate(
def _telemetry_clock(
video_path: Path, telemetry_points: T.Sequence[geo.Point]
) -> float | None:
) -> VideoClock | None:
"""
GPS seconds per video second of the video's own GPS clock: more than 1 if
the video is a timelapse, or None if it has no GPS clock
The video's own GPS clock, or None if it has none. It runs more than 1 GPS
second per video second if the video is a timelapse
"""
clock = GPXVideoExtractor._read_video_clock(telemetry_points)
if clock is None:
return None

if clock.rate != 1.0:
LOG.info(
"The GPS clock in %s runs %.1f times faster than the video (timelapse)",
"The GPS clock in %s runs %.3g times faster than the video",
video_path.name,
clock.rate,
)
return clock.rate
return clock


def _parse_filename_time(video_path: Path) -> datetime.datetime | None:
Expand Down Expand Up @@ -770,48 +770,36 @@ def _sample_single_video_by_interval(
telemetry_points = _read_telemetry(video_path)

# Frames are stamped with their capture times, which a timelapse packs
# several seconds of into each second of video: as the video's own GPS
# timestamps tell, and for a video without a GPS clock, at
# --video_duration_ratio
clock_rate = _telemetry_clock_rate(video_path, telemetry_points)
# several seconds of into each second of video: by the video's own GPS
# clock, as GPX sync places a GPX and distance samples are stamped, and for
# a video without a GPS clock, at --video_duration_ratio
clock = _telemetry_clock(video_path, telemetry_points)
# GPS rejected as noise gives no telemetry, but GPX sync reads its clock,
# and the frames are stamped by it, where GPX sync places them along the
# GPX, so that the geotagging stage finds them there
gpx_clock: VideoClock | None = None
if clock_rate is None:
gpx_clock = _gpx_sync_clock(video_path, geotag_options)
if gpx_clock is not None:
clock_rate = gpx_clock.rate
timeline: GPSTimeline | None = None
if clock_rate is None:
if clock is None:
clock = _gpx_sync_clock(video_path, geotag_options)
if clock is None:
rate = duration_ratio
else:
rate = clock_rate
rate = clock.rate
if duration_ratio != 1.0:
LOG.warning(
"Ignoring --video_duration_ratio %g for %s, whose own GPS clock runs at %.3gx",
duration_ratio,
video_path.name,
clock_rate,
clock.rate,
)
if gpx_clock is None:
# As distance samples are stamped, and as the geotagging stage
# times the video's track to place the samples along it
timeline = GPSTimeline.read(telemetry_points, clock_rate)

if clock_rate is None and video_sync is not None:
if clock is None and video_sync is not None:
_check_gpx_start_by_interval(video_path, geotag_options, video_sync)

probe = _probe_video(ffmpeg, video_path)

if start_time is None:
if timeline is not None:
if clock is not None:
start_time = datetime.datetime.fromtimestamp(
timeline.to_unix_time(0.0), tz=datetime.timezone.utc
)
elif gpx_clock is not None:
start_time = datetime.datetime.fromtimestamp(
gpx_clock.to_unix_time(0.0), tz=datetime.timezone.utc
clock.to_unix_time(0.0), tz=datetime.timezone.utc
)
else:
start_time = _extract_video_start_time(video_path, probe, telemetry_points)
Expand Down Expand Up @@ -843,11 +831,7 @@ def _sample_single_video_by_interval(
continue
# extract_frames() produces 1-based frame indices so we need to subtract 1 here
video_time = (frame_idx_1based - 1) * sample_interval
if timeline is None:
seconds = video_time * rate
else:
seconds = timeline.to_unix_time(video_time) - timeline.to_unix_time(0.0)
timestamp = start_time + datetime.timedelta(seconds=seconds)
timestamp = start_time + datetime.timedelta(seconds=video_time * rate)
exif_edit = ExifEdit(sample_paths[0])
exif_edit.add_date_time_original(timestamp)
exif_edit.add_gps_datetime(timestamp)
Expand Down Expand Up @@ -1057,11 +1041,11 @@ def _sample_single_video_by_distance(
last_unix_time = max(track_unix_times, default=None)

# Read from the video itself, a track with a GPS clock of its own paces the
# video, and stamps the frames along its timeline, as interval samples are.
# A GPX source warns of the options that its sync ignores itself, and its
# sync placed the track along the video's clock already: its timestamps
# stamp the frames as they are
track_timeline: GPSTimeline | None = None
# video, and its clock stamps the frames, as interval samples are. A GPX
# source warns of the options that its sync ignores itself, and its sync
# placed the track along the video's clock already: its timestamps stamp
# the frames as they are
track_clock: VideoClock | None = None
if not any(
option.source in factory.EXTERNAL_GPS_SOURCES for option in geotag_options
):
Expand All @@ -1073,14 +1057,12 @@ def _sample_single_video_by_distance(
start_time,
)
track_clock = GPXVideoExtractor._read_video_clock(video_metadata.points)
if track_clock is not None:
track_timeline = GPSTimeline.read(video_metadata.points, track_clock.rate)

# Frames along the timeline, or at points with an absolute timestamp, are
# timestamped from it below, so only the others need the start time
# Frames stamped by the clock, or at points with an absolute timestamp,
# are timestamped below, so only the others need the start time
if (
start_time is None
and track_timeline is None
and track_clock is None
and any(
interp.get_unix_time() is None
for _, interp in sample_points_by_frame_idx.values()
Expand Down Expand Up @@ -1130,8 +1112,8 @@ def _sample_single_video_by_distance(

gps_unix_time = (
interp.get_unix_time()
if track_timeline is None
else track_timeline.to_unix_time(interp.time)
if track_clock is None
else track_clock.to_unix_time(interp.time)
)
if gps_unix_time is not None:
assert first_unix_time is not None and last_unix_time is not None
Expand Down
21 changes: 11 additions & 10 deletions tests/integration/test_gopro.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,16 +31,17 @@
"MAPILLARY_TOOLS_GOPRO_GPS_PRECISION": "10000000",
"MAPILLARY_TOOLS_MAX_CAPTURE_SPEED_KMH": "2000000", # km/h
}
# The capture times come from the GPMF GPS timestamps (which put the start at
# 2019-11-18T23:42:10.009Z, without the first, 1.35 seconds stale), not from the
# container's creation time. This camera writes the creation time in local time,
# so reading the start time from there would timestamp every sample 8 hours off.
# The capture times come from the GPMF GPS clock (which puts the start at
# 2019-11-18T23:42:10.003Z, without the first timestamp, 1.36 seconds stale),
# not from the container's creation time. This camera writes the creation time
# in local time, so reading the start time from there would timestamp every
# sample 8 hours off.
EXPECTED_DESCS: T.List[T.Any] = [
{
"filename": "hero8.mp4/hero8_v_000001.jpg",
"filetype": "image",
"MAPAltitude": 9540.24,
"MAPCaptureTime": "2019_11_18_23_42_10_009",
"MAPCaptureTime": "2019_11_18_23_42_10_003",
"MAPCompassHeading": {
"TrueHeading": 123.93587938690177,
"MagneticHeading": 123.93587938690177,
Expand All @@ -55,7 +56,7 @@
"filename": "hero8.mp4/hero8_v_000002.jpg",
"filetype": "image",
"MAPAltitude": 7112.573717404068,
"MAPCaptureTime": "2019_11_18_23_42_12_052",
"MAPCaptureTime": "2019_11_18_23_42_12_003",
"MAPCompassHeading": {
"TrueHeading": 140.8665026186285,
"MagneticHeading": 140.8665026186285,
Expand All @@ -70,7 +71,7 @@
"filename": "hero8.mp4/hero8_v_000003.jpg",
"filetype": "image",
"MAPAltitude": 7463.642846094319,
"MAPCaptureTime": "2019_11_18_23_42_13_976",
"MAPCaptureTime": "2019_11_18_23_42_14_003",
"MAPCompassHeading": {
"TrueHeading": 138.44255851085705,
"MagneticHeading": 138.44255851085705,
Expand All @@ -85,7 +86,7 @@
"filename": "hero8.mp4/hero8_v_000004.jpg",
"filetype": "image",
"MAPAltitude": 6909.8168472111465,
"MAPCaptureTime": "2019_11_18_23_42_16_009",
"MAPCaptureTime": "2019_11_18_23_42_16_003",
"MAPCompassHeading": {
"TrueHeading": 142.23462669862568,
"MagneticHeading": 142.23462669862568,
Expand All @@ -100,7 +101,7 @@
"filename": "hero8.mp4/hero8_v_000005.jpg",
"filetype": "image",
"MAPAltitude": 7212.594480737465,
"MAPCaptureTime": "2019_11_18_23_42_17_992",
"MAPCaptureTime": "2019_11_18_23_42_18_003",
"MAPCompassHeading": {
"TrueHeading": 164.70819093235514,
"MagneticHeading": 164.70819093235514,
Expand All @@ -115,7 +116,7 @@
"filename": "hero8.mp4/hero8_v_000006.jpg",
"filetype": "image",
"MAPAltitude": 7274.361994963208,
"MAPCaptureTime": "2019_11_18_23_42_20_019",
"MAPCaptureTime": "2019_11_18_23_42_20_003",
"MAPCompassHeading": {
"TrueHeading": 139.71549328876722,
"MagneticHeading": 139.71549328876722,
Expand Down
21 changes: 10 additions & 11 deletions tests/integration/test_process.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -761,7 +761,7 @@ def test_process_video_geotag_source_with_gpx_outside_video(
assert descs[0]["error"]["vars"] == {
# Its first GPS timestamp is 23:44:42.590, a second before the rest
# place the start of the video
"image_time": "2019_11_18_23_44_43_624",
"image_time": "2019_11_18_23_44_43_613",
"gpx_start_time": "2025_03_14_07_00_00_000",
"gpx_end_time": "2025_03_14_07_01_33_624",
}
Expand All @@ -775,7 +775,7 @@ def test_process_video_from_gpx_with_interpolation_options(
images. They were ignored for videos.
"""
video_path = setup_data.join("gopro_data").join("max-360mode.mp4")
# 3.624 seconds before the GPS timestamps of the video place its start
# 3.613 seconds before the GPS clock of the video starts
start = datetime.datetime(2019, 11, 18, 23, 44, 40, tzinfo=datetime.timezone.utc)
points = "".join(
f'<trkpt lat="{37.0 + i * 0.0001:.7f}" lon="-122.0">'
Expand Down Expand Up @@ -809,20 +809,19 @@ def video_time_of_recorded(*options: str) -> float:
[point] = [p for p in describe(*options)["MAPGPSTrack"] if p[5] == recorded]
return point[0]

# In ms of video, which the GPS timestamps of the video pace at a second
# per second, give or take their jitter of some 40 ms
synced = video_time_of_recorded()
assert synced == pytest.approx(10000 - 3624, abs=50)
assert synced == pytest.approx(10000 - 3613, abs=1)

assert video_time_of_recorded("--interpolation_use_gpx_start_time") == (
pytest.approx(10000, abs=50)
assert video_time_of_recorded("--interpolation_use_gpx_start_time") == 10000
assert (
video_time_of_recorded(
"--interpolation_use_gpx_start_time", "--interpolation_offset_time=2.5"
)
== 7500
)
assert video_time_of_recorded(
"--interpolation_use_gpx_start_time", "--interpolation_offset_time=2.5"
) == pytest.approx(7500, abs=50)

shifted = video_time_of_recorded("--interpolation_offset_time=2")
assert shifted == pytest.approx(synced - 2000, abs=50)
assert shifted == pytest.approx(synced - 2000, abs=1)

# an offset that moves the track off the video fails it, as it does a
# track placed by the clock of the video
Expand Down
9 changes: 3 additions & 6 deletions tests/integration/test_upload.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@

import py.path
import pytest
from mapillary_tools.geotag.video_extractors.gpx import GPSTimeline, GPXVideoExtractor
from mapillary_tools.geotag.video_extractors.gpx import GPXVideoExtractor
from mapillary_tools.geotag.video_extractors.native import NativeVideoExtractor

from .fixtures import (
Expand Down Expand Up @@ -52,9 +52,7 @@ def test_upload_video_with_gpx_running_past_both_ends(
points = NativeVideoExtractor(Path(video_path)).extract().points
clock = GPXVideoExtractor._read_video_clock(points)
assert clock is not None
timeline = GPSTimeline.read(points, clock.rate)
assert timeline is not None
video_start = timeline.to_unix_time(0)
video_start = clock.to_unix_time(0)
# From 12.5 seconds before the video to about 47 seconds after it
trkpts = "".join(
f'<trkpt lat="{37.0 + i * 0.0001:.7f}" lon="-122.0"><time>'
Expand Down Expand Up @@ -85,8 +83,7 @@ def test_upload_video_with_gpx_running_past_both_ends(
assert first[0] == 0
assert first[2] == pytest.approx(37.0 + 12.5 * 0.0001, abs=1e-7)
assert first[5] == pytest.approx(video_start, abs=1e-3)
# The CAMM track of the uploaded video truncates the time between samples
assert last[0] == pytest.approx(duration * 1000, abs=2)
assert last[0] == pytest.approx(duration * 1000, abs=1)
assert last[5] == pytest.approx(video_start + duration, abs=1e-3)


Expand Down
Loading
Loading