Skip to content

simple

Modules:

  • device –
  • discovery –
  • models –

Classes:

  • Device –

    Simple synchronous API for interacting with Pupil Labs devices.

  • MatchedGazeEyesSceneItem –

    A matched triplet of scene video frame, eye video frame, and gaze data.

  • MatchedItem –

    A matched pair of scene video frame and gaze data.

  • SimpleVideoFrame –

    A simplified video frame representation.

Functions:

Device 

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

Bases: DeviceBase

Simple synchronous API for interacting with Pupil Labs devices.

This class provides a simplified, synchronous interface to Pupil Labs devices, wrapping the asynchronous API with a more user-friendly interface.

Important

Use [discover_devices][pupil_labs.realtime_api.simple.discovery. discover_devices] instead of initializing the class manually. See the simple_discovery_example example.

Parameters:

  • address (str) –

    IP address of the device.

  • port (int) –

    Port number of the device.

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

    Full service discovery name.

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

    DNS name of the device.

  • start_streaming_by_default (bool, default: False ) –

    Whether to start streaming automatically.

  • suppress_decoding_warnings (bool, default: True ) –

    Whether to suppress decoding warnings.

Methods:

Attributes:

Source code in src/pupil_labs/realtime_api/simple/device.py 
 69
 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
103
104
105
106
107
108
109
110
111
112
113
114
115
def __init__(
    self,
    address: str,
    port: int,
    full_name: str | None = None,
    dns_name: str | None = None,
    start_streaming_by_default: bool = False,
    suppress_decoding_warnings: bool = True,
) -> None:
    """Initialize a Device instance.

    Args:
        address: IP address of the device.
        port: Port number of the device.
        full_name: Full service discovery name.
        dns_name: DNS name of the device.
        start_streaming_by_default: Whether to start streaming automatically.
        suppress_decoding_warnings: Whether to suppress decoding warnings.

    """
    super().__init__(
        address,
        port,
        full_name=full_name,
        dns_name=dns_name,
        suppress_decoding_warnings=suppress_decoding_warnings,
    )
    self._status = self._get_status()
    self._start_background_worker(start_streaming_by_default)

    self._errors: list[str] = []

    self.stream_name_start_event_map = {
        SensorName.GAZE.value: self._EVENT.SHOULD_START_GAZE,
        SensorName.WORLD.value: self._EVENT.SHOULD_START_WORLD,
        SensorName.EYES.value: self._EVENT.SHOULD_START_EYES,
        SensorName.IMU.value: self._EVENT.SHOULD_START_IMU,
        SensorName.EYE_EVENTS.value: self._EVENT.SHOULD_START_EYE_EVENTS,
    }

    self.stream_name_stop_event_map = {
        SensorName.GAZE.value: self._EVENT.SHOULD_STOP_GAZE,
        SensorName.WORLD.value: self._EVENT.SHOULD_STOP_WORLD,
        SensorName.EYES.value: self._EVENT.SHOULD_STOP_EYES,
        SensorName.IMU.value: self._EVENT.SHOULD_STOP_IMU,
        SensorName.EYE_EVENTS.value: self._EVENT.SHOULD_STOP_EYE_EVENTS,
    }

battery_level_percent property 

battery_level_percent: int

Get the battery level of the connected phone in percentage.

battery_state property 

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

Get the battery state of the connected phone.

Possible values are "OK", "LOW", or "CRITICAL".

is_currently_streaming property 

is_currently_streaming: bool

Check if data streaming is currently active.

Returns:

  • bool ( bool ) –

    True if streaming is active, False otherwise.

memory_num_free_bytes property 

memory_num_free_bytes: int

Get the available memory of the connected phone in bytes.

memory_state property 

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

Get the memory state of the connected phone.

Possible values are "OK", "LOW", or "CRITICAL".

module_serial property 

module_serial: str | Literal['default'] | None

Returns None or "default" if no glasses are connected

Info

Only available on Neon.

phone_id property 

phone_id: str

Get the ID of the connected phone.

phone_ip property 

phone_ip: str

Get the IP address of the connected phone.

phone_name property 

phone_name: str

Get the name of the connected phone.

serial_number_glasses property 

serial_number_glasses: str | Literal['default'] | None

Returns None or "default" if no glasses are connected

Info

Only available on Pupil Invisible.

serial_number_scene_cam property 

serial_number_scene_cam: str | None

Returns None if no scene camera is connected

Info

Only available on Pupil Invisible.

version_glasses property 

version_glasses: str | None

Get the version of the connected glasses.

Returns:

  • str ( str | None ) –

    1 -> Pupil Invisible, 2 -> Neon, or None -> No glasses connected.

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 

close() -> None

Close the device connection and stop all background threads.

This method should be called when the device is no longer needed to free up resources.

Source code in src/pupil_labs/realtime_api/simple/device.py 
679
680
681
682
683
684
685
686
687
688
689
def close(self) -> None:
    """Close the device connection and stop all background threads.

    This method should be called when the device is no longer needed
    to free up resources.
    """
    if self._event_manager:
        if self.is_currently_streaming:
            self.streaming_stop()
        self._event_manager.trigger_threadsafe(self._EVENT.SHOULD_WORKER_CLOSE)
        self._auto_update_thread.join()

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,
    )

estimate_time_offset 

estimate_time_offset(number_of_measurements: int = 100, sleep_between_measurements_seconds: float | None = None) -> TimeEchoEstimates | None

Estimate the time offset between the host device and the client.

This uses the Time Echo protocol to estimate the clock offset between the device and the client.

Parameters:

  • number_of_measurements (int, default: 100 ) –

    Number of measurements to take.

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

    Optional sleep time between

Returns:

  • TimeEchoEstimates ( TimeEchoEstimates | None ) –

    Statistics for roundtrip durations and time offsets, or None if estimation failed or device doesn't support the protocol.

See Also

:mod:pupil_labs.realtime_api.time_echo for more details.

Source code in src/pupil_labs/realtime_api/simple/device.py 
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
def estimate_time_offset(
    self,
    number_of_measurements: int = 100,
    sleep_between_measurements_seconds: float | None = None,
) -> TimeEchoEstimates | None:
    """Estimate the time offset between the host device and the client.

    This uses the Time Echo protocol to estimate the clock offset between
    the device and the client.

    Args:
        number_of_measurements: Number of measurements to take.
        sleep_between_measurements_seconds: Optional sleep time between
        measurements.

    Returns:
        TimeEchoEstimates: Statistics for roundtrip durations and time offsets,
            or None if estimation failed or device doesn't support the protocol.

    See Also:
        :mod:`pupil_labs.realtime_api.time_echo` for more details.

    """
    if self._status.phone.time_echo_port is None:
        logger.warning(
            "You Pupil Invisible Companion app is out-of-date and does not yet "
            "support the Time Echo protocol. Upgrade to version 1.4.28 or newer."
        )
        return None
    estimator = TimeOffsetEstimator(
        self.phone_ip, self._status.phone.time_echo_port
    )
    return asyncio.run(
        estimator.estimate(
            number_of_measurements, sleep_between_measurements_seconds
        )
    )

eye_events_sensor 

eye_events_sensor() -> Sensor | None

Get the eye events sensor.

Source code in src/pupil_labs/realtime_api/simple/device.py 
215
216
217
def eye_events_sensor(self) -> Sensor | None:
    """Get the eye events sensor."""
    return self._status.direct_eye_events_sensor()

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,
    )

gaze_sensor 

gaze_sensor() -> Sensor | None

Get the gaze sensor.

Source code in src/pupil_labs/realtime_api/simple/device.py 
211
212
213
def gaze_sensor(self) -> Sensor | None:
    """Get the gaze sensor."""
    return self._status.direct_gaze_sensor()

get_calibration 

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.

Returns:

  • Calibration –

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

Raises:

Source code in src/pupil_labs/realtime_api/simple/device.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
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.

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

    Raises:
        DeviceError: If the request fails.

    """

    async def _get_calibration() -> Calibration:
        async with _DeviceAsync.convert_from(self) as control:
            return await control.get_calibration()

    return asyncio.run(_get_calibration())

get_errors 

get_errors() -> list[str]

Get a list of errors from the device.

Returns:

  • list[str] –

    list[str]: List of error messages.

Source code in src/pupil_labs/realtime_api/simple/device.py 
195
196
197
198
199
200
201
202
203
204
205
def get_errors(self) -> list[str]:
    """Get a list of errors from the device.

    Returns:
        list[str]: List of error messages.

    """
    errors = self._errors.copy()
    self._errors.clear()

    return errors

get_template 

get_template() -> Template

Get the template currently selected on the device.

Wraps pupil_labs.realtime_api.device.Device.get_template

Returns:

  • Template ( Template ) –

    The currently selected template.

Raises:

Source code in src/pupil_labs/realtime_api/simple/device.py 
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
def get_template(self) -> Template:
    """Get the template currently selected on the device.

    Wraps [pupil_labs.realtime_api.device.Device.get_template][]

    Returns:
        Template: The currently selected template.

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

    """

    async def _get_template() -> Template:
        async with _DeviceAsync.convert_from(self) as control:
            return await control.get_template()

    return asyncio.run(_get_template())

get_template_data 

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

Get the template data entered on the device.

Wraps pupil_labs.realtime_api.device.Device.get_template_data

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 result from the GET request.

Raises:

  • DeviceError –

    If the template's data could not be fetched.

Source code in src/pupil_labs/realtime_api/simple/device.py 
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
def get_template_data(self, template_format: TemplateDataFormat = "simple") -> Any:
    """Get the template data entered on the device.

    Wraps [pupil_labs.realtime_api.device.Device.get_template_data][]

    Args:
        template_format: 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 result from the GET request.

    Raises:
        DeviceError: If the template's data could not be fetched.

    """

    async def _get_template_data() -> Any:
        async with _DeviceAsync.convert_from(self) as control:
            return await control.get_template_data(template_format=template_format)

    return asyncio.run(_get_template_data())

post_template_data 

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

Send the data to the currently selected template.

Wraps pupil_labs.realtime_api.device.Device.post_template_data

Parameters:

  • template_data (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 from the POST request.

Raises:

Source code in src/pupil_labs/realtime_api/simple/device.py 
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
def post_template_data(
    self,
    template_data: dict[str, list[str]],
    template_format: TemplateDataFormat = "simple",
) -> Any:
    """Send the data to the currently selected template.

    Wraps [pupil_labs.realtime_api.device.Device.post_template_data][]

    Args:
        template_data: The template data to send.
        template_format: 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 from the POST request.

    Raises:
        DeviceError: If the data can not be sent.
        ValueError: If invalid data type.

    """

    async def _post_template_data() -> Any:
        async with _DeviceAsync.convert_from(self) as control:
            return await control.post_template_data(
                template_data, template_format=template_format
            )

    return asyncio.run(_post_template_data())

receive_eye_events 

receive_eye_events(timeout_seconds: float | None = None) -> FixationEventData | BlinkEventData | FixationOnsetEventData | None

Receive an eye event.

Parameters:

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

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

Returns:

Source code in src/pupil_labs/realtime_api/simple/device.py 
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
def receive_eye_events(
    self, timeout_seconds: float | None = None
) -> FixationEventData | BlinkEventData | FixationOnsetEventData | None:
    """Receive an eye event.

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

    Returns:
        FixationEventData | BlinkEventData | FixationOnsetEventData or None:
        The received eye event, or None if timeout was reached.

    """
    return cast(
        FixationEventData | BlinkEventData | FixationOnsetEventData,
        self._receive_item(SensorName.EYE_EVENTS.value, timeout_seconds),
    )

receive_eyes_video_frame 

receive_eyes_video_frame(timeout_seconds: float | None = None) -> SimpleVideoFrame | None

Receive an eye camera video frame.

Parameters:

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

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

Returns:

Source code in src/pupil_labs/realtime_api/simple/device.py 
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
def receive_eyes_video_frame(
    self, timeout_seconds: float | None = None
) -> SimpleVideoFrame | None:
    """Receive an eye camera video frame.

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

    Returns:
        SimpleVideoFrame or None: The received video frame, or None if timeout
        was reached.

    """
    return cast(
        SimpleVideoFrame,
        self._receive_item(SensorName.EYES.value, timeout_seconds),
    )

receive_gaze_datum 

receive_gaze_datum(timeout_seconds: float | None = None) -> GazeDataType | None

Receive a gaze data point.

Parameters:

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

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

Returns:

  • GazeDataType | None –

    GazeDataType or None: The received gaze data, or None if timeout was

  • GazeDataType | None –

    reached.

Source code in src/pupil_labs/realtime_api/simple/device.py 
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
def receive_gaze_datum(
    self, timeout_seconds: float | None = None
) -> GazeDataType | None:
    """Receive a gaze data point.

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

    Returns:
        GazeDataType or None: The received gaze data, or None if timeout was
        reached.

    """
    return cast(
        GazeDataType, self._receive_item(SensorName.GAZE.value, timeout_seconds)
    )

receive_imu_datum 

receive_imu_datum(timeout_seconds: float | None = None) -> IMUData | None

Receive an IMU data point.

Parameters:

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

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

Returns:

  • IMUData | None –

    IMUData or None: The received IMU data, or None if timeout was reached.

Source code in src/pupil_labs/realtime_api/simple/device.py 
459
460
461
462
463
464
465
466
467
468
469
470
def receive_imu_datum(self, timeout_seconds: float | None = None) -> IMUData | None:
    """Receive an IMU data point.

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

    Returns:
        IMUData or None: The received IMU data, or None if timeout was reached.

    """
    return cast(IMUData, self._receive_item(SensorName.IMU.value, timeout_seconds))

receive_matched_scene_and_eyes_video_frames_and_gaze 

receive_matched_scene_and_eyes_video_frames_and_gaze(timeout_seconds: float | None = None) -> MatchedGazeEyesSceneItem | None

Receive a matched triplet of scene video frame, eye video frame, and gaze.

Parameters:

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

    Maximum time to wait for a matched triplet. If None, wait indefinitely.

Returns:

Source code in src/pupil_labs/realtime_api/simple/device.py 
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
def receive_matched_scene_and_eyes_video_frames_and_gaze(
    self, timeout_seconds: float | None = None
) -> MatchedGazeEyesSceneItem | None:
    """Receive a matched triplet of scene video frame, eye video frame, and gaze.

    Args:
        timeout_seconds: Maximum time to wait for a matched triplet.
            If None, wait indefinitely.

    Returns:
        MatchedGazeEyesSceneItem or None: The matched triplet, or None if timeout
        was reached.

    """
    return cast(
        MatchedGazeEyesSceneItem,
        self._receive_item(MATCHED_GAZE_EYES_LABEL, timeout_seconds),
    )

receive_matched_scene_video_frame_and_gaze 

receive_matched_scene_video_frame_and_gaze(timeout_seconds: float | None = None) -> MatchedItem | None

Receive a matched pair of scene video frame and gaze data.

Parameters:

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

    Maximum time to wait for a matched pair. If None, wait indefinitely.

Returns:

  • MatchedItem | None –

    MatchedItem or None: The matched pair, or None if timeout was reached.

Source code in src/pupil_labs/realtime_api/simple/device.py 
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
def receive_matched_scene_video_frame_and_gaze(
    self, timeout_seconds: float | None = None
) -> MatchedItem | None:
    """Receive a matched pair of scene video frame and gaze data.

    Args:
        timeout_seconds: Maximum time to wait for a matched pair.
            If None, wait indefinitely.

    Returns:
        MatchedItem or None: The matched pair, or None if timeout was reached.

    """
    return cast(
        MatchedItem, self._receive_item(MATCHED_ITEM_LABEL, timeout_seconds)
    )

receive_scene_video_frame 

receive_scene_video_frame(timeout_seconds: float | None = None) -> SimpleVideoFrame | None

Receive a scene (world) video frame.

Parameters:

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

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

Returns:

Source code in src/pupil_labs/realtime_api/simple/device.py 
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
def receive_scene_video_frame(
    self, timeout_seconds: float | None = None
) -> SimpleVideoFrame | None:
    """Receive a scene (world) video frame.

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

    Returns:
        SimpleVideoFrame or None: The received video frame, or None if timeout was
        reached.

    """
    return cast(
        SimpleVideoFrame,
        self._receive_item(SensorName.WORLD.value, timeout_seconds),
    )

recording_cancel 

recording_cancel() -> None

Cancel the current recording without saving it.

Wraps pupil_labs.realtime_api.device.Device.recording_cancel

Raises:

  • DeviceError –

    If the recording could not be cancelled. Possible reasons

  • include –
    • Recording not running
Source code in src/pupil_labs/realtime_api/simple/device.py 
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
def recording_cancel(self) -> None:
    """Cancel the current recording without saving it.

    Wraps [pupil_labs.realtime_api.device.Device.recording_cancel][]

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

    """

    async def _cancel_recording() -> None:
        async with _DeviceAsync.convert_from(self) as control:
            return await control.recording_cancel()

    return asyncio.run(_cancel_recording())

recording_start 

recording_start() -> str

Start a recording on the device.

Wraps pupil_labs.realtime_api.device.Device.recording_start

Returns:

  • str ( str ) –

    ID of the started recording.

Raises:

  • DeviceError –

    If the 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/simple/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
def recording_start(self) -> str:
    """Start a recording on the device.

    Wraps [pupil_labs.realtime_api.device.Device.recording_start][]

    Returns:
        str: ID of the started recording.

    Raises:
        DeviceError: If the 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 def _start_recording() -> str:
        async with _DeviceAsync.convert_from(self) as control:
            return await control.recording_start()

    return asyncio.run(_start_recording())

recording_stop_and_save 

recording_stop_and_save() -> None

Stop and save the current recording.

Wraps pupil_labs.realtime_api.device.Device.recording_stop_and_save

Raises:

  • DeviceError –

    If the recording could not be stopped. Possible reasons

  • include –
    • Recording not running
    • Template has required fields
Source code in src/pupil_labs/realtime_api/simple/device.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
def recording_stop_and_save(self) -> None:
    """Stop and save the current recording.

    Wraps [pupil_labs.realtime_api.device.Device.recording_stop_and_save][]

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

    """

    async def _stop_and_save_recording() -> None:
        async with _DeviceAsync.convert_from(self) as control:
            return await control.recording_stop_and_save()

    return asyncio.run(_stop_and_save_recording())

send_event 

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/simple/device.py 
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
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.

    """

    async def _send_event() -> Event:
        async with _DeviceAsync.convert_from(self) as control:
            return await control.send_event(event_name, event_timestamp_unix_ns)

    return asyncio.run(_send_event())

start_stream_if_needed 

start_stream_if_needed(sensor: str) -> None

Start streaming if not already streaming.

Parameters:

  • sensor (str) –

    Sensor name to check.

Source code in src/pupil_labs/realtime_api/simple/device.py 
562
563
564
565
566
567
568
569
570
571
def start_stream_if_needed(self, sensor: str) -> None:
    """Start streaming if not already streaming.

    Args:
        sensor: Sensor name to check.

    """
    if not self._is_streaming_flags[sensor].is_set():
        logger.debug("receive_* called without being streaming")
        self.streaming_start(sensor)

streaming_start 

streaming_start(stream_name: str) -> None

Start streaming data from the specified sensor.

Parameters:

  • stream_name (str) –

    Name of the sensor to start streaming from. It can be one of

  • –

    py:attr:SensorName values or None, which will start all streams.

Raises:

  • ValueError –

    If the stream name is not recognized.

Source code in src/pupil_labs/realtime_api/simple/device.py 
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
def streaming_start(self, stream_name: str) -> None:
    """Start streaming data from the specified sensor.

    Args:
        stream_name: Name of the sensor to start streaming from. It can be one of
        :py:attr:`SensorName` values or None, which will start all streams.

    Raises:
        ValueError: If the stream name is not recognized.

    """
    if stream_name is None:
        for event in (
            self._EVENT.SHOULD_START_GAZE,
            self._EVENT.SHOULD_START_WORLD,
            self._EVENT.SHOULD_START_EYES,
            self._EVENT.SHOULD_START_IMU,
            self._EVENT.SHOULD_START_EYE_EVENTS,
        ):
            self._streaming_trigger_action(event)
        return

    event = self.stream_name_start_event_map[stream_name]
    self._streaming_trigger_action(event)

streaming_stop 

streaming_stop(stream_name: str | None = None) -> None

Stop streaming data from the specified sensor.

Parameters:

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

    Name of the sensor to start streaming from. It can be one of

  • –

    py:attr:SensorName values or None, which will stop all streams.

Raises:

  • ValueError –

    If the stream name is not recognized.

Source code in src/pupil_labs/realtime_api/simple/device.py 
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
def streaming_stop(self, stream_name: str | None = None) -> None:
    """Stop streaming data from the specified sensor.

    Args:
        stream_name: Name of the sensor to start streaming from. It can be one of
        :py:attr:`SensorName` values or None, which will stop all streams.

    Raises:
        ValueError: If the stream name is not recognized.

    """
    if stream_name is None:
        for event in (
            self._EVENT.SHOULD_STOP_GAZE,
            self._EVENT.SHOULD_STOP_WORLD,
            self._EVENT.SHOULD_STOP_EYES,
            self._EVENT.SHOULD_STOP_IMU,
            self._EVENT.SHOULD_STOP_EYE_EVENTS,
        ):
            self._streaming_trigger_action(event)
        return

    event = self.stream_name_stop_event_map[stream_name]
    self._streaming_trigger_action(event)

world_sensor 

world_sensor() -> Sensor | None

Get the world sensor.

Source code in src/pupil_labs/realtime_api/simple/device.py 
207
208
209
def world_sensor(self) -> Sensor | None:
    """Get the world sensor."""
    return self._status.direct_world_sensor()

MatchedGazeEyesSceneItem

Bases: NamedTuple

A matched triplet of scene video frame, eye video frame, and gaze data.

This class represents scene and eye video frames along with gaze data that occurred at approximately the same time.

Attributes:

eyes instance-attribute 

eyes: SimpleVideoFrame

Eye camera video frame.

gaze instance-attribute 

gaze: GazeDataType

Corresponding gaze data.

scene instance-attribute 

scene: SimpleVideoFrame

Scene video frame.

MatchedItem

Bases: NamedTuple

A matched pair of scene video frame and gaze data.

This class represents a scene video frame and gaze data point that occurred at approximately the same time.

Note

The name MatchedItem is maintained for backward compatibility. It represents a matched pair of scene video frame and gaze data.

Attributes:

frame instance-attribute 

frame: SimpleVideoFrame

Scene video frame.

gaze instance-attribute 

gaze: GazeDataType

Corresponding gaze data.

SimpleVideoFrame

Bases: NamedTuple

A simplified video frame representation.

This class provides a simplified representation of a video frame with BGR pixel data and timestamp information.

Attributes:

bgr_pixels instance-attribute 

bgr_pixels: BGRBuffer

BGR pixel data as a NumPy array.

timestamp_unix_seconds instance-attribute 

timestamp_unix_seconds: float

Timestamp in seconds since Unix epoch.

discover_devices 

discover_devices(search_duration_seconds: float) -> list[Device]

Discover all available devices on the local network.

This function searches for devices on the local network for the specified duration and returns all discovered devices.

Parameters:

  • search_duration_seconds (float) –

    How long to search for devices in seconds.

Returns:

  • list[Device] –

    list[Device]: List of discovered devices.

Example 
# Discover devices for 5 seconds
devices = discover_devices(5.0)
for device in devices:
    print(f"Found device: {device.phone_name}")
See Also

The asynchronous equivalent :func:pupil_labs.realtime_api.discovery.discover_devices

Source code in src/pupil_labs/realtime_api/simple/discovery.py 
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
def discover_devices(search_duration_seconds: float) -> list[Device]:
    """Discover all available devices on the local network.

    This function searches for devices on the local network for the specified
    duration and returns all discovered devices.

    Args:
        search_duration_seconds: How long to search for devices in seconds.

    Returns:
        list[Device]: List of discovered devices.

    Example:
        ```python
        # Discover devices for 5 seconds
        devices = discover_devices(5.0)
        for device in devices:
            print(f"Found device: {device.phone_name}")
        ```

    See Also:
        The asynchronous equivalent
        :func:`pupil_labs.realtime_api.discovery.discover_devices`

    """

    async def _discover() -> tuple[DiscoveredDeviceInfo, ...]:
        async with AsyncNetwork() as network:
            await asyncio.sleep(search_duration_seconds)
            return network.devices

    return [Device.from_discovered_device(dev) for dev in asyncio.run(_discover())]

discover_one_device 

discover_one_device(max_search_duration_seconds: float | None = 10.0) -> Device | None

Discover and return the first device found on the local network.

This function searches for devices on the local network and returns the first discovered device, or None if no device is found within the specified maximum search duration.

Parameters:

  • max_search_duration_seconds (float | None, default: 10.0 ) –

    Maximum time to search for a device in seconds. If None, search indefinitely. Default is 10.0 seconds.

Returns:

  • Device | None –

    Device or None: The first discovered device, or None if no device was found

  • Device | None –

    within the specified time limit.

Example 
# Try to find a device within 5 seconds
device = discover_one_device(5.0)
if device:
    print(f"Found device: {device.phone_name}")
else:
    print("No device found")
See Also

The asynchronous equivalent :func:pupil_labs.realtime_api.discovery.Network.wait_for_new_device

Source code in src/pupil_labs/realtime_api/simple/discovery.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def discover_one_device(
    max_search_duration_seconds: float | None = 10.0,
) -> Device | None:
    """Discover and return the first device found on the local network.

    This function searches for devices on the local network and returns
    the first discovered device, or None if no device is found within
    the specified maximum search duration.

    Args:
        max_search_duration_seconds: Maximum time to search for a device in seconds.
            If None, search indefinitely. Default is 10.0 seconds.

    Returns:
        Device or None: The first discovered device, or None if no device was found
        within the specified time limit.

    Example:
        ```python
        # Try to find a device within 5 seconds
        device = discover_one_device(5.0)
        if device:
            print(f"Found device: {device.phone_name}")
        else:
            print("No device found")
        ```

    See Also:
        The asynchronous equivalent
        :func:`pupil_labs.realtime_api.discovery.Network.wait_for_new_device`

    """

    async def _discover() -> DiscoveredDeviceInfo | None:
        async with AsyncNetwork() as network:
            return await network.wait_for_new_device(max_search_duration_seconds)

    device = asyncio.run(_discover())
    return None if device is None else Device.from_discovered_device(device)