fix

Author: robgee86

src/arduino/app_peripherals/camera/base_camera.py

@@ -46,7 +46,7 @@ def __init__(
         self.adjustments = adjustments
         self.logger = logger  # This will be overridden by subclasses if needed
         self.name = self.__class__.__name__  # This will be overridden by subclasses if needed
-        self._status: Literal['disconnected', 'connected', 'streaming', 'paused'] = "disconnected"
+        self._status: Literal["disconnected", "connected", "streaming", "paused"] = "disconnected"
 
         self._camera_lock = threading.Lock()
         self._is_started = False
@@ -66,10 +66,10 @@ def __init__(
         self._event_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="CameraEvent")
 
     @property
-    def status(self) -> Literal['disconnected', 'connected', 'streaming', 'paused']:
+    def status(self) -> Literal["disconnected", "connected", "streaming", "paused"]:
         """Read-only property for camera status."""
         return self._status
-    
+
     @property
     def _none_frame_threshold(self) -> int:
         """Heuristic: 750ms of empty frames based on current fps."""
@@ -255,7 +255,7 @@ def _read_frame(self) -> Optional[np.ndarray]:
         """
         pass
 
-    def _set_status(self, new_status: str, data: dict | None = None) -> None:
+    def _set_status(self, new_status: Literal["disconnected", "connected", "streaming", "paused"], data: dict | None = None) -> None:
         """
         Updates the current status of the camera and invokes the registered status
         changed callback in the background, if any.
@@ -271,15 +271,19 @@ def _set_status(self, new_status: str, data: dict | None = None) -> None:
             new_status (str): The name of the new status.
             data (dict): Additional data associated with the status change.
         """
+
+        if self.status == new_status:
+            return
+
         allowed_transitions = {
             "disconnected": ["connected"],
             "connected": ["disconnected", "streaming"],
             "streaming": ["paused", "disconnected"],
             "paused": ["streaming", "disconnected"],
         }
 
-        # If current status is not in the state machine, do nothing
-        if self._status not in allowed_transitions:
+        # If new status is not in the state machine, ignore it
+        if new_status not in allowed_transitions:
             return
 
         # Check if new_status is an allowed transition for the current status

tests/arduino/app_peripherals/camera/test_base_camera.py

@@ -403,7 +403,7 @@ def event_callback(event, data):
     camera.capture()
     camera.capture()  # Should emit "paused" event
     camera.frame = np.zeros((480, 640, 3), dtype=np.uint8)
-    
+
     assert camera.status == "paused"
 
     camera.capture()  # Should emit "streaming" event

tests/arduino/app_peripherals/camera/test_websocket_camera.py

@@ -43,23 +43,25 @@ def encoded_frame_json(encoded_frame_binary):
 def test_websocket_camera_init_default():
     """Test WebSocketCamera initialization with default parameters."""
     camera = WebSocketCamera()
-    assert camera.host == "0.0.0.0"
+    assert camera.address == "ws://0.0.0.0:8080"
     assert camera.port == 8080
     assert camera.timeout == 3
     assert camera.frame_format == "binary"
     assert camera.resolution == (640, 480)
     assert camera.fps == 10
+    assert camera.status == "disconnected"
 
 
 def test_websocket_camera_init_custom():
     """Test WebSocketCamera initialization with custom parameters."""
     camera = WebSocketCamera(host="127.0.0.1", port=9090, timeout=30, frame_format="json", resolution=(1920, 1080), fps=30)
-    assert camera.host == "127.0.0.1"
+    assert camera.address == "ws://127.0.0.1:9090"
     assert camera.port == 9090
     assert camera.timeout == 30
     assert camera.frame_format == "json"
     assert camera.resolution == (1920, 1080)
     assert camera.fps == 30
+    assert camera.status == "disconnected"
 
 
 def test_websocket_camera_start_stop():
@@ -73,13 +75,17 @@ def test_websocket_camera_start_stop():
         pytest.fail("Camera start failed")
 
     assert camera.is_started()
+    # Starting does not coincide with being connected in case of WebSocketCamera
+    # as that depends on client activity
+    assert camera.status == "disconnected"
 
     try:
         camera.stop()
     except Exception:
         pytest.fail("Camera stop failed")
 
     assert not camera.is_started()
+    assert camera.status == "disconnected"
 
 
 def test_websocket_camera_parse_message_binary(sample_frame, encoded_frame_binary):
@@ -161,7 +167,7 @@ def test_websocket_camera_read_frame_empty_queue():
 async def test_websocket_camera_capture_frame(encoded_frame_binary):
     """Test capturing frame from WebSocket camera."""
     with WebSocketCamera(port=0, frame_format="binary") as camera:
-        async with websockets.connect(f"{camera.protocol}://{camera.host}:{camera.port}") as ws:
+        async with websockets.connect(camera.address) as ws:
             # Skip welcome message
             await ws.recv()
 
@@ -183,15 +189,15 @@ async def test_websocket_camera_single_client():
 
     try:
         # Connect first client
-        async with websockets.connect(f"{camera.protocol}://{camera.host}:{camera.port}") as ws1:
+        async with websockets.connect(camera.address) as ws1:
             # First client should receive welcome message
             welcome = await ws1.recv()
             message = json.loads(welcome)
             assert message["status"] == "connected"
 
             # Try to connect second client while first is connected
             try:
-                async with websockets.connect(f"{camera.protocol}://{camera.host}:{camera.port}") as ws2:
+                async with websockets.connect(camera.address) as ws2:
                     # Second client should receive rejection message
                     rejection = await asyncio.wait_for(ws2.recv(), timeout=1.0)
                     message = json.loads(rejection)
@@ -207,7 +213,7 @@ async def test_websocket_camera_single_client():
 async def test_websocket_camera_welcome_message():
     """Test that welcome message is sent to connected client."""
     with WebSocketCamera(port=0) as camera:
-        async with websockets.connect(f"{camera.protocol}://{camera.host}:{camera.port}") as ws:
+        async with websockets.connect(camera.address) as ws:
             # Should receive welcome message
             welcome = await asyncio.wait_for(ws.recv(), timeout=1.0)
             message = json.loads(welcome)
@@ -220,7 +226,7 @@ async def test_websocket_camera_welcome_message():
 async def test_websocket_camera_receives_frames(encoded_frame_binary):
     """Test that server receives and queues frames from client."""
     with WebSocketCamera(port=0, frame_format="binary") as camera:
-        async with websockets.connect(f"{camera.protocol}://{camera.host}:{camera.port}") as ws:
+        async with websockets.connect(camera.address) as ws:
             # Skip welcome message
             await ws.recv()
 
@@ -241,7 +247,7 @@ async def test_websocket_camera_disconnects_client_on_stop():
     camera.start()
 
     try:
-        async with websockets.connect(f"{camera.protocol}://{camera.host}:{camera.port}") as ws:
+        async with websockets.connect(camera.address) as ws:
             # Client connected, receive welcome message
             welcome = await ws.recv()
             message = json.loads(welcome)
@@ -280,7 +286,7 @@ def test_websocket_camera_stop_without_client():
 async def test_websocket_camera_backpressure(sample_frame):
     """Test that old frames are dropped when new frames arrive faster than they're consumed."""
     with WebSocketCamera(port=0, frame_format="binary") as camera:
-        async with websockets.connect(f"{camera.protocol}://{camera.host}:{camera.port}") as ws:
+        async with websockets.connect(camera.address) as ws:
             await ws.recv()  # Skip welcome message
 
             _, buffer1 = cv2.imencode(".jpg", np.ones((480, 640, 3), dtype=np.uint8) * 1)
@@ -349,7 +355,7 @@ def event_listener(event_type, data):
 
     # This should emit connection and disconnection events
     async def client_task():
-        async with websockets.connect(f"{camera.protocol}://{camera.host}:{camera.port}"):
+        async with websockets.connect(camera.address):
             pass
 
     # Run client concurrently to properly test event handling
@@ -432,7 +438,7 @@ def event_listener(event_type, data):
 
     # This should emit a connection event but no disconnection event
     async def client_task():
-        async with websockets.connect(f"{camera.protocol}://{camera.host}:{camera.port}"):
+        async with websockets.connect(camera.address):
             pass
         await can_close.wait()