fix: pipelining

Author: robgee86

src/arduino/app_peripherals/camera/base_camera.py

@@ -23,20 +23,20 @@ class BaseCamera(ABC):
     providing a unified API regardless of the underlying camera protocol or type.
     """
 
-    def __init__(self, resolution: Optional[Tuple[int, int]] = None, fps: int = 10, 
-                 transformer: Optional[Callable[[np.ndarray], np.ndarray]] = None, **kwargs):
+    def __init__(self, resolution: Optional[Tuple[int, int]] = (640, 480), fps: int = 10, 
+                 adjuster: Optional[Callable[[np.ndarray], np.ndarray]] = None, **kwargs):
         """
         Initialize the camera base.
 
         Args:
             resolution (tuple, optional): Resolution as (width, height). None uses default resolution.
             fps (int): Frames per second for the camera.
-            transformer (callable, optional): Function to transform frames that takes a numpy array and returns a numpy array. Default: None
+            adjuster (callable, optional): Function pipeline to adjust frames that takes a numpy array and returns a numpy array. Default: None
             **kwargs: Additional camera-specific parameters.
         """
         self.resolution = resolution
         self.fps = fps
-        self.transformer = transformer
+        self.adjuster = adjuster
         self._is_started = False
         self._cap_lock = threading.Lock()
         self._last_capture_time = time.monotonic()
@@ -100,13 +100,11 @@ def _extract_frame(self) -> Optional[np.ndarray]:
 
             self._last_capture_time = time.monotonic()
 
-            if self.transformer is None:
-                return frame
-            
-            try:
-                frame = frame | self.transformer
-            except Exception as e:
-                raise CameraTransformError(f"Frame transformation failed ({self.transformer}): {e}")
+            if self.adjuster is not None:
+                try:
+                    frame = self.adjuster(frame)
+                except Exception as e:
+                    raise CameraTransformError(f"Frame transformation failed ({self.adjuster}): {e}")
 
             return frame
 

src/arduino/app_peripherals/camera/camera.py

@@ -32,7 +32,7 @@ def __new__(cls, source: Union[str, int] = 0, **kwargs) -> BaseCamera:
                     resolution (tuple, optional): Frame resolution as (width, height). 
                         Default: None (auto)
                     fps (int, optional): Target frames per second. Default: 10
-                    transformer (callable, optional): Function to transform frames that takes a
+                    adjuster (callable, optional): Function pipeline to adjust frames that takes a
                         numpy array and returns a numpy array. Default: None
                 V4L Camera Parameters:
                     device_index (int, optional): V4L device index override

src/arduino/app_peripherals/camera/examples/websocket_camera_proxy.py

@@ -24,7 +24,10 @@
 
 import os
 
+import numpy as np
+
 from arduino.app_peripherals.camera import Camera
+from arduino.app_utils.image.image_editor import ImageEditor
 
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', '..', '..'))
 
@@ -53,48 +56,37 @@ async def connect_output_tcp(output_host: str, output_port: int):
     """Connect to the output TCP server."""
     global output_writer, output_reader
 
-    logger.info(f"Connecting to TCP server at {output_host}:{output_port}...")
+    logger.info(f"Connecting to output server at {output_host}:{output_port}...")
 
     try:
         output_reader, output_writer = await asyncio.open_connection(
             output_host, output_port
         )
-        logger.info("TCP connection established successfully")
-        
-        return True
+        logger.info("Connected successfully to output server")
 
     except Exception as e:
-        logger.error(f"Failed to connect to TCP server: {e}")
-        return False
+        raise Exception(f"Failed to connect to output server: {e}")
 
 
-async def forward_frame(frame, quality: int):
+async def forward_frame(frame: np.ndarray):
     """Forward a frame to the output TCP server as raw JPEG."""
     global output_writer
 
     if not output_writer or output_writer.is_closing():
         return
 
     try:
-        # Frame is already a PIL.Image.Image in JPEG format
-        # Convert PIL image to bytes
-        import io
-        img_bytes = io.BytesIO()
-        frame.save(img_bytes, format='JPEG', quality=quality)
-        frame_data = img_bytes.getvalue()
-        
-        # Send raw JPEG binary data
-        output_writer.write(frame_data)
+        output_writer.write(frame.tobytes())
         await output_writer.drain()
 
     except ConnectionResetError:
-        logger.warning("TCP connection reset while forwarding frame")
+        logger.warning("Output connection reset while forwarding frame")
         output_writer = None
     except Exception as e:
         logger.error(f"Error forwarding frame: {e}")
 
 
-async def camera_loop(fps: int, quality: int):
+async def camera_loop(fps: int):
     """Main camera capture and forwarding loop."""
     global running, camera
 
@@ -111,21 +103,26 @@ async def camera_loop(fps: int, quality: int):
         try:
             # Read frame from WebSocketCamera
             frame = camera.capture()
-            
-            if frame is not None:
-                # Rate limiting
-                current_time = time.time()
-                time_since_last = current_time - last_frame_time
-                if time_since_last < frame_interval:
-                    await asyncio.sleep(frame_interval - time_since_last)
-                
-                last_frame_time = time.time()
-                
-                # Forward frame if output TCP connection is available
-                await forward_frame(frame, quality)
-            else:
+            # frame = ImageEditor.compress_to_jpeg(frame, 80.1)
+            if frame is None:
                 # No frame available, small delay to avoid busy waiting
                 await asyncio.sleep(0.01)
+                continue
+            
+            # Rate limiting
+            current_time = time.time()
+            time_since_last = current_time - last_frame_time
+            if time_since_last < frame_interval:
+                await asyncio.sleep(frame_interval - time_since_last)
+            
+            last_frame_time = time.time()
+            
+            if output_writer is None or output_writer.is_closing():
+                # Output connection is not available, give room to the other tasks
+                await asyncio.sleep(0.01)
+            else:
+                # Forward frame if output connection is available
+                await forward_frame(frame)
 
         except Exception as e:
             logger.error(f"Error in camera loop: {e}")
@@ -138,18 +135,16 @@ async def maintain_output_connection(output_host: str, output_port: int, reconne
 
     while running:
         try:
-            # Establish connection
-            if await connect_output_tcp(output_host, output_port):
-                logger.info("TCP connection established, maintaining...")
+            await connect_output_tcp(output_host, output_port)
+            
+            # Keep monitoring
+            while running and output_writer and not output_writer.is_closing():
+                await asyncio.sleep(1.0)
 
-                # Keep connection alive
-                while running and output_writer and not output_writer.is_closing():
-                    await asyncio.sleep(1.0)
-                    
-                logger.info("TCP connection lost")
+            logger.info("Lost connection to output server")
 
         except Exception as e:
-            logger.error(f"TCP connection error: {e}")
+            logger.error(e)
         finally:
             # Clean up connection
             if output_writer:
@@ -163,7 +158,7 @@ async def maintain_output_connection(output_host: str, output_port: int, reconne
 
         # Wait before reconnecting
         if running:
-            logger.info(f"Reconnecting to TCP server in {reconnect_delay} seconds...")
+            logger.info(f"Reconnecting to output server in {reconnect_delay} seconds...")
             await asyncio.sleep(reconnect_delay)
 
 
@@ -172,10 +167,12 @@ async def main():
     global running, camera
 
     parser = argparse.ArgumentParser(description="WebSocket Camera Proxy")
+    parser.add_argument("--input-host", default="localhost",
+                       help="WebSocketCamera input host (default: localhost)")
     parser.add_argument("--input-port", type=int, default=8080,
                        help="WebSocketCamera input port (default: 8080)")
-    parser.add_argument("--output-host", default="0.0.0.0",
-                       help="Output TCP server host (default: 0.0.0.0)")
+    parser.add_argument("--output-host", default="localhost",
+                       help="Output TCP server host (default: localhost)")
     parser.add_argument("--output-port", type=int, default=5000,
                        help="Output TCP server port (default: 5000)")
     parser.add_argument("--fps", type=int, default=30,
@@ -204,11 +201,12 @@ async def main():
     logger.info(f"Target FPS: {args.fps}")
 
     from arduino.app_utils.image.image_editor import compressed_to_jpeg
-    camera = Camera("ws://0.0.0.0:5000", transformer=compressed_to_jpeg(80))
+    camera = Camera(f"ws://{args.input_host}:{args.input_port}", adjuster=compressed_to_jpeg(80))
+    # camera = Camera(f"ws://{args.input_host}:{args.input_port}")
 
     try:
         # Start camera input and output connection tasks
-        camera_task = asyncio.create_task(camera_loop(args.fps, args.quality))
+        camera_task = asyncio.create_task(camera_loop(args.fps))
         connection_task = asyncio.create_task(maintain_output_connection(args.output_host, args.output_port, reconnect_delay))
 
         # Run both tasks concurrently

src/arduino/app_peripherals/camera/examples/websocket_client_streamer.py

@@ -13,6 +13,9 @@
 import sys
 import time
 
+from arduino.app_peripherals.camera import Camera
+from arduino.app_utils.image.image_editor import ImageEditor
+
 logging.basicConfig(
     level=logging.INFO,
     format='%(asctime)s - %(levelname)s - %(message)s'
@@ -82,44 +85,33 @@ async def stop(self):
                 logger.warning(f"Error closing WebSocket: {e}")
 
         if self.camera:
-            self.camera.release()
-            logger.info("Camera released")
+            self.camera.stop()
+            logger.info("Camera stopped")
 
         logger.info("Webcam streamer stopped")
 
     async def _camera_loop(self):
         """Main camera capture loop."""
         logger.info(f"Opening camera {self.camera_id}...")
-        self.camera = cv2.VideoCapture(self.camera_id)
+        self.camera = Camera(self.camera_id, resolution=(FRAME_WIDTH, FRAME_HEIGHT), fps=self.fps)
+        self.camera.start()
 
-        if not self.camera.isOpened():
+        if not self.camera.is_started():
             logger.error(f"Failed to open camera {self.camera_id}")
             return
 
-        self.camera.set(cv2.CAP_PROP_FPS, self.fps)
-        self.camera.set(cv2.CAP_PROP_FRAME_WIDTH, FRAME_WIDTH)
-        self.camera.set(cv2.CAP_PROP_FRAME_HEIGHT, FRAME_HEIGHT)
-
-        # Verify the resolution was set correctly
-        actual_width = int(self.camera.get(cv2.CAP_PROP_FRAME_WIDTH))
-        actual_height = int(self.camera.get(cv2.CAP_PROP_FRAME_HEIGHT))
-        actual_fps = self.camera.get(cv2.CAP_PROP_FPS)
-
-        if actual_width != FRAME_WIDTH or actual_height != FRAME_HEIGHT:
-            logger.warning(f"Camera resolution mismatch! Requested {FRAME_WIDTH}x{FRAME_HEIGHT}, got {actual_width}x{actual_height}")
-        
         logger.info("Camera opened successfully")
 
         last_frame_time = time.time()
 
         while self.running:
             try:
-                ret, frame = self.camera.read()
-                if not ret:
+                frame = self.camera.capture()
+                if frame is None:
                     logger.warning("Failed to capture frame")
                     await asyncio.sleep(0.1)
                     continue
-                
+
                 # Rate limiting to enforce frame rate
                 current_time = time.time()
                 time_since_last = current_time - last_frame_time
@@ -135,6 +127,8 @@ async def _camera_loop(self):
                         logger.warning("WebSocket connection lost during frame send")
                         self.websocket = None
 
+                await asyncio.sleep(0.001)
+                
             except Exception as e:
                 logger.error(f"Error in camera loop: {e}")
                 await asyncio.sleep(1.0)
@@ -191,9 +185,6 @@ async def _handle_websocket_messages(self):
                         logger.info(f"Server goodbye: {data.get('message', 'Disconnecting')}")
                         break
 
-                    elif data.get("status") == "dropping_frames":
-                        logger.warning(f"Server warning: {data.get('message', 'Dropping frames!')}")
-                    
                     elif data.get("error"):
                         logger.warning(f"Server error: {data.get('message', 'Unknown error')}")
                         if data.get("code") == 1000:  # Server busy
@@ -216,36 +207,18 @@ async def _send_frame(self, frame):
         try:
             if self.server_frame_format == "binary":
                 # Encode frame as JPEG and send binary data
-                encode_params = [cv2.IMWRITE_JPEG_QUALITY, self.quality]
-                success, encoded_frame = cv2.imencode('.jpg', frame, encode_params)
-                
-                if not success:
-                    logger.warning("Failed to encode frame")
-                    return
-                
+                encoded_frame = ImageEditor.compress_to_jpeg(frame)
                 await self.websocket.send(encoded_frame.tobytes())
 
             elif self.server_frame_format == "base64":
                 # Encode frame as JPEG and send base64 data
-                encode_params = [cv2.IMWRITE_JPEG_QUALITY, self.quality]
-                success, encoded_frame = cv2.imencode('.jpg', frame, encode_params)
-                
-                if not success:
-                    logger.warning("Failed to encode frame")
-                    return
-                
+                encoded_frame = ImageEditor.compress_to_jpeg(frame)
                 frame_b64 = base64.b64encode(encoded_frame.tobytes()).decode('utf-8')
                 await self.websocket.send(frame_b64)
 
             elif self.server_frame_format == "json":
                 # Encode frame as JPEG, base64 encode and wrap in JSON
-                encode_params = [cv2.IMWRITE_JPEG_QUALITY, self.quality]
-                success, encoded_frame = cv2.imencode('.jpg', frame, encode_params)
-                
-                if not success:
-                    logger.warning("Failed to encode frame")
-                    return
-                
+                encoded_frame = ImageEditor.compress_to_jpeg(frame)
                 frame_b64 = base64.b64encode(encoded_frame.tobytes()).decode('utf-8')
                 message = json.dumps({"image": frame_b64})
                 await self.websocket.send(message)
@@ -269,7 +242,7 @@ def signal_handler(signum, frame):
 async def main():
     """Main function."""
     parser = argparse.ArgumentParser(description="WebSocket Camera Client Streamer")
-    parser.add_argument("--host", default="127.0.0.1", help="WebSocket server host (default: 127.0.0.1)")
+    parser.add_argument("--host", default="localhost", help="WebSocket server host (default: localhost)")
     parser.add_argument("--port", type=int, default=8080, help="WebSocket server port (default: 8080)")
     parser.add_argument("--camera", type=int, default=0, help="Camera device ID (default: 0)")
     parser.add_argument("--fps", type=int, default=30, help="Target FPS (default: 30)")

src/arduino/app_peripherals/camera/v4l_camera.py

@@ -129,6 +129,17 @@ def _open_camera(self) -> None:
                     f"Camera {self.camera_id} resolution set to {actual_width}x{actual_height} "
                     f"instead of requested {self.resolution[0]}x{self.resolution[1]}"
                 )
+                self.resolution = (actual_width, actual_height)
+        
+        if self.fps:
+            self._cap.set(cv2.CAP_PROP_FPS, self.fps)
+
+            actual_fps = int(self._cap.get(cv2.CAP_PROP_FPS))
+            if actual_fps != self.fps:
+                logger.warning(
+                    f"Camera {self.camera_id} FPS set to {actual_fps} instead of requested {self.fps}"
+                )
+                self.fps = actual_fps
 
         logger.info(f"Opened V4L camera {self.camera_id}")