refactor: Camera args

Author: robgee86

src/arduino/app_peripherals/camera/README.md

@@ -72,9 +72,9 @@ See the arduino.app_utils.image module for more supported adjustments.
 ## Camera Types
 The Camera class provides automatic camera type detection based on the format of its source argument. keyword arguments will be propagated to the underlying implementation.
 
-Note: constructor arguments (except source) must be provided in keyword format to forward them correctly to the specific camera implementations.
+Note: Camera's constructor arguments (except those in its signature) must be provided in keyword format to forward them correctly to the specific camera implementations.
 
-The underlying camera implementations can be instantiated explicitly (V4LCamera, IPCamera and WebSocketCamera), if needed.
+The underlying camera implementations can also be instantiated explicitly (V4LCamera, IPCamera and WebSocketCamera), if needed.
 
 ### V4L Cameras
 For local USB cameras and V4L-compatible devices.

src/arduino/app_peripherals/camera/base_camera.py

@@ -5,7 +5,7 @@
 import threading
 import time
 from abc import ABC, abstractmethod
-from typing import Optional, Tuple, Callable
+from typing import Optional, Callable
 import numpy as np
 
 from arduino.app_utils import Logger
@@ -25,9 +25,9 @@ class BaseCamera(ABC):
 
     def __init__(
         self,
-        resolution: Optional[Tuple[int, int]] = (640, 480),
+        resolution: tuple[int, int] = (640, 480),
         fps: int = 10,
-        adjustments: Optional[Callable[[np.ndarray], np.ndarray]] = None,
+        adjustments: Callable[[np.ndarray], np.ndarray] = None,
     ):
         """
         Initialize the camera base.
@@ -87,7 +87,7 @@ def capture(self) -> Optional[np.ndarray]:
             return None
         return frame
 
-    def _extract_frame(self) -> Optional[np.ndarray]:
+    def _extract_frame(self) -> np.ndarray | None:
         """Extract a frame with FPS throttling and post-processing."""
         with self._camera_lock:
             # FPS throttling

src/arduino/app_peripherals/camera/camera.py

@@ -2,9 +2,11 @@
 #
 # SPDX-License-Identifier: MPL-2.0
 
-from typing import Union
+from collections.abc import Callable
 from urllib.parse import urlparse
 
+import numpy as np
+
 from .base_camera import BaseCamera
 from .errors import CameraConfigError
 
@@ -25,7 +27,14 @@ class Camera:
     format to forward them correctly to the specific camera implementations.
     """
 
-    def __new__(cls, source: Union[str, int] = 0, **kwargs) -> BaseCamera:
+    def __new__(
+            cls,
+            source: str | int = 0,
+            resolution: tuple[int, int] = (640, 480),
+            fps: int = 10,
+            adjustments: Callable[[np.ndarray], np.ndarray] = None,
+            **kwargs
+    ) -> BaseCamera:
         """Create a camera instance based on the source type.
 
         Args:
@@ -34,13 +43,12 @@ def __new__(cls, source: Union[str, int] = 0, **kwargs) -> BaseCamera:
                 - 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 for input streams (e.g., "ws://0.0.0.0:8080")
+            resolution (tuple, optional): Frame resolution as (width, height).
+                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
             **kwargs: Camera-specific configuration parameters grouped by type:
-                Common Parameters:
-                    resolution (tuple, optional): Frame resolution as (width, height).
-                        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 (int, optional): V4L device index override. Default: 0.
                 IP Camera Parameters:
@@ -88,26 +96,26 @@ def __new__(cls, source: Union[str, int] = 0, **kwargs) -> BaseCamera:
             # V4L Camera
             from .v4l_camera import V4LCamera
 
-            return V4LCamera(source, **kwargs)
+            return V4LCamera(source, resolution=resolution, fps=fps, adjustments=adjustments, **kwargs)
         elif isinstance(source, str):
             parsed = urlparse(source)
             if parsed.scheme in ["http", "https", "rtsp"]:
                 # IP Camera
                 from .ip_camera import IPCamera
 
-                return IPCamera(source, **kwargs)
+                return IPCamera(source, resolution=resolution, fps=fps, adjustments=adjustments, **kwargs)
             elif parsed.scheme in ["ws", "wss"]:
                 # WebSocket Camera - extract host and port from URL
                 from .websocket_camera import WebSocketCamera
 
                 host = parsed.hostname or "localhost"
                 port = parsed.port or 8080
-                return WebSocketCamera(host=host, port=port, **kwargs)
+                return WebSocketCamera(host=host, port=port, resolution=resolution, fps=fps, adjustments=adjustments, **kwargs)
             elif source.startswith("/dev/video") or source.isdigit():
                 # V4L device path or index as string
                 from .v4l_camera import V4LCamera
 
-                return V4LCamera(source, **kwargs)
+                return V4LCamera(source, resolution=resolution, fps=fps, adjustments=adjustments, **kwargs)
             else:
                 raise CameraConfigError(f"Unsupported camera source: {source}")
         else:

src/arduino/app_peripherals/camera/examples/1_initialize.py

@@ -13,5 +13,6 @@
 camera = Camera(2, resolution=(640, 480), fps=15)  # Infers camera type
 v4l = V4LCamera(2, (640, 480), 15)  # Explicitly requests V4L camera
 
-# Note: constructor arguments (except source) must be provided in keyword
-# format to forward them correctly to the specific camera implementations.
+# Note: Camera's constructor arguments (except those in its signature)
+# must be provided in keyword format to forward them correctly to the
+# specific camera implementations.

src/arduino/app_peripherals/camera/ip_camera.py

@@ -5,8 +5,8 @@
 import cv2
 import numpy as np
 import requests
-from typing import Callable, Optional, Tuple
 from urllib.parse import urlparse
+from collections.abc import Callable
 
 from arduino.app_utils import Logger
 
@@ -27,12 +27,12 @@ class IPCamera(BaseCamera):
     def __init__(
         self,
         url: str,
-        username: Optional[str] = None,
-        password: Optional[str] = None,
+        username: str | None = None,
+        password: str | None = None,
         timeout: int = 10,
-        resolution: Optional[Tuple[int, int]] = (640, 480),
+        resolution: tuple[int, int] = (640, 480),
         fps: int = 10,
-        adjustments: Optional[Callable[[np.ndarray], np.ndarray]] = None,
+        adjustments: Callable[[np.ndarray], np.ndarray] = None,
     ):
         """
         Initialize IP camera.
@@ -126,7 +126,7 @@ def _close_camera(self) -> None:
             self._cap.release()
             self._cap = None
 
-    def _read_frame(self) -> Optional[np.ndarray]:
+    def _read_frame(self) -> np.ndarray | None:
         """Read a frame from the IP camera with automatic reconnection."""
         if self._cap is None:
             logger.info(f"No connection to IP camera {self.url}, attempting to reconnect")

src/arduino/app_peripherals/camera/v4l_camera.py

@@ -6,7 +6,7 @@
 import re
 import cv2
 import numpy as np
-from typing import Callable, Optional, Tuple, Union, Dict
+from collections.abc import Callable
 
 from arduino.app_utils import Logger
 
@@ -26,10 +26,10 @@ class V4LCamera(BaseCamera):
 
     def __init__(
         self,
-        device: Union[str, int] = 0,
-        resolution: Optional[Tuple[int, int]] = (640, 480),
+        device: str | int = 0,
+        resolution: tuple[int, int] = (640, 480),
         fps: int = 10,
-        adjustments: Optional[Callable[[np.ndarray], np.ndarray]] = None,
+        adjustments: Callable[[np.ndarray], np.ndarray] = None,
     ):
         """
         Initialize V4L camera.
@@ -49,7 +49,7 @@ def __init__(
 
         self._cap = None
 
-    def _resolve_camera_id(self, device: Union[str, int]) -> int:
+    def _resolve_camera_id(self, device: str | int) -> int:
         """
         Resolve camera identifier to a numeric device ID.
 
@@ -83,7 +83,7 @@ def _resolve_camera_id(self, device: Union[str, int]) -> int:
 
         raise CameraOpenError(f"Cannot resolve camera identifier: {device}")
 
-    def _get_video_devices_by_index(self) -> Dict[int, str]:
+    def _get_video_devices_by_index(self) -> dict[int, str]:
         """
         Map camera indices to device numbers by reading /dev/v4l/by-id/.
 
@@ -160,7 +160,7 @@ def _close_camera(self) -> None:
             self._cap.release()
             self._cap = None
 
-    def _read_frame(self) -> Optional[np.ndarray]:
+    def _read_frame(self) -> np.ndarray | None:
         """Read a frame from the V4L camera."""
         if self._cap is None:
             return None

src/arduino/app_peripherals/camera/websocket_camera.py

@@ -7,11 +7,11 @@
 import threading
 import queue
 import time
-from typing import Callable, Optional, Tuple, Union
 import numpy as np
 import cv2
 import websockets
 import asyncio
+from collections.abc import Callable
 from concurrent.futures import CancelledError, TimeoutError
 
 from arduino.app_utils import Logger
@@ -48,9 +48,9 @@ def __init__(
         port: int = 8080,
         timeout: int = 10,
         frame_format: str = "binary",
-        resolution: Optional[Tuple[int, int]] = (640, 480),
+        resolution: tuple[int, int] = (640, 480),
         fps: int = 10,
-        adjustments: Optional[Callable[[np.ndarray], np.ndarray]] = None,
+        adjustments: Callable[[np.ndarray], np.ndarray] = None,
     ):
         """
         Initialize WebSocket camera server.
@@ -78,7 +78,7 @@ def __init__(
         self._loop = None
         self._server_thread = None
         self._stop_event = asyncio.Event()
-        self._client: Optional[websockets.ServerConnection] = None
+        self._client: websockets.ServerConnection = None
         self._client_lock = asyncio.Lock()
 
     def _open_camera(self) -> None:
@@ -195,7 +195,7 @@ async def _ws_handler(self, conn: websockets.ServerConnection) -> None:
                     self._client = None
                     logger.info(f"Client removed: {client_addr}")
 
-    async def _parse_message(self, message) -> Optional[np.ndarray]:
+    async def _parse_message(self, message) -> np.ndarray | None:
         """Parse WebSocket message to extract frame."""
         try:
             if self.frame_format == "base64":
@@ -297,7 +297,7 @@ async def _set_async_stop_event(self):
                 await self._client.close()
                 self._stop_event.set()
 
-    def _read_frame(self) -> Optional[np.ndarray]:
+    def _read_frame(self) -> np.ndarray | None:
         """Read a frame from the queue."""
         try:
             # Get frame with short timeout to avoid blocking
@@ -306,7 +306,7 @@ def _read_frame(self) -> Optional[np.ndarray]:
         except queue.Empty:
             return None
 
-    def _send_message_to_client(self, message: Union[str, bytes, dict]) -> None:
+    def _send_message_to_client(self, message: str | bytes | dict) -> None:
         """
         Send a message to the connected client (if any).
 
@@ -333,7 +333,7 @@ def _send_message_to_client(self, message: Union[str, bytes, dict]) -> None:
             logger.error(f"Error sending message to client: {e}")
             raise
 
-    async def _send_to_client(self, message: Union[str, bytes, dict]) -> None:
+    async def _send_to_client(self, message: str | bytes | dict) -> None:
         """Send message to a single client."""
         if isinstance(message, dict):
             message = json.dumps(message)