Simple Examples

Code examples that use the Simple API.

Note

The examples require Python 3.9+ to run.

Find one or more devices

from pupil_labs.realtime_api.simple import discover_devices, discover_one_device

# Look for devices. Returns as soon as it has found the first device.
print("Looking for the next best device...\n\t", end="")
print(discover_one_device(max_search_duration_seconds=10.0))

# List all devices that could be found within 10 seconds
print("Starting 10 second search...\n\t", end="")
print(discover_devices(search_duration_seconds=10.0))

Remote control devices

Get current status

from pupil_labs.realtime_api.simple import discover_one_device

# Look for devices. Returns as soon as it has found the first device.
print("Looking for the next best device...")
device = discover_one_device(max_search_duration_seconds=10)
if device is None:
    print("No device found.")
    raise SystemExit(-1)

# Device status is fetched on initialization and kept up-to-date in the background

print(f"Phone IP address: {device.phone_ip}")
print(f"Phone name: {device.phone_name}")
print(f"Phone unique ID: {device.phone_id}")

print(f"Battery level: {device.battery_level_percent}%")
print(f"Battery state: {device.battery_state}")

print(f"Free storage: {device.memory_num_free_bytes / 1024**3}GB")
print(f"Storage level: {device.memory_state}")

print(f"Connected glasses: SN {device.serial_number_glasses}")
print(f"Connected scene camera: SN {device.serial_number_scene_cam}")

device.close()  # explicitly stop auto-update

Automatic status updates

The Device class monitors a Neon / Pupil Invisible Companion device in the background and mirrors its state accordingly.

from pupil_labs.realtime_api.simple import discover_one_device

# Look for devices. Returns as soon as it has found the first device.
print("Looking for the next best device...")
device = discover_one_device(max_search_duration_seconds=10)
if device is None:
    print("No device found.")
    raise SystemExit(-1)

scene_camera = device.world_sensor()
connected = False if scene_camera is None else scene_camera.connected
print("Scene camera connected:", connected)

input("(Dis)connect the scene camera and hit enter...")

scene_camera = device.world_sensor()
connected = False if scene_camera is None else scene_camera.connected
print("Scene camera connected:", connected)

device.close()

Send event

An event without an explicit timestamp, will be timestamped on arrival at the Neon / Pupil Invisible Companion device.

import time

from pupil_labs.realtime_api.simple import discover_one_device

# Look for devices. Returns as soon as it has found the first device.
print("Looking for the next best device...")
device = discover_one_device(max_search_duration_seconds=10)
if device is None:
    print("No device found.")
    raise SystemExit(-1)

print(device.send_event("test event; timestamped at arrival"))

# send event with current timestamp
print(
    device.send_event(
        "test event; timestamped by the client, relying on NTP for sync",
        event_timestamp_unix_ns=time.time_ns(),
    )
)

# Estimate clock offset between Companion device and client script
# (only needs to be done once)
estimate = device.estimate_time_offset()
clock_offset_ns = round(estimate.time_offset_ms.mean * 1_000_000)
print(f"Clock offset: {clock_offset_ns:_d} ns")

# send event with current timestamp, but correct it manual for possible clock offset
current_time_ns_in_client_clock = time.time_ns()
current_time_ns_in_companion_clock = current_time_ns_in_client_clock - clock_offset_ns
print(
    device.send_event(
        "test event; timestamped by the client, manual clock offset correction",
        event_timestamp_unix_ns=current_time_ns_in_companion_clock,
    )
)


device.close()

Start, stop and save, and cancel recordings

import time

from pupil_labs.realtime_api.simple import discover_one_device

# Look for devices. Returns as soon as it has found the first device.
print("Looking for the next best device...")
device = discover_one_device(max_search_duration_seconds=10)
if device is None:
    print("No device found.")
    raise SystemExit(-1)

print(f"Starting recording")
recording_id = device.recording_start()
print(f"Started recording with id {recording_id}")

time.sleep(5)

device.recording_stop_and_save()

print("Recording stopped and saved")
# device.recording_cancel()  # uncomment to cancel recording

device.close()

Templates

You can programmatically fill the template. This allows you to also define the recording name if the template is created correctly.

import os

import beaupy

from pupil_labs.realtime_api.models import InvalidTemplateAnswersError, TemplateItem
from pupil_labs.realtime_api.simple import discover_one_device

# handle KeyboardInterrupts ourselves
beaupy.Config.raise_on_interrupt = True

# Look for devices. Returns as soon as it has found the first device.
print("Looking for the next best device...")
device = discover_one_device(max_search_duration_seconds=10)
if device is None:
    print("No device found.")
    raise SystemExit(-1)

# Fetch current template definition
template = device.get_template()

# Fetch data filled on the template
data = device.get_template_data()

LINE = "\u2500" * os.get_terminal_size().columns
RED = "\033[31m"
RESET = "\033[0m"

print(f"[{template.name}] Data pre-filled:")
print(LINE)
print("\n".join(f"{k}\t{v}" for k, v in data.items()))


def prompt_checkbox_answer(item: TemplateItem, current_value):
    ticked = []
    for i, choice in enumerate(item.choices):
        current_value: list
        if choice in (current_value or []):
            current_value.remove(choice)
            ticked.append(i)
    choices = beaupy.select_multiple(
        item.choices,
        ticked_indices=ticked,
    )
    return choices


def prompt_radio_answer(item: TemplateItem, current_value):
    cursor_index = 0
    if current_value and current_value[0] in item.choices:
        cursor_index = item.choices.index(current_value[0])

    choice = beaupy.select(item.choices, cursor_index=cursor_index)
    template_input = []
    if choice is not None:
        template_input = [choice]
    return template_input


def prompt_string_answer(item: TemplateItem, current_value):
    placeholder = item.help_text if item.help_text and item.help_text != [""] else None
    current_value = (
        placeholder if not current_value or current_value == [""] else current_value
    )
    return beaupy.prompt(
        f"Enter value for '{item.title}': ",
        initial_value="" if current_value is None else str(current_value),
    )


# Filling a template
questionnaire = {}
if template:
    try:
        for item in template.items:
            if item.widget_type in ("SECTION_HEADER", "PAGE_BREAK"):
                continue
            print(LINE)
            print(
                f"{'* ' if item.required else ''}"
                + f"ID: {item.id} - Title: {item.title} "
                + f"- Input Type: {item.input_type}"
            )
            current_value = data.get(str(item.id))
            while True:
                question = template.get_question_by_id(item.id)
                if item.widget_type == "CHECKBOX_LIST":
                    template_input = prompt_checkbox_answer(item, current_value)
                elif item.widget_type == "RADIO_LIST":
                    template_input = prompt_radio_answer(item, current_value)
                else:
                    template_input = prompt_string_answer(item, current_value)

                try:
                    print(template_input)
                    errors = question.validate_answer(template_input)
                    if not errors:
                        questionnaire[str(item.id)] = template_input
                        break
                    else:
                        print(f"Errors: {errors}")
                except InvalidTemplateAnswersError as e:
                    print(f"{RED}Validation failed for: {template_input}")
                    for error in e.errors:
                        print(f"    {error['msg']}")
                    print(LINE + RESET)
    except KeyboardInterrupt:
        print("\nKeyboardInterrupt detected. Skipping the rest of the template")

print(LINE)
# Sending the template
if questionnaire:
    device.post_template_data(questionnaire)

# Fetch new data filled on the template
data = device.get_template_data()

# Iterate to check filled data
print(f"[{template.name}] Data post:")
print(LINE)
print("\n".join(f"{k}\t{v}" for k, v in data.items()))

device.close()

Streaming

Gaze data

from pupil_labs.realtime_api.simple import discover_one_device

# Look for devices. Returns as soon as it has found the first device.
print("Looking for the next best device...")
device = discover_one_device(max_search_duration_seconds=10)
if device is None:
    print("No device found.")
    raise SystemExit(-1)

# device.streaming_start()  # optional, if not called, stream is started on-demand

try:
    while True:
        print(device.receive_gaze_datum())
except KeyboardInterrupt:
    pass
finally:
    print("Stopping...")
    # device.streaming_stop()  # optional, if not called, stream is stopped on close
    device.close()  # explicitly stop auto-update

Scene camera video

import cv2
import numpy as np

# Workaround for https://github.com/opencv/opencv/issues/21952
cv2.imshow("cv/av bug", np.zeros(1))
cv2.destroyAllWindows()

from pupil_labs.realtime_api.simple import discover_one_device  # noqa


def main():
    # Look for devices. Returns as soon as it has found the first device.
    print("Looking for the next best device...")
    device = discover_one_device(max_search_duration_seconds=10)
    if device is None:
        print("No device found.")
        raise SystemExit(-1)

    print(f"Connecting to {device}...")

    try:
        while True:
            bgr_pixels, frame_datetime = device.receive_scene_video_frame()
            draw_time(bgr_pixels, frame_datetime)
            cv2.imshow("Scene Camera - Press ESC to quit", bgr_pixels)
            if cv2.waitKey(1) & 0xFF == 27:
                break
    except KeyboardInterrupt:
        pass
    finally:
        print("Stopping...")
        device.close()  # explicitly stop auto-update


def draw_time(frame, time):
    frame_txt_font_name = cv2.FONT_HERSHEY_SIMPLEX
    frame_txt_font_scale = 1.0
    frame_txt_thickness = 1

    # first line: frame index
    frame_txt = str(time)

    cv2.putText(
        frame,
        frame_txt,
        (20, 50),
        frame_txt_font_name,
        frame_txt_font_scale,
        (255, 255, 255),
        thickness=frame_txt_thickness,
        lineType=cv2.LINE_8,
    )


if __name__ == "__main__":
    main()

Eyes camera video

Note

Only available when connecting to a Neon Companion app

import cv2
import numpy as np

# Workaround for https://github.com/opencv/opencv/issues/21952
cv2.imshow("cv/av bug", np.zeros(1))
cv2.destroyAllWindows()

from pupil_labs.realtime_api.simple import discover_one_device  # noqa


def main():
    # Look for devices. Returns as soon as it has found the first device.
    print("Looking for the next best device...")
    device = discover_one_device(max_search_duration_seconds=10)
    if device is None:
        print("No device found.")
        raise SystemExit(-1)

    print(f"Connecting to {device}...")

    try:
        while True:
            bgr_pixels, frame_datetime = device.receive_eyes_video_frame()
            draw_time(bgr_pixels, frame_datetime)
            cv2.imshow("Eyes Camera - Press ESC to quit", bgr_pixels)
            if cv2.waitKey(1) & 0xFF == 27:
                break
    except KeyboardInterrupt:
        pass
    finally:
        print("Stopping...")
        device.close()  # explicitly stop auto-update


def draw_time(frame, time):
    frame_txt_font_name = cv2.FONT_HERSHEY_SIMPLEX
    frame_txt_font_scale = 1.0
    frame_txt_thickness = 1

    # first line: frame index
    frame_txt = str(time)

    cv2.putText(
        frame,
        frame_txt,
        (20, 50),
        frame_txt_font_name,
        frame_txt_font_scale,
        (255, 255, 255),
        thickness=frame_txt_thickness,
        lineType=cv2.LINE_8,
    )


if __name__ == "__main__":
    main()

Camera calibration

Note

Only available when connecting to a Neon Companion app

from pupil_labs.realtime_api.simple import discover_one_device

# Look for devices. Returns as soon as it has found the first device.
print("Looking for the next best device...")
device = discover_one_device(max_search_duration_seconds=10)
if device is None:
    print("No device found.")
    raise SystemExit(-1)

# Device status is fetched on initialization and kept up-to-date in the background

calibration = device.get_calibration()

print("Scene camera matrix:")
print(calibration["scene_camera_matrix"][0])
print("\nScene distortion coefficients:")
print(calibration["scene_distortion_coefficients"][0])

print("\nRight camera matrix:")
print(calibration["right_camera_matrix"][0])
print("\nRight distortion coefficients:")
print(calibration["right_distortion_coefficients"][0])

print("\nLeft camera matrix:")
print(calibration["left_camera_matrix"][0])
print("\nLeft distortion coefficients:")
print(calibration["left_distortion_coefficients"][0])

device.close()

Scene camera video with overlayed gaze

import cv2
import numpy as np

# Workaround for https://github.com/opencv/opencv/issues/21952
cv2.imshow("cv/av bug", np.zeros(1))
cv2.destroyAllWindows()

from pupil_labs.realtime_api.simple import discover_one_device  # noqa


def main():
    # Look for devices. Returns as soon as it has found the first device.
    print("Looking for the next best device...")
    device = discover_one_device(max_search_duration_seconds=10)
    if device is None:
        print("No device found.")
        raise SystemExit(-1)

    print(f"Connecting to {device}...")

    try:
        while True:
            frame, gaze = device.receive_matched_scene_video_frame_and_gaze()
            cv2.circle(
                frame.bgr_pixels,
                (int(gaze.x), int(gaze.y)),
                radius=80,
                color=(0, 0, 255),
                thickness=15,
            )

            cv2.imshow("Scene camera with gaze overlay", frame.bgr_pixels)
            if cv2.waitKey(1) & 0xFF == 27:
                break
    except KeyboardInterrupt:
        pass
    finally:
        print("Stopping...")
        device.close()  # explicitly stop auto-update


if __name__ == "__main__":
    main()

Scene camera video with overlayed eyes video and gaze circle

Note

Only available when connecting to a Neon Companion app

import cv2
import numpy as np

# Workaround for https://github.com/opencv/opencv/issues/21952
cv2.imshow("cv/av bug", np.zeros(1))
cv2.destroyAllWindows()

from pupil_labs.realtime_api.simple import discover_one_device  # noqa


def main():
    # Look for devices. Returns as soon as it has found the first device.
    print("Looking for the next best device...")
    device = discover_one_device(max_search_duration_seconds=10)
    if device is None:
        print("No device found.")
        raise SystemExit(-1)

    print(f"Connecting to {device}...")

    try:
        while True:
            matched = device.receive_matched_scene_and_eyes_video_frames_and_gaze()
            if not matched:
                print(
                    "Not able to find a match! Note: Pupil Invisible does not support "
                    "streaming eyes video"
                )
                continue

            cv2.circle(
                matched.scene.bgr_pixels,
                (int(matched.gaze.x), int(matched.gaze.y)),
                radius=80,
                color=(0, 0, 255),
                thickness=15,
            )

            # Render eyes video into the scene video
            height, width, _ = matched.eyes.bgr_pixels.shape
            matched.scene.bgr_pixels[:height, :width, :] = matched.eyes.bgr_pixels

            cv2.imshow(
                "Scene camera with eyes and gaze overlay", matched.scene.bgr_pixels
            )
            if cv2.waitKey(1) & 0xFF == 27:
                break
    except KeyboardInterrupt:
        pass
    finally:
        print("Stopping...")
        device.close()  # explicitly stop auto-update


if __name__ == "__main__":
    main()

Time Offset Estimation

See pupil_labs.realtime_api.time_echo for details.

from pupil_labs.realtime_api.simple import discover_one_device

# Look for devices. Returns as soon as it has found the first device.
print("Looking for the next best device...")
device = discover_one_device(max_search_duration_seconds=10)
if device is None:
    raise SystemExit("No device found.")

estimate = device.estimate_time_offset()
if estimate is None:
    device.close()
    raise SystemExit("Pupil Companion app is too old")

print(f"Mean time offset: {estimate.time_offset_ms.mean} ms")
print(f"Mean roundtrip duration: {estimate.roundtrip_duration_ms.mean} ms")

device.close()