Example usage

To run the examples, install the package using the examples extras requirement:

pip install -e pupil-core-network-client[examples]

Connect, and start and stop a recording

import argparse
import time

import pupil_labs.pupil_core_network_client as pcnc


def main(address: str, port: int):
    device = pcnc.Device(address, port)

    print("Version:", device.request_version())
    print("Start recording:", device.request_recording_start())
    print("Waiting 5 seconds..")
    time.sleep(5.0)
    print("Stop recording:", device.request_recording_stop())


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("-a", "--address", type=str, default="127.0.0.1")
    parser.add_argument("-p", "--port", type=int, default=50020)
    args = parser.parse_args()

    main(args.address, args.port)

Print the estimated clock offset and current Pupil time

import argparse

import pupil_labs.pupil_core_network_client as pcnc


def main(address: str, port: int):
    device = pcnc.Device(address, port)

    print(
        f"Measured offset between client clock ({device.client_clock}) and pupil "
        f"time:\n\t{device.clock_offset_statistics}"
    )

    print("Current pupil time:", device.current_pupil_time())


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("-a", "--address", type=str, default="127.0.0.1")
    parser.add_argument("-p", "--port", type=int, default=50020)
    args = parser.parse_args()

    main(args.address, args.port)

Start the Annotation Plugin and send custom annotions

import argparse
import time

import pupil_labs.pupil_core_network_client as pcnc


def main(address: str, port: int):
    device = pcnc.Device(address, port)

    print(
        "Requesting annotation plugin start...",
        device.request_plugin_start("Annotation_Capture"),
    )
    print("Wait a second for the request to be processed...")
    time.sleep(1.0)

    print(
        "Sending annotation with automatic timestamp",
        device.send_annotation(label="automatic timestamp"),
    )

    print(
        "Sending annotation with custom timestamp",
        # Timestamps should be in Pupil time. By subtracting 5 seconds, the annotation
        # will be associated with a point in time 5 seconds ago.
        device.send_annotation(
            label="custom timestamp", timestamp=device.current_pupil_time() - 5.0
        ),
    )

    print(
        "Sending annotation with custom fields",
        # Custom fields can be passed as keyword arguments and will be saved to their
        # dedicated column during the Pupil Player export
        device.send_annotation(
            label="custom fields", trial=5, subject="me", condition="eye tracking"
        ),
    )


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("-a", "--address", type=str, default="127.0.0.1")
    parser.add_argument("-p", "--port", type=int, default=50020)
    args = parser.parse_args()

    main(args.address, args.port)

Stream Video From Pupil Capture

import argparse
import contextlib
import logging
import time

import pupil_labs.pupil_core_network_client as pcnc


def main(address: str, port: int, max_frame_rate_hz: int):
    device = pcnc.Device(address, port)
    device.send_notification(
        {"subject": "frame_publishing.set_format", "format": "bgr"}
    )

    with contextlib.suppress(KeyboardInterrupt):
        with device.subscribe_in_background("frame.world", buffer_size=1) as sub:
            while True:
                message = sub.recv_new_message()
                print(
                    f"[{message.payload['timestamp']}] {message.payload['topic']} "
                    f"{message.payload['index']}"
                )
                time.sleep(1 / max_frame_rate_hz)


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("-a", "--address", type=str, default="127.0.0.1")
    parser.add_argument("-p", "--port", type=int, default=50020)
    parser.add_argument("-fps", "--max-frame-rate", type=int, default=10)
    parser.add_argument("--debug", action="store_true")
    args = parser.parse_args()
    logging.basicConfig(level=logging.DEBUG if args.debug else logging.WARNING)

    main(args.address, args.port, args.max_frame_rate)

Stream Video to Pupil Capture

This example shows how start world and eye plugins and streaming BGR data to Pupil Capture’s HMD Streaming backend.

import argparse
import contextlib
import time

import numpy as np

import pupil_labs.pupil_core_network_client as pcnc

current_image = np.zeros((400, 600, 3), dtype=np.uint8)


def main(address: str, port: int, frame_rate_hz: int):
    device = pcnc.Device(address, port)

    source_class_name = "HMD_Streaming_Source"
    frame_topic = "hmd_streaming.custom"
    device.request_plugin_start(source_class_name, args={"topics": (frame_topic,)})
    for eye_id in range(2):
        device.request_plugin_start_eye_process(
            eye_id, source_class_name, args={"topics": (frame_topic,)}
        )

    with contextlib.suppress(KeyboardInterrupt):
        # Enable sending messages directly to the IPC backbone instead of having
        # Pupil Remote forward every message one by one and waiting for a response each
        # time. This allows sending messages with a much higher rate.
        with device.high_frequency_message_sending():
            increasing_index = 0
            while True:
                send_image(
                    device,
                    gray_image(increasing_index),
                    frame_topic,
                    increasing_index,
                    timestamp=device.current_pupil_time(),
                )
                increasing_index += 1
                time.sleep(1 / frame_rate_hz)


def send_image(device: pcnc.Device, image, topic: str, index: int, timestamp: float):
    height, width, depth = image.shape
    device.send_message(
        {
            "format": "bgr",
            "projection_matrix": [  # dummy pin-hole camera intrinsics
                [1000, 0.0, width / 2.0],
                [0.0, 1000, height / 2.0],
                [0.0, 0.0, 1.0],
            ],
            "topic": topic,
            "width": width,
            "height": height,
            "index": index,
            "timestamp": timestamp,
            "__raw_data__": [image],
        }
    )


def gray_image(color_seed: int):
    """Return an image with a gray value between 85 and 170"""
    return current_image + (color_seed % 85) + 85


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("-a", "--address", type=str, default="127.0.0.1")
    parser.add_argument("-p", "--port", type=int, default=50020)
    parser.add_argument("-fps", "--frame-rate", type=int, default=60)
    args = parser.parse_args()

    main(args.address, args.port, args.frame_rate)