Merge pull request #48 from domstoppable/main

Adds a function for calibration retrieval

Author: dourvaris

docs/examples/simple.rst

@@ -91,6 +91,16 @@ Eyes camera video
    :emphasize-lines: 18
    :linenos:
 
+Camera calibration
+------------------
+
+.. note:: Only available when connecting to a Neon Companion app
+
+.. literalinclude:: ../../examples/simple/camera_calibration.py
+   :language: python
+   :emphasize-lines: 12
+   :linenos:
+
 
 .. _stream_video_with_overlayed_gaze_example_simple:
 

examples/simple/camera_calibration.py

@@ -0,0 +1,29 @@
+from pupil_labs.realtime_api.simple import discover_one_device
+
+# Look for devices. Returns as soon as it has found the first device.
+print("Looking for the next best device...")
+device = discover_one_device(max_search_duration_seconds=10)
+if device is None:
+    print("No device found.")
+    raise SystemExit(-1)
+
+# Device status is fetched on initialization and kept up-to-date in the background
+
+calibration = device.get_calibration()
+
+print("Scene camera matrix:")
+print(calibration["scene_camera_matrix"][0])
+print("\nScene distortion coefficients:")
+print(calibration["scene_distortion_coefficients"][0])
+
+print("\nRight camera matrix:")
+print(calibration["right_camera_matrix"][0])
+print("\nRight distortion coefficients:")
+print(calibration["right_distortion_coefficients"][0])
+
+print("\nLeft camera matrix:")
+print(calibration["left_camera_matrix"][0])
+print("\nLeft distortion coefficients:")
+print(calibration["left_distortion_coefficients"][0])
+
+device.close()

src/pupil_labs/realtime_api/device.py

@@ -6,6 +6,7 @@
 import typing as T
 
 import aiohttp
+import numpy as np
 import websockets
 
 import pupil_labs  # noqa: F401
@@ -165,6 +166,35 @@ async def __aexit__(
     def _create_client_session(self):
         self.session = aiohttp.ClientSession()
 
+    async def get_calibration(self) -> np.ndarray:
+        """
+        :raises pupil_labs.realtime_api.device.DeviceError: if the request fails
+        """
+        async with self.session.get(self.api_url(APIPath.CALIBRATION)) as response:
+            if response.status != 200:
+                raise DeviceError(response.status, "Failed to fetch calibration")
+
+            raw_data = await response.read()
+            return np.frombuffer(
+                raw_data,
+                np.dtype(
+                    [
+                        ("version", "u1"),
+                        ("serial", "6a"),
+                        ("scene_camera_matrix", "(3,3)d"),
+                        ("scene_distortion_coefficients", "8d"),
+                        ("scene_extrinsics_affine_matrix", "(4,4)d"),
+                        ("right_camera_matrix", "(3,3)d"),
+                        ("right_distortion_coefficients", "8d"),
+                        ("right_extrinsics_affine_matrix", "(4,4)d"),
+                        ("left_camera_matrix", "(3,3)d"),
+                        ("left_distortion_coefficients", "8d"),
+                        ("left_extrinsics_affine_matrix", "(4,4)d"),
+                        ("crc", "u4"),
+                    ]
+                ),
+            )
+
 
 class StatusUpdateNotifier:
     def __init__(self, device: Device, callbacks: T.List[UpdateCallback]) -> None:

src/pupil_labs/realtime_api/models.py

@@ -19,6 +19,7 @@ class APIPath(enum.Enum):
     RECORDING_STOP_AND_SAVE = "/recording:stop_and_save"
     RECORDING_CANCEL = "/recording:cancel"
     EVENT = "/event"
+    CALIBRATION = "/../calibration.bin"
 
     def full_address(
         self, address: str, port: int, protocol: str = "http", prefix: str = "/api"

src/pupil_labs/realtime_api/simple/device.py

@@ -114,6 +114,13 @@ def world_sensor(self) -> T.Optional[Sensor]:
     def gaze_sensor(self) -> T.Optional[Sensor]:
         return self._status.direct_gaze_sensor()
 
+    def get_calibration(self):
+        async def _get_calibration():
+            async with _DeviceAsync.convert_from(self) as control:
+                return await control.get_calibration()
+
+        return asyncio.run(_get_calibration())
+
     def recording_start(self) -> str:
         """Wraps :py:meth:`pupil_labs.realtime_api.device.Device.recording_start`