Merge branch 'main' into simple_eye_events

Author: domstoppable

CHANGES.rst

@@ -1,3 +1,16 @@
+1.5.0
+###########
+- Auto-start only necessary streams in Simple API
+- Adds error information
+
+1.4.0
+###########
+- Adds eye events (blinks, fixations)
+
+1.3.6
+###########
+- Adds eyelid data
+
 1.3.5
 ###########
 - Fixes streaming bug when audio is enabled

examples/async/start_stop_recordings.py

@@ -1,12 +1,17 @@
 import asyncio
 
 from pupil_labs.realtime_api import Device, Network, StatusUpdateNotifier
-from pupil_labs.realtime_api.models import Recording
+from pupil_labs.realtime_api.models import Recording, Sensor
 
 
-async def print_recording(component):
+async def on_status_update(component):
     if isinstance(component, Recording):
-        print(f"Update: {component.message}")
+        if component.action == "ERROR":
+            print(f"Error : {component.message}")
+
+    elif isinstance(component, Sensor):
+        if component.stream_error:
+            print(f"Stream error in sensor {component.sensor}")
 
 
 async def main():
@@ -18,7 +23,7 @@ async def main():
 
     async with Device.from_discovered_device(dev_info) as device:
         # get update when recording is fully started
-        notifier = StatusUpdateNotifier(device, callbacks=[print_recording])
+        notifier = StatusUpdateNotifier(device, callbacks=[on_status_update])
         await notifier.receive_updates_start()
         recording_id = await device.recording_start()
         print(f"Initiated recording with id {recording_id}")

examples/async/stream_video_with_overlayed_fixations.py

@@ -75,7 +75,7 @@ async def enqueue_sensor_data(sensor: T.AsyncIterator, queue: asyncio.Queue) ->
 
 
 async def match_and_draw(queue_video, queue_eye_events):
-    fixation_history = deque(maxlen=5)
+    fixation_history = deque(maxlen=10)
     fixation_counter = 0
 
     blink = None
@@ -88,6 +88,9 @@ async def match_and_draw(queue_video, queue_eye_events):
         while not queue_eye_events.empty():
             _, eye_event = await queue_eye_events.get()
             if isinstance(eye_event, FixationEventData):
+                if eye_event.event_type == 0:
+                    continue
+
                 fixation_history.append(
                     {
                         "id": fixation_counter,

examples/simple/start_stop_recordings.py

@@ -13,7 +13,13 @@
 recording_id = device.recording_start()
 print(f"Started recording with id {recording_id}")
 
-time.sleep(5)
+# Check for errors while recording runs
+start_time = time.time()
+while time.time() - start_time < 5:
+    for e in device.get_errors():
+        print("Error:", e)
+
+    time.sleep(1)
 
 device.recording_stop_and_save()
 

src/pupil_labs/realtime_api/models.py

@@ -128,6 +128,7 @@ class Sensor(T.NamedTuple):
     params: T.Optional[str] = None
     port: T.Optional[int] = None
     protocol: str = "rtsp"
+    stream_error: bool = True
 
     @property
     def url(self) -> T.Optional[str]:

src/pupil_labs/realtime_api/simple/device.py

@@ -10,7 +10,15 @@
 from ..base import DeviceBase
 from ..device import Device as _DeviceAsync
 from ..device import StatusUpdateNotifier
-from ..models import Component, Event, Sensor, Status, Template, TemplateDataFormat
+from ..models import (
+    Component,
+    Event,
+    Recording,
+    Sensor,
+    Status,
+    Template,
+    TemplateDataFormat,
+)
 from ..streaming import (
     BlinkEventData,
     FixationEventData,
@@ -60,6 +68,8 @@ def __init__(
         self._status = self._get_status()
         self._start_background_worker(start_streaming_by_default)
 
+        self._errors: T.List[str] = []
+
     @property
     def phone_name(self) -> str:
         return self._status.phone.device_name
@@ -107,6 +117,12 @@ def serial_number_scene_cam(self) -> T.Optional[str]:
         """Returns ``None`` if no scene camera is connected"""
         return self._status.hardware.world_camera_serial
 
+    def get_errors(self) -> T.List[str]:
+        errors = self._errors.copy()
+        self._errors.clear()
+
+        return errors
+
     def world_sensor(self) -> T.Optional[Sensor]:
         return self._status.direct_world_sensor()
 
@@ -282,9 +298,16 @@ def receive_matched_scene_and_eyes_video_frames_and_gaze(
     def _receive_item(
         self, sensor: str, timeout_seconds: T.Optional[float] = None
     ) -> T.Optional[T.Union[VideoFrame, GazeDataType]]:
-        if not self.is_currently_streaming:
-            logger.debug("receive_* called without being streaming")
-            self.streaming_start()
+        if sensor == MATCHED_ITEM_LABEL:
+            self.start_stream_if_needed(Sensor.Name.GAZE.value)
+            self.start_stream_if_needed(Sensor.Name.WORLD.value)
+        elif sensor == MATCHED_GAZE_EYES_LABEL:
+            self.start_stream_if_needed(Sensor.Name.GAZE.value)
+            self.start_stream_if_needed(Sensor.Name.EYES.value)
+            self.start_stream_if_needed(Sensor.Name.WORLD.value)
+        else:
+            self.start_stream_if_needed(sensor)
+
         try:
             return self._most_recent_item[sensor].popleft()
         except IndexError:
@@ -295,23 +318,67 @@ def _receive_item(
                 return self._most_recent_item[sensor].popleft()
             return None
 
-    def streaming_start(self):
-        self._streaming_trigger_action(self._EVENT.SHOULD_STREAMS_START)
+    def start_stream_if_needed(self, sensor: str):
+        if not self._is_streaming_flags[sensor].is_set():
+            logger.debug("receive_* called without being streaming")
+            self.streaming_start(sensor)
+
+    def streaming_start(self, stream_name: str):
+        if stream_name is None:
+            for event in (
+                self._EVENT.SHOULD_START_GAZE,
+                self._EVENT.SHOULD_START_WORLD,
+                self._EVENT.SHOULD_START_EYES,
+                self._EVENT.SHOULD_START_IMU,
+            ):
+                self._streaming_trigger_action(event)
+            return
+
+        event = None
+        if stream_name == Sensor.Name.GAZE.value:
+            event = self._EVENT.SHOULD_START_GAZE
+        elif stream_name == Sensor.Name.WORLD.value:
+            event = self._EVENT.SHOULD_START_WORLD
+        elif stream_name == Sensor.Name.EYES.value:
+            event = self._EVENT.SHOULD_START_EYES
+        elif stream_name == Sensor.Name.IMU.value:
+            event = self._EVENT.SHOULD_START_IMU
+
+        self._streaming_trigger_action(event)
+
+    def streaming_stop(self, stream_name: str = None):
+        if stream_name is None:
+            for event in (
+                self._EVENT.SHOULD_STOP_GAZE,
+                self._EVENT.SHOULD_STOP_WORLD,
+                self._EVENT.SHOULD_STOP_EYES,
+                self._EVENT.SHOULD_STOP_IMU,
+            ):
+                self._streaming_trigger_action(event)
+            return
+
+        event = None
+        if stream_name == Sensor.Name.GAZE.value:
+            event = self._EVENT.SHOULD_STOP_GAZE
+        elif stream_name == Sensor.Name.WORLD.value:
+            event = self._EVENT.SHOULD_STOP_WORLD
+        elif stream_name == Sensor.Name.EYES.value:
+            event = self._EVENT.SHOULD_STOP_EYES
+        elif stream_name == Sensor.Name.IMU.value:
+            event = self._EVENT.SHOULD_STOP_IMU
 
-    def streaming_stop(self):
-        self._streaming_trigger_action(self._EVENT.SHOULD_STREAMS_STOP)
+        self._streaming_trigger_action(event)
 
     def _streaming_trigger_action(self, action):
         if self._event_manager and self._background_loop:
             logger.debug(f"Sending {action.name} trigger")
             self._event_manager.trigger_threadsafe(action)
         else:
-            logger.debug(f"Could send {action.name} trigger")
+            logger.debug(f"Could not send {action.name} trigger")
 
     @property
     def is_currently_streaming(self) -> bool:
-        is_streaming = self._is_streaming_flag.is_set()
-        return is_streaming
+        return any(flag.is_set() for flag in self._is_streaming_flags.values())
 
     def estimate_time_offset(
         self,
@@ -349,8 +416,14 @@ def __del__(self):
 
     class _EVENT(enum.Enum):
         SHOULD_WORKER_CLOSE = "should worker close"
-        SHOULD_STREAMS_START = "should stream start"
-        SHOULD_STREAMS_STOP = "should streams stop"
+        SHOULD_START_GAZE = "should start gaze"
+        SHOULD_START_WORLD = "should start world"
+        SHOULD_START_EYES = "should start eyes"
+        SHOULD_START_IMU = "should start imu"
+        SHOULD_STOP_GAZE = "should stop gaze"
+        SHOULD_STOP_WORLD = "should stop world"
+        SHOULD_STOP_EYES = "should stop eyes"
+        SHOULD_STOP_IMU = "should stop imu"
 
     def _start_background_worker(self, start_streaming_by_default):
         self._event_manager = None
@@ -377,13 +450,18 @@ def _start_background_worker(self, start_streaming_by_default):
         self._cached_eyes_for_matching: EyesCacheType = collections.deque(maxlen=200)
 
         event_auto_update_started = threading.Event()
-        self._is_streaming_flag = threading.Event()
+        self._is_streaming_flags = {
+            Sensor.Name.GAZE.value: threading.Event(),
+            Sensor.Name.WORLD.value: threading.Event(),
+            Sensor.Name.EYES.value: threading.Event(),
+            Sensor.Name.IMU.value: threading.Event(),
+        }
         self._auto_update_thread = threading.Thread(
             target=self._auto_update,
             kwargs=dict(
                 device_weakref=weakref.ref(self),  # weak ref to avoid cycling ref
                 auto_update_started_flag=event_auto_update_started,
-                is_streaming_flag=self._is_streaming_flag,
+                is_streaming_flags=self._is_streaming_flags,
                 start_streaming_by_default=start_streaming_by_default,
             ),
             name=f"{self} auto-update thread",
@@ -409,7 +487,7 @@ async def _get_status():
     def _auto_update(
         device_weakref: weakref.ReferenceType,
         auto_update_started_flag: threading.Event,
-        is_streaming_flag: threading.Event,
+        is_streaming_flags: T.Dict[str, threading.Event],
         start_streaming_by_default: bool = False,
     ):
         stream_managers = {
@@ -450,6 +528,14 @@ async def _process_status_changes(changed: Component):
                 else:
                     logger.debug(f"Unhandled DIRECT sensor {changed.sensor}")
 
+            elif isinstance(changed, Recording) and changed.action == "ERROR":
+                device_weakref()._errors.append(changed.message)
+
+            elif isinstance(changed, Sensor) and changed.stream_error:
+                error = f"Stream error in sensor {changed.sensor}"
+                if error not in device_weakref()._errors:
+                    device_weakref()._errors.append(error)
+
         async def _auto_update_until_closed():
             async with _DeviceAsync.convert_from(device_weakref()) as device:
                 event_manager = _AsyncEventManager(Device._EVENT)
@@ -467,28 +553,59 @@ async def _auto_update_until_closed():
                 auto_update_started_flag.set()
                 if start_streaming_by_default:
                     logger.debug("Streaming started by default")
-                    is_streaming_flag.set()
+                    start_stream(Sensor.Name.GAZE.value)
+                    start_stream(Sensor.Name.WORLD.value)
+                    start_stream(Sensor.Name.EYES.value)
+                    start_stream(Sensor.Name.IMU.value)
 
                 while True:
                     logger.debug("Background worker waiting for event...")
                     event = await event_manager.wait_for_first_event()
                     logger.debug(f"Background worker received {event}")
+
                     if event is Device._EVENT.SHOULD_WORKER_CLOSE:
                         break
-                    elif event is Device._EVENT.SHOULD_STREAMS_START:
-                        for manager in stream_managers.values():
-                            manager.should_be_streaming = True
-                        is_streaming_flag.set()
-                        logger.debug("Streaming started")
-                    elif event is Device._EVENT.SHOULD_STREAMS_STOP:
-                        for manager in stream_managers.values():
-                            manager.should_be_streaming = False
-                        is_streaming_flag.clear()
-                        logger.debug("Streaming stopped")
-                    else:
+
+                    try:
+                        func = event_func_map[event]
+                        stream = event_stream_map[event]
+                        func(stream.value)
+                    except KeyError:
                         raise RuntimeError(f"Unhandled {event!r}")
 
                 await notifier.receive_updates_stop()
                 device_weakref()._event_manager = None
 
+        def start_stream(stream_name):
+            is_streaming_flags[stream_name].set()
+            stream_managers[stream_name].should_be_streaming = True
+            logger.debug(f"Streaming started {stream_name}")
+
+        def stop_stream(stream_name):
+            stream_managers[stream_name].should_be_streaming = False
+            is_streaming_flags[stream_name].clear()
+            logger.debug(f"Streaming stopped {stream_name}")
+
+        event_func_map = {
+            Device._EVENT.SHOULD_START_GAZE: start_stream,
+            Device._EVENT.SHOULD_START_WORLD: start_stream,
+            Device._EVENT.SHOULD_START_EYES: start_stream,
+            Device._EVENT.SHOULD_START_IMU: start_stream,
+            Device._EVENT.SHOULD_STOP_GAZE: stop_stream,
+            Device._EVENT.SHOULD_STOP_WORLD: stop_stream,
+            Device._EVENT.SHOULD_STOP_EYES: stop_stream,
+            Device._EVENT.SHOULD_STOP_IMU: stop_stream,
+        }
+
+        event_stream_map = {
+            Device._EVENT.SHOULD_START_GAZE: Sensor.Name.GAZE,
+            Device._EVENT.SHOULD_START_WORLD: Sensor.Name.WORLD,
+            Device._EVENT.SHOULD_START_EYES: Sensor.Name.EYES,
+            Device._EVENT.SHOULD_START_IMU: Sensor.Name.IMU,
+            Device._EVENT.SHOULD_STOP_GAZE: Sensor.Name.GAZE,
+            Device._EVENT.SHOULD_STOP_WORLD: Sensor.Name.WORLD,
+            Device._EVENT.SHOULD_STOP_EYES: Sensor.Name.EYES,
+            Device._EVENT.SHOULD_STOP_IMU: Sensor.Name.IMU,
+        }
+
         return asyncio.run(_auto_update_until_closed())