Sample a single video stream by interval, as by distance

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:

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")