refactor: clearer APIs and doc

Author: robgee86

src/arduino/app_peripherals/camera/__init__.py

@@ -3,10 +3,16 @@
 # SPDX-License-Identifier: MPL-2.0
 
 from .camera import Camera
+from .v4l_camera import V4LCamera
+from .ip_camera import IPCamera
+from .websocket_camera import WebSocketCamera
 from .errors import *
 
 __all__ = [
     "Camera",
+    "V4LCamera",
+    "IPCamera",
+    "WebSocketCamera",
     "CameraError",
     "CameraReadError",
     "CameraOpenError",

src/arduino/app_peripherals/camera/base_camera.py

@@ -23,52 +23,57 @@ class BaseCamera(ABC):
     providing a unified API regardless of the underlying camera protocol or type.
     """
 
-    def __init__(self, resolution: Optional[Tuple[int, int]] = (640, 480), fps: int = 10, 
-                 adjustments: Optional[Callable[[np.ndarray], np.ndarray]] = None, **kwargs):
+    def __init__(
+        self,
+        resolution: Optional[Tuple[int, int]] = (640, 480),
+        fps: int = 10, 
+        adjustments: Optional[Callable[[np.ndarray], np.ndarray]] = None,
+    ):
         """
         Initialize the camera base.
 
         Args:
             resolution (tuple, optional): Resolution as (width, height). None uses default resolution.
-            fps (int): Frames per second for the camera.
+            fps (int): Frames per second to capture from the camera.
             adjustments (callable, optional): Function or 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.adjustments = adjustments
+        self.logger = logger  # This will be overridden by subclasses if needed
+
+        self._camera_lock = threading.Lock()
         self._is_started = False
-        self._cap_lock = threading.Lock()
         self._last_capture_time = time.monotonic()
-        self.desired_interval = 1.0 / fps if fps > 0 else 0
+        self._desired_interval = 1.0 / fps if fps > 0 else 0
 
     def start(self) -> None:
         """Start the camera capture."""
-        with self._cap_lock:
+        with self._camera_lock:
             if self._is_started:
                 return
 
             try:
                 self._open_camera()
                 self._is_started = True
                 self._last_capture_time = time.monotonic()
-                logger.info(f"Successfully started {self.__class__.__name__}")
+                self.logger.info(f"Successfully started {self.__class__.__name__}")
             except Exception as e:
                 raise CameraOpenError(f"Failed to start camera: {e}")
 
     def stop(self) -> None:
         """Stop the camera and release resources."""
-        with self._cap_lock:
+        with self._camera_lock:
             if not self._is_started:
                 return
 
             try:
                 self._close_camera()
                 self._is_started = False
-                logger.info(f"Stopped {self.__class__.__name__}")
+                self.logger.info(f"Stopped {self.__class__.__name__}")
             except Exception as e:
-                logger.warning(f"Error stopping camera: {e}")
+                self.logger.warning(f"Error stopping camera: {e}")
 
     def capture(self) -> Optional[np.ndarray]:
         """
@@ -85,13 +90,13 @@ def capture(self) -> Optional[np.ndarray]:
     def _extract_frame(self) -> Optional[np.ndarray]:
         """Extract a frame with FPS throttling and post-processing."""
         # FPS throttling
-        if self.desired_interval > 0:
+        if self._desired_interval > 0:
             current_time = time.monotonic()
             elapsed = current_time - self._last_capture_time
-            if elapsed < self.desired_interval:
-                time.sleep(self.desired_interval - elapsed)
+            if elapsed < self._desired_interval:
+                time.sleep(self._desired_interval - elapsed)
 
-        with self._cap_lock:
+        with self._camera_lock:
             if not self._is_started:
                 return None
 

src/arduino/app_peripherals/camera/camera.py

@@ -15,6 +15,14 @@ class Camera:
     
     This class serves as both a factory and a wrapper, automatically creating
     the appropriate camera implementation based on the provided configuration.
+
+    Supports:
+        - V4L Cameras (local cameras connected to the system), the default
+        - IP Cameras (network-based cameras via RTSP, HLS)
+        - WebSocket Cameras (input streams via WebSocket client)
+
+    Note: constructor arguments (except source) must be provided in keyword
+    format to forward them correctly to the specific camera implementations.
     """
 
     def __new__(cls, source: Union[str, int] = 0, **kwargs) -> BaseCamera:
@@ -23,22 +31,20 @@ def __new__(cls, source: Union[str, int] = 0, **kwargs) -> BaseCamera:
         Args:
             source (Union[str, int]): Camera source identifier. Supports:
                 - int: V4L camera index (e.g., 0, 1)
-                - str: Camera index as string (e.g., "0", "1") for V4L
-                - str: Device path (e.g., "/dev/video0") for V4L
+                - str: V4L camera index (e.g., "0", "1") or device path (e.g., "/dev/video0")
                 - str: URL for IP cameras (e.g., "rtsp://...", "http://...")
-                - str: WebSocket URL (e.g., "ws://0.0.0.0:8080")
+                - str: WebSocket URL for input streams (e.g., "ws://0.0.0.0:8080")
             **kwargs: Camera-specific configuration parameters grouped by type:
                 Common Parameters:
                     resolution (tuple, optional): Frame resolution as (width, height). 
-                        Default: None (auto)
+                        Default: (640, 480)
                     fps (int, optional): Target frames per second. Default: 10
                     adjustments (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
-                    capture_format (str, optional): Video capture format (e.g., 'MJPG', 'YUYV')
-                    buffer_size (int, optional): Number of frames to buffer
+                    device (int, optional): V4L device index override. Default: 0.
                 IP Camera Parameters:
+                    url (str): Camera stream URL
                     username (str, optional): Authentication username
                     password (str, optional): Authentication password
                     timeout (float, optional): Connection timeout in seconds. Default: 10.0
@@ -54,9 +60,10 @@ def __new__(cls, source: Union[str, int] = 0, **kwargs) -> BaseCamera:
             
         Raises:
             CameraConfigError: If source type is not supported or parameters are invalid
+            CameraOpenError: If the camera cannot be opened
             
         Examples:
-            V4L/USB Camera:
+            V4L Camera:
             
             ```python
             camera = Camera(0, resolution=(640, 480), fps=30)
@@ -67,17 +74,16 @@ def __new__(cls, source: Union[str, int] = 0, **kwargs) -> BaseCamera:
             
             ```python
             camera = Camera("rtsp://192.168.1.100:554/stream", username="admin", password="secret", timeout=15.0)
-            camera = Camera("http://192.168.1.100:8080/video", retry_attempts=5)
+            camera = Camera("http://192.168.1.100:8080/video.mp4")
             ```
             
             WebSocket Camera:
             
             ```python   
-            camera = Camera("ws://0.0.0.0:8080", frame_format="json", max_queue_size=20)
-            camera = Camera("ws://192.168.1.100:8080", ping_interval=30)
+            camera = Camera("ws://0.0.0.0:8080", frame_format="json")
+            camera = Camera("ws://192.168.1.100:8080", timeout=5)
             ```
         """
-        # Dynamic imports to avoid circular dependencies
         if isinstance(source, int) or (isinstance(source, str) and source.isdigit()):
             # V4L Camera
             from .v4l_camera import V4LCamera

src/arduino/app_peripherals/camera/ip_camera.py

@@ -5,13 +5,13 @@
 import cv2
 import numpy as np
 import requests
-from typing import Optional
+from typing import Callable, Optional, Tuple
 from urllib.parse import urlparse
 
 from arduino.app_utils import Logger
 
 from .camera import BaseCamera
-from .errors import CameraOpenError
+from .errors import CameraConfigError, CameraOpenError
 
 logger = Logger("IPCamera")
 
@@ -24,44 +24,58 @@ class IPCamera(BaseCamera):
     Can handle authentication and various streaming protocols.
     """
 
-    def __init__(self, url: str, username: Optional[str] = None, 
-                 password: Optional[str] = None, timeout: int = 10, **kwargs):
+    def __init__(
+        self,
+        url: str,
+        username: Optional[str] = None, 
+        password: Optional[str] = None,
+        timeout: int = 10,
+        resolution: Optional[Tuple[int, int]] = (640, 480),
+        fps: int = 10, 
+        adjustments: Optional[Callable[[np.ndarray], np.ndarray]] = None,
+    ):
         """
         Initialize IP camera.
 
         Args:
-            url: Camera stream URL (rtsp://, http://, https://)
+            url: Camera stream URL (i.e. rtsp://..., http://..., https://...)
             username: Optional authentication username
             password: Optional authentication password  
             timeout: Connection timeout in seconds
-            **kwargs: Additional camera parameters propagated to BaseCamera
+            resolution (tuple, optional): Resolution as (width, height). None uses default resolution.
+            fps (int): Frames per second to capture from the camera.
+            adjustments (callable, optional): Function or function pipeline to adjust frames that takes
+                a numpy array and returns a numpy array. Default: None
         """
-        super().__init__(**kwargs)
+        super().__init__(resolution, fps, adjustments)
         self.url = url
         self.username = username
         self.password = password
         self.timeout = timeout
+        self.logger = logger
+
         self._cap = None
+        
         self._validate_url()
 
     def _validate_url(self) -> None:
         """Validate the camera URL format."""
         try:
             parsed = urlparse(self.url)
             if parsed.scheme not in ['http', 'https', 'rtsp']:
-                raise CameraOpenError(f"Unsupported URL scheme: {parsed.scheme}")
+                raise CameraConfigError(f"Unsupported URL scheme: {parsed.scheme}")
         except Exception as e:
-            raise CameraOpenError(f"Invalid URL format: {e}")
+            raise CameraConfigError(f"Invalid URL format: {e}")
 
     def _open_camera(self) -> None:
         """Open the IP camera connection."""
-        auth_url = self._build_authenticated_url()
+        url = self._build_url()
 
         # Test connectivity first for HTTP streams
         if self.url.startswith(('http://', 'https://')):
             self._test_http_connectivity()
 
-        self._cap = cv2.VideoCapture(auth_url)
+        self._cap = cv2.VideoCapture(url)
         self._cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)  # Reduce buffer to get latest frames
         if not self._cap.isOpened():
             raise CameraOpenError(f"Failed to open IP camera: {self.url}")
@@ -75,17 +89,15 @@ def _open_camera(self) -> None:
 
         logger.info(f"Opened IP camera: {self.url}")
 
-    def _build_authenticated_url(self) -> str:
+    def _build_url(self) -> str:
         """Build URL with authentication if credentials provided."""
+        # If no username or password provided as parameters, return original URL
         if not self.username or not self.password:
             return self.url
 
         parsed = urlparse(self.url)
-        if parsed.username and parsed.password:
-            # URL already has credentials
-            return self.url
-        
-        # Add credentials to URL
+
+        # Override any URL credentials if credentials are provided
         auth_netloc = f"{self.username}:{self.password}@{parsed.hostname}"
         if parsed.port:
             auth_netloc += f":{parsed.port}"