Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
84 changes: 73 additions & 11 deletions mapillary_tools/geotag/video_extractors/gpx.py
Original file line number Diff line number Diff line change
Expand Up @@ -441,10 +441,10 @@ def _place_without_video_gps(
Place the GPX track on the timeline of a video that has no GPS clock to
sync it with: by the start time of the video, when sampling was given
one, and otherwise by assuming that the track starts with the video,
with a warning when their lengths differ (see _check_length). No other
time of the video, such as its creation time, which the camera's own
clock stamps, places the track. It runs at --video_duration_ratio GPS
seconds per second of video.
which is refused or warned of when their lengths differ (see
_check_length). No other time of the video, such as its creation time,
which the camera's own clock stamps, places the track. It runs at
--video_duration_ratio GPS seconds per second of video.

Return the clock of the video the track was placed by: the start time,
or None if the track was assumed to start with the video.
Expand Down Expand Up @@ -732,16 +732,19 @@ def _check_coverage(

def _check_length(self, gpx_points: T.Sequence[geo.Point], ratio: float) -> None:
"""
Warn when the GPX track, assumed to start with the video because the
video has no GPS clock to place it by, lasts longer or shorter than the
video by more than _GPX_LENGTH_TOLERANCE_SECONDS: the start of the video
then needs aligning with the track. A longer track warns only if it also
Check the GPX track, assumed to start with the video because the video
has no GPS clock to place it by, against the length of the video, which
spans **ratio** GPS seconds per second. A track that runs on for longer
than the whole video may be refused (see _check_assumed_start).

Otherwise, warn when the track lasts longer or shorter than the video
by more than _GPX_LENGTH_TOLERANCE_SECONDS: the start of the video then
needs aligning with the track. A longer track warns only if it also
lasts more than _MIN_LONG_GPX_RATIO times as long as the video, and may
be that of a timelapse. The video spans **ratio** GPS seconds per second.
be that of a timelapse.

The track is taken to be right, so neither is refused.
--interpolation_offset_time places the start of the video along the
track, which leaves nothing to warn of.
track, which leaves nothing to refuse or warn of.
"""
if self.interpolation is not None and self.interpolation.offset_time:
return
Expand All @@ -751,6 +754,8 @@ def _check_length(self, gpx_points: T.Sequence[geo.Point], ratio: float) -> None
return

video_span, gpx_duration = durations
self._check_assumed_start(gpx_points, video_span, gpx_duration, ratio)

longer = (
_GPX_LENGTH_TOLERANCE_SECONDS < gpx_duration - video_span
and _MIN_LONG_GPX_RATIO * video_span < gpx_duration
Expand Down Expand Up @@ -801,6 +806,63 @@ def _check_length(self, gpx_points: T.Sequence[geo.Point], ratio: float) -> None
LOG.warning(message)
self.warnings.append(warning)

def _check_assumed_start(
self,
gpx_points: T.Sequence[geo.Point],
video_span: float,
gpx_duration: float,
ratio: float,
) -> None:
"""
Refuse a GPX track assumed to start with the video that runs on for
longer than the whole video, which spans **video_span** GPS seconds at
**ratio** GPS seconds per second: it was not recorded for the video
alone, and the video could start anywhere along it.

Sampling refuses it: --video_start_time places the video along the
track, and --video_duration_ratio spans it at the pace of a timelapse.
process takes neither, and refuses only a track that lasts longer than
_MAX_RECORDING_SECONDS, a log of several trips whose start is that of
the first. A shorter one may be the track of a timelapse, which runs on
past the end of the video by design: process uploads it whole, with a
warning to align them (see _check_length).
"""
excess = gpx_duration - video_span
if excess <= video_span:
return
if self.video_sync is None and gpx_duration < _MAX_RECORDING_SECONDS:
return

message = (
f"The {self.file_format} track in {self.gpx_path} lasts {_format_duration(gpx_duration)} "
f"and {self._describe_video_span(video_span, ratio)}, so the video "
"could start anywhere along the track. "
)
if self.video_sync is None:
# process, which has no --video_start_time
message += (
"A track that long covers more than this video: trim the "
f"{self.file_format} to the video, or sample the video with "
"video_process and --video_start_time"
)
# Not --interpolation_use_gpx_start_time, which starts the track with
# the video too, and refuses it for the same reason
elif gpx_duration < _MAX_RECORDING_SECONDS:
message += (
"If the video is a timelapse, specify --video_duration_ratio. If "
"the GPS logger was started with the video and left running, "
"specify --video_start_time "
f"{build_capture_time(min(p.time for p in gpx_points))}, when the "
"track starts. If the track covers more than this video, specify "
"the time the video started as --video_start_time instead"
)
else:
message += (
"A track that long covers more than this video: specify the time "
"the video started as --video_start_time"
)
raise exceptions.MapillaryGPXVideoStartUnknownError(message)

def _check_given_start(
self, gpx_points: T.Sequence[geo.Point], ratio: float
) -> None:
Expand Down
11 changes: 7 additions & 4 deletions mapillary_tools/sample_video.py
Original file line number Diff line number Diff line change
Expand Up @@ -718,10 +718,11 @@ def _track_file_start_time(
of the first source that geotags images, which the geotagging stage places
the interval samples along by their capture times: the start of the track,
which is assumed to start with the video, as GPX sync places a track on
such a video (see GPXVideoExtractor._place_without_video_gps), with a
warning when their lengths differ. None without such a track, or when
--interpolation_use_gpx_start_time starts the track with the video, which
the geotagging stage does.
such a video (see GPXVideoExtractor._place_without_video_gps): refused
if it runs on for longer than the whole video, and warned of if their
lengths differ (see GPXVideoExtractor._check_length). None without such a
track, or when --interpolation_use_gpx_start_time starts the track with
the video, which the geotagging stage does.
"""
for option in geotag_options:
extractor_class = _TRACK_FILE_EXTRACTORS.get(option.source)
Expand Down Expand Up @@ -751,6 +752,8 @@ def _track_file_start_time(
return None
if not gpx_points:
return None
# Refused or warned of as in distance sampling, which places the track
# by GPX sync
extractor._check_length(
gpx_points, 1.0 if video_sync is None else video_sync.duration_ratio
)
Expand Down
33 changes: 33 additions & 0 deletions tests/integration/test_process.py
Original file line number Diff line number Diff line change
Expand Up @@ -1715,6 +1715,39 @@ def test_process_video_with_a_gpx_far_longer_than_it_warns(
assert desc["MAPGPSTrack"][0][0] == 0


def test_process_video_with_a_gpx_longer_than_a_day_fails(
setup_data: py.path.local,
):
"""
A track that lasts more than a day is a log of several trips, whose start
is that of the first: process does not assume that the video starts with
it, as sampling does not for any track that runs on past the video for
longer than the video lasts
"""
video_path = setup_data.join("videos").join("sample-5s.mp4")
start = datetime.datetime(2025, 3, 14, 7, 0, 0, tzinfo=datetime.timezone.utc)
gpx_file = setup_data.join("track.gpx")
# Two trips two days apart
_write_track_parts(
gpx_file,
[(start, 10, 37.0), (start + datetime.timedelta(days=2), 10, 37.1)],
)

[desc] = run_process_for_descs(
[
*["--geotag_source", "gpx"],
*["--geotag_source_path", str(gpx_file)],
str(video_path),
]
)

assert desc["error"]["type"] == "MapillaryGPXVideoStartUnknownError"
message = desc["error"]["message"]
assert "lasts 2.0 days" in message
assert "could start anywhere along the track" in message
assert "trim the GPX to the video" in message


def test_process_image_with_a_placeholder_capture_time_and_offset(
tmpdir: py.path.local,
):
Expand Down
56 changes: 48 additions & 8 deletions tests/integration/test_video_process.py
Original file line number Diff line number Diff line change
Expand Up @@ -503,14 +503,15 @@ def test_video_process_by_distance_places_a_gpx_by_the_video_start_time(
@pytest.mark.parametrize(
"creation_time", ["2025-03-14T07:00:36Z", None], ids=["creation time", "none"]
)
def test_video_process_assumes_a_long_gpx_starts_with_the_video(
def test_video_process_refuses_to_guess_where_the_video_starts(
setup_data: py.path.local, sampling: list[str], creation_time: str | None
):
"""
Without --video_start_time, a video without GPS of its own is assumed to
start with the GPX, which is taken to be right: never by its creation time,
which the camera's clock stamps, even where the track covers it. A track
much longer than the video warns to align the two, and is not refused.
Without --video_start_time, a GPX far longer than the video could start
anywhere along it, so assuming that it starts with the video is a guess:
refused, by distance and by interval alike. Nor does the creation time of
the video, which the camera's clock stamps, place it, even where the track
covers it.
"""
pytest_skip_if_not_ffmpeg_installed()

Expand All @@ -523,14 +524,54 @@ def test_video_process_assumes_a_long_gpx_starts_with_the_video(
video_path = str(
_with_creation_time(py.path.local(video_path), creation_time, dir)
)
with pytest.raises(subprocess.CalledProcessError) as ex:
run_video_process(
[
*sampling,
"--geotag_source=gpx",
*["--geotag_source_path", str(gpx_file)],
video_path,
str(Path(dir, "samples")),
],
capture_output=True,
text=True,
)

assert 7 == ex.value.returncode, ex.value.stderr
assert "could start anywhere along the track" in ex.value.stderr
assert "specify --video_start_time 2025_03_14_07_00_00_000" in ex.value.stderr


@pytest.mark.parametrize(
"sampling",
[
["--video_sample_distance=3"],
["--video_sample_distance=-1", "--video_sample_interval=1"],
],
ids=["by distance", "by interval"],
)
def test_video_process_takes_a_long_gpx_that_fits_at_the_duration_ratio(
setup_data: py.path.local, sampling: list[str]
):
"""
At 30 GPS seconds per second, the 5.76 seconds of video span 173 seconds,
past which the 93-second track does not run on: assumed to start with the
video, as the track of a timelapse
"""
pytest_skip_if_not_ffmpeg_installed()

gpx_file = setup_data.join("gpx").join("sf_30km_h.gpx")

with tempfile.TemporaryDirectory() as dir:
desc_path = Path(dir, "descs.json")
result = run_video_process(
[
*sampling,
"--video_duration_ratio=30",
"--geotag_source=gpx",
*["--geotag_source_path", str(gpx_file)],
*["--desc_path", str(desc_path)],
video_path,
str(setup_data.join("videos").join("sample-5s.mp4")),
str(Path(dir, "samples")),
],
capture_output=True,
Expand All @@ -542,8 +583,7 @@ def test_video_process_assumes_a_long_gpx_starts_with_the_video(
assert all("error" not in desc for desc in descs), descs
capture_times = sorted(desc["MAPCaptureTime"] for desc in descs)
assert "2025_03_14_07_00_00_000" <= capture_times[0], capture_times
assert capture_times[-1] < "2025_03_14_07_00_07_000", capture_times
assert "Please align the video start time with the GPX track" in result.stderr
assert "could start anywhere along the track" not in result.stderr


def test_video_process_by_distance_with_a_gpx_that_misses_the_video(
Expand Down
Loading
Loading