diff --git a/dimos/hardware/sensors/camera/test_webcam_backend.py b/dimos/hardware/sensors/camera/test_webcam_backend.py new file mode 100644 index 0000000000..01db0c7143 --- /dev/null +++ b/dimos/hardware/sensors/camera/test_webcam_backend.py @@ -0,0 +1,53 @@ +# Copyright 2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import cv2 +import pytest + +from dimos.hardware.sensors.camera.webcam import Webcam + + +@pytest.fixture +def webcam(mocker, request): + camera = Webcam(camera_index=request.param) + mocker.patch("dimos.hardware.sensors.camera.webcam.threading.Thread") + yield camera + camera.stop() + + +@pytest.mark.parametrize( + ("platform", "webcam", "device", "backend"), + [ + ("linux", "/dev/video0", "/dev/video0", cv2.CAP_V4L2), + ("linux", "/dev/v4l/by-id/usb-camera", "/dev/v4l/by-id/usb-camera", cv2.CAP_V4L2), + ("linux", 2, 2, cv2.CAP_ANY), + ("linux", "2", 2, cv2.CAP_ANY), + ("linux", "-1", -1, cv2.CAP_ANY), + ("linux", "-2", -2, cv2.CAP_ANY), + ("linux", "+2", 2, cv2.CAP_ANY), + ("linux", " 2 ", 2, cv2.CAP_ANY), + ("linux", " -1 ", -1, cv2.CAP_ANY), + ("linux", "rtsp://camera/stream", "rtsp://camera/stream", cv2.CAP_ANY), + ("darwin", 0, 0, cv2.CAP_ANY), + ("darwin", "/dev/video0", "/dev/video0", cv2.CAP_ANY), + ], + indirect=["webcam"], +) +def test_capture_backend(platform, webcam, device, backend, mocker): + video_capture = mocker.patch("cv2.VideoCapture") + mocker.patch("sys.platform", platform) + + webcam.start() + + video_capture.assert_called_once_with(device, backend) diff --git a/dimos/hardware/sensors/camera/webcam.py b/dimos/hardware/sensors/camera/webcam.py index e647f32d32..177cfd34ed 100644 --- a/dimos/hardware/sensors/camera/webcam.py +++ b/dimos/hardware/sensors/camera/webcam.py @@ -13,11 +13,12 @@ # limitations under the License. from functools import cache +import sys import threading import time -from typing import Literal +from typing import Annotated, Any, Literal -from pydantic import Field +from pydantic import BeforeValidator, Field from reactivex import create from reactivex.observable import Observable @@ -27,8 +28,20 @@ from dimos.utils.reactive import backpressure +def _parse_camera_device(value: Any) -> Any: + if isinstance(value, str): + try: + return int(value) + except ValueError: + pass + return value + + +CameraDevice = Annotated[int | str, BeforeValidator(_parse_camera_device)] + + class WebcamConfig(CameraConfig): - camera_index: int = 0 # /dev/videoN + camera_index: CameraDevice = 0 # Index or device path such as /dev/v4l/by-id/... width: int = 640 height: int = 480 fps: float = 15.0 @@ -77,8 +90,13 @@ def start(self): # type: ignore[no-untyped-def] if self._capture_thread and self._capture_thread.is_alive(): return - # Open the video capture - self._capture = cv2.VideoCapture(self.config.camera_index) # type: ignore[assignment] + # Device paths otherwise let FFmpeg open the camera, which cannot apply + # the requested capture dimensions and frame rate through set(). + device = self.config.camera_index + backend = cv2.CAP_ANY + if sys.platform == "linux" and isinstance(device, str) and device.startswith("/dev/"): + backend = cv2.CAP_V4L2 + self._capture = cv2.VideoCapture(device, backend) # type: ignore[assignment] if not self._capture.isOpened(): # type: ignore[attr-defined] raise RuntimeError(f"Failed to open camera {self.config.camera_index}")