feat: implement a (simple) state machine for state management

Author: robgee86

src/arduino/app_peripherals/camera/base_camera.py

@@ -6,7 +6,7 @@
 import time
 from abc import ABC, abstractmethod
 from concurrent.futures import ThreadPoolExecutor
-from typing import Optional, Callable
+from typing import Literal, Optional, Callable
 import numpy as np
 
 from arduino.app_utils import Logger
@@ -46,6 +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._camera_lock = threading.Lock()
         self._is_started = False
@@ -59,12 +60,16 @@ def __init__(
 
         # Stream interruption detection
         self._consecutive_none_frames = 0
-        self._stream_paused = False
 
         # Event handling
-        self._on_event_cb: Callable[[str, dict], None] | None = None
+        self._on_status_changed_cb: Callable[[str, dict], None] | None = None
         self._event_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="CameraEvent")
 
+    @property
+    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."""
@@ -146,14 +151,11 @@ def capture(self) -> Optional[np.ndarray]:
             frame = self._read_frame()
             if frame is None:
                 self._consecutive_none_frames += 1
-                if self._consecutive_none_frames >= self._none_frame_threshold and not self._stream_paused:
-                    self._stream_paused = True
-                    self._emit_event("paused")
+                if self._consecutive_none_frames >= self._none_frame_threshold:
+                    self._set_status("paused")
                 return None
 
-            if self._stream_paused:
-                self._stream_paused = False
-                self._emit_event("resumed")
+            self._set_status("streaming")
 
             self._consecutive_none_frames = 0
 
@@ -190,48 +192,48 @@ def is_started(self) -> bool:
         """Check if the camera has been started."""
         return self._is_started
 
-    def on_event(self, callback: Callable[[str, dict | None], None] | None):
+    def on_status_changed(self, callback: Callable[[str, dict], None] | None):
         """Registers or removes a callback to be triggered on camera lifecycle events.
 
-        When a camera lifecycle event will happen, the provided callback function will be invoked.
+        When a camera status changes, the provided callback function will be invoked.
         If None is provided, the callback will be removed.
 
         Args:
-            callback (Callable[[str, dict | None], None]): A callback that will be called every time a camera
-                lifecycle event will happen with the event name and any associated data. The event
-                names depend on the actual camera implementation being used. Some common events are:
-                - 'disconnected': The camera has been disconnected.
+            callback (Callable[[str, dict], None]): A callback that will be called every time the
+                camera status changes with the new status and any associated data. The status names
+                depend on the actual camera implementation being used. Some common events are:
                 - 'connected': The camera has been reconnected.
+                - 'disconnected': The camera has been disconnected.
+                - 'streaming': The stream is streaming.
                 - 'paused': The stream has been paused and is temporarily unavailable.
-                - 'resumed': The stream has resumed after being paused.
             callback (None): To unregister the current callback, if any.
 
         Example:
-            def on_event(event: str, data: dict):
-                print(f"Camera is now: {event}")
+            def on_status(status: str, data: dict):
+                print(f"Camera is now: {status}")
                 print(f"Data: {data}")
                 # Here you can add your code to react to the event
 
-            camera.on_event(on_event)
+            camera.on_status_changed(on_status)
         """
         if callback is None:
-            self._on_event_cb = None
+            self._on_status_changed_cb = None
         else:
 
-            def _callback_wrapper(event: str, data: dict):
+            def _callback_wrapper(new_status: str, data: dict):
                 try:
-                    callback(event, data)
+                    callback(new_status, data)
                 except Exception as e:
-                    self.logger.error(f"Callback for event '{event}' failed with error: {e}")
+                    self.logger.error(f"Callback for '{new_status}' status failed with error: {e}")
 
-            self._on_event_cb = _callback_wrapper
+            self._on_status_changed_cb = _callback_wrapper
 
     @abstractmethod
     def _open_camera(self) -> None:
         """
         Open the camera connection.
 
-        Must be implemented by subclasses and events should be emitted accordingly.
+        Must be implemented by subclasses and status changes should be emitted accordingly.
         """
         pass
 
@@ -240,7 +242,7 @@ def _close_camera(self) -> None:
         """
         Close the camera connection.
 
-        Must be implemented by subclasses and events should be emitted accordingly.
+        Must be implemented by subclasses and status changes should be emitted accordingly.
         """
         pass
 
@@ -253,16 +255,38 @@ def _read_frame(self) -> Optional[np.ndarray]:
         """
         pass
 
-    def _emit_event(self, event: str, data: dict | None = None) -> None:
+    def _set_status(self, new_status: str, data: dict | None = None) -> None:
         """
-        Invoke the registered event callback in the background, if any.
+        Updates the current status of the camera and invokes the registered status
+        changed callback in the background, if any.
+
+        Only allowed states and transitions are considered, other states are ignored.
+        Allowed states are:
+            - disconnected
+            - connected
+            - streaming
+            - paused
 
         Args:
-            event (str): The name of the event.
-            data (dict): Additional data associated with the event.
+            new_status (str): The name of the new status.
+            data (dict): Additional data associated with the status change.
         """
-        if self._on_event_cb is not None:
-            self._event_executor.submit(self._on_event_cb, event, data if data is not None else {})
+        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:
+            return
+
+        # Check if new_status is an allowed transition for the current status
+        if new_status in allowed_transitions[self._status]:
+            self._status = new_status
+            if self._on_status_changed_cb is not None:
+                self._event_executor.submit(self._on_status_changed_cb, new_status, data if data is not None else {})
 
     def __enter__(self):
         """Context manager entry."""

src/arduino/app_peripherals/camera/ip_camera.py

@@ -89,7 +89,7 @@ def _open_camera(self) -> None:
             if not ret and frame is None:
                 raise CameraOpenError(f"Read test failed for IP camera at {self.url}")
 
-            self._emit_event("connected", {"camera_url": self.url})
+            self._set_status("connected", {"camera_url": self.url})
 
         except CameraOpenError:
             if self._cap is not None:
@@ -138,7 +138,7 @@ def _close_camera(self) -> None:
         if self._cap is not None:
             self._cap.release()
             self._cap = None
-            self._emit_event("disconnected", {"camera_url": self.url})
+            self._set_status("disconnected", {"camera_url": self.url})
 
     def _read_frame(self) -> np.ndarray | None:
         """Read a frame from the IP camera with automatic reconnection."""

src/arduino/app_peripherals/camera/v4l_camera.py

@@ -188,7 +188,7 @@ def _open_camera(self) -> None:
             if not ret and frame is None:
                 raise CameraOpenError(f"Read test failed for camera {self.name}")
 
-            self._emit_event("connected", {"camera_name": self.name, "camera_path": self.v4l_path})
+            self._set_status("connected", {"camera_name": self.name, "camera_path": self.v4l_path})
 
         except CameraOpenError:
             if self._cap is not None:
@@ -207,7 +207,7 @@ def _close_camera(self) -> None:
         if self._cap is not None:
             self._cap.release()
             self._cap = None
-            self._emit_event("disconnected", {"camera_name": self.name, "camera_path": self.v4l_path})
+            self._set_status("disconnected", {"camera_name": self.name, "camera_path": self.v4l_path})
 
     def _read_frame(self) -> np.ndarray | None:
         """

src/arduino/app_peripherals/camera/websocket_camera.py

@@ -178,7 +178,7 @@ async def _ws_handler(self, conn: websockets.ServerConnection) -> None:
             # Accept the client
             self._client = conn
 
-        self._emit_event("connected", {"client_address": client_addr})
+        self._set_status("connected", {"client_address": client_addr})
 
         self.logger.debug(f"Client connected: {client_addr}")
 
@@ -218,7 +218,7 @@ async def _ws_handler(self, conn: websockets.ServerConnection) -> None:
             async with self._client_lock:
                 if self._client == conn:
                     self._client = None
-                    self._emit_event("disconnected", {"client_address": client_addr})
+                    self._set_status("disconnected", {"client_address": client_addr})
                     self.logger.debug(f"Client disconnected: {client_addr}")
 
     def _parse_message(self, message: str | bytes) -> np.ndarray | None:

tests/arduino/app_peripherals/camera/test_base_camera.py

@@ -37,15 +37,15 @@ def _open_camera(self):
         if self.should_fail_open:
             raise RuntimeError(self.open_error_message)
         else:
-            self._emit_event("connected")
+            self._set_status("connected")
 
     def _close_camera(self):
         """Mock implementation of _close_camera."""
         self.close_call_count += 1
         if self.should_fail_close:
             raise RuntimeError(self.close_error_message)
         else:
-            self._emit_event("disconnected")
+            self._set_status("disconnected")
 
     def _read_frame(self):
         """Mock implementation that returns a dummy frame."""
@@ -388,28 +388,39 @@ def test_events():
     def event_callback(event, data):
         events.append((event, data))
 
-    camera.on_event(event_callback)
+    camera.on_status_changed(event_callback)
 
     camera.start()  # Should emit "connected" event
 
-    camera.capture()
+    assert camera.status == "connected"
+
+    camera.capture()  # Should emit "streaming" event
+
+    assert camera.status == "streaming"
 
     camera.frame = None
     camera.capture()
     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 "resumed" event
+    camera.capture()  # Should emit "streaming" event
+
+    assert camera.status == "streaming"
 
     camera.stop()  # Should emit "disconnected" event
 
+    assert camera.status == "disconnected"
+
     # The events list is modified from another thread, so a brief sleep
     # helps ensure the main thread sees the appended items before asserting.
     time.sleep(0.1)
 
-    assert len(events) == 4
+    assert len(events) == 5
     assert "connected" in events[0][0]
-    assert "paused" in events[1][0]
-    assert "resumed" in events[2][0]
-    assert "disconnected" in events[3][0]
+    assert "streaming" in events[1][0]
+    assert "paused" in events[2][0]
+    assert "streaming" in events[3][0]
+    assert "disconnected" in events[4][0]

tests/arduino/app_peripherals/camera/test_ip_camera.py

@@ -312,7 +312,7 @@ def test_events(mock_videocapture, mock_requests):
     def event_callback(event, data):
         events.append((event, data))
 
-    camera.on_event(event_callback)
+    camera.on_status_changed(event_callback)
 
     camera.start()
     camera.stop()

tests/arduino/app_peripherals/camera/test_v4l_camera.py

@@ -493,7 +493,7 @@ def test_events(self, mock_videocapture, mock_successful_connect):
         def event_callback(event, data):
             events.append((event, data))
 
-        camera.on_event(event_callback)
+        camera.on_status_changed(event_callback)
 
         camera.start()
         camera.stop()

tests/arduino/app_peripherals/camera/test_websocket_camera.py

@@ -344,7 +344,7 @@ def event_listener(event_type, data):
         events.append((event_type, data))
 
     camera = WebSocketCamera(port=0)
-    camera.on_event(event_listener)
+    camera.on_status_changed(event_listener)
     camera.start()
 
     # This should emit connection and disconnection events
@@ -396,7 +396,7 @@ def event_listener(event_type, data):
         events.append((event_type, data))
 
     camera = WebSocketCamera(port=0)
-    camera.on_event(event_listener)
+    camera.on_status_changed(event_listener)
     camera.start()
 
     await asyncio.sleep(0.1)
@@ -425,7 +425,7 @@ def event_listener(event_type, data):
         events.append((event_type, data))
 
     camera = WebSocketCamera(port=0, timeout=1)  # Reduced timeout for faster stop() call
-    camera.on_event(event_listener)
+    camera.on_status_changed(event_listener)
     camera.start()
 
     can_close = asyncio.Event()