From e57865fb8cd0d479a35c0978670ce11ff38a4548 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Tue, 20 Oct 2026 15:21:53 +0200 Subject: [PATCH] Sample a single video stream by interval, as by distance Interval sampling mapped every video stream of a video to one output, and ffmpeg numbers the frames of all of them in one sequence. A GoPro MAX .360 video has two video streams, so its frames came out interleaved, and frame i, taken at (i - 1) // 2 intervals, was stamped (i - 1) intervals into the video. Half of them were stamped past its end. On a 5-minute .360 video sampled every second, the frames that could be geotagged were placed a median of 73 m from where they were taken, and the other half failed. Interval sampling now samples the video stream with the highest resolution, as distance sampling does. A video with a single video stream, which is nearly every video, is sampled as before and its samples keep their names. --- mapillary_tools/sample_video.py | 34 ++++++++++++++++++++++++++++----- tests/unit/test_sample_video.py | 34 +++++++++++++++++++++++++++++++++ 2 files changed, 63 insertions(+), 5 deletions(-) diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py index e945d252..6ec33da8 100644 --- a/mapillary_tools/sample_video.py +++ b/mapillary_tools/sample_video.py @@ -560,6 +560,24 @@ def _extract_video_start_time( return _creation_time_to_start_time(video_path, probe) +def _interval_stream_specifier(probe: ffmpeglib.Probe) -> str: + """ + The video stream to sample by interval: the one with the highest + resolution, as for sampling by distance. ffmpeg numbers the frames of all + the streams it samples in one sequence, so with two (a GoPro MAX .360 video + has two) each frame would be numbered, and stamped, as if twice as late. + A video with a single video stream is sampled as "v", which names its + samples as it always has. + """ + video_streams = probe.probe_video_streams() + if len(video_streams) <= 1: + return "v" + + video_stream = probe.probe_video_with_max_resolution() + assert video_stream is not None + return str(video_stream["index"]) + + def _sample_single_video_by_interval( video_path: Path, sample_dir: Path, @@ -569,18 +587,24 @@ def _sample_single_video_by_interval( ) -> None: ffmpeg = ffmpeglib.FFMPEG(constants.FFMPEG_PATH, constants.FFPROBE_PATH) + probe = ffmpeglib.Probe(ffmpeg.probe_format_and_streams(video_path)) + if start_time is None: - start_time = _extract_video_start_time( - video_path, ffmpeglib.Probe(ffmpeg.probe_format_and_streams(video_path)) - ) + start_time = _extract_video_start_time(video_path, probe) if start_time is None: raise exceptions.MapillaryVideoError( f"Unable to extract video start time from {video_path}" ) + stream_specifier = _interval_stream_specifier(probe) + with wip_dir_context(wip_sample_dir(sample_dir), sample_dir) as wip_dir: - ffmpeg.extract_frames_by_interval(video_path, wip_dir, sample_interval) - frame_samples = ffmpeglib.FFMPEG.sort_selected_samples(wip_dir, video_path) + ffmpeg.extract_frames_by_interval( + video_path, wip_dir, sample_interval, stream_specifier=stream_specifier + ) + frame_samples = ffmpeglib.FFMPEG.sort_selected_samples( + wip_dir, video_path, selected_stream_specifiers=[stream_specifier] + ) for frame_idx_1based, sample_paths in frame_samples: assert len(sample_paths) == 1 if sample_paths[0] is None: diff --git a/tests/unit/test_sample_video.py b/tests/unit/test_sample_video.py index cf4760b7..fa08869f 100644 --- a/tests/unit/test_sample_video.py +++ b/tests/unit/test_sample_video.py @@ -131,6 +131,40 @@ def test_sample_video_with_start_time(tmpdir: py.path.local, setup_mock): _validate_interval([Path(s) for s in samples], video_start_time) +def test_sample_the_video_stream_with_the_highest_resolution( + tmpdir: py.path.local, setup_mock +): + """ + By interval as by distance: ffmpeg would number the frames of two streams + in one sequence, and they would be stamped as if twice as late. + """ + root = _PWD.joinpath("data/mock_sample_video") + probe = json.loads(root.joinpath("videos", "hello.mp4").read_text()) + [video_stream] = [s for s in probe["streams"] if s["codec_type"] == "video"] + probe["streams"].append({**video_stream, "index": 2, "width": 5760}) + video_dir = Path(tmpdir.mkdir("videos")) + video_dir.joinpath("hello.mp4").write_text(json.dumps(probe)) + sample_dir = tmpdir.mkdir("sampled_video_frames") + + sample_video.sample_video( + video_dir, + Path(sample_dir), + video_sample_distance=-1, + video_sample_interval=2, + rerun=True, + ) + + samples = sorted(Path(s) for s in sample_dir.join("hello.mp4").listdir()) + assert len(samples), "expect samples but got none" + for idx, sample in enumerate(samples): + assert sample.name == f"hello_2_{idx + 1:06d}.jpg" + assert exif_read.ExifRead( + sample + ).extract_capture_time() == PROBE_START_TIME + datetime.timedelta( + seconds=2 * idx + ) + + def test_sample_video_from_gps_clock(tmpdir: py.path.local, setup_mock, monkeypatch): """A video's own GPS clock wins over the container's creation time.""" root = _PWD.joinpath("data/mock_sample_video")