Skip to content

realtime_api

pupil_labs.realtime_api package.

Python Client for the Pupil Labs Real-Time API

Modules:

Classes:

Functions:

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

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)

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.

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
    )

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

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.

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)

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

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.

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

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

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

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)

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

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

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)

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

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
            )

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)

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)

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)

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)