Merge pull request #91 from pupil-labs/camera_controls

Exposed endpoints for camera control

Author: domstoppable

examples/async/device_camera_controls.py

@@ -0,0 +1,53 @@
+import asyncio
+import logging
+
+from rich import print  # noqa: A004
+
+from pupil_labs.realtime_api import Device, DeviceError, Network, receive_video_frames
+
+
+async def main():
+    async with Network() as network:
+        dev_info = await network.wait_for_new_device(timeout_seconds=5)
+
+    if dev_info is None:
+        print("No device could be found! Abort")
+        return
+    else:
+        print(f"Connecting to {dev_info.addresses[0]}:{dev_info.port}")
+
+    async with Device.from_discovered_device(dev_info) as device:
+        # Initiate video stream before querying camera state to ensure camera is active
+        # You could instead simply open the scene video preview in the companion app
+        status = await device.get_status()
+        sensor_world = status.direct_world_sensor()
+        frames_itr = receive_video_frames(sensor_world.url)
+        await anext(frames_itr)
+
+        state = None
+        try:
+            state = await device.get_camera_state()
+            print("Current state:", state)
+        except DeviceError as err:
+            print(err)
+            print("Open the scene video preview in the companion app")
+
+        try:
+            await device.set_camera_state(
+                ae_mode="auto",
+                man_exp=50,
+                gain=50,
+                brightness=0,
+                contrast=70,
+                gamma=300,
+                validate_with_state=state,
+            )
+        except DeviceError as err:
+            print(err)
+
+        await frames_itr.aclose()
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level="DEBUG")
+    asyncio.run(main())

examples/simple/camera_controls.py

@@ -0,0 +1,94 @@
+import time
+
+import cv2
+from rich import print  # noqa: A004
+
+from pupil_labs.realtime_api.simple import discover_one_device
+
+
+def set_state_1(device, camera_state):
+    print("Updating camera to state 1")
+    device.set_camera_state(
+        ae_mode="manual",
+        man_exp=500,
+        gain=64,
+        brightness=0,
+        contrast=32,
+        gamma=300,
+        validate_with_state=camera_state,
+    )
+
+
+def set_state_2(device, camera_state):
+    print("Updating camera to state 2")
+    device.set_camera_state(
+        ae_mode="manual",
+        man_exp=1000,
+        gain=75,
+        brightness=-20,
+        contrast=50,
+        gamma=200,
+        validate_with_state=camera_state,
+    )
+
+
+def main():
+    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)
+
+    # Initiate video stream before querying camera state to ensure camera is active
+    # You could instead simply open the scene video preview in the companion app
+    device.receive_scene_video_frame()
+
+    print("Retrieving camera state...")
+    camera_state = device.get_camera_state()
+    print(f"Current camera state: {camera_state}")
+
+    set_state_1(device, camera_state)
+    last_state = 1
+    last_tick_time = time.time()
+    while True:
+        bgr_pixels, frame_datetime = device.receive_scene_video_frame()
+        draw_time(bgr_pixels, frame_datetime)
+        cv2.imshow("Scene Camera - Press ESC to quit", bgr_pixels)
+
+        if time.time() - last_tick_time > 4:
+            if last_state == 1:
+                set_state_2(device, camera_state)
+                last_state = 2
+            else:
+                set_state_1(device, camera_state)
+                last_state = 1
+            last_tick_time = time.time()
+
+        if cv2.waitKey(1) & 0xFF == 27:
+            break
+
+    device.close()
+
+
+def draw_time(frame, timestamp):
+    frame_txt_font_name = cv2.FONT_HERSHEY_SIMPLEX
+    frame_txt_font_scale = 1.0
+    frame_txt_thickness = 1
+
+    # first line: frame index
+    frame_txt = str(timestamp)
+
+    cv2.putText(
+        frame,
+        frame_txt,
+        (20, 50),
+        frame_txt_font_name,
+        frame_txt_font_scale,
+        (255, 255, 255),
+        thickness=frame_txt_thickness,
+        lineType=cv2.LINE_8,
+    )
+
+
+if __name__ == "__main__":
+    main()

pyproject.toml

@@ -33,6 +33,16 @@ dependencies = [
     "zeroconf>=0.146.1",
 ]
 
+[project.optional-dependencies]
+examples = [
+    "opencv-python",
+    "rich",
+    "librosa",
+    "speechrecognition",
+    "openai-whisper",
+]
+
+
 [project.urls]
 Homepage = "https://pupil-labs.github.io/pl-realtime-api"
 Documentation = "https://pupil-labs.github.io/pl-realtime-api"
@@ -69,7 +79,6 @@ dev = [
     "tomli>=2.0; python_version < '3.11'",
     "types-deprecated>=1.2.15.20250304",
 ]
-examples = ["opencv-python", "rich", "librosa", "speechrecognition", "openai-whisper"]
 
 [build-system]
 requires = ["hatchling", "uv-dynamic-versioning"]
@@ -88,7 +97,7 @@ source = ["pupil_labs"]
 
 [tool.deptry.per_rule_ignores]
 DEP001 = ["pupil_labs", "cv2", "whisper", "speech_recognition", "pyaudio", "audio_player", "aiortsp", "dpkt"]
-DEP002 = ["opencv-python"]
+DEP002 = ["opencv-python", "openai-whisper"]
 DEP003 = ["pupil_labs", "typing_extensions", "pydantic_core", "aiortsp", "setuptools"]
 DEP004 = ["cv2", "mkdocs", "rich", "librosa", "whisper", "speech_recognition", "pytest"]
 

src/pupil_labs/realtime_api/__init__.py

@@ -7,6 +7,7 @@
 
 import importlib.metadata
 
+from .camera_control import CameraState
 from .device import APIPath, Device, DeviceError, StatusUpdateNotifier
 from .discovery import Network, discover_devices
 from .streaming import (
@@ -42,6 +43,7 @@
     "APIPath",
     "AudioFrame",
     "BlinkEventData",
+    "CameraState",
     "Device",
     "DeviceError",
     "DualMonocularGazeData",

src/pupil_labs/realtime_api/camera_control.py

@@ -0,0 +1,87 @@
+from collections.abc import Sequence
+from dataclasses import dataclass, fields
+from enum import Enum
+from typing import Literal, cast
+
+from typing_extensions import NotRequired, TypedDict
+
+
+class ChangeRequestParameters(TypedDict):
+    ae_mode: NotRequired[Literal["auto", "manual"]]
+    man_exp: NotRequired[int]
+    gain: NotRequired[int]
+    brightness: NotRequired[int]
+    contrast: NotRequired[int]
+    gamma: NotRequired[int]
+    camera: Literal["world"]
+
+
+@dataclass(frozen=True)
+class ControlStateEnum:
+    current_value: str
+    allowed_values: Sequence[str]
+
+    def validate(self, value: str) -> None:
+        if value not in self.allowed_values:
+            raise ValueError(f"`{value}` not in allowed_values={self.allowed_values}")
+
+
+class ControlStateEnumResponse(TypedDict):
+    control_name: str
+    current_value: str
+    allowed_values: Sequence[str]
+    value_type: str
+
+
+@dataclass(frozen=True)
+class ControlStateInteger:
+    current_value: int
+    value_min: int
+    value_max: int
+
+    def validate(self, value: int) -> None:
+        if not (self.value_min <= value <= self.value_max):
+            raise ValueError(
+                f"`{value!r}` not in range [{self.value_min!r}, {self.value_max!r}]"
+            )
+
+
+class ControlStateIntegerResponse(TypedDict):
+    control_name: str
+    current_value: int
+    value_min: int
+    value_max: int
+    value_type: str
+
+
+class ControlStateResponseEnvelope(TypedDict):
+    message: str
+    result: ControlStateIntegerResponse | ControlStateEnumResponse
+
+
+@dataclass(frozen=True)
+class CameraState:
+    ae_mode: ControlStateEnum
+    man_exp: ControlStateInteger
+    gain: ControlStateInteger
+    brightness: ControlStateInteger
+    contrast: ControlStateInteger
+    gamma: ControlStateInteger
+
+    @classmethod
+    def state_class_by_attr(
+        cls, name: str
+    ) -> type[ControlStateEnum] | type[ControlStateInteger]:
+        return cast(
+            type[ControlStateEnum] | type[ControlStateInteger],
+            {f.name: f.type for f in fields(cls)}[name],
+        )
+
+
+class Control(Enum):
+    AUTOEXPOSURE_MODE = "ae_mode"
+    MANUAL_EXPOSURE_TIME = "man_exp"
+    BRIGHTNESS = "brightness"
+    CONTRAST = "contrast"
+    GAMMA = "gamma"
+    GAIN = "gain"