Module API

The main entrypoint of the API is the Device class. You can use it to

• connect to a Pupil Capture or Pupil Service instance,

• start and stop recordings

• start and stop calibrations

• start plugins

• send messages to the IPC backend

• send notifications

• send annotations

class pupil_labs.pupil_core_network_client.Device(address: str = '127.0.0.1', port: int = 50020, client_clock: ~typing.Callable[[], float] = <built-in function monotonic>, should_auto_reconnect: bool = False)

Bases: object

client_clock: Callable[[], float]

Client clock function. Returns time in seconds.

clock_offset_statistics: ClockOffsetStatistics

Statistic results of the clock offset estimation

connect()

current_pupil_time() → float

disconnect()

estimate_client_to_remote_clock_offset(num_measurements: int = 10) → ClockOffsetStatistics

Returns the clock offset after multiple measurements to reduce the effect of varying network delay.

Since the network connection to Pupil Capture/Service is not necessarily stable, one has to assume that the delays to send and receive commands are not symmetrical and might vary. To reduce the possible clock-offset estimation error, this function repeats the measurement multiple times and returns the mean clock offset. The variance of these measurements is expected to be higher for remote connections (two different computers) than for local connections (script and Core software running on the same computer). You can easily extend this function to perform further statistical analysis on your clock-offset measurements to examine the accuracy of the time sync.

Description taken and code adopted from pupil helpers remote_annotations.py

high_frequency_message_sending()

Context manager that improves the efficiency of send_message()

Instead of sending the message to Pupil Remote via the REQ socket and waiting for a response, the device will use a PUB socket connected directly to the Pupil Capture IPC. As a result, the client is able to send messages with a considerable higher frequency, e.g. streaming scene and eye videos to the HMD video backend.

Example:

device = Device()
with device.high_frequency_message_sending():
    device.send_message(...)

property is_connected

measure_one_client_to_remote_clock_offset()

Calculates the offset between the Pupil Core software clock and a local clock

Requesting the remote pupil time takes time. This delay needs to be considered when calculating the clock offset. We measure the local time before (A) and after (B) the request and assume that the remote pupil time was measured at (A+B)/2, i.e. the midpoint between A and B. As a result, we have two measurements from two different clocks that were taken assumingly at the same point in time. The difference between them (“clock offset”) allows us, given a new local clock measurement, to infer the corresponding time on the remote clock.

Description taken and code adopted from pupil helpers remote_annotations.py

request_calibration_start() → str

request_calibration_stop() → str

request_current_pupil_time() → float

request_plugin_start(plugin_class_name: str, args: dict | None = None) → str

request_plugin_start_eye_process(eye_id: Literal[0, 1], plugin_class_name: str, args: dict | None = None) → str

request_recording_start(session_name: str | None = None) → str

request_recording_stop() → str

request_version() → str

send_annotation(label: str, timestamp: float | None = None, **kwargs) → str

send_message(payload: dict) → str

send_notification(notification: dict) → str

Sends notification to Pupil Remote

subscribe(topics: Union[str, Sequence[str]]) → Subscription

subscribe_in_background(topics: Union[str, Sequence[str]], buffer_size: int | None = None) → Subscription

The device class calculates the clock offset between the client clock and the pupil time base of the connected Pupil Core software. To account for varying network delay, it performs multiple measurements. The resulting statistics are exposed via the clock_offset_statistics attibute.

class pupil_labs.pupil_core_network_client.ClockOffsetStatistics(mean_offset, std_offset, num_measurements)

Bases: NamedTuple

mean_offset: float

Clock offset mean, in seconds

num_measurements: int

Number of measurements (at least 2)

std_offset: float

Clock offset standard deviation, in seconds

Subscriptions are implemented in pupil_labs.pupil_core_network_client.subscription. Use pupil_labs.pupil_core_network_client.Device.subscribe() and pupil_labs.pupil_core_network_client.Device.subscribe_in_background() as entry points.

class pupil_labs.pupil_core_network_client.subscription.BackgroundSubscription(*args, buffer_size: int | None, **kwargs)

Bases: Subscription

Process subscription in background and buffer recent messages

connect()

disconnect()

property has_new_message: bool

property is_connected

recv_new_message(timeout_ms: int | None = None) → Message

Recv a message with topic, payload. Topic is a utf-8 encoded string. Returned as unicode object. Payload is a msgpack serialized dict. Returned as a python dict. Any addional message frames will be added as a list in the payload dict with key: ‘__raw_data__’ .

class pupil_labs.pupil_core_network_client.subscription.Message(topic, payload)

Bases: NamedTuple

payload: dict

Message payload

property raw_data: Optional[Sequence[ByteString]]

topic: str

Message topic

class pupil_labs.pupil_core_network_client.subscription.Subscription(address: str = '127.0.0.1', *, port: int, topics: Union[str, Sequence[str]])

Bases: object

connect()

disconnect()

property has_new_message: bool

property is_connected

recv_new_message(timeout_ms: int | None = None) → pupil_labs.pupil_core_network_client.subscription.Message | None

Recv a message with topic, payload. Topic is a utf-8 encoded string. Returned as unicode object. Payload is a msgpack serialized dict. Returned as a python dict. Any addional message frames will be added as a list in the payload dict with key: ‘__raw_data__’ .