feat: add recording to camera

Author: robgee86

src/arduino/app_peripherals/camera/base_camera.py

@@ -42,6 +42,8 @@ def __init__(
             auto_reconnect (bool, optional): Enable automatic reconnection on failure. Default: True.
         """
         self.resolution = resolution
+        if fps <= 0:
+            raise ValueError("FPS must be a positive integer")
         self.fps = fps
         self.adjustments = adjustments
         self.logger = logger  # This will be overridden by subclasses if needed
@@ -176,6 +178,9 @@ def stream(self):
 
         Yields:
             np.ndarray: Video frames as numpy arrays.
+
+        Raises:
+            CameraReadError: If the camera is not started.
         """
         if not self.is_started():
             raise CameraReadError(f"Attempted to acquire stream from {self.name} before starting it.")
@@ -184,9 +189,107 @@ def stream(self):
             frame = self.capture()
             if frame is not None:
                 yield frame
-            else:
-                # Avoid busy-waiting if no frame available
-                time.sleep(0.001)
+
+    def record(self, duration) -> np.ndarray:
+        """
+        Record video for a specified duration and return it as a numpy array of raw frames.
+
+        Args:
+            duration (float): Recording duration in seconds.
+
+        Returns:
+            np.ndarray: numpy array of raw frames.
+
+        Raises:
+            CameraReadError: If camera is not started or any read error occurs.
+            ValueError: If duration is not positive.
+            MemoryError: If memory allocation for the full recording fails.
+        """
+        if duration <= 0:
+            raise ValueError("Duration must be positive")
+
+        total_frames = int(self.fps * duration)
+
+        # Get shape and dtype from first frame
+        first_frame = self.capture()
+        if first_frame is None:
+            raise CameraOpenError("Failed to inspect the video stream for metadata.")
+        frame_shape = first_frame.shape
+        frame_dtype = first_frame.dtype
+
+        try:
+            frames = np.zeros((total_frames, *frame_shape), dtype=frame_dtype)
+        except Exception as e:
+            raise MemoryError(f"Could not allocate memory for {total_frames} frames: {e}")
+
+        count = 1
+        frames[0] = first_frame
+        while count < total_frames:
+            frame = self.capture()
+            if frame is not None:
+                frames[count] = frame
+                count += 1
+
+        if not frames.any():
+            raise CameraReadError("No frames captured during recording.")
+
+        return frames[:count]
+
+    def record_avi(self, duration) -> np.ndarray:
+        """
+        Record video for a specified duration and return as MJPEG in AVI container.
+
+        Args:
+            duration (float): Recording duration in seconds.
+
+        Returns:
+            np.ndarray: AVI file containing MJPEG video.
+
+        Raises:
+            CameraReadError: If camera is not started or any read error occurs.
+            ValueError: If duration is not positive.
+            MemoryError: If memory allocation for the full recording fails.
+        """
+        if duration <= 0:
+            raise ValueError("Duration must be positive")
+
+        import os
+        import tempfile
+        import cv2
+
+        total_frames = int(self.fps * duration)
+
+        # Get width and height from first frame
+        first_frame = self.capture()
+        if first_frame is None:
+            raise CameraOpenError("Failed to inspect the video stream for metadata.")
+        height, width = first_frame.shape[:2]
+
+        # Write MJPEG AVI to a temp file
+        with tempfile.NamedTemporaryFile(suffix=".avi", delete=False) as tmp:
+            filename = tmp.name
+
+        avi_data = np.empty(0, dtype=np.uint8)
+        try:
+            fourcc = cv2.VideoWriter.fourcc(*"MJPG")
+            out = cv2.VideoWriter(filename, fourcc, self.fps, (width, height))
+            frame = first_frame
+            for i in range(total_frames):
+                if frame is not None:
+                    if frame.dtype != np.uint8:
+                        frame = _to_uint8(frame)
+                    out.write(frame)
+                
+                if i < total_frames - 1:
+                    frame = self.capture()
+            
+            out.release()
+            with open(filename, "rb") as f:
+                avi_data = f.read()
+        finally:
+            os.remove(filename)
+
+        return np.frombuffer(avi_data, dtype=np.uint8)
 
     def is_started(self) -> bool:
         """Check if the camera has been started."""
@@ -300,3 +403,16 @@ def __enter__(self):
     def __exit__(self, exc_type, exc_val, exc_tb):
         """Context manager exit."""
         self.stop()
+
+
+def _to_uint8(frame) -> np.ndarray:
+    """Normalize and convert to uint8."""
+    if np.issubdtype(frame.dtype, np.floating):
+        # We adopt the OpenCV convention: float images are in [0, 1]
+        frame = np.clip(frame * 255, 0, 255)
+    
+    elif np.issubdtype(frame.dtype, np.integer) and frame.dtype != np.uint8:
+        info = np.iinfo(frame.dtype)
+        frame = (frame.astype(np.float32) - info.min) / (info.max - info.min) * 255
+    
+    return frame.astype(np.uint8)

src/arduino/app_peripherals/camera/examples/3_capture_video.py

@@ -9,13 +9,19 @@
 from arduino.app_peripherals.camera import Camera
 
 
-# Capture a video for 5 seconds at 15 FPS
 camera = Camera(fps=15)
 camera.start()
 
+# You can capture a video by capturing frames in a loop
 start_time = time.time()
 while time.time() - start_time < 5:
     image: np.ndarray = camera.capture()
     # You can process the image here if needed, e.g save it
 
+# Or you can obtain the same in a single sentence
+recording: np.ndarray = camera.record(duration=5)
+
+# Or you can ask for an AVI recording
+recording: np.ndarray = camera.record_avi(duration=5)
+
 camera.stop()

src/arduino/app_peripherals/camera/websocket_camera.py

@@ -58,7 +58,6 @@ def __init__(
         Initialize WebSocket camera server.
 
         Args:
-            host (str): Host address to bind the server to (default: "0.0.0.0")
             port (int): Port to bind the server to (default: 8080)
             timeout (int): Connection timeout in seconds (default: 10)
             frame_format (str): Expected frame format from clients ("binary", "base64", "json") (default: "binary")

tests/arduino/app_peripherals/camera/test_base_camera.py

@@ -5,6 +5,8 @@
 import pytest
 import time
 import numpy as np
+import tempfile
+import cv2
 
 from arduino.app_peripherals.camera import BaseCamera, CameraTransformError
 from arduino.app_peripherals.usb_camera import CameraReadError
@@ -78,6 +80,12 @@ def test_base_camera_init_custom():
     assert not camera.is_started()
 
 
+def test_base_camera_init_invalid():
+    """Test BaseCamera initialization with invalid parameters."""
+    with pytest.raises(ValueError):
+        MockedCamera(fps=0)
+
+
 def test_is_started_state_transitions():
     """Test is_started return value through different state transitions."""
     camera = MockedCamera()
@@ -344,20 +352,6 @@ def test_context_manager_with_exception():
     assert camera.close_call_count == 1
 
 
-def test_capture_no_throttling():
-    """Test capture behavior with fps=0 (no throttling)."""
-    camera = MockedCamera(fps=0)
-    camera.start()
-
-    start_time = time.monotonic()
-    frame = camera.capture()
-    elapsed = time.monotonic() - start_time
-
-    assert elapsed < 0.01
-    assert frame is not None
-    assert camera.read_call_count == 1
-
-
 def test_capture_multiple_adjustments():
     """Test that adjustment pipelines are applied correctly."""
 
@@ -424,3 +418,94 @@ def event_callback(event, data):
     assert "paused" in events[2][0]
     assert "streaming" in events[3][0]
     assert "disconnected" in events[4][0]
+
+def test_record_zero_duration():
+    camera = MockedCamera()
+    camera.start()
+    with pytest.raises(ValueError):
+        camera.record(0)
+    camera.stop()
+
+def test_record():
+    camera = MockedCamera(fps=5)
+    camera.frame = np.ones((480, 640, 3), dtype=np.uint8)
+    camera.start()
+    
+    duration = 1.0
+    expected_frames = int(camera.fps * duration)
+    frames = camera.record(duration)
+    
+    assert isinstance(frames, np.ndarray)
+    assert frames.shape[0] == expected_frames
+    assert frames.shape[1:] == camera.frame.shape
+    assert frames.dtype == camera.frame.dtype
+    assert np.all(frames == camera.frame)
+    
+    camera.stop()
+
+def test_record_avi():
+    camera = MockedCamera(fps=5)
+    camera.frame = np.ones((480, 640, 3), dtype=np.uint8)
+    camera.start()
+
+    duration = 1.0
+    expected_frames = int(camera.fps * duration)
+    avi_bytes = camera.record_avi(duration)
+
+    assert isinstance(avi_bytes, np.ndarray)
+    assert avi_bytes.dtype == np.uint8
+    assert avi_bytes.size > 0
+    
+    with tempfile.NamedTemporaryFile(suffix='.avi') as tmp:
+        tmp.write(avi_bytes.tobytes())
+        tmp.flush()
+
+        read_count = 0
+        cap = cv2.VideoCapture(tmp.name)
+        while True:
+            ret, frame = cap.read()
+            if not ret:
+                break
+            assert frame is not None
+            assert frame.dtype == np.uint8
+            read_count += 1
+        
+        cap.release()
+    
+    assert read_count == expected_frames
+    
+    camera.stop()
+
+def test_record_avi_uint8_conversion():
+    camera = MockedCamera(fps=5)
+    # Use float32 frame, should be converted to uint8 in AVI
+    camera.frame = np.ones((10, 10, 3), dtype=np.float64)
+    camera.start()
+
+    duration = 1.0
+    expected_frames = int(camera.fps * duration)
+    avi_bytes = camera.record_avi(duration)
+    
+    assert isinstance(avi_bytes, np.ndarray)
+    assert avi_bytes.dtype == np.uint8
+    assert avi_bytes.size > 0
+    
+    with tempfile.NamedTemporaryFile(suffix='.avi') as tmp:
+        tmp.write(avi_bytes.tobytes())
+        tmp.flush()
+        
+        read_count = 0
+        cap = cv2.VideoCapture(tmp.name)
+        while True:
+            ret, frame = cap.read()
+            if not ret:
+                break
+            assert frame is not None
+            assert frame.dtype == np.uint8
+            read_count += 1
+        
+        cap.release()
+
+    assert read_count == expected_frames
+    
+    camera.stop()