|
2 | 2 |
|
3 | 3 | """Tests for io_utils extensionless video file handling (D99228861).""" |
4 | 4 |
|
| 5 | +import os |
5 | 6 | import tempfile |
6 | 7 | import unittest |
7 | 8 | from unittest.mock import MagicMock, patch |
8 | 9 |
|
| 10 | +import numpy as np |
| 11 | +import torch |
| 12 | + |
9 | 13 | from sam3.model.io_utils import load_video_frames |
10 | 14 |
|
11 | 15 |
|
@@ -107,6 +111,63 @@ def test_dummy_video_pattern(self) -> None: |
107 | 111 | self.assertEqual(h, 480) |
108 | 112 | self.assertEqual(w, 640) |
109 | 113 |
|
| 114 | + def test_cv2_video_file_loader_scales_before_normalization(self) -> None: |
| 115 | + """OpenCV video loading should match normalized decoded uint8 frames.""" |
| 116 | + try: |
| 117 | + import cv2 |
| 118 | + except ImportError as exc: |
| 119 | + self.skipTest(f"OpenCV is required for this test: {exc}") |
| 120 | + |
| 121 | + image_size = 8 |
| 122 | + img_mean = (0.5, 0.25, 0.75) |
| 123 | + img_std = (0.5, 0.25, 0.25) |
| 124 | + frames_rgb = [ |
| 125 | + np.full((image_size, image_size, 3), (255, 0, 128), dtype=np.uint8), |
| 126 | + np.full((image_size, image_size, 3), (16, 192, 64), dtype=np.uint8), |
| 127 | + ] |
| 128 | + |
| 129 | + with tempfile.TemporaryDirectory() as tmpdir: |
| 130 | + video_path = os.path.join(tmpdir, "tiny.avi") |
| 131 | + writer = cv2.VideoWriter( |
| 132 | + video_path, |
| 133 | + cv2.VideoWriter_fourcc(*"MJPG"), |
| 134 | + 2.0, |
| 135 | + (image_size, image_size), |
| 136 | + ) |
| 137 | + if not writer.isOpened(): |
| 138 | + self.skipTest("OpenCV could not create a temporary MJPG video") |
| 139 | + |
| 140 | + for frame_rgb in frames_rgb: |
| 141 | + writer.write(cv2.cvtColor(frame_rgb, cv2.COLOR_RGB2BGR)) |
| 142 | + writer.release() |
| 143 | + |
| 144 | + decoded_frames = [] |
| 145 | + cap = cv2.VideoCapture(video_path) |
| 146 | + while True: |
| 147 | + ret, frame_bgr = cap.read() |
| 148 | + if not ret: |
| 149 | + break |
| 150 | + decoded_frames.append(cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2RGB)) |
| 151 | + cap.release() |
| 152 | + self.assertEqual(len(decoded_frames), len(frames_rgb)) |
| 153 | + |
| 154 | + expected = np.stack(decoded_frames, axis=0).astype(np.float32) / 255.0 |
| 155 | + expected = torch.from_numpy(expected).permute(0, 3, 1, 2) |
| 156 | + expected -= torch.tensor(img_mean, dtype=torch.float16).view(1, 3, 1, 1) |
| 157 | + expected /= torch.tensor(img_std, dtype=torch.float16).view(1, 3, 1, 1) |
| 158 | + |
| 159 | + frames, height, width = load_video_frames( |
| 160 | + video_path=video_path, |
| 161 | + image_size=image_size, |
| 162 | + offload_video_to_cpu=True, |
| 163 | + img_mean=img_mean, |
| 164 | + img_std=img_std, |
| 165 | + video_loader_type="cv2", |
| 166 | + ) |
| 167 | + |
| 168 | + self.assertEqual((height, width), (image_size, image_size)) |
| 169 | + torch.testing.assert_close(frames, expected, rtol=0, atol=1e-6) |
| 170 | + |
110 | 171 | @patch("sam3.model.io_utils.load_video_frames_from_video_file") |
111 | 172 | def test_unknown_extension_routes_to_video_loader( |
112 | 173 | self, mock_load_video: MagicMock |
|
0 commit comments