Add video device index (#51)

* Add video device index

* Add unit tests

Author: openminddev

src/om1_vlm/video/video_stream.py

@@ -38,6 +38,8 @@ class VideoStream:
         By default (640, 480)
     jpeg_quality : int, optional
         JPEG quality for encoding frames, by default 70
+    device_index : int, optional
+        Index of the camera device to use, by default 0
     """
 
     def __init__(
@@ -47,6 +49,7 @@ def __init__(
         fps: Optional[int] = 30,
         resolution: Optional[Tuple[int, int]] = (640, 480),
         jpeg_quality: int = 70,
+        device_index: int = 0,
     ):
         self._video_thread: Optional[threading.Thread] = None
 
@@ -55,6 +58,7 @@ def __init__(
         self.register_frame_callback(frame_callback)
 
         # Video capture device
+        self.device_index = device_index
         self._cap = None
 
         self.running: bool = True
@@ -96,6 +100,13 @@ def on_video(self):
             camindex = 0 if devices else 0
         else:
             camindex = "/dev/video" + str(devices[0][0]) if devices else "/dev/video0"
+
+        if self.device_index != 0:
+            if platform.system() == "Darwin":
+                camindex = self.device_index
+            else:
+                camindex = f"/dev/video{self.device_index}"
+
         logger.info(f"Using camera: {camindex}")
 
         self._cap = cv2.VideoCapture(camindex)

tests/om1_vlm/video/test_video_stream.py

@@ -360,3 +360,197 @@ def callback(frame_data):
                 assert (
                     ts > 1600000000
                 ), "Timestamp should be a reasonable Unix timestamp"
+
+
+def test_device_index_default():
+    """Test that default device_index (0) works correctly."""
+    captured_device_indices = []
+
+    def mock_video_capture(device_index):
+        captured_device_indices.append(device_index)
+        return MockVideoCapture(device_index)
+
+    mock_devices = [(0, "Mock Camera"), (1, "Another Camera")]
+
+    with (
+        patch(
+            "om1_vlm.video.video_stream.enumerate_video_devices",
+            return_value=mock_devices,
+        ),
+        patch("platform.system", return_value="Linux"),
+        patch("cv2.VideoCapture", side_effect=mock_video_capture),
+        patch("cv2.imencode", return_value=(True, b"fake_image_data")),
+    ):
+        # Test default device_index (0)
+        stream = VideoStream(device_index=0)
+        stream.start()
+
+        # Wait briefly for initialization
+        time.sleep(0.1)
+        stream.stop()
+
+        # Verify that the default device was used (Linux uses /dev/video0 for device 0)
+        assert len(captured_device_indices) > 0, "VideoCapture should have been called"
+        assert (
+            captured_device_indices[0] == "/dev/video0"
+        ), f"Expected /dev/video0, got {captured_device_indices[0]}"
+
+
+def test_device_index_custom_linux():
+    """Test that custom device_index works correctly on Linux."""
+    captured_device_indices = []
+
+    def mock_video_capture(device_index):
+        captured_device_indices.append(device_index)
+        return MockVideoCapture(device_index)
+
+    mock_devices = [(0, "Mock Camera"), (1, "Another Camera"), (2, "Third Camera")]
+
+    with (
+        patch(
+            "om1_vlm.video.video_stream.enumerate_video_devices",
+            return_value=mock_devices,
+        ),
+        patch("platform.system", return_value="Linux"),
+        patch("cv2.VideoCapture", side_effect=mock_video_capture),
+        patch("cv2.imencode", return_value=(True, b"fake_image_data")),
+    ):
+        # Test custom device_index (2)
+        stream = VideoStream(device_index=2)
+        stream.start()
+
+        # Wait briefly for initialization
+        time.sleep(0.1)
+        stream.stop()
+
+        # Verify that the correct device was used (Linux uses /dev/video2 for device 2)
+        assert len(captured_device_indices) > 0, "VideoCapture should have been called"
+        assert (
+            captured_device_indices[0] == "/dev/video2"
+        ), f"Expected /dev/video2, got {captured_device_indices[0]}"
+
+
+def test_device_index_custom_macos():
+    """Test that custom device_index works correctly on macOS."""
+    captured_device_indices = []
+
+    def mock_video_capture(device_index):
+        captured_device_indices.append(device_index)
+        return MockVideoCapture(device_index)
+
+    mock_devices = [(0, "Mock Camera"), (1, "Another Camera"), (2, "Third Camera")]
+
+    with (
+        patch(
+            "om1_vlm.video.video_stream.enumerate_video_devices",
+            return_value=mock_devices,
+        ),
+        patch("platform.system", return_value="Darwin"),
+        patch("cv2.VideoCapture", side_effect=mock_video_capture),
+        patch("cv2.imencode", return_value=(True, b"fake_image_data")),
+    ):
+        # Test custom device_index (1)
+        stream = VideoStream(device_index=1)
+        stream.start()
+
+        # Wait briefly for initialization
+        time.sleep(0.1)
+        stream.stop()
+
+        # Verify that the correct device was used (macOS uses device_index directly)
+        assert len(captured_device_indices) > 0, "VideoCapture should have been called"
+        assert (
+            captured_device_indices[0] == 1
+        ), f"Expected 1, got {captured_device_indices[0]}"
+
+
+def test_device_index_zero_macos():
+    """Test that device_index 0 works correctly on macOS."""
+    captured_device_indices = []
+
+    def mock_video_capture(device_index):
+        captured_device_indices.append(device_index)
+        return MockVideoCapture(device_index)
+
+    mock_devices = [(0, "Mock Camera"), (1, "Another Camera")]
+
+    with (
+        patch(
+            "om1_vlm.video.video_stream.enumerate_video_devices",
+            return_value=mock_devices,
+        ),
+        patch("platform.system", return_value="Darwin"),
+        patch("cv2.VideoCapture", side_effect=mock_video_capture),
+        patch("cv2.imencode", return_value=(True, b"fake_image_data")),
+    ):
+        # Test device_index 0 on macOS
+        stream = VideoStream(device_index=0)
+        stream.start()
+
+        # Wait briefly for initialization
+        time.sleep(0.1)
+        stream.stop()
+
+        # Verify that device 0 was used (macOS uses device_index directly)
+        assert len(captured_device_indices) > 0, "VideoCapture should have been called"
+        assert (
+            captured_device_indices[0] == 0
+        ), f"Expected 0, got {captured_device_indices[0]}"
+
+
+def test_device_index_no_devices_available():
+    """Test device_index behavior when no devices are available."""
+    captured_device_indices = []
+
+    def mock_video_capture(device_index):
+        captured_device_indices.append(device_index)
+        return MockVideoCapture(device_index)
+
+    # Empty device list
+    mock_devices = []
+
+    with (
+        patch(
+            "om1_vlm.video.video_stream.enumerate_video_devices",
+            return_value=mock_devices,
+        ),
+        patch("platform.system", return_value="Linux"),
+        patch("cv2.VideoCapture", side_effect=mock_video_capture),
+        patch("cv2.imencode", return_value=(True, b"fake_image_data")),
+    ):
+        # Test with custom device_index when no devices are available
+        stream = VideoStream(device_index=1)
+        stream.start()
+
+        # Wait briefly for initialization
+        time.sleep(0.1)
+        stream.stop()
+
+        # Should still use the specified device_index
+        assert len(captured_device_indices) > 0, "VideoCapture should have been called"
+        assert (
+            captured_device_indices[0] == "/dev/video1"
+        ), f"Expected /dev/video1, got {captured_device_indices[0]}"
+
+
+def test_device_index_initialization():
+    """Test that device_index is properly stored during initialization."""
+    # Test default device_index
+    stream = VideoStream()
+    assert stream.device_index == 0, "Default device_index should be 0"
+
+    # Test custom device_index
+    stream_custom = VideoStream(device_index=3)
+    assert (
+        stream_custom.device_index == 3
+    ), "Custom device_index should be stored correctly"
+
+    # Test with other parameters
+    stream_complex = VideoStream(
+        device_index=5, fps=60, resolution=(1920, 1080), jpeg_quality=90
+    )
+    assert (
+        stream_complex.device_index == 5
+    ), "Device index should be preserved with other parameters"
+    assert stream_complex.fps == 60, "Other parameters should also be preserved"
+    assert stream_complex.resolution == (1920, 1080), "Resolution should be preserved"