Skip to content

Asynchronous API

Device Discovery

discovery

Classes:

  • Network –

    Network discovery client for finding devices.

Functions:

Network 

Network()

Network discovery client for finding devices.

This class manages device discovery on the local network using Zeroconf/Bonjour. It maintains a list of discovered devices and provides methods to access them.

Attributes:

  • _devices (dict | None) –

    A dictionary of discovered devices, where the keys are device names and the values are DiscoveredDeviceInfo objects.

  • _new_devices (Queue) –

    A queue to hold newly discovered devices.

  • _aiozeroconf (AsyncZeroconf | None) –

    An instance of AsyncZeroconf for network discovery.

  • _aiobrowser (AsyncServiceBrowser | None) –

    An instance of AsyncServiceBrowser for browsing services on the network.

  • _open (bool) –

    A flag indicating whether the network discovery client is open.

Methods:

Source code in src/pupil_labs/realtime_api/discovery.py 
33
34
35
36
37
38
39
40
41
42
def __init__(self) -> None:
    self._devices: dict | None = {}
    self._new_devices: asyncio.Queue[DiscoveredDeviceInfo] = asyncio.Queue()
    self._aiozeroconf: AsyncZeroconf | None = AsyncZeroconf()
    self._aiobrowser: AsyncServiceBrowser | None = AsyncServiceBrowser(
        self._aiozeroconf.zeroconf,
        "_http._tcp.local.",
        handlers=[self._handle_service_change],
    )
    self._open: bool = True

devices property 

devices: tuple[DiscoveredDeviceInfo, ...]

Return a tuple of discovered devices.

close async 

close() -> None

Close all network resources.

This method stops the Zeroconf browser, closes connections, and clears the device list.

Source code in src/pupil_labs/realtime_api/discovery.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
async def close(self) -> None:
    """Close all network resources.

    This method stops the Zeroconf browser, closes connections, and clears
    the device list.
    """
    if self._open:
        await self._aiobrowser.async_cancel() if self._aiobrowser else None
        await self._aiozeroconf.async_close() if self._aiozeroconf else None
        if self._devices:
            self._devices.clear()
            self._devices = None
        while not self._new_devices.empty():
            self._new_devices.get_nowait()
        self._aiobrowser = None
        self._aiozeroconf = None
        self._open = False

wait_for_new_device async 

wait_for_new_device(timeout_seconds: float | None = None) -> DiscoveredDeviceInfo | None

Wait for a new device to be discovered.

Parameters:

  • timeout_seconds (float | None, default: None ) –

    Maximum time to wait for a new device. If None, wait indefinitely.

Returns:

  • DiscoveredDeviceInfo | None –

    Optional[DiscoveredDeviceInfo]: The newly discovered device, or None if the timeout was reached.

Source code in src/pupil_labs/realtime_api/discovery.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
async def wait_for_new_device(
    self, timeout_seconds: float | None = None
) -> DiscoveredDeviceInfo | None:
    """Wait for a new device to be discovered.

    Args:
        timeout_seconds: Maximum time to wait for a new device.
            If None, wait indefinitely.

    Returns:
        Optional[DiscoveredDeviceInfo]: The newly discovered device,
            or None if the timeout was reached.

    """
    try:
        return await asyncio.wait_for(self._new_devices.get(), timeout_seconds)
    except asyncio.TimeoutError:
        return None

discover_devices async 

discover_devices(timeout_seconds: float | None = None) -> AsyncIterator[DiscoveredDeviceInfo]

Use Bonjour to find devices in the local network that serve the Realtime API.

This function creates a temporary network discovery client and yields discovered devices as they are found.

Parameters:

  • timeout_seconds (float | None, default: None ) –

    Stop after timeout_seconds. If None, run discovery forever.

Yields:

Example 
async for device in discover_devices(timeout_seconds=10.0):
    print(f"Found device: {device.name} at {device.addresses[0]}:{device.port}")
Source code in src/pupil_labs/realtime_api/discovery.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
async def discover_devices(
    timeout_seconds: float | None = None,
) -> AsyncIterator[DiscoveredDeviceInfo]:
    """Use Bonjour to find devices in the local network that serve the Realtime API.

    This function creates a temporary network discovery client and yields
    discovered devices as they are found.

    Args:
        timeout_seconds: Stop after ``timeout_seconds``. If ``None``, run discovery
            forever.

    Yields:
        DiscoveredDeviceInfo: Information about discovered devices.

    Example:
        ```python
        async for device in discover_devices(timeout_seconds=10.0):
            print(f"Found device: {device.name} at {device.addresses[0]}:{device.port}")
        ```

    """
    async with Network() as network:
        while True:
            if timeout_seconds is not None and timeout_seconds <= 0.0:
                return
            t0 = time.perf_counter()
            device = await network.wait_for_new_device(timeout_seconds)
            if device is None:
                return  # timeout reached
            else:
                yield device
            if timeout_seconds is not None:
                timeout_seconds -= time.perf_counter() - t0

is_valid_service_name 

is_valid_service_name(name: str) -> bool

Check if the service name is valid for Realtime API

Source code in src/pupil_labs/realtime_api/discovery.py 
210
211
212
def is_valid_service_name(name: str) -> bool:
    """Check if the service name is valid for Realtime API"""
    return name.split(":")[0] == "PI monitor"

Remote Control

device

Classes:

  • Device –

    Class representing a Pupil Labs device.

  • DeviceError –

    Exception raised when a device operation fails.

  • StatusUpdateNotifier –

    Helper class for handling device status update callbacks.

Attributes:

UpdateCallback module-attribute 

UpdateCallback = UpdateCallbackSync | UpdateCallbackAsync

Type annotation for synchronous and asynchronous callbacks

UpdateCallbackAsync module-attribute 

UpdateCallbackAsync = Callable[[Component], Awaitable[None]]

Type annotation for asynchronous update callbacks

See Also

:class:~pupil_labs.realtime_api.models.Component

UpdateCallbackSync module-attribute 

UpdateCallbackSync = Callable[[Component], None]

Type annotation for synchronous update callbacks

See Also

:class:~pupil_labs.realtime_api.models.Component

Device 

Device(*args: Any, **kwargs: Any)

Bases: DeviceBase

Class representing a Pupil Labs device.

This class provides methods to interact with the device, such as starting and stopping recordings, sending events, and fetching device status. It also provides a context manager for automatically closing the device session.

Methods:

Attributes:

  • active_session (ClientSession) –

    Returns the active session, raising an error if it's None.

  • session (ClientSession | None) –

    The HTTP session used for making requests.

  • template_definition (Template | None) –

    The template definition currently selected on the device.

Source code in src/pupil_labs/realtime_api/device.py 
74
75
76
77
def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Initialize the Device class."""
    super().__init__(*args, **kwargs)
    self._create_client_session()

active_session property 

active_session: ClientSession

Returns the active session, raising an error if it's None.

session instance-attribute 

session: ClientSession | None

The HTTP session used for making requests.

template_definition class-attribute instance-attribute 

template_definition: Template | None = None

The template definition currently selected on the device.

api_url 

api_url(path: APIPath, protocol: str = 'http', prefix: str = '/api') -> str

Construct a full API URL for the given path.

Parameters:

  • path (APIPath) –

    API path to access.

  • protocol (str, default: 'http' ) –

    Protocol to use (http).

  • prefix (str, default: '/api' ) –

    API URL prefix.

Returns:

  • str –

    Complete URL for the API endpoint.

Source code in src/pupil_labs/realtime_api/base.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def api_url(
    self, path: APIPath, protocol: str = "http", prefix: str = "/api"
) -> str:
    """Construct a full API URL for the given path.

    Args:
        path: API path to access.
        protocol: Protocol to use (http).
        prefix: API URL prefix.

    Returns:
        Complete URL for the API endpoint.

    """
    return path.full_address(
        self.address, self.port, protocol=protocol, prefix=prefix
    )

close async 

close() -> None

Close the connection to the device.

Source code in src/pupil_labs/realtime_api/device.py 
336
337
338
339
async def close(self) -> None:
    """Close the connection to the device."""
    await self.active_session.close()
    self.session = None

convert_from classmethod 

convert_from(other: T) -> DeviceType

Convert another device instance to this type.

Parameters:

  • other (T) –

    Device instance to convert.

Returns:

Source code in src/pupil_labs/realtime_api/base.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
@classmethod
def convert_from(cls: type[DeviceType], other: T) -> DeviceType:
    """Convert another device instance to this type.

    Args:
        other: Device instance to convert.

    Returns:
        Converted device instance.

    """
    return cls(
        other.address,
        other.port,
        full_name=other.full_name,
        dns_name=other.dns_name,
    )

from_discovered_device classmethod 

from_discovered_device(device: DiscoveredDeviceInfo) -> DeviceType

Create a device instance from discovery information.

Parameters:

Returns:

Source code in src/pupil_labs/realtime_api/base.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
@classmethod
def from_discovered_device(
    cls: type[DeviceType], device: DiscoveredDeviceInfo
) -> DeviceType:
    """Create a device instance from discovery information.

    Args:
        device: Discovered device information.

    Returns:
        Device instance

    """
    return cls(
        device.addresses[0],
        device.port,
        full_name=device.name,
        dns_name=device.server,
    )

get_calibration async 

get_calibration() -> Calibration

Get the current cameras calibration data.

Note that Pupil Invisible and Neon are calibration free systems, this refers to the intrinsincs and extrinsics of the cameras and is only available for Neon.

Returns:

  • Calibration –

    pupil_labs.neon_recording.calib.Calibration: The calibration data.

Raises:

Source code in src/pupil_labs/realtime_api/device.py 
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
async def get_calibration(self) -> Calibration:
    """Get the current cameras calibration data.

    Note that Pupil Invisible and Neon are calibration free systems, this refers to
    the intrinsincs and extrinsics of the cameras and is only available for Neon.

    Returns:
        pupil_labs.neon_recording.calib.Calibration: The calibration data.

    Raises:
        DeviceError: If the request fails.

    """
    async with self.active_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 cast(Calibration, Calibration.from_buffer(raw_data))

get_status async 

get_status() -> Status

Get the current status of the device.

Returns:

  • Status ( Status ) –

    The current device status.

Raises:

Source code in src/pupil_labs/realtime_api/device.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
async def get_status(self) -> Status:
    """Get the current status of the device.

    Returns:
        Status: The current device status.

    Raises:
        DeviceError: If the request fails.

    """
    async with self.active_session.get(self.api_url(APIPath.STATUS)) as response:
        confirmation = await response.json()
        if response.status != 200:
            raise DeviceError(response.status, confirmation["message"])
        result = confirmation["result"]
        logger.debug(f"[{self}.get_status] Received status: {result}")
        return Status.from_dict(result)

get_template async 

get_template() -> Template

Get the template currently selected on device.

Returns:

  • Template ( Template ) –

    The currently selected template.

Raises:

Source code in src/pupil_labs/realtime_api/device.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
async def get_template(self) -> Template:
    """Get the template currently selected on device.

    Returns:
        Template: The currently selected template.

    Raises:
        DeviceError: If the template can't be fetched.

    """
    async with self.active_session.get(
        self.api_url(APIPath.TEMPLATE_DEFINITION)
    ) as response:
        confirmation = await response.json()
        if response.status != 200:
            raise DeviceError(response.status, confirmation["message"])
        result = confirmation["result"]
        logger.debug(f"[{self}.get_template_def] Received template def: {result}")
        self.template_definition = Template(**result)
        return self.template_definition

get_template_data async 

get_template_data(template_format: TemplateDataFormat = 'simple') -> Any

Get the template data entered on device.

Parameters:

  • template_format (TemplateDataFormat, default: 'simple' ) –

    Format of the returned data. - "api" returns the data as is from the api e.g., {"item_uuid": ["42"]} - "simple" returns the data parsed e.g., {"item_uuid": 42}

Returns:

  • Any –

    The template data in the requested format.

Raises:

Source code in src/pupil_labs/realtime_api/device.py 
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
async def get_template_data(
    self, template_format: TemplateDataFormat = "simple"
) -> Any:
    """Get the template data entered on device.

    Args:
        template_format (TemplateDataFormat): Format of the returned data.
            - "api" returns the data as is from the api e.g., {"item_uuid": ["42"]}
            - "simple" returns the data parsed e.g., {"item_uuid": 42}

    Returns:
        The template data in the requested format.

    Raises:
        DeviceError: If the template's data could not be fetched.
        AssertionError: If an invalid format is provided.

    """
    assert template_format in get_args(TemplateDataFormat), (
        f"format should be one of {TemplateDataFormat}"
    )

    async with self.active_session.get(
        self.api_url(APIPath.TEMPLATE_DATA)
    ) as response:
        confirmation = await response.json()
        if response.status != 200:
            raise DeviceError(response.status, confirmation["message"])
        result = confirmation["result"]
        logger.debug(
            f"[{self}.get_template_data] Received data's template: {result}"
        )
        if template_format == "api":
            return result
        elif template_format == "simple":
            template = await self.get_template()
            return template.convert_from_api_to_simple_format(result)

post_template_data async 

post_template_data(template_answers: dict[str, list[str]], template_format: TemplateDataFormat = 'simple') -> Any

Set the data for the currently selected template.

Parameters:

  • template_answers (dict[str, list[str]]) –

    The template data to send.

  • template_format (TemplateDataFormat, default: 'simple' ) –

    Format of the input data. - "api" accepts the data as in realtime api format e.g., {"item_uuid": ["42"]} - "simple" accepts the data in parsed format e.g., {"item_uuid": 42}

Returns:

  • Any –

    The result of the operation.

Raises:

Source code in src/pupil_labs/realtime_api/device.py 
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
async def post_template_data(
    self,
    template_answers: dict[str, list[str]],
    template_format: TemplateDataFormat = "simple",
) -> Any:
    """Set the data for the currently selected template.

    Args:
        template_answers: The template data to send.
        template_format (TemplateDataFormat): Format of the input data.
            - "api" accepts the data as in realtime api format e.g.,
                {"item_uuid": ["42"]}
            - "simple" accepts the data in parsed format e.g., {"item_uuid": 42}

    Returns:
        The result of the operation.

    Raises:
        DeviceError: If the data can not be sent.
        ValueError: If invalid data type.
        AssertionError: If an invalid format is provided.

    """
    assert template_format in get_args(TemplateDataFormat), (
        f"format should be one of {TemplateDataFormat}"
    )

    self.template_definition = await self.get_template()

    if template_format == "simple":
        template_answers = (
            self.template_definition.convert_from_simple_to_api_format(
                template_answers
            )
        )

    pre_populated_data = await self.get_template_data(template_format="api")
    errors = self.template_definition.validate_answers(
        pre_populated_data | template_answers, template_format="api"
    )
    if errors:
        raise ValueError(errors)

    # workaround for issue with api as it fails when passing in an empty list
    # ie. it wants [""] instead of []
    template_answers = {
        key: value or [""] for key, value in template_answers.items()
    }

    async with self.active_session.post(
        self.api_url(APIPath.TEMPLATE_DATA), json=template_answers
    ) as response:
        confirmation = await response.json()
        if response.status != 200:
            raise DeviceError(response.status, confirmation["message"])
        result = confirmation["result"]
        logger.debug(f"[{self}.get_template_data] Send data's template: {result}")
        return result

recording_cancel async 

recording_cancel() -> None

Cancel the current recording without saving it.

Raises:

  • DeviceError –

    If the recording could not be cancelled. Possible reasons include: - Recording not running

Source code in src/pupil_labs/realtime_api/device.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
async def recording_cancel(self) -> None:
    """Cancel the current recording without saving it.

    Raises:
        DeviceError: If the recording could not be cancelled.
            Possible reasons include:
            - Recording not running

    """
    async with self.active_session.post(
        self.api_url(APIPath.RECORDING_CANCEL)
    ) as response:
        confirmation = await response.json()
        logger.debug(f"[{self}.stop_recording] Received response: {confirmation}")
        if response.status != 200:
            raise DeviceError(response.status, confirmation["message"])

recording_start async 

recording_start() -> str

Start a recording on the device.

Returns:

  • str ( str ) –

    ID of the started recording.

Raises:

  • DeviceError –

    If recording could not be started. Possible reasons include: - Recording already running - Template has required fields - Low battery - Low storage - No wearer selected - No workspace selected - Setup bottom sheets not completed

Source code in src/pupil_labs/realtime_api/device.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
async def recording_start(self) -> str:
    """Start a recording on the device.

    Returns:
        str: ID of the started recording.

    Raises:
        DeviceError: If recording could not be started. Possible reasons include:
            - Recording already running
            - Template has required fields
            - Low battery
            - Low storage
            - No wearer selected
            - No workspace selected
            - Setup bottom sheets not completed

    """
    async with self.active_session.post(
        self.api_url(APIPath.RECORDING_START)
    ) as response:
        confirmation = await response.json()
        logger.debug(f"[{self}.start_recording] Received response: {confirmation}")
        if response.status != 200:
            raise DeviceError(response.status, confirmation["message"])
        return cast(str, confirmation["result"]["id"])

recording_stop_and_save async 

recording_stop_and_save() -> None

Stop and save the current recording.

Raises:

  • DeviceError –

    If recording could not be stopped. Possible reasons include: - Recording not running - Template has required fields

Source code in src/pupil_labs/realtime_api/device.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
async def recording_stop_and_save(self) -> None:
    """Stop and save the current recording.

    Raises:
        DeviceError: If recording could not be stopped. Possible reasons include:
            - Recording not running
            - Template has required fields

    """
    async with self.active_session.post(
        self.api_url(APIPath.RECORDING_STOP_AND_SAVE)
    ) as response:
        confirmation = await response.json()
        logger.debug(f"[{self}.stop_recording] Received response: {confirmation}")
        if response.status != 200:
            raise DeviceError(response.status, confirmation["message"])

send_event async 

send_event(event_name: str, event_timestamp_unix_ns: int | None = None) -> Event

Send an event to the device.

Parameters:

  • event_name (str) –

    Name of the event.

  • event_timestamp_unix_ns (int | None, default: None ) –

    Optional timestamp in unix nanoseconds. If None, the current time will be used.

Returns:

  • Event ( Event ) –

    The created event.

Raises:

Source code in src/pupil_labs/realtime_api/device.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
async def send_event(
    self, event_name: str, event_timestamp_unix_ns: int | None = None
) -> Event:
    """Send an event to the device.

    Args:
        event_name: Name of the event.
        event_timestamp_unix_ns: Optional timestamp in unix nanoseconds.
            If None, the current time will be used.

    Returns:
        Event: The created event.

    Raises:
        DeviceError: If sending the event fails.

    """
    event: dict[str, Any] = {"name": event_name}
    if event_timestamp_unix_ns is not None:
        event["timestamp"] = event_timestamp_unix_ns

    async with self.active_session.post(
        self.api_url(APIPath.EVENT), json=event
    ) as response:
        confirmation = await response.json()
        logger.debug(f"[{self}.send_event] Received response: {confirmation}")
        if response.status != 200:
            raise DeviceError(response.status, confirmation["message"])
        confirmation["result"]["name"] = (
            event_name  # As the API does not return the name yet
        )
        return Event.from_dict(confirmation["result"])

status_updates async 

status_updates() -> AsyncIterator[Component]

Stream status updates from the device.

Yields:

Auto-reconnect, see: https://websockets.readthedocs.io/en/stable/reference/asyncio/client.html#websockets.asyncio.client.connect

Source code in src/pupil_labs/realtime_api/device.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
async def status_updates(self) -> AsyncIterator[Component]:
    """Stream status updates from the device.

    Yields:
        Component: Status update components as they arrive.

    Auto-reconnect, see:
        https://websockets.readthedocs.io/en/stable/reference/asyncio/client.html#websockets.asyncio.client.connect

    """
    websocket_status_endpoint = self.api_url(APIPath.STATUS, protocol="ws")
    async for websocket in websockets.connect(websocket_status_endpoint):
        try:
            async for message_raw in websocket:
                message_json = json.loads(message_raw)
                try:
                    component = parse_component(message_json)
                except UnknownComponentError:
                    logger.warning(f"Dropping unknown component: {component}")
                    continue
                yield component
        except websockets.ConnectionClosed:
            logger.debug("Websocket connection closed. Reconnecting...")
            continue
        except asyncio.CancelledError:
            logger.debug("status_updates() cancelled")
            break

DeviceError

Bases: Exception

Exception raised when a device operation fails.

StatusUpdateNotifier 

StatusUpdateNotifier(device: Device, callbacks: list[UpdateCallback])

Helper class for handling device status update callbacks.

This class manages the streaming of status updates from a device and dispatches them to registered callbacks.

Attributes:

  • _auto_update_task (Task | None) –

    Task for the update loop.

  • _device (Device) –

    The device to get updates from.

  • _callbacks (list[UpdateCallback]) –

    List of callbacks to invoke.

Methods:

Source code in src/pupil_labs/realtime_api/device.py 
408
409
410
411
def __init__(self, device: Device, callbacks: list[UpdateCallback]) -> None:
    self._auto_update_task: asyncio.Task | None = None
    self._device = device
    self._callbacks = callbacks

receive_updates_start async 

receive_updates_start() -> None

Start receiving status updates.

This method starts the background task that receives updates and dispatches them to registered callbacks.

Source code in src/pupil_labs/realtime_api/device.py 
413
414
415
416
417
418
419
420
421
422
async def receive_updates_start(self) -> None:
    """Start receiving status updates.

    This method starts the background task that receives updates
    and dispatches them to registered callbacks.
    """
    if self._auto_update_task is not None:
        logger.debug("Auto-update already started!")
        return
    self._auto_update_task = asyncio.create_task(self._auto_update())

receive_updates_stop async 

receive_updates_stop() -> None

Stop receiving status updates.

This method cancels the background task that receives updates.

Source code in src/pupil_labs/realtime_api/device.py 
424
425
426
427
428
429
430
431
432
433
434
435
436
async def receive_updates_stop(self) -> None:
    """Stop receiving status updates.

    This method cancels the background task that receives updates.
    """
    if self._auto_update_task is None:
        logger.debug("Auto-update is not running!")
        return
    self._auto_update_task.cancel()
    with contextlib.suppress(asyncio.CancelledError):
        # wait for the task to be cancelled
        await self._auto_update_task
    self._auto_update_task = None

Streaming

Gaze Data

gaze

Classes:

  • DualMonocularGazeData –

    Experimental class for dual monocular gaze data.

  • EyestateEyelidGazeData –

    Gaze data with additional eyelid state information.

  • EyestateGazeData –

    Gaze data with additional eye state information.

  • GazeData –

    Basic gaze data with position, timestamp and indicator of glasses worn status.

  • Point –

    A point in 2D space, represented by x and y coordinates.

  • RTSPGazeStreamer –

    Stream and parse gaze data from an RTSP source.

Functions:

Attributes:

GazeDataType module-attribute 

GazeDataType = GazeData | DualMonocularGazeData | EyestateGazeData | EyestateEyelidGazeData

Type alias for various gaze data types.

DualMonocularGazeData

Bases: NamedTuple

Experimental class for dual monocular gaze data.

Contains separate gaze points for left and right eyes.

Methods:

  • from_raw –

    Create a DualMonocularGazeData instance from raw data.

Attributes:

datetime property 

datetime: datetime

Get the timestamp as a datetime object.

left instance-attribute 

left: Point

Gaze point for the left eye.

right instance-attribute 

right: Point

Gaze point for the right eye.

timestamp_unix_ns property 

timestamp_unix_ns: int

Get the timestamp in nanoseconds since Unix epoch.

timestamp_unix_seconds instance-attribute 

timestamp_unix_seconds: float

Timestamp in seconds since Unix epoch.

worn instance-attribute 

worn: bool

Whether the glasses are being worn.

from_raw classmethod 

from_raw(data: RTSPData) -> DualMonocularGazeData

Create a DualMonocularGazeData instance from raw data.

Parameters:

  • data (RTSPData) –

    The raw data received from the RTSP stream.

Returns:

  • DualMonocularGazeData ( DualMonocularGazeData ) –

    An instance of DualMonocularGazeData with the parsed values.

Source code in src/pupil_labs/realtime_api/streaming/gaze.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
@classmethod
def from_raw(cls, data: RTSPData) -> "DualMonocularGazeData":
    """Create a DualMonocularGazeData instance from raw data.

    Args:
        data (RTSPData): The raw data received from the RTSP stream.

    Returns:
        DualMonocularGazeData: An instance of DualMonocularGazeData with the parsed
            values.

    """
    x1, y1, worn, x2, y2 = struct.unpack("!ffBff", data.raw)
    return cls(
        Point(x1, y1), Point(x2, y2), worn == 255, data.timestamp_unix_seconds
    )

EyestateEyelidGazeData

Bases: NamedTuple

Gaze data with additional eyelid state information.

Contains gaze point, pupil diameter, eyeball center coordinates, optical axis coordinates, as well as eyelid angles and aperture for both left and right eyes.

Methods:

  • from_raw –

    Create an EyestateEyelidGazeData instance from raw data.

Attributes:

datetime property 

datetime: datetime

Get the timestamp as a datetime object.

eyeball_center_left_x instance-attribute 

eyeball_center_left_x: float

X coordinate of the eyeball center for the left eye.

eyeball_center_left_y instance-attribute 

eyeball_center_left_y: float

Y coordinate of the eyeball center for the left eye.

eyeball_center_left_z instance-attribute 

eyeball_center_left_z: float

Z coordinate of the eyeball center for the left eye.

eyeball_center_right_x instance-attribute 

eyeball_center_right_x: float

X coordinate of the eyeball center for the right eye.

eyeball_center_right_y instance-attribute 

eyeball_center_right_y: float

Y coordinate of the eyeball center for the right eye.

eyeball_center_right_z instance-attribute 

eyeball_center_right_z: float

Z coordinate of the eyeball center for the right eye.

eyelid_angle_bottom_left instance-attribute 

eyelid_angle_bottom_left: float

Angle of the bottom eyelid for the left eye(rad).

eyelid_angle_bottom_right instance-attribute 

eyelid_angle_bottom_right: float

Angle of the bottom eyelid for the right eye (rad).

eyelid_angle_top_left instance-attribute 

eyelid_angle_top_left: float

Angle of the top eyelid for the left eye(rad).

eyelid_angle_top_right instance-attribute 

eyelid_angle_top_right: float

Angle of the top eyelid for the right eye (rad).

eyelid_aperture_left instance-attribute 

eyelid_aperture_left: float

Aperture of the eyelid for the left eye (mm).

eyelid_aperture_right instance-attribute 

eyelid_aperture_right: float

Aperture of the eyelid for the right eye (mm).

optical_axis_left_x instance-attribute 

optical_axis_left_x: float

X coordinate of the optical axis for the left eye.

optical_axis_left_y instance-attribute 

optical_axis_left_y: float

Y coordinate of the optical axis for the left eye.

optical_axis_left_z instance-attribute 

optical_axis_left_z: float

Z coordinate of the optical axis for the left eye.

optical_axis_right_x instance-attribute 

optical_axis_right_x: float

X coordinate of the optical axis for the right eye.

optical_axis_right_y instance-attribute 

optical_axis_right_y: float

Y coordinate of the optical axis for the right eye.

optical_axis_right_z instance-attribute 

optical_axis_right_z: float

Z coordinate of the optical axis for the right eye.

pupil_diameter_left instance-attribute 

pupil_diameter_left: float

Pupil diameter for the left eye.

pupil_diameter_right instance-attribute 

pupil_diameter_right: float

Pupil diameter for the right eye.

timestamp_unix_ns property 

timestamp_unix_ns: int

Get the timestamp in nanoseconds since Unix epoch.

timestamp_unix_seconds instance-attribute 

timestamp_unix_seconds: float

Timestamp in seconds since Unix epoch.

worn instance-attribute 

worn: bool

Whether the glasses are being worn.

x instance-attribute 

x: float

X coordinate of the gaze point.

y instance-attribute 

y: float

Y coordinate of the gaze point.

from_raw classmethod 

from_raw(data: RTSPData) -> EyestateEyelidGazeData

Create an EyestateEyelidGazeData instance from raw data.

Parameters:

  • data (RTSPData) –

    The raw data received from the RTSP stream.

Returns:

  • EyestateEyelidGazeData ( EyestateEyelidGazeData ) –

    An instance of EyestateEyelidGazeData with the parsed values.

Source code in src/pupil_labs/realtime_api/streaming/gaze.py 
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
@classmethod
def from_raw(cls, data: RTSPData) -> "EyestateEyelidGazeData":
    """Create an EyestateEyelidGazeData instance from raw data.

    Args:
        data (RTSPData): The raw data received from the RTSP stream.

    Returns:
        EyestateEyelidGazeData: An instance of EyestateEyelidGazeData with the
            parsed values.

    """
    (
        x,
        y,
        worn,
        pupil_diameter_left,
        eyeball_center_left_x,
        eyeball_center_left_y,
        eyeball_center_left_z,
        optical_axis_left_x,
        optical_axis_left_y,
        optical_axis_left_z,
        pupil_diam_right,
        eyeball_center_right_x,
        eyeball_center_right_y,
        eyeball_center_right_z,
        optical_axis_right_x,
        optical_axis_right_y,
        optical_axis_right_z,
        eyelid_angle_top_left,
        eyelid_angle_bottom_left,
        eyelid_aperture_left,
        eyelid_angle_top_right,
        eyelid_angle_bottom_right,
        eyelid_aperture_right,
    ) = struct.unpack("!ffBffffffffffffffffffff", data.raw)
    return cls(
        x,
        y,
        worn == 255,
        pupil_diameter_left,
        eyeball_center_left_x,
        eyeball_center_left_y,
        eyeball_center_left_z,
        optical_axis_left_x,
        optical_axis_left_y,
        optical_axis_left_z,
        pupil_diam_right,
        eyeball_center_right_x,
        eyeball_center_right_y,
        eyeball_center_right_z,
        optical_axis_right_x,
        optical_axis_right_y,
        optical_axis_right_z,
        eyelid_angle_top_left,
        eyelid_angle_bottom_left,
        eyelid_aperture_left,
        eyelid_angle_top_right,
        eyelid_angle_bottom_right,
        eyelid_aperture_right,
        data.timestamp_unix_seconds,
    )

EyestateGazeData

Bases: NamedTuple

Gaze data with additional eye state information.

Contains gaze point, pupil diameter, eyeball center coordinates, and optical axis coordinates for both left and right eyes.

Methods:

  • from_raw –

    Create an EyestateGazeData instance from raw data.

Attributes:

datetime property 

datetime: datetime

Get the timestamp as a datetime object.

eyeball_center_left_x instance-attribute 

eyeball_center_left_x: float

X coordinate of the eyeball center for the left eye.

eyeball_center_left_y instance-attribute 

eyeball_center_left_y: float

Y coordinate of the eyeball center for the left eye.

eyeball_center_left_z instance-attribute 

eyeball_center_left_z: float

Z coordinate of the eyeball center for the left eye.

eyeball_center_right_x instance-attribute 

eyeball_center_right_x: float

X coordinate of the eyeball center for the right eye.

eyeball_center_right_y instance-attribute 

eyeball_center_right_y: float

Y coordinate of the eyeball center for the right eye.

eyeball_center_right_z instance-attribute 

eyeball_center_right_z: float

Z coordinate of the eyeball center for the right eye.

optical_axis_left_x instance-attribute 

optical_axis_left_x: float

X coordinate of the optical axis for the left eye.

optical_axis_left_y instance-attribute 

optical_axis_left_y: float

Y coordinate of the optical axis for the left eye.

optical_axis_left_z instance-attribute 

optical_axis_left_z: float

Z coordinate of the optical axis for the left eye.

optical_axis_right_x instance-attribute 

optical_axis_right_x: float

X coordinate of the optical axis for the right eye.

optical_axis_right_y instance-attribute 

optical_axis_right_y: float

Y coordinate of the optical axis for the right eye.

optical_axis_right_z instance-attribute 

optical_axis_right_z: float

Z coordinate of the optical axis for the right eye.

pupil_diameter_left instance-attribute 

pupil_diameter_left: float

Pupil diameter for the left eye.

pupil_diameter_right instance-attribute 

pupil_diameter_right: float

Pupil diameter for the right eye.

timestamp_unix_ns property 

timestamp_unix_ns: int

Get the timestamp in nanoseconds since Unix epoch.

timestamp_unix_seconds instance-attribute 

timestamp_unix_seconds: float

Timestamp in seconds since Unix epoch.

worn instance-attribute 

worn: bool

Whether the glasses are being worn.

x instance-attribute 

x: float

X coordinate of the gaze point.

y instance-attribute 

y: float

Y coordinate of the gaze point.

from_raw classmethod 

from_raw(data: RTSPData) -> EyestateGazeData

Create an EyestateGazeData instance from raw data.

Parameters:

  • data (RTSPData) –

    The raw data received from the RTSP stream.

Returns:

  • EyestateGazeData ( EyestateGazeData ) –

    An instance of EyestateGazeData with the parsed values.

Source code in src/pupil_labs/realtime_api/streaming/gaze.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
@classmethod
def from_raw(cls, data: RTSPData) -> "EyestateGazeData":
    """Create an EyestateGazeData instance from raw data.

    Args:
        data (RTSPData): The raw data received from the RTSP stream.

    Returns:
        EyestateGazeData: An instance of EyestateGazeData with the parsed values.

    """
    (
        x,
        y,
        worn,
        pupil_diameter_left,
        eyeball_center_left_x,
        eyeball_center_left_y,
        eyeball_center_left_z,
        optical_axis_left_x,
        optical_axis_left_y,
        optical_axis_left_z,
        pupil_diam_right,
        eyeball_center_right_x,
        eyeball_center_right_y,
        eyeball_center_right_z,
        optical_axis_right_x,
        optical_axis_right_y,
        optical_axis_right_z,
    ) = struct.unpack("!ffBffffffffffffff", data.raw)
    return cls(
        x,
        y,
        worn == 255,
        pupil_diameter_left,
        eyeball_center_left_x,
        eyeball_center_left_y,
        eyeball_center_left_z,
        optical_axis_left_x,
        optical_axis_left_y,
        optical_axis_left_z,
        pupil_diam_right,
        eyeball_center_right_x,
        eyeball_center_right_y,
        eyeball_center_right_z,
        optical_axis_right_x,
        optical_axis_right_y,
        optical_axis_right_z,
        data.timestamp_unix_seconds,
    )

GazeData

Bases: NamedTuple

Basic gaze data with position, timestamp and indicator of glasses worn status.

Represents the 2D gaze point on the scene camera coordinates with a timestamp in nanoseconds unix epoch and an indicator of whether the glasses are being worn.

Methods:

  • from_raw –

    Create a GazeData instance from raw data.

Attributes:

datetime property 

datetime: datetime

Get the timestamp as a datetime object.

timestamp_unix_ns property 

timestamp_unix_ns: int

Get the timestamp in nanoseconds since Unix epoch.

timestamp_unix_seconds instance-attribute 

timestamp_unix_seconds: float

Timestamp in seconds since Unix epoch.

worn instance-attribute 

worn: bool

Whether the glasses are being worn.

x instance-attribute 

x: float

"X coordinate of the gaze point

y instance-attribute 

y: float

Y coordinate of the gaze point.

from_raw classmethod 

from_raw(data: RTSPData) -> GazeData

Create a GazeData instance from raw data.

Parameters:

  • data (RTSPData) –

    The raw data received from the RTSP stream.

Returns:

  • GazeData ( GazeData ) –

    An instance of GazeData with the parsed values.

Source code in src/pupil_labs/realtime_api/streaming/gaze.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
@classmethod
def from_raw(cls, data: RTSPData) -> "GazeData":
    """Create a GazeData instance from raw data.

    Args:
        data (RTSPData): The raw data received from the RTSP stream.

    Returns:
        GazeData: An instance of GazeData with the parsed values.

    """
    x, y, worn = struct.unpack("!ffB", data.raw)
    return cls(x, y, worn == 255, data.timestamp_unix_seconds)

Point

Bases: NamedTuple

A point in 2D space, represented by x and y coordinates.

RTSPGazeStreamer 

RTSPGazeStreamer(*args: Any, **kwargs: Any)

Bases: RTSPRawStreamer

Stream and parse gaze data from an RTSP source.

This class extends RTSPRawStreamer to parse raw RTSP data into structured gaze data objects. The specific type of gaze data is determined by the length of the raw data packet.

Methods:

  • receive –

    Receive and parse gaze data from the RTSP stream.

Attributes:

  • encoding (str) –

    Get the encoding of the RTSP stream.

  • reader (_WallclockRTSPReader) –

    Get the underlying RTSP reader.

Source code in src/pupil_labs/realtime_api/streaming/base.py 
78
79
80
def __init__(self, *args: Any, **kwargs: Any) -> None:
    self._reader = _WallclockRTSPReader(*args, **kwargs)
    self._encoding = None

encoding property 

encoding: str

Get the encoding of the RTSP stream.

Returns:

  • str ( str ) –

    The encoding name in lowercase.

Raises:

reader property 

reader: _WallclockRTSPReader

Get the underlying RTSP reader.

receive async 

receive() -> AsyncIterator[GazeDataType]

Receive and parse gaze data from the RTSP stream.

Yields:

  • 9 bytes: GazeData (basic gaze position)
  • 17 bytes: DualMonocularGazeData (left and right eye positions)
  • 65 bytes: EyestateGazeData (gaze with eye state)
  • 89 bytes: EyestateEyelidGazeData (gaze with eye state and eyelid info)

Raises:

  • KeyError –

    If the data length does not match any known format.

  • Exception –

    If there is an error parsing the gaze data.

Source code in src/pupil_labs/realtime_api/streaming/gaze.py 
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
async def receive(  # type: ignore[override]
    self,
) -> AsyncIterator[GazeDataType]:
    """Receive and parse gaze data from the RTSP stream.

    Yields:
        GazeDataType

        Parsed gaze data of various types. The type of gaze data object is
        determined by the length of the raw data packet:
    - 9 bytes: GazeData (basic gaze position)
    - 17 bytes: DualMonocularGazeData (left and right eye positions)
    - 65 bytes: EyestateGazeData (gaze with eye state)
    - 89 bytes: EyestateEyelidGazeData (gaze with eye state and eyelid info)

    Raises:
        KeyError: If the data length does not match any known format.
        Exception: If there is an error parsing the gaze data.

    """
    data_class_by_raw_len = {
        9: GazeData,
        17: DualMonocularGazeData,
        65: EyestateGazeData,
        89: EyestateEyelidGazeData,
    }
    async for data in super().receive():
        try:
            cls = data_class_by_raw_len[len(data.raw)]
            yield cls.from_raw(data)  # type: ignore[attr-defined]
        except KeyError:
            logger.exception(f"Raw gaze data has unexpected length: {data}")
            raise
        except Exception:
            logger.exception(f"Unable to parse gaze data {data}")
            raise

receive_gaze_data async 

receive_gaze_data(url: str, *args: Any, **kwargs: Any) -> AsyncIterator[GazeDataType]

Receive gaze data from an RTSP stream.

This is a convenience function that creates an RTSPGazeStreamer and yields parsed gaze data.

Parameters:

  • url (str) –

    RTSP URL to connect to.

  • *args (Any, default: () ) –

    Additional positional arguments passed to RTSPGazeStreamer.

  • **kwargs (Any, default: {} ) –

    Additional keyword arguments passed to RTSPGazeStreamer.

Yields:

Source code in src/pupil_labs/realtime_api/streaming/gaze.py 
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
async def receive_gaze_data(
    url: str, *args: Any, **kwargs: Any
) -> AsyncIterator[GazeDataType]:
    """Receive gaze data from an RTSP stream.

    This is a convenience function that creates an RTSPGazeStreamer and yields
    parsed gaze data.

    Args:
        url: RTSP URL to connect to.
        *args: Additional positional arguments passed to RTSPGazeStreamer.
        **kwargs: Additional keyword arguments passed to RTSPGazeStreamer.

    Yields:
        GazeDataType: Parsed gaze data of various types.
        The type of gaze data object is determined by the length of the raw data packet:
        - 9 bytes: GazeData (basic gaze position)
        - 17 bytes: DualMonocularGazeData (left and right eye positions)
        - 65 bytes: EyestateGazeData (gaze with eye state)
        - 89 bytes: EyestateEyelidGazeData (gaze with eye state and eyelid info)

    """
    async with RTSPGazeStreamer(url, *args, **kwargs) as streamer:
        async for datum in streamer.receive():
            yield cast(GazeDataType, datum)

IMU Data

imu

Classes:

  • Data3D –

    3D data point with x, y, z coordinates.

  • IMUData –

    Data from the Inertial Measurement Unit (IMU).

  • Quaternion –

    Quaternion data point with x, y, z, w coordinates.

  • RTSPImuStreamer –

    Stream and parse IMU data from an RTSP source.

Functions:

Data3D

Bases: NamedTuple

3D data point with x, y, z coordinates.

IMUData

Bases: NamedTuple

Data from the Inertial Measurement Unit (IMU).

Contains gyroscope, accelerometer, and rotation data from the IMU sensor.

Attributes:

accel_data instance-attribute 

accel_data: Data3D

Accelerometer data in m/s².

datetime property 

datetime: datetime

Get timestamp as a datetime object.

gyro_data instance-attribute 

gyro_data: Data3D

Gyroscope data in deg/s.

quaternion instance-attribute 

quaternion: Quaternion

Rotation represented as a quaternion.

timestamp_unix_nanoseconds property 

timestamp_unix_nanoseconds: int

Get timestamp in nanoseconds since Unix epoch.

Deprecated

This class property is deprecated and will be removed in future versions. Use timestamp_unix_ns() instead.

timestamp_unix_ns property 

timestamp_unix_ns: int

Get timestamp in nanoseconds since Unix epoch.

timestamp_unix_seconds instance-attribute 

timestamp_unix_seconds: float

Timestamp in seconds since Unix epoch.

Quaternion

Bases: NamedTuple

Quaternion data point with x, y, z, w coordinates.

RTSPImuStreamer 

RTSPImuStreamer(*args: Any, **kwargs: Any)

Bases: RTSPRawStreamer

Stream and parse IMU data from an RTSP source.

This class extends RTSPRawStreamer to parse raw RTSP data into structured IMU data objects.

Methods:

  • receive –

    Receive and parse IMU data from the RTSP stream.

Attributes:

  • encoding (str) –

    Get the encoding of the RTSP stream.

  • reader (_WallclockRTSPReader) –

    Get the underlying RTSP reader.

Source code in src/pupil_labs/realtime_api/streaming/base.py 
78
79
80
def __init__(self, *args: Any, **kwargs: Any) -> None:
    self._reader = _WallclockRTSPReader(*args, **kwargs)
    self._encoding = None

encoding property 

encoding: str

Get the encoding of the RTSP stream.

Returns:

  • str ( str ) –

    The encoding name in lowercase.

Raises:

reader property 

reader: _WallclockRTSPReader

Get the underlying RTSP reader.

receive async 

receive() -> AsyncIterator[IMUData]

Receive and parse IMU data from the RTSP stream.

This method parses the raw binary data into IMUData objects by using the protobuf deserializer.

Yields:

Raises:

  • Exception –

    If there is an error parsing the IMU data.

Source code in src/pupil_labs/realtime_api/streaming/imu.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
async def receive(  # type: ignore[override]
    self,
) -> AsyncIterator[IMUData]:
    """Receive and parse IMU data from the RTSP stream.

    This method parses the raw binary data into IMUData objects by using
    the protobuf deserializer.

    Yields:
        IMUData: Parsed IMU data.

    Raises:
        Exception: If there is an error parsing the IMU data.

    """
    async for data in super().receive():
        try:
            imu_packet = ImuPacket()
            imu_packet.ParseFromString(data.raw)
            imu_data = IMUPacket_to_IMUData(imu_packet)
            yield imu_data
        except Exception:
            logger.exception(f"Unable to parse imu data {data}")
            raise

IMUPacket_to_IMUData 

IMUPacket_to_IMUData(imu_packet: ImuPacket) -> IMUData

Create an IMUData instance from a protobuf IMU packet.

Parameters:

  • imu_packet (ImuPacket) –

    Protobuf IMU packet.

Returns:

  • IMUData ( IMUData ) –

    Converted IMU data.

Source code in src/pupil_labs/realtime_api/streaming/imu.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def IMUPacket_to_IMUData(imu_packet: "ImuPacket") -> IMUData:  # type: ignore[no-any-unimported]
    """Create an IMUData instance from a protobuf IMU packet.

    Args:
        imu_packet: Protobuf IMU packet.

    Returns:
        IMUData: Converted IMU data.

    """
    gyro_data = Data3D(
        x=imu_packet.gyroData.x,
        y=imu_packet.gyroData.y,
        z=imu_packet.gyroData.z,
    )
    accel_data = Data3D(
        x=imu_packet.accelData.x,
        y=imu_packet.accelData.y,
        z=imu_packet.accelData.z,
    )
    quaternion = Quaternion(
        x=imu_packet.rotVecData.x,
        y=imu_packet.rotVecData.y,
        z=imu_packet.rotVecData.z,
        w=imu_packet.rotVecData.w,
    )
    imu_data = IMUData(
        gyro_data=gyro_data,
        accel_data=accel_data,
        quaternion=quaternion,
        timestamp_unix_seconds=imu_packet.tsNs / 1e9,
    )
    return imu_data

receive_imu_data async 

receive_imu_data(url: str, *args: Any, **kwargs: Any) -> AsyncIterator[IMUData]

Receive IMU data from a given RTSP URL.

Parameters:

  • url (str) –

    RTSP URL to connect to.

  • *args (Any, default: () ) –

    Additional arguments for the streamer.

  • **kwargs (Any, default: {} ) –

    Additional keyword arguments for the streamer.

Yields:

Source code in src/pupil_labs/realtime_api/streaming/imu.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
async def receive_imu_data(
    url: str, *args: Any, **kwargs: Any
) -> AsyncIterator[IMUData]:
    """Receive IMU data from a given RTSP URL.

    Args:
        url: RTSP URL to connect to.
        *args: Additional arguments for the streamer.
        **kwargs: Additional keyword arguments for the streamer.

    Yields:
        IMUData: Parsed IMU data from the RTSP stream.

    """
    async with RTSPImuStreamer(url, *args, **kwargs) as streamer:
        assert isinstance(streamer, RTSPImuStreamer)
        async for datum in streamer.receive():
            yield cast(IMUData, datum)

Eye Events

eye_events

Classes:

Functions:

BlinkEventData

Bases: NamedTuple

Data for a blink event.

Represents a detected blink event with timing information.

Methods:

  • from_raw –

    Create a BlinkEventData instance from raw RTSP data.

Attributes:

datetime property 

datetime: datetime

Get the timestamp as a datetime object.

end_time_ns instance-attribute 

end_time_ns: int

End time of the blink in nanoseconds.

event_type instance-attribute 

event_type: int

Type of event (4 -> blink events).

rtp_ts_unix_seconds instance-attribute 

rtp_ts_unix_seconds: float

RTP timestamp in seconds since Unix epoch.

start_time_ns instance-attribute 

start_time_ns: int

Start time of the blink in nanoseconds.

timestamp_unix_ns property 

timestamp_unix_ns: int

Get the timestamp in nanoseconds since the Unix epoch.

from_raw classmethod 

from_raw(data: RTSPData) -> BlinkEventData

Create a BlinkEventData instance from raw RTSP data.

Source code in src/pupil_labs/realtime_api/streaming/eye_events.py 
27
28
29
30
31
32
33
34
35
@classmethod
def from_raw(cls, data: RTSPData) -> "BlinkEventData":
    """Create a BlinkEventData instance from raw RTSP data."""
    (
        event_type,
        start_time_ns,
        end_time_ns,
    ) = struct.unpack("!iqq", data.raw)
    return cls(event_type, start_time_ns, end_time_ns, data.timestamp_unix_seconds)

FixationEventData

Bases: NamedTuple

Data for a fixation or saccade event.

Represents a completed fixation or saccade event with detailed information.

Methods:

  • from_raw –

    Create a FixationEventData instance from raw RTSP data.

Attributes:

amplitude_angle_deg instance-attribute 

amplitude_angle_deg: float

Amplitude in degrees.

amplitude_pixels instance-attribute 

amplitude_pixels: float

Amplitude in pixels.

datetime property 

datetime: datetime

Get the timestamp as a datetime object.

end_gaze_x instance-attribute 

end_gaze_x: float

End gaze x-coordinate in pixels.

end_gaze_y instance-attribute 

end_gaze_y: float

End gaze y-coordinate in pixels.

end_time_ns instance-attribute 

end_time_ns: int

End time of the event in nanoseconds.

event_type instance-attribute 

event_type: int

Type of event (0 for saccade, 1 for fixation).

max_velocity instance-attribute 

max_velocity: float

Maximum velocity in pixels per degree.

mean_gaze_x instance-attribute 

mean_gaze_x: float

Mean gaze x-coordinate in pixels.

mean_gaze_y instance-attribute 

mean_gaze_y: float

Mean gaze y-coordinate in pixels.

mean_velocity instance-attribute 

mean_velocity: float

Mean velocity in pixels per degree.

rtp_ts_unix_seconds instance-attribute 

rtp_ts_unix_seconds: float

RTP timestamp in seconds since Unix epoch.

start_gaze_x instance-attribute 

start_gaze_x: float

Start gaze x-coordinate in pixels.

start_gaze_y instance-attribute 

start_gaze_y: float

Start gaze y-coordinate in pixels.

start_time_ns instance-attribute 

start_time_ns: int

Start time of the event in nanoseconds.

timestamp_unix_ns property 

timestamp_unix_ns: int

Get the timestamp in nanoseconds since the Unix epoch.

from_raw classmethod 

from_raw(data: RTSPData) -> FixationEventData

Create a FixationEventData instance from raw RTSP data.

Source code in src/pupil_labs/realtime_api/streaming/eye_events.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@classmethod
def from_raw(cls, data: RTSPData) -> "FixationEventData":
    """Create a FixationEventData instance from raw RTSP data."""
    (
        event_type,
        start_time_ns,
        end_time_ns,
        start_gaze_x,
        start_gaze_y,
        end_gaze_x,
        end_gaze_y,
        mean_gaze_x,
        mean_gaze_y,
        amplitude_pixels,
        amplitude_angle_deg,
        mean_velocity,
        max_velocity,
    ) = struct.unpack("!iqqffffffffff", data.raw)
    return cls(
        event_type,
        start_time_ns,
        end_time_ns,
        start_gaze_x,
        start_gaze_y,
        end_gaze_x,
        end_gaze_y,
        mean_gaze_x,
        mean_gaze_y,
        amplitude_pixels,
        amplitude_angle_deg,
        mean_velocity,
        max_velocity,
        data.timestamp_unix_seconds,
    )

FixationOnsetEventData

Bases: NamedTuple

Data for a fixation or saccade onset event.

Represents the beginning of a fixation or saccade event.

Attributes:

datetime property 

datetime: datetime

Get the timestamp as a datetime object.

event_type instance-attribute 

event_type: int

Type of event (2 for saccade onset, 3 for fixation onset).

rtp_ts_unix_seconds instance-attribute 

rtp_ts_unix_seconds: float

RTP timestamp in seconds since Unix epoch.

start_time_ns instance-attribute 

start_time_ns: int

Start time of the event in nanoseconds.

timestamp_unix_ns property 

timestamp_unix_ns: int

Get the timestamp in nanoseconds since the Unix epoch.

RTSPEyeEventStreamer 

RTSPEyeEventStreamer(*args: Any, **kwargs: Any)

Bases: RTSPRawStreamer

Stream and parse eye events from an RTSP source.

This class extends RTSPRawStreamer to parse raw RTSP data into structured eye event data objects.

Methods:

  • receive –

    Receive and parse eye events from the RTSP stream.

Attributes:

  • encoding (str) –

    Get the encoding of the RTSP stream.

  • reader (_WallclockRTSPReader) –

    Get the underlying RTSP reader.

Source code in src/pupil_labs/realtime_api/streaming/base.py 
78
79
80
def __init__(self, *args: Any, **kwargs: Any) -> None:
    self._reader = _WallclockRTSPReader(*args, **kwargs)
    self._encoding = None

encoding property 

encoding: str

Get the encoding of the RTSP stream.

Returns:

  • str ( str ) –

    The encoding name in lowercase.

Raises:

reader property 

reader: _WallclockRTSPReader

Get the underlying RTSP reader.

receive async 

receive() -> AsyncIterator[FixationEventData | FixationOnsetEventData | BlinkEventData]

Receive and parse eye events from the RTSP stream.

Yields:

Raises:

  • KeyError –

    If the event type is not recognized.

  • Exception –

    If there is an error parsing the event data.

Source code in src/pupil_labs/realtime_api/streaming/eye_events.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
async def receive(  # type: ignore[override]
    self,
) -> AsyncIterator[FixationEventData | FixationOnsetEventData | BlinkEventData]:
    """Receive and parse eye events from the RTSP stream.

    Yields:
        FixationEventData | FixationOnsetEventData | BlinkEventData: Parsed eye
            event data.

    Raises:
        KeyError: If the event type is not recognized.
        Exception: If there is an error parsing the event data.

    """
    data_class_by_type = {
        0: FixationEventData,
        1: FixationEventData,
        2: FixationOnsetEventData,
        3: FixationOnsetEventData,
        4: BlinkEventData,
        5: None,  # KEEPALIVE MSG, SKIP
    }
    async for data in super().receive():
        try:
            event_type = struct.unpack_from("!i", data.raw)[0]
            cls = data_class_by_type[event_type]
            if cls is not None:
                yield cls.from_raw(data)  # type: ignore[attr-defined]
        except KeyError:
            logger.exception(f"Raw eye event data has unexpected type: {data}")
            raise
        except Exception:
            logger.exception(f"Unable to parse eye event data {data}")
            raise

receive_eye_events_data async 

receive_eye_events_data(url: str, *args: Any, **kwargs: Any) -> AsyncIterator[FixationEventData | FixationOnsetEventData | BlinkEventData]

Receive eye events data from an RTSP stream.

This is a convenience function that creates an RTSPEyeEventStreamer and yields parsed eye event data.

Parameters:

  • url (str) –

    RTSP URL to connect to.

  • *args (Any, default: () ) –

    Additional positional arguments passed to RTSPEyeEventStreamer.

  • **kwargs (Any, default: {} ) –

    Additional keyword arguments passed to RTSPEyeEventStreamer.

Yields:

Source code in src/pupil_labs/realtime_api/streaming/eye_events.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
async def receive_eye_events_data(
    url: str, *args: Any, **kwargs: Any
) -> AsyncIterator[FixationEventData | FixationOnsetEventData | BlinkEventData]:
    """Receive eye events data from an RTSP stream.

    This is a convenience function that creates an RTSPEyeEventStreamer and yields
    parsed eye event data.

    Args:
        url: RTSP URL to connect to.
        *args: Additional positional arguments passed to RTSPEyeEventStreamer.
        **kwargs: Additional keyword arguments passed to RTSPEyeEventStreamer.

    Yields:
        FixationEventData: Parsed fixation event data.

    """
    async with RTSPEyeEventStreamer(url, *args, **kwargs) as streamer:
        async for datum in streamer.receive():
            yield cast(
                FixationEventData | FixationOnsetEventData | BlinkEventData, datum
            )

Scene Video

video

Classes:

Functions:

Attributes:

  • BGRBuffer –

    Type annotation for raw BGR image buffers of the scene camera

BGRBuffer module-attribute 

BGRBuffer = NDArray[uint8]

Type annotation for raw BGR image buffers of the scene camera

RTSPVideoFrameStreamer 

RTSPVideoFrameStreamer(*args: Any, **kwargs: Any)

Bases: RTSPRawStreamer

Stream and decode video frames from an RTSP source.

This class extends RTSPRawStreamer to parse raw RTSP data into video frames using the pupil_labs.video and pyav library for decoding.

Attributes:

  • _sprop_parameter_set_payloads (list[ByteString] | None) –

    Cached SPS/PPS parameters for the H.264 codec.

Methods:

  • receive –

    Receive and decode video frames from the RTSP stream.

Source code in src/pupil_labs/realtime_api/streaming/video.py 
93
94
95
def __init__(self, *args: Any, **kwargs: Any) -> None:
    super().__init__(*args, **kwargs)
    self._sprop_parameter_set_payloads: list[ByteString] | None = None

encoding property 

encoding: str

Get the encoding of the RTSP stream.

Returns:

  • str ( str ) –

    The encoding name in lowercase.

Raises:

reader property 

reader: _WallclockRTSPReader

Get the underlying RTSP reader.

sprop_parameter_set_payloads property 

sprop_parameter_set_payloads: list[ByteString] | None

Get the SPS/PPS parameter set payloads for the H.264 codec.

These parameters are extracted from the SDP data and are required for initializing the H.264 decoder.

Returns:

  • list[ByteString] | None –

    list[ByteString]: List of parameter set payloads.

Raises:

receive async 

receive() -> AsyncIterator[VideoFrame]

Receive and decode video frames from the RTSP stream.

Source code in src/pupil_labs/realtime_api/streaming/video.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
async def receive(self) -> AsyncIterator[VideoFrame]:  # type: ignore[override]
    """Receive and decode video frames from the RTSP stream."""
    codec = None
    frame_timestamp = None

    async for data in super().receive():
        if not codec:
            try:
                codec = av.CodecContext.create(self.encoding, "r")
                if self.sprop_parameter_set_payloads:
                    for param in self.sprop_parameter_set_payloads:
                        codec.parse(param)
            except SDPDataNotAvailableError as err:
                logger.debug(
                    f"Session description protocol data not available yet: {err}"
                )
                continue
            except av.codec.codec.UnknownCodecError:
                logger.exception(
                    "Unknown codec error: "
                    "Please try clearing the app's storage and cache."
                )
                raise
        # if pkt is the start of a new fragmented frame, parse will return a packet
        # containing the data from the previous fragments
        for packet in codec.parse(extract_payload_from_nal_unit(data.raw)):
            # use timestamp of previous packets
            for av_frame in codec.decode(packet):  # type: ignore[attr-defined]
                if frame_timestamp is None:
                    raise ValueError("No timestamp available for the video frame.")
                yield VideoFrame(av_frame, frame_timestamp)

        frame_timestamp = data.timestamp_unix_seconds

VideoFrame

Bases: NamedTuple

A video frame with timestamp information.

This class represents a video frame from the scene camera with associated timestamp information. The Class inherits VideoFrame from py.av library.

Methods:

  • bgr_buffer –

    Convert the video frame to a BGR buffer.

  • to_ndarray –

    Convert the video frame to a NumPy array.

Attributes:

av_frame instance-attribute 

av_frame: VideoFrame

The video frame.

datetime property 

datetime: datetime

Get timestamp as a datetime object.

timestamp_unix_ns property 

timestamp_unix_ns: int

Get timestamp in nanoseconds since Unix epoch.

timestamp_unix_seconds instance-attribute 

timestamp_unix_seconds: float

Timestamp in seconds since Unix epoch.

bgr_buffer 

bgr_buffer() -> BGRBuffer

Convert the video frame to a BGR buffer.

This method converts the video frame to a BGR buffer, which is a NumPy array with the shape (height, width, 3) and dtype uint8. The BGR format is commonly used in computer vision applications.

Returns:

  • BGRBuffer ( BGRBuffer ) –

    The BGR buffer as a NumPy array.

Source code in src/pupil_labs/realtime_api/streaming/video.py 
46
47
48
49
50
51
52
53
54
55
56
57
def bgr_buffer(self) -> BGRBuffer:
    """Convert the video frame to a BGR buffer.

    This method converts the video frame to a BGR buffer, which is a
    NumPy array with the shape (height, width, 3) and dtype uint8.
    The BGR format is commonly used in computer vision applications.

    Returns:
        BGRBuffer: The BGR buffer as a NumPy array.

    """
    return self.to_ndarray(format="bgr24")

to_ndarray 

to_ndarray(*args: Any, **kwargs: Any) -> NDArray

Convert the video frame to a NumPy array.

Source code in src/pupil_labs/realtime_api/streaming/video.py 
42
43
44
def to_ndarray(self, *args: Any, **kwargs: Any) -> npt.NDArray:
    """Convert the video frame to a NumPy array."""
    return self.av_frame.to_ndarray(*args, **kwargs)

receive_video_frames async 

receive_video_frames(url: str, *args: Any, **kwargs: Any) -> AsyncIterator[VideoFrame]

Receive video frames from an RTSP stream.

This is a convenience function that creates an RTSPVideoFrameStreamer and yields video frames.

Parameters:

  • url (str) –

    RTSP URL to connect to.

  • *args (Any, default: () ) –

    Additional positional arguments passed to RTSPVideoFrameStreamer.

  • **kwargs (Any, default: {} ) –

    Additional keyword arguments passed to RTSPVideoFrameStreamer.

Yields:

Source code in src/pupil_labs/realtime_api/streaming/video.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
async def receive_video_frames(
    url: str, *args: Any, **kwargs: Any
) -> AsyncIterator[VideoFrame]:
    """Receive video frames from an RTSP stream.

    This is a convenience function that creates an RTSPVideoFrameStreamer and yields
    video frames.

    Args:
        url: RTSP URL to connect to.
        *args: Additional positional arguments passed to RTSPVideoFrameStreamer.
        **kwargs: Additional keyword arguments passed to RTSPVideoFrameStreamer.

    Yields:
        VideoFrame: Parsed video frames.

    """
    async with RTSPVideoFrameStreamer(url, *args, **kwargs) as streamer:
        async for datum in streamer.receive():
            yield cast(VideoFrame, datum)

extract_payload_from_nal_unit 

extract_payload_from_nal_unit(unit: ByteString) -> ByteString

Extract and process payload from a Network Abstraction Layer (NAL) unit.

This function extracts the payload from a NAL unit, handling various formats: - Prepends NAL unit start code to payload if necessary - Handles fragmented units (of type FU-A)

The implementation follows RFC 3984 specifications for H.264 NAL units.

Parameters:

  • unit (ByteString) –

    The NAL unit as a ByteString.

Returns:

  • ByteString ( ByteString ) –

    The processed payload, potentially with start code prepended.

Raises:

  • ValueError –

    If the first bit is not zero (forbidden_zero_bit).

References

Inspired by https://github.com/runtheops/rtsp-rtp/blob/master/transport/primitives/nal_unit.py Rewritten due to license incompatibility. See RFC 3984 (https://www.ietf.org/rfc/rfc3984.txt) for detailed NAL unit specifications.

Source code in src/pupil_labs/realtime_api/streaming/nal_unit.py 
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def extract_payload_from_nal_unit(unit: ByteString) -> ByteString:
    """Extract and process payload from a Network Abstraction Layer (NAL) unit.

    This function extracts the payload from a NAL unit, handling various formats:
    - Prepends NAL unit start code to payload if necessary
    - Handles fragmented units (of type FU-A)

    The implementation follows RFC 3984 specifications for H.264 NAL units.

    Args:
        unit: The NAL unit as a ByteString.

    Returns:
        ByteString: The processed payload, potentially with start code prepended.

    Raises:
        ValueError: If the first bit is not zero (forbidden_zero_bit).

    References:
        Inspired by https://github.com/runtheops/rtsp-rtp/blob/master/transport/primitives/nal_unit.py
        Rewritten due to license incompatibility.
        See RFC 3984 (https://www.ietf.org/rfc/rfc3984.txt) for detailed NAL unit
        specifications.

    """
    start_code = b"\x00\x00\x00\x01"
    offset = 0
    # slice to keep ByteString type; indexing would return int in native byte order
    first_byte_raw = unit[:1]
    # ensure network order for conversion to uint8
    first_byte = struct.unpack("!B", first_byte_raw)[0]
    is_first_bit_one = (first_byte & 0b10000000) != 0
    if is_first_bit_one:
        # See section 1.3 of https://www.ietf.org/rfc/rfc3984.txt
        raise ValueError("First bit must be zero (forbidden_zero_bit)")

    nal_type = first_byte & 0b00011111
    if nal_type == 28:
        # Fragmentation unit FU-A
        # https://www.ietf.org/rfc/rfc3984.txt
        # Section 5.8.
        fu_header_raw = unit[1:2]  # get second byte while retaining ByteString type
        fu_header = struct.unpack("!B", fu_header_raw)[0]
        offset = 2  # skip first two bytes

        is_fu_start_bit_one = (fu_header & 0b10000000) != 0
        if is_fu_start_bit_one:
            # reconstruct header of a non-fragmented NAL unit
            first_byte_bits_1_to_3 = first_byte & 0b11100000
            # NAL type of non-fragmented NAL unit:
            fu_header_bits_4_to_8 = fu_header & 0b00011111
            reconstructed_header = first_byte_bits_1_to_3 + fu_header_bits_4_to_8
            start_code += bytes((reconstructed_header,))  # convert int to ByteString
        else:
            # do not prepend start code to payload since we are in the middle of a
            # fragmented unit
            start_code = b""

    return start_code + unit[offset:]

Raw RTSP Data

base

Classes:

Functions:

RTSPData

Bases: NamedTuple

Container for RTSP data with timestamp information.

Attributes:

datetime property 

datetime: datetime

Get the timestamp as a datetime object.

raw instance-attribute 

raw: ByteString

Raw binary data received from the RTSP stream.

timestamp_unix_ns property 

timestamp_unix_ns: int

Get the timestamp in nanoseconds since the Unix epoch.

timestamp_unix_seconds instance-attribute 

timestamp_unix_seconds: float

Timestamp in seconds since the Unix epoch from RTCP SR packets.

RTSPRawStreamer 

RTSPRawStreamer(*args: Any, **kwargs: Any)

Stream raw data from an RTSP source.

This class connects to an RTSP source and provides access to the raw data with timestamps synchronized to the device's clock.

All constructor arguments are forwarded to the underlying aiortsp.rtsp.reader.RTSPReader.

Methods:

  • receive –

    Receive raw data from the RTSP stream.

Attributes:

  • encoding (str) –

    Get the encoding of the RTSP stream.

  • reader (_WallclockRTSPReader) –

    Get the underlying RTSP reader.

Source code in src/pupil_labs/realtime_api/streaming/base.py 
78
79
80
def __init__(self, *args: Any, **kwargs: Any) -> None:
    self._reader = _WallclockRTSPReader(*args, **kwargs)
    self._encoding = None

encoding property 

encoding: str

Get the encoding of the RTSP stream.

Returns:

  • str ( str ) –

    The encoding name in lowercase.

Raises:

reader property 

reader: _WallclockRTSPReader

Get the underlying RTSP reader.

receive async 

receive() -> AsyncIterator[RTSPData]

Receive raw data from the RTSP stream.

This method yields RTSPData objects containing the raw data and corresponding timestamps.

Source code in src/pupil_labs/realtime_api/streaming/base.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
95
async def receive(self) -> AsyncIterator[RTSPData]:
    """Receive raw data from the RTSP stream.

    This method yields RTSPData objects containing the raw data and
    corresponding timestamps.
    """
    async for pkt in self.reader.iter_packets():
        try:
            timestamp_seconds = self.reader.absolute_timestamp_from_packet(pkt)
        except _UnknownClockoffsetError:
            # The absolute timestamp is not known yet.
            # Waiting for the first RTCP SR packet...
            continue
        yield RTSPData(pkt.data, timestamp_seconds)

SDPDataNotAvailableError

Bases: Exception

Exception raised when SDP data is not available or incomplete.

This exception is raised when attempting to access SDP (Session Description Protocol) data that is not yet available or is missing required fields.

receive_raw_rtsp_data async 

receive_raw_rtsp_data(url: str, *args: Any, **kwargs: Any) -> AsyncIterator[RTSPData]

Receive raw data from an RTSP stream.

This is a convenience function that creates an RTSPRawStreamer and yields timestamped data packets.

Parameters:

  • url (str) –

    RTSP URL to connect to.

  • *args (Any, default: () ) –

    Additional positional arguments passed to RTSPRawStreamer.

  • **kwargs (Any, default: {} ) –

    Additional keyword arguments passed to RTSPRawStreamer.

Yields:

Source code in src/pupil_labs/realtime_api/streaming/base.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
async def receive_raw_rtsp_data(
    url: str, *args: Any, **kwargs: Any
) -> AsyncIterator[RTSPData]:
    """Receive raw data from an RTSP stream.

    This is a convenience function that creates an RTSPRawStreamer and yields
    timestamped data packets.

    Args:
        url: RTSP URL to connect to.
        *args: Additional positional arguments passed to RTSPRawStreamer.
        **kwargs: Additional keyword arguments passed to RTSPRawStreamer.

    Yields:
        RTSPData: Timestamped RTSP data packets.

    """
    async with RTSPRawStreamer(url, *args, **kwargs) as streamer:
        async for datum in streamer.receive():
            yield cast(RTSPData, datum)

Time Echo Protocol

time_echo

Manual time offset estimation via the Pupil Labs Time Echo protocol.

The Realtime Network API host device timestamps its data with nanoseconds since the Unix epoch(January 1, 1970, 00:00:00 UTC). This clock is kept in sync by the operating system through NTP Network Time Protocol. For some use cases, this sync is not good enough. For more accurate time syncs, the Time Echo protocol allows the estimation of the direct offset between the host's and the client's clocks.

The Time Echo protocol works in the following way:

  1. The API host (Neon / Pupil Invisible Companion app) opens a TCP server at an specific port
  2. The client connects to the host address and port
  3. The client sends its current time (t1) in milliseconds as an uint64 in network byte order to the host
  4. The host responds with the time echo, two uint64 values in network byte order
  5. The first value is equal to the sent client time (t1)
  6. The second value corresponds to the host's time in milliseconds (tH)
  7. The client calculates the duration of steps 3 and 4 (roundtrip time) by measuring the client time before sending the request (t1) and after receiving the echo (t2)
  8. The protocol assumes that the transport duration is symmetric. It will assume that tH was measured at the same time as the midpoint betwee t1 and t2.

  9. To calculate the offset between the host's and client's clock, we subtract tH from the client's midpoint (t1 + t2) / 2::

    offset_ms = ((t1 + t2) / 2) - tH

  10. This measurement can be repeated multiple times to make the time offset estimation more robust.

To convert client to host time, subtract the offset::

host_time_ms = client_time_ms() - offset_ms

This is particularly helpful to accurately timestamp local events, e.g. a stimulus presentation.

To convert host to client time, add the offset::

client_time_ms = host_time_ms() + offset_ms

This is particularly helpful to convert the received data into the client's time domain.

Classes:

  • Estimate –

    Provides easy access to statistics over a collection of measurements.

  • TimeEcho –

    Measurement of a single time echo.

  • TimeEchoEstimates –

    Provides estimates for the roundtrip duration and time offsets.

  • TimeOffsetEstimator –

    Estimates the time offset between PC and Companion using the Time Echo protocol.

Functions:

  • time_ms –

    Return milliseconds since Unix epoch_ (January 1, 1970, 00:00:00 UTC)

Attributes:

TimeFunction module-attribute 

TimeFunction = Callable[[], int]

Returns time in milliseconds

Estimate 

Estimate(measurements: Iterable[int])

Provides easy access to statistics over a collection of measurements.

This class calculates descriptive statistics (mean, standard deviation, median) over a collection of measurements.

Attributes:

  • measurements (tuple[int]) –

    The raw measurements.

  • mean (float) –

    Mean value of the measurements.

  • std (float) –

    Standard deviation of the measurements.

  • median (float) –

    Median value of the measurements.

Source code in src/pupil_labs/realtime_api/time_echo.py 
88
89
90
91
92
def __init__(self, measurements: Iterable[int]) -> None:
    self.measurements = tuple(measurements)
    self._mean = statistics.mean(self.measurements)
    self._std = statistics.stdev(self.measurements)
    self._median = statistics.median(self.measurements)

mean property 

mean: float

Mean value of the measurements.

median property 

median: float

Median value of the measurements.

std property 

std: float

Standard deviation of the measurements.

TimeEcho

Bases: NamedTuple

Measurement of a single time echo.

Attributes:

roundtrip_duration_ms instance-attribute 

roundtrip_duration_ms: int

Round trip duration of the time echo, in milliseconds.

time_offset_ms instance-attribute 

time_offset_ms: int

Time offset between host and client, in milliseconds.

TimeEchoEstimates

Bases: NamedTuple

Provides estimates for the roundtrip duration and time offsets.

Attributes:

roundtrip_duration_ms instance-attribute 

roundtrip_duration_ms: Estimate

Statistics for roundtrip durations.

time_offset_ms instance-attribute 

time_offset_ms: Estimate

Statistics for time offsets.

TimeOffsetEstimator 

TimeOffsetEstimator(address: str, port: int)

Estimates the time offset between PC and Companion using the Time Echo protocol.

This class implements the Time Echo protocol to estimate the time offset between the client and host clocks.

Attributes:

  • address (str) –

    Host address.

  • port (int) –

    Host port for the Time Echo protocol.

Methods:

  • estimate –

    Estimate the time offset between client and host.

  • request_time_echo –

    Request a time echo, measure the roundtrip time and estimate the time offset.

Source code in src/pupil_labs/realtime_api/time_echo.py 
145
146
147
def __init__(self, address: str, port: int) -> None:
    self.address = address
    self.port = port

estimate async 

estimate(number_of_measurements: int = 100, sleep_between_measurements_seconds: float | None = None, time_fn_ms: TimeFunction = time_ms) -> TimeEchoEstimates | None

Estimate the time offset between client and host.

Parameters:

  • number_of_measurements (int, default: 100 ) –

    Number of measurements to take.

  • sleep_between_measurements_seconds (float | None, default: None ) –

    Optional sleep time between measurements.

  • time_fn_ms (TimeFunction, default: time_ms ) –

    Function that returns the current time in milliseconds.

Returns:

  • TimeEchoEstimates ( TimeEchoEstimates | None ) –

    Statistics for roundtrip durations and time offsets, or None if estimation failed.

Source code in src/pupil_labs/realtime_api/time_echo.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
async def estimate(
    self,
    number_of_measurements: int = 100,
    sleep_between_measurements_seconds: float | None = None,
    time_fn_ms: TimeFunction = time_ms,
) -> TimeEchoEstimates | None:
    """Estimate the time offset between client and host.

    Args:
        number_of_measurements: Number of measurements to take.
        sleep_between_measurements_seconds: Optional sleep time between
            measurements.
        time_fn_ms: Function that returns the current time in milliseconds.

    Returns:
        TimeEchoEstimates: Statistics for roundtrip durations and time offsets,
            or None if estimation failed.

    """
    measurements = collections.defaultdict(list)

    try:
        logger.debug(f"Connecting to {self.address}:{self.port}...")
        reader, writer = await asyncio.open_connection(self.address, self.port)
    except ConnectionError:
        logger.exception("Could not connect to Time Echo server")
        return None

    try:
        rt, offset = await self.request_time_echo(time_fn_ms, reader, writer)
        logger.debug(
            f"Dropping first measurement (roundtrip: {rt} ms, offset: {offset} ms)"
        )
        logger.info(f"Measuring {number_of_measurements} times...")
        for _ in range(number_of_measurements):
            try:
                rt, offset = await self.request_time_echo(
                    time_fn_ms, reader, writer
                )
                measurements["roundtrip"].append(rt)
                measurements["offset"].append(offset)
                if sleep_between_measurements_seconds is not None:
                    await asyncio.sleep(sleep_between_measurements_seconds)
            except ValueError as err:
                logger.warning(err)
    finally:
        writer.close()
        await writer.wait_closed()
        logger.debug(f"Connection closed {writer.is_closing()}")

    try:
        estimates = TimeEchoEstimates(
            roundtrip_duration_ms=Estimate(measurements["roundtrip"]),
            time_offset_ms=Estimate(measurements["offset"]),
        )
    except statistics.StatisticsError:
        logger.exception("Not enough valid samples were collected")
        return None

    return estimates

request_time_echo async staticmethod 

request_time_echo(time_fn_ms: TimeFunction, reader: StreamReader, writer: StreamWriter) -> TimeEcho

Request a time echo, measure the roundtrip time and estimate the time offset.

Parameters:

  • time_fn_ms (TimeFunction) –

    Function that returns the current time in milliseconds.

  • reader (StreamReader) –

    Stream reader for receiving responses.

  • writer (StreamWriter) –

    Stream writer for sending requests.

Returns:

  • TimeEcho ( TimeEcho ) –

    Roundtrip duration and time offset.

Raises:

Source code in src/pupil_labs/realtime_api/time_echo.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
@staticmethod
async def request_time_echo(
    time_fn_ms: TimeFunction,
    reader: asyncio.StreamReader,
    writer: asyncio.StreamWriter,
) -> TimeEcho:
    """Request a time echo, measure the roundtrip time and estimate the time offset.

    Args:
        time_fn_ms: Function that returns the current time in milliseconds.
        reader: Stream reader for receiving responses.
        writer: Stream writer for sending requests.

    Returns:
        TimeEcho: Roundtrip duration and time offset.

    Raises:
        ValueError: If the response is invalid.

    """
    before_ms = time_fn_ms()
    before_ms_bytes = struct.pack("!Q", before_ms)
    writer.write(before_ms_bytes)
    await writer.drain()
    validation_server_ms_bytes = await reader.read(16)
    after_ms = time_fn_ms()
    if len(validation_server_ms_bytes) != 16:
        raise ValueError(
            "Dropping invalid measurement. Expected response of length 16 "
            f"(got {len(validation_server_ms_bytes)})"
        )
    validation_ms, server_ms = struct.unpack("!QQ", validation_server_ms_bytes)
    logger.debug(
        f"Response: {validation_ms} {server_ms} ({validation_server_ms_bytes!r})"
    )
    if validation_ms != before_ms:
        raise ValueError(
            "Dropping invalid measurement. Expected validation timestamp: "
            f"{before_ms} (got {validation_ms})"
        )
    server_ts_in_client_time_ms = round((before_ms + after_ms) / 2)
    offset_ms = server_ts_in_client_time_ms - server_ms
    return TimeEcho(after_ms - before_ms, offset_ms)

time_ms 

time_ms() -> int

Return milliseconds since Unix epoch_ (January 1, 1970, 00:00:00 UTC)

Source code in src/pupil_labs/realtime_api/time_echo.py 
128
129
130
def time_ms() -> int:
    """Return milliseconds since `Unix epoch`_ (January 1, 1970, 00:00:00 UTC)"""
    return time_ns() // 1_000_000

models

Classes:

  • APIPath –

    API endpoint paths for the Realtime API.

  • ConnectionType –

    Enumeration of connection types.

  • DiscoveredDeviceInfo –

    Information about a discovered device on the network.

  • Event –

    Event information from the device.

  • Hardware –

    Information about the Hardware connected (eye tracker).

  • InvalidTemplateAnswersError –

    Exception raised when template answers fail validation.

  • NetworkDevice –

    Information about devices discovered by the host device, not the client.

  • Phone –

    Information relative to the Companion Device.

  • Recording –

    Information about a recording.

  • Sensor –

    Information about a sensor on the device.

  • SensorName –

    Enumeration of sensor types.

  • Status –

    Represents the Companion's Device full status

  • Template –

    Template Class for data collection.

  • TemplateItem –

    Individual item/ question in a Template.

  • UnknownComponentError –

    Exception raised when a component cannot be parsed.

Functions:

Attributes:

Component module-attribute 

Component = Phone | Hardware | Sensor | Recording | NetworkDevice | Event

Type annotation for :class:Status components.

ComponentRaw module-attribute 

ComponentRaw = dict[str, Any]

Type annotation for json-parsed responses from the REST and Websocket API.

TemplateDataFormat module-attribute 

TemplateDataFormat = Literal['api', 'simple']

Format specification for template data.

TemplateItemInputType module-attribute 

TemplateItemInputType = Literal['any', 'integer', 'float']

Type of input data for a template item.

TemplateItemWidgetType module-attribute 

TemplateItemWidgetType = Literal['TEXT', 'PARAGRAPH', 'RADIO_LIST', 'CHECKBOX_LIST', 'SECTION_HEADER', 'PAGE_BREAK']

Type of widget to display for a template item.

APIPath

Bases: Enum

API endpoint paths for the Realtime API.

This enum defines the various API endpoints that can be accessed through the Realtime API.

Methods:

  • full_address –

    Construct a full URL for this API endpoint.

full_address 

full_address(address: str, port: int, protocol: str = 'http', prefix: str = '/api') -> str

Construct a full URL for this API endpoint.

Source code in src/pupil_labs/realtime_api/models.py 
47
48
49
50
51
def full_address(
    self, address: str, port: int, protocol: str = "http", prefix: str = "/api"
) -> str:
    """Construct a full URL for this API endpoint."""
    return f"{protocol}://{address}:{port}" + prefix + self.value

ConnectionType

Bases: Enum

Enumeration of connection types.

DiscoveredDeviceInfo

Bases: NamedTuple

Information about a discovered device on the network.

Attributes:

  • addresses (list[str]) –

    IP addresses, e.g. ['192.168.0.2'].

  • name (str) –

    Full mDNS service name.

  • port (int) –

    Port number, e.g. 8080.

  • server (str) –

    mDNS server name. e.g. 'neon.local.' or 'pi.local.'.

addresses instance-attribute 

addresses: list[str]

IP addresses, e.g. ['192.168.0.2'].

name instance-attribute 

name: str

Full mDNS service name.

Follows 'PI monitor:<phone name>:<hardware id>._http._tcp.local.' naming pattern.

port instance-attribute 

port: int

Port number, e.g. 8080.

server instance-attribute 

server: str

mDNS server name. e.g. 'neon.local.' or 'pi.local.'.

Event

Bases: NamedTuple

Event information from the device.

Methods:

  • from_dict –

    Create an Event from a dictionary.

Attributes:

datetime property 

datetime: datetime

Get the event time as a datetime object.

Returns:

  • datetime ( datetime ) –

    Event time as a Python datetime.

name instance-attribute 

name: str | None

Name of the event.

recording_id instance-attribute 

recording_id: str | None

ID of the recording this event belongs to.

timestamp instance-attribute 

timestamp: int

Unix epoch timestamp in nanoseconds.

from_dict classmethod 

from_dict(event_dict: dict[str, Any]) -> Event

Create an Event from a dictionary.

Parameters:

  • event_dict (dict[str, Any]) –

    Dictionary containing event data.

Returns:

  • Event ( Event ) –

    New Event instance.

Source code in src/pupil_labs/realtime_api/models.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
@classmethod
def from_dict(cls, event_dict: dict[str, Any]) -> "Event":
    """Create an Event from a dictionary.

    Args:
        event_dict: Dictionary containing event data.

    Returns:
        Event: New Event instance.

    """
    return cls(
        name=event_dict.get("name"),
        recording_id=event_dict.get("recording_id"),
        timestamp=event_dict["timestamp"],
    )

Hardware

Bases: NamedTuple

Information about the Hardware connected (eye tracker).

Attributes:

glasses_serial class-attribute instance-attribute 

glasses_serial: str = 'unknown'

Serial number of the glasses. For Pupil Invisible devices.

module_serial class-attribute instance-attribute 

module_serial: str = 'unknown'

Serial number of the module. For Neon devices.

version class-attribute instance-attribute 

version: str = 'unknown'

Hardware version. 1-> Pupil Invisible 2-> Neon

world_camera_serial class-attribute instance-attribute 

world_camera_serial: str = 'unknown'

Serial number of the world camera. For Pupil Invisible devices.

InvalidTemplateAnswersError 

InvalidTemplateAnswersError(template: Template | TemplateItem, answers: dict[str, Any], errors: list[ErrorDetails])

Bases: Exception

Exception raised when template answers fail validation.

Attributes:

  • template (Template | TemplateItem) –

    Template or item that failed validation.

  • errors (list[dict]) –

    List of validation errors.

  • answers (dict) –

    The answers that failed validation.

Source code in src/pupil_labs/realtime_api/models.py 
936
937
938
939
940
941
942
943
944
def __init__(
    self,
    template: Template | TemplateItem,
    answers: dict[str, Any],
    errors: list[ErrorDetails],
) -> None:
    self.template = template
    self.errors = errors
    self.answers = answers

NetworkDevice

Bases: NamedTuple

Information about devices discovered by the host device, not the client.

This class represents device information made available via the websocket update connection by the host device (exposed via pupil_labs.realtime_api.device.Device.status_updates. Devices discovered directly by this library are represented as DiscoveredDeviceInfo and returned by discover_devices Network.

Attributes:

  • connected (bool) –

    Whether the device is connected.

  • device_id (str) –

    Unique device identifier.

  • device_name (str) –

    Human-readable device name (can be modified by the user in the Companion App

  • ip (str) –

    IP address of the device.

connected instance-attribute 

connected: bool

Whether the device is connected.

device_id instance-attribute 

device_id: str

Unique device identifier.

device_name instance-attribute 

device_name: str

Human-readable device name (can be modified by the user in the Companion App settings).

ip instance-attribute 

ip: str

IP address of the device.

Phone

Bases: NamedTuple

Information relative to the Companion Device.

Attributes:

battery_level instance-attribute 

battery_level: int

Battery percentage (0-100)

battery_state instance-attribute 

battery_state: Literal['OK', 'LOW', 'CRITICAL']

Battery state.

device_id instance-attribute 

device_id: str

Unique device identifier.

device_name instance-attribute 

device_name: str

Human-readable device name.

ip instance-attribute 

ip: str

IP address of the phone.

memory instance-attribute 

memory: int

Available memory in bytes.

memory_state instance-attribute 

memory_state: Literal['OK', 'LOW', 'CRITICAL']

Memory state.

time_echo_port class-attribute instance-attribute 

time_echo_port: int | None = None

Port for time synchronization, if available.

Recording

Bases: NamedTuple

Information about a recording.

Attributes:

action instance-attribute 

action: str

Current recording action.

id instance-attribute 

id: str

Unique recording identifier.

message instance-attribute 

message: str

Status message.

rec_duration_ns instance-attribute 

rec_duration_ns: int

Recording duration in nanoseconds.

rec_duration_seconds property 

rec_duration_seconds: float

Get the recording duration in seconds.

Sensor

Bases: NamedTuple

Information about a sensor on the device.

Attributes:

  • conn_type (str) –

    Connection type (see ConnectionType Enum).

  • connected (bool) –

    Whether the sensor is connected.

  • ip (str | None) –

    IP address of the sensor.

  • params (str | None) –

    Additional parameters.

  • port (int | None) –

    Port number.

  • protocol (str) –

    Protocol used for the connection.

  • sensor (str) –

    Sensor type (see Name Enum).

  • stream_error (bool) –

    Whether the stream errors.

  • url (str | None) –

    Get the URL for accessing this sensor's data stream.

conn_type instance-attribute 

conn_type: str

Connection type (see ConnectionType Enum).

connected class-attribute instance-attribute 

connected: bool = False

Whether the sensor is connected.

ip class-attribute instance-attribute 

ip: str | None = None

IP address of the sensor.

params class-attribute instance-attribute 

params: str | None = None

Additional parameters.

port class-attribute instance-attribute 

port: int | None = None

Port number.

protocol class-attribute instance-attribute 

protocol: str = 'rtsp'

Protocol used for the connection.

sensor instance-attribute 

sensor: str

Sensor type (see Name Enum).

stream_error class-attribute instance-attribute 

stream_error: bool = True

Whether the stream errors.

url property 

url: str | None

Get the URL for accessing this sensor's data stream.

Returns:

  • str | None –

    str | None: URL if connected, None otherwise.

SensorName

Bases: Enum

Enumeration of sensor types.

Status dataclass 

Status(phone: Phone, hardware: Hardware, sensors: list[Sensor], recording: Recording | None)

Represents the Companion's Device full status

Methods:

Attributes:

hardware instance-attribute 

hardware: Hardware

Information about glasses connected, won't be present if not connected

phone instance-attribute 

phone: Phone

Information about the connected phone. Always present.

recording instance-attribute 

recording: Recording | None

Current recording, if any.

sensors instance-attribute 

sensors: list[Sensor]

"List of sensor information.

direct_eye_events_sensor 

direct_eye_events_sensor() -> Sensor | None

Get blinks, fixations sensor.

Only available on Neon with Companion App version 2.9 or newer.

Source code in src/pupil_labs/realtime_api/models.py 
460
461
462
463
464
465
466
467
468
469
470
471
def direct_eye_events_sensor(self) -> Sensor | None:
    """Get blinks, fixations _sensor_.

    Only available on Neon with Companion App version 2.9 or newer.
    """
    return next(
        self.matching_sensors(SensorName.EYE_EVENTS, ConnectionType.DIRECT),
        Sensor(
            sensor=SensorName.EYE_EVENTS.value,
            conn_type=ConnectionType.DIRECT.value,
        ),
    )

direct_eyes_sensor 

direct_eyes_sensor() -> Sensor | None

Get the eye camera sensor with direct connection. Only available on Neon.

Source code in src/pupil_labs/realtime_api/models.py 
453
454
455
456
457
458
def direct_eyes_sensor(self) -> Sensor | None:
    """Get the eye camera sensor with direct connection. Only available on Neon."""
    return next(
        self.matching_sensors(SensorName.EYES, ConnectionType.DIRECT),
        Sensor(sensor=SensorName.EYES.value, conn_type=ConnectionType.DIRECT.value),
    )

direct_gaze_sensor 

direct_gaze_sensor() -> Sensor | None

Get the gaze sensor with direct connection.

Source code in src/pupil_labs/realtime_api/models.py 
439
440
441
442
443
444
def direct_gaze_sensor(self) -> Sensor | None:
    """Get the gaze sensor with direct connection."""
    return next(
        self.matching_sensors(SensorName.GAZE, ConnectionType.DIRECT),
        Sensor(sensor=SensorName.GAZE.value, conn_type=ConnectionType.DIRECT.value),
    )

direct_imu_sensor 

direct_imu_sensor() -> Sensor | None

Get the IMU sensor with direct connection.

Source code in src/pupil_labs/realtime_api/models.py 
446
447
448
449
450
451
def direct_imu_sensor(self) -> Sensor | None:
    """Get the IMU sensor with direct connection."""
    return next(
        self.matching_sensors(SensorName.IMU, ConnectionType.DIRECT),
        Sensor(sensor=SensorName.IMU.value, conn_type=ConnectionType.DIRECT.value),
    )

direct_world_sensor 

direct_world_sensor() -> Sensor | None

Get the scene camera sensor with direct connection.

Note

Pupil Invisible devices, the world camera can be detached

Source code in src/pupil_labs/realtime_api/models.py 
425
426
427
428
429
430
431
432
433
434
435
436
437
def direct_world_sensor(self) -> Sensor | None:
    """Get the scene camera sensor with direct connection.

    Note:
        Pupil Invisible devices, the world camera can be detached

    """
    return next(
        self.matching_sensors(SensorName.WORLD, ConnectionType.DIRECT),
        Sensor(
            sensor=SensorName.WORLD.value, conn_type=ConnectionType.DIRECT.value
        ),
    )

from_dict classmethod 

from_dict(status_json_result: list[ComponentRaw]) -> Status

Create a Status from a list of raw components.

Parameters:

  • status_json_result (list[ComponentRaw]) –

    List of raw component dictionaries.

Returns:

  • Status ( Status ) –

    New Status instance.

Source code in src/pupil_labs/realtime_api/models.py 
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
@classmethod
def from_dict(cls, status_json_result: list[ComponentRaw]) -> "Status":
    """Create a Status from a list of raw components.

    Args:
        status_json_result: List of raw component dictionaries.

    Returns:
        Status: New Status instance.

    """
    phone = None
    recording = None
    hardware = Hardware()
    sensors = []
    for dct in status_json_result:
        try:
            component = parse_component(dct)
        except UnknownComponentError:
            logger.warning(f"Dropping unknown component: {dct}")
            continue
        if isinstance(component, Phone):
            phone = component
        elif isinstance(component, Hardware):
            hardware = component
        elif isinstance(component, Sensor):
            sensors.append(component)
        elif isinstance(component, Recording):
            recording = component
        elif isinstance(component, NetworkDevice):
            pass  # no need to handle NetworkDevice updates here
        else:
            logger.warning(f"Unknown model class: {type(component).__name__}")
    sensors.sort(key=lambda s: (not s.connected, s.conn_type, s.sensor))
    if not phone:
        raise ValueError("Status data must include a 'Phone' component.")
    return cls(phone, hardware, sensors, recording)

matching_sensors 

matching_sensors(name: SensorName, connection: ConnectionType) -> Iterator[Sensor]

Find sensors matching specified criteria.

Parameters:

  • name (SensorName) –

    Sensor name to match, or ANY to match any name.

  • connection (ConnectionType) –

    Connection type to match, or ANY to match any type.

Yields:

  • Sensor ( Sensor ) –

    Sensors matching the criteria.

Source code in src/pupil_labs/realtime_api/models.py 
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
def matching_sensors(
    self, name: SensorName, connection: ConnectionType
) -> Iterator[Sensor]:
    """Find sensors matching specified criteria.

    Args:
        name: Sensor name to match, or ANY to match any name.
        connection: Connection type to match, or ANY to match any type.

    Yields:
        Sensor: Sensors matching the criteria.

    """
    for sensor in self.sensors:
        if name is not SensorName.ANY and sensor.sensor != name.value:
            continue
        if (
            connection is not ConnectionType.ANY
            and sensor.conn_type != connection.value
        ):
            continue
        yield sensor

update 

update(component: Component) -> None

Update Component.

Parameters:

  • component (Component) –

    Component to update.

Source code in src/pupil_labs/realtime_api/models.py 
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
def update(self, component: Component) -> None:
    """Update Component.

    Args:
        component: Component to update.

    """
    if isinstance(component, Phone):
        self.phone = component
    elif isinstance(component, Hardware):
        self.hardware = component
    elif isinstance(component, Recording):
        self.recording = component
    elif isinstance(component, Sensor):
        for idx, sensor in enumerate(self.sensors):
            if (
                sensor.sensor == component.sensor
                and sensor.conn_type == component.conn_type
            ):
                self.sensors[idx] = component
                break

Template

Template Class for data collection.

Methods:

Attributes:

archived_at class-attribute instance-attribute 

archived_at: datetime | None = None

Archival timestamp (if archived).

created_at instance-attribute 

created_at: datetime

Creation timestamp.

description class-attribute instance-attribute 

description: str | None = None

Template description.

id instance-attribute 

id: UUID

Unique identifier.

is_default_template class-attribute instance-attribute 

is_default_template: bool = True

Whether this is the default template for the Workspace

items class-attribute instance-attribute 

items: list[TemplateItem] = field(default_factory=list)

List of template items.

label_ids class-attribute instance-attribute 

label_ids: list[UUID] = field(default_factory=list, metadata={'readonly': True})

Associated label IDs.

name instance-attribute 

name: str

Template name.

published_at class-attribute instance-attribute 

published_at: datetime | None = None

Publication timestamp.

recording_ids class-attribute instance-attribute 

recording_ids: list[UUID] | None = None

Associated recording IDs.

recording_name_format instance-attribute 

recording_name_format: list[str]

Format for recording name.

updated_at instance-attribute 

updated_at: datetime

Last update timestamp.

convert_from_api_to_simple_format 

convert_from_api_to_simple_format(data: dict[str, list[str]]) -> dict[str, Any]

Convert data from API format to simple format.

Parameters:

Returns:

  • dict ( dict[str, Any] ) –

    Data in simple format.

Source code in src/pupil_labs/realtime_api/models.py 
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
def convert_from_api_to_simple_format(
    self, data: dict[str, list[str]]
) -> dict[str, Any]:
    """Convert data from API format to simple format.

    Args:
        data: Data in API format.

    Returns:
        dict: Data in simple format.

    """
    simple_format = {}
    for question_id, value in data.items():
        question = self.get_question_by_id(question_id)
        if question is None:
            logger.warning(
                f"Skipping unknown question ID '{question_id}' during API to "
                f"simple conversion."
            )
            continue
        processed_value: Any
        if question.widget_type in {"CHECKBOX_LIST", "RADIO_LIST"}:
            if question.choices is None:
                logger.warning(
                    f"Question {question_id} (type {question.widget_type}) "
                    f"has no choices defined."
                )
                processed_value = []
            elif value == [""] and "" not in question.choices:
                processed_value = []
        else:
            if not value:
                value = [""]

            value_str = value[0]
            if question.input_type != "any":
                processed_value = (
                    None if value_str == "" else question._value_type(value_str)
                )
            else:
                processed_value = value

        simple_format[question_id] = processed_value
    return simple_format

convert_from_simple_to_api_format 

convert_from_simple_to_api_format(data: dict[str, Any]) -> dict[str, list[Any]]

Convert data from simple format to API format.

Parameters:

  • data (dict[str, Any]) –

    Data in simple format.

Returns:

Source code in src/pupil_labs/realtime_api/models.py 
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
def convert_from_simple_to_api_format(
    self, data: dict[str, Any]
) -> dict[str, list[Any]]:
    """Convert data from simple format to API format.

    Args:
        data: Data in simple format.

    Returns:
        dict: Data in API format.

    """
    api_format = {}
    for question_id, value in data.items():
        if value is None:
            value = ""
        if not isinstance(value, list):
            value = [value]

        api_format[question_id] = value
    return api_format

get_question_by_id 

get_question_by_id(question_id: str | UUID) -> TemplateItem | None

Get a template item by ID.

Parameters:

  • question_id (str | UUID) –

    ID of the template item.

Returns:

  • TemplateItem | None –

    TemplateItem | None: The template item, or None if not found.

Source code in src/pupil_labs/realtime_api/models.py 
775
776
777
778
779
780
781
782
783
784
785
786
787
788
def get_question_by_id(self, question_id: str | UUID) -> TemplateItem | None:
    """Get a template item by ID.

    Args:
        question_id: ID of the template item.

    Returns:
        TemplateItem | None: The template item, or None if not found.

    """
    for item in self.items:
        if str(item.id) == str(question_id):
            return item
    return None

validate_answers 

validate_answers(answers: dict[str, list[str]], template_format: TemplateDataFormat, raise_exception: bool = True) -> list[ErrorDetails]

Validate answers for this Template.

Parameters:

  • answers (dict[str, list[str]]) –

    Answers to validate.

  • raise_exception (bool, default: True ) –

    Whether to raise an exception on validation failure.

  • template_format (TemplateDataFormat) –

    Format of the answers ("simple" or "api").

Returns:

  • list ( list[ErrorDetails] ) –

    List of validation errors, or empty list if validation succeeded.

Raises:

Source code in src/pupil_labs/realtime_api/models.py 
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
def validate_answers(
    self,
    answers: dict[str, list[str]],
    template_format: TemplateDataFormat,
    raise_exception: bool = True,
) -> list[ErrorDetails]:
    """Validate answers for this Template.

    Args:
        answers: Answers to validate.
        raise_exception: Whether to raise an exception on validation failure.
        template_format: Format of the answers ("simple" or "api").

    Returns:
        list: List of validation errors, or empty list if validation succeeded.

    Raises:
        InvalidTemplateAnswersError: If validation fails and raise_exception is
        True.

    """
    AnswerModel = self._create_answer_model(template_format=template_format)
    errors = []
    try:
        AnswerModel(**answers)
    except ValidationError as e:
        errors = e.errors()

    for error in errors:
        question_id = error["loc"][0]
        question = self.get_question_by_id(str(question_id))
        if question:
            error["question"] = asdict(question)  # type: ignore[typeddict-unknown-key]

    if errors and raise_exception:
        raise InvalidTemplateAnswersError(self, answers, errors)
    return errors

TemplateItem

Individual item/ question in a Template.

Methods:

Attributes:

choices instance-attribute 

choices: list[str] | None

Available choices for selection items (e.g., radio or checkbox).

help_text instance-attribute 

help_text: str | None

Help or description text for the item.

id instance-attribute 

id: UUID

Unique identifier for the template item.

input_type instance-attribute 

input_type: TemplateItemInputType

Type of input data for this item.

required instance-attribute 

required: bool

Whether the item is required or not.

title instance-attribute 

title: str

Title or question text for the template item.

widget_type instance-attribute 

widget_type: TemplateItemWidgetType

Type of widget to display for this item.

validate_answer 

validate_answer(answer: Any, template_format: TemplateDataFormat = 'simple', raise_exception: bool = True) -> list[ErrorDetails]

Validate an answer for this template item.

Parameters:

  • answer (Any) –

    Answer to validate.

  • template_format (TemplateDataFormat, default: 'simple' ) –

    Format of the template ("simple" or "api").

  • raise_exception (bool, default: True ) –

    Whether to raise an exception on validation failure.

Returns:

  • list ( list[ErrorDetails] ) –

    List of validation errors, or empty list if validation succeeded.

Raises:

Source code in src/pupil_labs/realtime_api/models.py 
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
def validate_answer(
    self,
    answer: Any,
    template_format: TemplateDataFormat = "simple",
    raise_exception: bool = True,
) -> list[ErrorDetails]:
    """Validate an answer for this template item.

    Args:
        answer: Answer to validate.
        template_format: Format of the template ("simple" or "api").
        raise_exception: Whether to raise an exception on validation failure.

    Returns:
        list: List of validation errors, or empty list if validation succeeded.

    Raises:
        InvalidTemplateAnswersError: If validation fails and raise_exception is
        True.

    """
    validator = self._pydantic_validator(template_format=template_format)
    if validator is None:
        logger.warning(
            f"Skipping validation for {self.widget_type} item: {self.title}"
        )
        return []
    answers_model_def = {str(self.id): validator}
    model = create_model(
        f"TemplateItem_{self.id}_Answer",
        **answers_model_def,
        __config__=ConfigDict(extra="forbid"),
    )  # type: ignore[call-overload]
    errors = []
    try:
        model.__pydantic_validator__.validate_assignment(
            model.model_construct(), str(self.id), answer
        )
    except ValidationError as e:
        errors = e.errors()

    if errors and raise_exception:
        raise InvalidTemplateAnswersError(self, answers_model_def, errors)
    return errors

UnknownComponentError

Bases: ValueError

Exception raised when a component cannot be parsed.

allow_empty 

allow_empty(v: str) -> str | None

Convert empty strings to None.

Source code in src/pupil_labs/realtime_api/models.py 
864
865
866
867
868
def allow_empty(v: str) -> str | None:
    """Convert empty strings to None."""
    if v == "":
        return None
    return v

make_template_answer_model_base 

make_template_answer_model_base(template_: Template) -> type[BaseModel]

Create a base class for template answer models.

Parameters:

  • template_ (Template) –

    Template to create the model for.

Returns:

  • type ( type[BaseModel] ) –

    Base class for template answer models.

Source code in src/pupil_labs/realtime_api/models.py 
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
def make_template_answer_model_base(template_: Template) -> type[BaseModel]:
    """Create a base class for template answer models.

    Args:
        template_: Template to create the model for.

    Returns:
        type: Base class for template answer models.

    """

    class TemplateAnswerModelBase(BaseModel):
        template: ClassVar[Template] = template_
        model_config = ConfigDict(extra="forbid")

        def get(self, item_id: str) -> Any | None:
            return self.__dict__.get(item_id)

        def __repr__(self) -> str:
            args = []
            for item_id, _validator in self.model_fields.items():
                question = self.template.get_question_by_id(item_id)
                if not question:
                    raise ValueError(
                        f"Question with ID {item_id} not found in template."
                    )
                infos = map(
                    str,
                    [
                        question.title,
                        question.widget_type,
                        question.input_type,
                        question.choices,
                    ],
                )
                line = (
                    f"    {item_id}={self.__dict__[item_id]!r}, # {' - '.join(infos)}"
                )
                args.append(line)
            args_str = "\n".join(args)

            return f"Template_{self.template.id}_AnswerModel(\n" + args_str + "\n)"

        __str__ = __repr__

    return TemplateAnswerModelBase

not_empty 

not_empty(v: str) -> str

Validate that a string is not empty.

Source code in src/pupil_labs/realtime_api/models.py 
857
858
859
860
861
def not_empty(v: str) -> str:
    """Validate that a string is not empty."""
    if not len(v) > 0:
        raise ValueError("value is required")
    return v

option_in_allowed_values 

option_in_allowed_values(value: Any, allowed: list[str] | None) -> Any

Validate that a value is in a list of allowed values.

Source code in src/pupil_labs/realtime_api/models.py 
871
872
873
874
875
def option_in_allowed_values(value: Any, allowed: list[str] | None) -> Any:
    """Validate that a value is in a list of allowed values."""
    if allowed is None or value not in allowed:
        raise ValueError(f"{value!r} is not a valid choice from: {allowed}")
    return value

parse_component 

parse_component(raw: ComponentRaw) -> Component

Initialize an explicitly modelled representation

(:obj:pupil_labs.realtime_api.models.Component) from the json-parsed dictionary (:obj:pupil_labs.realtime_api.models.ComponentRaw) received from the API.

Parameters:

  • raw (ComponentRaw) –

    Dictionary containing component data.

Returns:

  • Component ( Component ) –

    Parsed component instance.

Raises:

Source code in src/pupil_labs/realtime_api/models.py 
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
def parse_component(raw: ComponentRaw) -> Component:
    """Initialize an explicitly modelled representation

    (:obj:`pupil_labs.realtime_api.models.Component`) from the json-parsed dictionary
    (:obj:`pupil_labs.realtime_api.models.ComponentRaw`) received from the API.

    Args:
        raw (ComponentRaw): Dictionary containing component data.

    Returns:
        Component: Parsed component instance.

    Raises:
        UnknownComponentError: If the component name cannot be mapped to an
        explicitly modelled class or the contained data does not fit the modelled
        fields.

    """
    model_name = raw["model"]
    data = raw["data"]
    try:
        model_class = _model_class_map[model_name]
        return _init_cls_with_annotated_fields_only(model_class, data)
    except KeyError as err:
        raise UnknownComponentError(
            f"Could not generate component for {model_name} from {data}"
        ) from err

base

Classes:

  • DeviceBase –

    Abstract base class representing Realtime API host devices.

Attributes:

  • DeviceType –

    Type annotation for concrete sub-classes of :class:`DeviceBase

DeviceType module-attribute 

DeviceType = TypeVar('DeviceType', bound='DeviceBase')

Type annotation for concrete sub-classes of :class:DeviceBase <pupil_labs.realtime_api.base.DeviceBase>.

DeviceBase 

DeviceBase(address: str, port: int, full_name: str | None = None, dns_name: str | None = None, suppress_decoding_warnings: bool = True)

Bases: ABC

Abstract base class representing Realtime API host devices.

This class provides the foundation for device implementations that connect to the Realtime API.

Attributes:

  • address (str) –

    REST API server address.

  • port (int) –

    REST API server port.

  • full_name (str | None) –

    Full service discovery name.

  • dns_name (str | None) –

    REST API server DNS name, e.g.neon.local / pi.local..

Parameters:

  • address (str) –

    REST API server address.

  • port (int) –

    REST API server port.

  • full_name (str | None, default: None ) –

    Full service discovery name.

  • dns_name (str | None, default: None ) –

    REST API server DNS name, e.g.neon.local / pi.local..

  • suppress_decoding_warnings (bool, default: True ) –

    Whether to suppress libav decoding warnings.

Methods:

Source code in src/pupil_labs/realtime_api/base.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def __init__(
    self,
    address: str,
    port: int,
    full_name: str | None = None,
    dns_name: str | None = None,
    suppress_decoding_warnings: bool = True,
):
    """Initialize the DeviceBase instance.

    Args:
        address (str): REST API server address.
        port (int): REST API server port.
        full_name (str | None): Full service discovery name.
        dns_name (str | None): REST API server DNS name,
            e.g.``neon.local / pi.local.``.
        suppress_decoding_warnings: Whether to suppress libav decoding warnings.

    """
    self.address: str = address
    self.port: int = port
    self.full_name: str | None = full_name
    self.dns_name: str | None = dns_name
    if suppress_decoding_warnings:
        # suppress decoding warnings due to incomplete data transmissions
        logging.getLogger("libav.h264").setLevel(logging.CRITICAL)
        logging.getLogger("libav.swscaler").setLevel(logging.ERROR)

api_url 

api_url(path: APIPath, protocol: str = 'http', prefix: str = '/api') -> str

Construct a full API URL for the given path.

Parameters:

  • path (APIPath) –

    API path to access.

  • protocol (str, default: 'http' ) –

    Protocol to use (http).

  • prefix (str, default: '/api' ) –

    API URL prefix.

Returns:

  • str –

    Complete URL for the API endpoint.

Source code in src/pupil_labs/realtime_api/base.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def api_url(
    self, path: APIPath, protocol: str = "http", prefix: str = "/api"
) -> str:
    """Construct a full API URL for the given path.

    Args:
        path: API path to access.
        protocol: Protocol to use (http).
        prefix: API URL prefix.

    Returns:
        Complete URL for the API endpoint.

    """
    return path.full_address(
        self.address, self.port, protocol=protocol, prefix=prefix
    )

convert_from classmethod 

convert_from(other: T) -> DeviceType

Convert another device instance to this type.

Parameters:

  • other (T) –

    Device instance to convert.

Returns:

Source code in src/pupil_labs/realtime_api/base.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
@classmethod
def convert_from(cls: type[DeviceType], other: T) -> DeviceType:
    """Convert another device instance to this type.

    Args:
        other: Device instance to convert.

    Returns:
        Converted device instance.

    """
    return cls(
        other.address,
        other.port,
        full_name=other.full_name,
        dns_name=other.dns_name,
    )

from_discovered_device classmethod 

from_discovered_device(device: DiscoveredDeviceInfo) -> DeviceType

Create a device instance from discovery information.

Parameters:

Returns:

Source code in src/pupil_labs/realtime_api/base.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
@classmethod
def from_discovered_device(
    cls: type[DeviceType], device: DiscoveredDeviceInfo
) -> DeviceType:
    """Create a device instance from discovery information.

    Args:
        device: Discovered device information.

    Returns:
        Device instance

    """
    return cls(
        device.addresses[0],
        device.port,
        full_name=device.name,
        dns_name=device.server,
    )