[Backport release/0.0.14] fix(camera): select V4L2 for Linux device paths

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)

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