Browse Source

Add USB camera support (V4L2)
- Add USB camera type to external camera service
- Auto-detect available V4L2 devices on Linux
- New API endpoint: GET /api/v1/printers/usb-cameras
- Use ffmpeg for USB camera capture and streaming
- Add "USB Camera (V4L2)" option in Settings UI
- Debounce camera URL input to avoid saving on every keystroke

Closes #143

maziggy 4 months ago
parent
commit
9d6164ab1c

+ 7 - 5
CHANGELOG.md

@@ -2,9 +2,15 @@
 
 
 All notable changes to Bambuddy will be documented in this file.
 All notable changes to Bambuddy will be documented in this file.
 
 
-## [0.1.7] - 2026-01-26
+## [0.1.6] - Not released
 
 
 ### New Features
 ### New Features
+- **USB Camera Support** - Connect USB webcams directly to your Bambuddy host:
+  - New "USB Camera (V4L2)" option in external camera settings
+  - Auto-detection of available USB cameras via V4L2
+  - API endpoint to list connected USB cameras (`GET /api/v1/printers/usb-cameras`)
+  - Works with any V4L2-compatible camera on Linux
+  - Uses ffmpeg for frame capture and streaming
 - **Build Plate Empty Detection** - Automatically detect if objects are on the build plate before printing:
 - **Build Plate Empty Detection** - Automatically detect if objects are on the build plate before printing:
   - Per-printer toggle to enable/disable plate detection
   - Per-printer toggle to enable/disable plate detection
   - Multi-reference calibration: Store up to 5 reference images of empty plates (different plate types)
   - Multi-reference calibration: Store up to 5 reference images of empty plates (different plate types)
@@ -17,10 +23,6 @@ All notable changes to Bambuddy will be documented in this file.
   - Split button UI: Main button opens calibration modal, chevron toggles detection on/off
   - Split button UI: Main button opens calibration modal, chevron toggles detection on/off
   - Green visual indicator when plate detection is enabled
   - Green visual indicator when plate detection is enabled
   - Included in backup/restore
   - Included in backup/restore
-
-## [0.1.6] - 2026-01-24
-
-### New Features
 - **Project Import/Export** - Export and import projects with full file support (Issue #152):
 - **Project Import/Export** - Export and import projects with full file support (Issue #152):
   - Export single project as ZIP (includes project settings, BOM, and all files from linked library folders)
   - Export single project as ZIP (includes project settings, BOM, and all files from linked library folders)
   - Export all projects as JSON for metadata-only backup
   - Export all projects as JSON for metadata-only backup

+ 1 - 1
README.md

@@ -57,7 +57,7 @@
 ### 📊 Monitoring & Control
 ### 📊 Monitoring & Control
 - Real-time printer status via WebSocket
 - Real-time printer status via WebSocket
 - Live camera streaming (MJPEG) & snapshots with multi-viewer support
 - Live camera streaming (MJPEG) & snapshots with multi-viewer support
-- External network camera support (MJPEG, RTSP, HTTP snapshot) with layer-based timelapse
+- External camera support (MJPEG, RTSP, HTTP snapshot, USB/V4L2) with layer-based timelapse
 - **Build plate empty detection** - Auto-pause print if objects detected on plate (multi-reference calibration, ROI adjustment)
 - **Build plate empty detection** - Auto-pause print if objects detected on plate (multi-reference calibration, ROI adjustment)
 - Fan status monitoring (part cooling, auxiliary, chamber)
 - Fan status monitoring (part cooling, auxiliary, chamber)
 - Printer control (stop, pause, resume, chamber light)
 - Printer control (stop, pause, resume, chamber light)

+ 2 - 2
backend/app/api/routes/camera.py

@@ -663,8 +663,8 @@ async def test_external_camera(
 
 
     Args:
     Args:
         printer_id: Printer ID (for authorization)
         printer_id: Printer ID (for authorization)
-        url: Camera URL to test
-        camera_type: Camera type ("mjpeg", "rtsp", "snapshot")
+        url: Camera URL or USB device path to test
+        camera_type: Camera type ("mjpeg", "rtsp", "snapshot", "usb")
 
 
     Returns:
     Returns:
         Dict with {success: bool, error?: str, resolution?: str}
         Dict with {success: bool, error?: str, resolution?: str}

+ 16 - 0
backend/app/api/routes/printers.py

@@ -68,6 +68,22 @@ async def create_printer(
     return printer
     return printer
 
 
 
 
+@router.get("/usb-cameras")
+async def list_usb_cameras():
+    """List available USB cameras connected to the system.
+
+    Returns a list of detected V4L2 video devices with their info.
+    Only works on Linux systems with V4L2 support.
+
+    Returns:
+        List of dicts with {device: str, name: str, capabilities: list, formats?: list}
+    """
+    from backend.app.services.external_camera import list_usb_cameras
+
+    cameras = list_usb_cameras()
+    return {"cameras": cameras}
+
+
 @router.get("/{printer_id}", response_model=PrinterResponse)
 @router.get("/{printer_id}", response_model=PrinterResponse)
 async def get_printer(printer_id: int, db: AsyncSession = Depends(get_db)):
 async def get_printer(printer_id: int, db: AsyncSession = Depends(get_db)):
     """Get a specific printer."""
     """Get a specific printer."""

+ 12 - 0
backend/app/api/routes/settings.py

@@ -977,6 +977,18 @@ async def import_backup(
                             is_active_val = is_active_val.lower() == "true"
                             is_active_val = is_active_val.lower() == "true"
                         existing.is_active = is_active_val
                         existing.is_active = is_active_val
 
 
+                    # Restore external camera settings
+                    existing.external_camera_url = printer_data.get("external_camera_url")
+                    existing.external_camera_type = printer_data.get("external_camera_type")
+                    existing.external_camera_enabled = printer_data.get("external_camera_enabled", False)
+
+                    # Restore plate detection settings
+                    existing.plate_detection_enabled = printer_data.get("plate_detection_enabled", False)
+                    existing.plate_detection_roi_x = printer_data.get("plate_detection_roi_x")
+                    existing.plate_detection_roi_y = printer_data.get("plate_detection_roi_y")
+                    existing.plate_detection_roi_w = printer_data.get("plate_detection_roi_w")
+                    existing.plate_detection_roi_h = printer_data.get("plate_detection_roi_h")
+
                     restored["printers"] += 1
                     restored["printers"] += 1
                 else:
                 else:
                     skipped["printers"] += 1
                     skipped["printers"] += 1

+ 1 - 1
backend/app/schemas/printer.py

@@ -12,7 +12,7 @@ class PrinterBase(BaseModel):
     location: str | None = None  # Group/location name
     location: str | None = None  # Group/location name
     auto_archive: bool = True
     auto_archive: bool = True
     external_camera_url: str | None = None
     external_camera_url: str | None = None
-    external_camera_type: str | None = None  # "mjpeg", "rtsp", "snapshot"
+    external_camera_type: str | None = None  # "mjpeg", "rtsp", "snapshot", "usb"
     external_camera_enabled: bool = False
     external_camera_enabled: bool = False
 
 
 
 

+ 240 - 6
backend/app/services/external_camera.py

@@ -1,10 +1,11 @@
-"""External network camera service.
+"""External camera service.
 
 
-Supports MJPEG streams, RTSP streams (via ffmpeg), and HTTP snapshot URLs.
+Supports MJPEG streams, RTSP streams (via ffmpeg), HTTP snapshot URLs, and USB cameras.
 """
 """
 
 
 import asyncio
 import asyncio
 import logging
 import logging
+import re
 import shutil
 import shutil
 from collections.abc import AsyncGenerator
 from collections.abc import AsyncGenerator
 from pathlib import Path
 from pathlib import Path
@@ -14,6 +15,69 @@ import aiohttp
 logger = logging.getLogger(__name__)
 logger = logging.getLogger(__name__)
 
 
 
 
+def list_usb_cameras() -> list[dict]:
+    """List available USB cameras (V4L2 devices on Linux).
+
+    Returns:
+        List of dicts with {device: str, name: str, capabilities: list}
+    """
+    cameras = []
+    video_devices = sorted(Path("/dev").glob("video*"))
+
+    for device in video_devices:
+        device_path = str(device)
+        info = {"device": device_path, "name": device.name, "capabilities": []}
+
+        # Try to get device info via v4l2-ctl
+        v4l2_ctl = shutil.which("v4l2-ctl")
+        if v4l2_ctl:
+            import subprocess
+
+            try:
+                result = subprocess.run(
+                    [v4l2_ctl, "-d", device_path, "--info"],
+                    capture_output=True,
+                    text=True,
+                    timeout=5,
+                )
+                if result.returncode == 0:
+                    # Parse device name from output
+                    for line in result.stdout.splitlines():
+                        if "Card type" in line:
+                            info["name"] = line.split(":", 1)[1].strip()
+                        elif "Driver name" in line:
+                            info["driver"] = line.split(":", 1)[1].strip()
+
+                    # Check if device supports video capture
+                    result = subprocess.run(
+                        [v4l2_ctl, "-d", device_path, "--list-formats"],
+                        capture_output=True,
+                        text=True,
+                        timeout=5,
+                    )
+                    if result.returncode == 0 and result.stdout.strip():
+                        info["capabilities"].append("capture")
+                        # Parse available formats
+                        formats = re.findall(r"'(\w+)'", result.stdout)
+                        info["formats"] = list(set(formats))
+
+            except (subprocess.TimeoutExpired, Exception) as e:
+                logger.debug(f"v4l2-ctl failed for {device_path}: {e}")
+
+        # Only include devices that look like video capture devices
+        # Skip metadata devices (typically odd numbered like video1, video3)
+        try:
+            device_num = int(device.name.replace("video", ""))
+            # Even numbered devices are usually capture, odd are metadata
+            # But also check if we got capabilities
+            if info.get("capabilities") or device_num % 2 == 0:
+                cameras.append(info)
+        except ValueError:
+            cameras.append(info)
+
+    return cameras
+
+
 def get_ffmpeg_path() -> str | None:
 def get_ffmpeg_path() -> str | None:
     """Get the path to ffmpeg executable."""
     """Get the path to ffmpeg executable."""
     # Try shutil.which first
     # Try shutil.which first
@@ -31,8 +95,8 @@ async def capture_frame(url: str, camera_type: str, timeout: int = 15) -> bytes
     """Capture single frame from external camera.
     """Capture single frame from external camera.
 
 
     Args:
     Args:
-        url: Camera URL (MJPEG stream, RTSP URL, or HTTP snapshot URL)
-        camera_type: "mjpeg", "rtsp", or "snapshot"
+        url: Camera URL (MJPEG stream, RTSP URL, HTTP snapshot URL, or USB device path)
+        camera_type: "mjpeg", "rtsp", "snapshot", or "usb"
         timeout: Connection timeout in seconds
         timeout: Connection timeout in seconds
 
 
     Returns:
     Returns:
@@ -45,11 +109,77 @@ async def capture_frame(url: str, camera_type: str, timeout: int = 15) -> bytes
         return await _capture_rtsp_frame(url, timeout)
         return await _capture_rtsp_frame(url, timeout)
     elif camera_type == "snapshot":
     elif camera_type == "snapshot":
         return await _capture_snapshot(url, timeout)
         return await _capture_snapshot(url, timeout)
+    elif camera_type == "usb":
+        return await _capture_usb_frame(url, timeout)
     else:
     else:
         logger.warning(f"Unknown camera type: {camera_type}")
         logger.warning(f"Unknown camera type: {camera_type}")
         return None
         return None
 
 
 
 
+async def _capture_usb_frame(device: str, timeout: int) -> bytes | None:
+    """Capture frame from USB camera using ffmpeg."""
+    ffmpeg = get_ffmpeg_path()
+    if not ffmpeg:
+        logger.error("ffmpeg not found - required for USB camera capture")
+        return None
+
+    # Validate device path
+    if not device.startswith("/dev/video"):
+        logger.error(f"Invalid USB device path: {device}")
+        return None
+
+    if not Path(device).exists():
+        logger.error(f"USB device does not exist: {device}")
+        return None
+
+    # Use ffmpeg to grab a single frame from USB camera
+    cmd = [
+        ffmpeg,
+        "-f",
+        "v4l2",
+        "-i",
+        device,
+        "-frames:v",
+        "1",
+        "-f",
+        "image2pipe",
+        "-vcodec",
+        "mjpeg",
+        "-q:v",
+        "2",
+        "-",
+    ]
+
+    try:
+        logger.debug(f"Running USB capture: {' '.join(cmd)}")
+        process = await asyncio.create_subprocess_exec(
+            *cmd,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+
+        stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout)
+
+        if process.returncode != 0:
+            logger.error(f"ffmpeg USB capture failed: {stderr.decode()[:200]}")
+            return None
+
+        if not stdout or len(stdout) < 100:
+            logger.error("ffmpeg returned empty or too small frame from USB camera")
+            return None
+
+        return stdout
+
+    except TimeoutError:
+        logger.warning(f"USB frame capture timed out after {timeout}s")
+        if process:
+            process.kill()
+        return None
+    except Exception as e:
+        logger.error(f"USB frame capture failed: {e}")
+        return None
+
+
 async def _capture_mjpeg_frame(url: str, timeout: int) -> bytes | None:
 async def _capture_mjpeg_frame(url: str, timeout: int) -> bytes | None:
     """Extract single frame from MJPEG stream."""
     """Extract single frame from MJPEG stream."""
     try:
     try:
@@ -225,8 +355,8 @@ async def generate_mjpeg_stream(url: str, camera_type: str, fps: int = 10) -> As
     """Generator yielding MJPEG frames for streaming.
     """Generator yielding MJPEG frames for streaming.
 
 
     Args:
     Args:
-        url: Camera URL
-        camera_type: "mjpeg", "rtsp", or "snapshot"
+        url: Camera URL or USB device path
+        camera_type: "mjpeg", "rtsp", "snapshot", or "usb"
         fps: Target frames per second
         fps: Target frames per second
 
 
     Yields:
     Yields:
@@ -248,6 +378,11 @@ async def generate_mjpeg_stream(url: str, camera_type: str, fps: int = 10) -> As
         async for frame in _stream_rtsp(url, fps):
         async for frame in _stream_rtsp(url, fps):
             yield _format_mjpeg_frame(frame)
             yield _format_mjpeg_frame(frame)
 
 
+    elif camera_type == "usb":
+        # Use ffmpeg to stream from USB camera
+        async for frame in _stream_usb(url, fps):
+            yield _format_mjpeg_frame(frame)
+
     elif camera_type == "snapshot":
     elif camera_type == "snapshot":
         # Poll snapshot URL at interval
         # Poll snapshot URL at interval
         while True:
         while True:
@@ -407,3 +542,102 @@ async def _stream_rtsp(url: str, fps: int) -> AsyncGenerator[bytes, None]:
             except TimeoutError:
             except TimeoutError:
                 process.kill()
                 process.kill()
                 await process.wait()
                 await process.wait()
+
+
+async def _stream_usb(device: str, fps: int) -> AsyncGenerator[bytes, None]:
+    """Stream frames from USB camera via ffmpeg."""
+    ffmpeg = get_ffmpeg_path()
+    if not ffmpeg:
+        logger.error("ffmpeg not found - required for USB camera streaming")
+        return
+
+    # Validate device path
+    if not device.startswith("/dev/video"):
+        logger.error(f"Invalid USB device path: {device}")
+        return
+
+    if not Path(device).exists():
+        logger.error(f"USB device does not exist: {device}")
+        return
+
+    # ffmpeg command to stream from USB camera (v4l2)
+    cmd = [
+        ffmpeg,
+        "-f",
+        "v4l2",
+        "-framerate",
+        str(fps),
+        "-i",
+        device,
+        "-f",
+        "mjpeg",
+        "-q:v",
+        "5",
+        "-r",
+        str(fps),
+        "-",
+    ]
+
+    process = None
+    try:
+        logger.info(f"Starting USB camera stream from {device} at {fps} fps")
+        process = await asyncio.create_subprocess_exec(
+            *cmd,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+
+        # Give ffmpeg a moment to start and check for immediate failures
+        await asyncio.sleep(0.5)
+        if process.returncode is not None:
+            stderr = await process.stderr.read()
+            logger.error(f"ffmpeg USB stream failed immediately: {stderr.decode()[:300]}")
+            return
+
+        buffer = b""
+        jpeg_start = b"\xff\xd8"
+        jpeg_end = b"\xff\xd9"
+
+        while True:
+            try:
+                chunk = await asyncio.wait_for(process.stdout.read(8192), timeout=30.0)
+
+                if not chunk:
+                    break
+
+                buffer += chunk
+
+                # Extract complete frames
+                while True:
+                    start_idx = buffer.find(jpeg_start)
+                    if start_idx == -1:
+                        buffer = buffer[-2:] if len(buffer) > 2 else buffer
+                        break
+
+                    if start_idx > 0:
+                        buffer = buffer[start_idx:]
+
+                    end_idx = buffer.find(jpeg_end, 2)
+                    if end_idx == -1:
+                        break
+
+                    frame = buffer[: end_idx + 2]
+                    buffer = buffer[end_idx + 2 :]
+                    yield frame
+
+            except TimeoutError:
+                logger.warning("USB stream read timeout")
+                break
+
+    except asyncio.CancelledError:
+        logger.info("USB stream cancelled")
+    except Exception as e:
+        logger.error(f"USB stream error: {e}")
+    finally:
+        if process and process.returncode is None:
+            process.terminate()
+            try:
+                await asyncio.wait_for(process.wait(), timeout=2.0)
+            except TimeoutError:
+                process.kill()
+                await process.wait()

+ 48 - 0
backend/tests/integration/test_camera_api.py

@@ -408,3 +408,51 @@ class TestCameraAPI:
         response = await async_client.get("/api/v1/printers/99999/camera/plate-detection/references/0/thumbnail")
         response = await async_client.get("/api/v1/printers/99999/camera/plate-detection/references/0/thumbnail")
 
 
         assert response.status_code == 404
         assert response.status_code == 404
+
+    # ========================================================================
+    # USB Camera Endpoint
+    # ========================================================================
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_list_usb_cameras_returns_list(self, async_client: AsyncClient):
+        """Verify USB cameras endpoint returns a list of cameras."""
+        response = await async_client.get("/api/v1/printers/usb-cameras")
+
+        assert response.status_code == 200
+        result = response.json()
+        assert "cameras" in result
+        assert isinstance(result["cameras"], list)
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_list_usb_cameras_structure(self, async_client: AsyncClient):
+        """Verify USB cameras endpoint returns proper structure for each camera."""
+        with patch("backend.app.services.external_camera.list_usb_cameras") as mock_list:
+            mock_list.return_value = [
+                {"device": "/dev/video0", "name": "Logitech Webcam C920", "index": 0},
+                {"device": "/dev/video2", "name": "USB Camera", "index": 2},
+            ]
+
+            response = await async_client.get("/api/v1/printers/usb-cameras")
+
+        assert response.status_code == 200
+        result = response.json()
+        assert len(result["cameras"]) == 2
+        assert result["cameras"][0]["device"] == "/dev/video0"
+        assert result["cameras"][0]["name"] == "Logitech Webcam C920"
+        assert result["cameras"][1]["device"] == "/dev/video2"
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_list_usb_cameras_empty_on_non_linux(self, async_client: AsyncClient):
+        """Verify USB cameras endpoint returns empty list on non-Linux systems."""
+        with patch("backend.app.services.external_camera.list_usb_cameras") as mock_list:
+            # Simulate non-Linux system (no /dev/video* devices)
+            mock_list.return_value = []
+
+            response = await async_client.get("/api/v1/printers/usb-cameras")
+
+        assert response.status_code == 200
+        result = response.json()
+        assert result["cameras"] == []

+ 83 - 0
backend/tests/integration/test_settings_api.py

@@ -370,3 +370,86 @@ class TestSettingsAPI:
         assert "per_printer_mapping_expanded" in result
         assert "per_printer_mapping_expanded" in result
         # Default is False as defined in schema
         # Default is False as defined in schema
         assert isinstance(result["per_printer_mapping_expanded"], bool)
         assert isinstance(result["per_printer_mapping_expanded"], bool)
+
+    # ========================================================================
+    # Backup/Restore tests
+    # ========================================================================
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_backup_includes_external_camera_settings(self, async_client: AsyncClient, printer_factory):
+        """Verify backup includes external camera settings for printers."""
+        # Create a printer with external camera settings
+        _printer = await printer_factory(
+            name="Camera Test Printer",
+            external_camera_url="/dev/video0",
+            external_camera_type="usb",
+            external_camera_enabled=True,
+        )
+
+        # Request backup with printers
+        response = await async_client.get("/api/v1/settings/backup?include_printers=true")
+
+        assert response.status_code == 200
+        backup = response.json()
+
+        # Find the printer in the backup
+        assert "printers" in backup
+        printer_data = next((p for p in backup["printers"] if p["name"] == "Camera Test Printer"), None)
+        assert printer_data is not None
+
+        # Verify external camera fields are included
+        assert "external_camera_url" in printer_data
+        assert "external_camera_type" in printer_data
+        assert "external_camera_enabled" in printer_data
+        assert printer_data["external_camera_url"] == "/dev/video0"
+        assert printer_data["external_camera_type"] == "usb"
+        assert printer_data["external_camera_enabled"] is True
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_restore_external_camera_settings_overwrite(self, async_client: AsyncClient, printer_factory):
+        """Verify restore with overwrite updates external camera settings."""
+        import io
+
+        # Create a printer without camera settings
+        printer = await printer_factory(
+            name="Restore Test",
+            external_camera_url=None,
+            external_camera_type=None,
+            external_camera_enabled=False,
+        )
+
+        # Create backup data with camera settings
+        backup_data = {
+            "version": "1.0",
+            "included": ["printers"],
+            "printers": [
+                {
+                    "name": "Restore Test",
+                    "serial_number": printer.serial_number,
+                    "ip_address": printer.ip_address,
+                    "external_camera_url": "/dev/video1",
+                    "external_camera_type": "usb",
+                    "external_camera_enabled": True,
+                }
+            ],
+        }
+
+        # Restore with overwrite
+        import json
+
+        files = {"file": ("backup.json", io.BytesIO(json.dumps(backup_data).encode()), "application/json")}
+        response = await async_client.post("/api/v1/settings/restore?overwrite=true", files=files)
+
+        assert response.status_code == 200
+        result = response.json()
+        assert result["success"] is True
+
+        # Verify the printer was updated
+        response = await async_client.get(f"/api/v1/printers/{printer.id}")
+        assert response.status_code == 200
+        updated_printer = response.json()
+        assert updated_printer["external_camera_url"] == "/dev/video1"
+        assert updated_printer["external_camera_type"] == "usb"
+        assert updated_printer["external_camera_enabled"] is True

+ 42 - 0
backend/tests/unit/services/test_external_camera.py

@@ -215,3 +215,45 @@ class TestRtspUrlHandling:
         # Only the URL differs
         # Only the URL differs
         assert cmd_rtsp[:-1] == cmd_rtsps[:-1]
         assert cmd_rtsp[:-1] == cmd_rtsps[:-1]
         assert cmd_rtsp[-1] != cmd_rtsps[-1]
         assert cmd_rtsp[-1] != cmd_rtsps[-1]
+
+
+class TestUsbCameraHandling:
+    """Tests for USB camera support."""
+
+    def test_list_usb_cameras_returns_list(self):
+        """Verify list_usb_cameras returns a list (may be empty if no cameras)."""
+        from backend.app.services.external_camera import list_usb_cameras
+
+        result = list_usb_cameras()
+        assert isinstance(result, list)
+
+    def test_list_usb_cameras_dict_structure(self):
+        """Verify each camera entry has expected fields."""
+        from backend.app.services.external_camera import list_usb_cameras
+
+        result = list_usb_cameras()
+        for camera in result:
+            assert "device" in camera
+            assert "name" in camera
+            assert camera["device"].startswith("/dev/video")
+
+    @pytest.mark.asyncio
+    async def test_capture_frame_usb_type_accepted(self):
+        """Verify 'usb' camera type is accepted."""
+        from backend.app.services.external_camera import capture_frame
+
+        # Non-existent device should fail gracefully
+        result = await capture_frame("/dev/video999", "usb", timeout=1)
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_capture_frame_usb_invalid_device_path(self):
+        """Verify invalid USB device paths are rejected."""
+        from backend.app.services.external_camera import capture_frame
+
+        # Invalid device path (not /dev/video*)
+        result = await capture_frame("/dev/sda1", "usb", timeout=1)
+        assert result is None
+
+        result = await capture_frame("http://example.com", "usb", timeout=1)
+        assert result is None

+ 46 - 9
frontend/src/pages/SettingsPage.tsx

@@ -526,9 +526,45 @@ export function SettingsPage() {
     }
     }
   };
   };
 
 
-  const handleUpdatePrinterCamera = (printerId: number, updates: { url?: string; type?: string; enabled?: boolean }) => {
-    const data: Partial<{ external_camera_url: string | null; external_camera_type: string | null; external_camera_enabled: boolean }> = {};
-    if (updates.url !== undefined) data.external_camera_url = updates.url || null;
+  // Local state for camera URL inputs (to avoid saving on every keystroke)
+  const [localCameraUrls, setLocalCameraUrls] = useState<Record<number, string>>({});
+  const cameraUrlSaveTimeoutRef = useRef<Record<number, ReturnType<typeof setTimeout>>>({});
+
+  // Initialize local camera URLs from printer data
+  useEffect(() => {
+    if (printers) {
+      const urls: Record<number, string> = {};
+      printers.forEach(p => {
+        if (p.external_camera_url && localCameraUrls[p.id] === undefined) {
+          urls[p.id] = p.external_camera_url;
+        }
+      });
+      if (Object.keys(urls).length > 0) {
+        setLocalCameraUrls(prev => ({ ...prev, ...urls }));
+      }
+    }
+  }, [printers]);
+
+  const handleCameraUrlChange = (printerId: number, url: string) => {
+    // Update local state immediately for responsive UI
+    setLocalCameraUrls(prev => ({ ...prev, [printerId]: url }));
+
+    // Clear existing timeout for this printer
+    if (cameraUrlSaveTimeoutRef.current[printerId]) {
+      clearTimeout(cameraUrlSaveTimeoutRef.current[printerId]);
+    }
+
+    // Debounce the save (800ms delay)
+    cameraUrlSaveTimeoutRef.current[printerId] = setTimeout(() => {
+      updatePrinterMutation.mutate({
+        id: printerId,
+        data: { external_camera_url: url || null }
+      });
+    }, 800);
+  };
+
+  const handleUpdatePrinterCamera = (printerId: number, updates: { type?: string; enabled?: boolean }) => {
+    const data: Partial<{ external_camera_type: string | null; external_camera_enabled: boolean }> = {};
     if (updates.type !== undefined) data.external_camera_type = updates.type || null;
     if (updates.type !== undefined) data.external_camera_type = updates.type || null;
     if (updates.enabled !== undefined) data.external_camera_enabled = updates.enabled;
     if (updates.enabled !== undefined) data.external_camera_enabled = updates.enabled;
     updatePrinterMutation.mutate({ id: printerId, data });
     updatePrinterMutation.mutate({ id: printerId, data });
@@ -1002,7 +1038,7 @@ export function SettingsPage() {
               <div className="border-t border-bambu-dark-tertiary pt-4 mt-4">
               <div className="border-t border-bambu-dark-tertiary pt-4 mt-4">
                 <h3 className="text-sm font-medium text-white mb-2">External Cameras</h3>
                 <h3 className="text-sm font-medium text-white mb-2">External Cameras</h3>
                 <p className="text-xs text-bambu-gray mb-3">
                 <p className="text-xs text-bambu-gray mb-3">
-                  Configure network cameras to replace the built-in printer camera. Supports MJPEG streams, RTSP, and HTTP snapshots. When enabled, the external camera is used for live view and finish photos.
+                  Configure external cameras to replace the built-in printer camera. Supports MJPEG streams, RTSP, HTTP snapshots, and USB cameras (V4L2). When enabled, the external camera is used for live view and finish photos.
                 </p>
                 </p>
 
 
                 {printers && printers.length > 0 ? (
                 {printers && printers.length > 0 ? (
@@ -1026,9 +1062,9 @@ export function SettingsPage() {
                           <div className="space-y-2 mt-2">
                           <div className="space-y-2 mt-2">
                             <input
                             <input
                               type="text"
                               type="text"
-                              placeholder="Camera URL (rtsp://... or http://...)"
-                              value={printer.external_camera_url || ''}
-                              onChange={(e) => handleUpdatePrinterCamera(printer.id, { url: e.target.value })}
+                              placeholder={printer.external_camera_type === 'usb' ? 'Device path (/dev/video0)' : 'Camera URL (rtsp://... or http://...)'}
+                              value={localCameraUrls[printer.id] ?? printer.external_camera_url ?? ''}
+                              onChange={(e) => handleCameraUrlChange(printer.id, e.target.value)}
                               className="w-full px-3 py-2 bg-bambu-dark-secondary border border-bambu-dark-tertiary rounded text-white text-sm focus:border-bambu-green focus:outline-none"
                               className="w-full px-3 py-2 bg-bambu-dark-secondary border border-bambu-dark-tertiary rounded text-white text-sm focus:border-bambu-green focus:outline-none"
                             />
                             />
                             <div className="flex gap-2">
                             <div className="flex gap-2">
@@ -1040,12 +1076,13 @@ export function SettingsPage() {
                                 <option value="mjpeg">MJPEG Stream</option>
                                 <option value="mjpeg">MJPEG Stream</option>
                                 <option value="rtsp">RTSP Stream</option>
                                 <option value="rtsp">RTSP Stream</option>
                                 <option value="snapshot">HTTP Snapshot</option>
                                 <option value="snapshot">HTTP Snapshot</option>
+                                <option value="usb">USB Camera (V4L2)</option>
                               </select>
                               </select>
                               <Button
                               <Button
                                 size="sm"
                                 size="sm"
                                 variant="secondary"
                                 variant="secondary"
-                                onClick={() => handleTestExternalCamera(printer.id, printer.external_camera_url || '', printer.external_camera_type || 'mjpeg')}
-                                disabled={extCameraTestLoading[printer.id] || !printer.external_camera_url}
+                                onClick={() => handleTestExternalCamera(printer.id, localCameraUrls[printer.id] ?? printer.external_camera_url ?? '', printer.external_camera_type || 'mjpeg')}
+                                disabled={extCameraTestLoading[printer.id] || !(localCameraUrls[printer.id] ?? printer.external_camera_url)}
                               >
                               >
                                 {extCameraTestLoading[printer.id] ? (
                                 {extCameraTestLoading[printer.id] ? (
                                   <Loader2 className="w-4 h-4 animate-spin" />
                                   <Loader2 className="w-4 h-4 animate-spin" />

File diff suppressed because it is too large
+ 0 - 0
static/assets/index-BraSLW7a.js


+ 1 - 1
static/index.html

@@ -23,7 +23,7 @@
 
 
     <!-- Splash screens for iOS -->
     <!-- Splash screens for iOS -->
     <link rel="apple-touch-startup-image" href="/img/android-chrome-512x512.png" />
     <link rel="apple-touch-startup-image" href="/img/android-chrome-512x512.png" />
-    <script type="module" crossorigin src="/assets/index-CsmyN56S.js"></script>
+    <script type="module" crossorigin src="/assets/index-BraSLW7a.js"></script>
     <link rel="stylesheet" crossorigin href="/assets/index-BI0S_RVt.css">
     <link rel="stylesheet" crossorigin href="/assets/index-BI0S_RVt.css">
   </head>
   </head>
   <body>
   <body>

Some files were not shown because too many files changed in this diff