Skip to content

Connect to a Device

One of the first steps you need to carry to leverage the API is to connect to one device. This library offers you ways to find one or multiple devices connected to your local network. See below the different ways to connect to a device:

Using the discover_devices:

discover_devices.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
import asyncio
import contextlib

from pupil_labs.realtime_api.discovery import Network, discover_devices


async def main():
    async with Network() as network:
        print("Looking for the next best device...\n\t", end="")
        print(await network.wait_for_new_device(timeout_seconds=5))

        print("---")
        print("All devices after searching for additional 5 seconds:")
        await asyncio.sleep(5)
        print(network.devices)

    print("---")
    print("Starting new, indefinitive search... hit ctrl-c to stop.")
    # optionally set timeout_seconds argument to limit search duration
    async for device_info in discover_devices():
        print(f"\t{device_info}")


if __name__ == "__main__":
    with contextlib.suppress(KeyboardInterrupt):
        asyncio.run(main())

from pupil_labs.realtime_api.device import Device

# This address is just an example. Find out the actual IP address of your device!
ip = "192.168.1.169"
device = Device(address=ip, port="8080")

Make sure the Companion App is running! If no device can be found, please have a look at the troubleshooting section.

Below you can find a link to the full code example and the API referece for the returned Device object.

Device Information & Automatic Status Updates

Once connected, the Device object, alows you to retrieve Status updates by calling get_status.

This Status represents the full Companion's Device state, including sub-classes representing:

get_status.py
    async with Device.from_discovered_device(dev_info) as device:
        status = await device.get_status()

        print(f"Device IP address: {status.phone.ip}")
        print(f"Battery level: {status.phone.battery_level} %")

        print(f"Connected glasses: SN {status.hardware.glasses_serial}")
        print(f"Connected scene camera: SN {status.hardware.world_camera_serial}")

        world = status.direct_world_sensor()
        print(f"World sensor: connected={world.connected} url={world.url}")

        gaze = status.direct_gaze_sensor()
        print(f"Gaze sensor: connected={gaze.connected} url={gaze.url}")

Device IP address: 192.168.1.60
Battery level: 78 %
Connected glasses: SN -1
Connected scene camera: SN -1
World sensor: connected=True url=rtsp://192.168.1.60:8086/?camera=world&audioenable=on
Gaze sensor: connected=True url=rtsp://192.168.1.60:8086/?camera=gaze&audioenable=on


    async with Device.from_discovered_device(dev_info) as device:
        duration = 20
        print(f"Starting auto-update for {duration} seconds")
        # callbacks can be awaitable, too
        notifier = StatusUpdateNotifier(device, callbacks=[print_component])
        await notifier.receive_updates_start()
        await asyncio.sleep(duration)
        print("Stopping auto-update")
        await notifier.receive_updates_stop()

Starting auto-update for 20 seconds
Phone(battery_level=77, battery_state='OK', device_id='d55a33b5ab845785', device_name='Neon Companion', ip='192.168.1.60', memory=99877777408, memory_state='OK', time_echo_port=12321)
Hardware(version='2.0', glasses_serial='-1', world_camera_serial='-1', module_serial='841684')
Sensor(sensor='imu', conn_type='WEBSOCKET', connected=True, ip='192.168.1.60', params='camera=imu&audioenable=off', port=8686, protocol='rtsp', stream_error=False)
Sensor(sensor='imu', conn_type='DIRECT', connected=True, ip='192.168.1.60', params='camera=imu&audioenable=on', port=8086, protocol='rtsp', stream_error=False)
Sensor(sensor='world', conn_type='WEBSOCKET', connected=True, ip='192.168.1.60', params='camera=world&audioenable=off', port=8686, protocol='rtsp', stream_error=False)

Refer to the Device API documentation for more details.

Device

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
DeviceBase

DeviceBase 

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

Bases: ABC

Abstract base class representing Realtime API host devices.

This class provides the foundation for device implementations that connect to the Realtime API.

Attributes:

  • address (str) –

    REST API server address.

  • port (int) –

    REST API server port.

  • full_name (str | None) –

    Full service discovery name.

  • dns_name (str | None) –

    REST API server DNS name, e.g.neon.local / pi.local..

Parameters:

  • address (str) –

    REST API server address.

  • port (int) –

    REST API server port.

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

    Full service discovery name.

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

    REST API server DNS name, e.g.neon.local / pi.local..

  • suppress_decoding_warnings (bool, default: True ) –

    Whether to suppress libav decoding warnings.

Methods:

Source code in src/pupil_labs/realtime_api/base.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def __init__(
    self,
    address: str,
    port: int,
    full_name: str | None = None,
    dns_name: str | None = None,
    suppress_decoding_warnings: bool = True,
):
    """Initialize the DeviceBase instance.

    Args:
        address (str): REST API server address.
        port (int): REST API server port.
        full_name (str | None): Full service discovery name.
        dns_name (str | None): REST API server DNS name,
            e.g.``neon.local / pi.local.``.
        suppress_decoding_warnings: Whether to suppress libav decoding warnings.

    """
    self.address: str = address
    self.port: int = port
    self.full_name: str | None = full_name
    self.dns_name: str | None = dns_name
    if suppress_decoding_warnings:
        # suppress decoding warnings due to incomplete data transmissions
        logging.getLogger("libav.h264").setLevel(logging.CRITICAL)
        logging.getLogger("libav.swscaler").setLevel(logging.ERROR)

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
    )

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

Status dataclass 

Status(phone: Phone, hardware: Hardware, sensors: list[Sensor], recording: Recording | None)

Represents the Companion's Device full status

Methods:

Attributes:

hardware instance-attribute 

hardware: Hardware

Information about glasses connected, won't be present if not connected

phone instance-attribute 

phone: Phone

Information about the connected phone. Always present.

recording instance-attribute 

recording: Recording | None

Current recording, if any.

sensors instance-attribute 

sensors: list[Sensor]

"List of sensor information.

direct_eye_events_sensor 

direct_eye_events_sensor() -> Sensor | None

Get blinks, fixations sensor.

Only available on Neon with Companion App version 2.9 or newer.

Source code in src/pupil_labs/realtime_api/models.py 
460
461
462
463
464
465
466
467
468
469
470
471
def direct_eye_events_sensor(self) -> Sensor | None:
    """Get blinks, fixations _sensor_.

    Only available on Neon with Companion App version 2.9 or newer.
    """
    return next(
        self.matching_sensors(SensorName.EYE_EVENTS, ConnectionType.DIRECT),
        Sensor(
            sensor=SensorName.EYE_EVENTS.value,
            conn_type=ConnectionType.DIRECT.value,
        ),
    )

direct_eyes_sensor 

direct_eyes_sensor() -> Sensor | None

Get the eye camera sensor with direct connection. Only available on Neon.

Source code in src/pupil_labs/realtime_api/models.py 
453
454
455
456
457
458
def direct_eyes_sensor(self) -> Sensor | None:
    """Get the eye camera sensor with direct connection. Only available on Neon."""
    return next(
        self.matching_sensors(SensorName.EYES, ConnectionType.DIRECT),
        Sensor(sensor=SensorName.EYES.value, conn_type=ConnectionType.DIRECT.value),
    )

direct_gaze_sensor 

direct_gaze_sensor() -> Sensor | None

Get the gaze sensor with direct connection.

Source code in src/pupil_labs/realtime_api/models.py 
439
440
441
442
443
444
def direct_gaze_sensor(self) -> Sensor | None:
    """Get the gaze sensor with direct connection."""
    return next(
        self.matching_sensors(SensorName.GAZE, ConnectionType.DIRECT),
        Sensor(sensor=SensorName.GAZE.value, conn_type=ConnectionType.DIRECT.value),
    )

direct_imu_sensor 

direct_imu_sensor() -> Sensor | None

Get the IMU sensor with direct connection.

Source code in src/pupil_labs/realtime_api/models.py 
446
447
448
449
450
451
def direct_imu_sensor(self) -> Sensor | None:
    """Get the IMU sensor with direct connection."""
    return next(
        self.matching_sensors(SensorName.IMU, ConnectionType.DIRECT),
        Sensor(sensor=SensorName.IMU.value, conn_type=ConnectionType.DIRECT.value),
    )

direct_world_sensor 

direct_world_sensor() -> Sensor | None

Get the scene camera sensor with direct connection.

Note

Pupil Invisible devices, the world camera can be detached

Source code in src/pupil_labs/realtime_api/models.py 
425
426
427
428
429
430
431
432
433
434
435
436
437
def direct_world_sensor(self) -> Sensor | None:
    """Get the scene camera sensor with direct connection.

    Note:
        Pupil Invisible devices, the world camera can be detached

    """
    return next(
        self.matching_sensors(SensorName.WORLD, ConnectionType.DIRECT),
        Sensor(
            sensor=SensorName.WORLD.value, conn_type=ConnectionType.DIRECT.value
        ),
    )

from_dict classmethod 

from_dict(status_json_result: list[ComponentRaw]) -> Status

Create a Status from a list of raw components.

Parameters:

  • status_json_result (list[ComponentRaw]) –

    List of raw component dictionaries.

Returns:

  • Status ( Status ) –

    New Status instance.

Source code in src/pupil_labs/realtime_api/models.py 
342
343
344
345
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
371
372
373
374
375
376
377
378
@classmethod
def from_dict(cls, status_json_result: list[ComponentRaw]) -> "Status":
    """Create a Status from a list of raw components.

    Args:
        status_json_result: List of raw component dictionaries.

    Returns:
        Status: New Status instance.

    """
    phone = None
    recording = None
    hardware = Hardware()
    sensors = []
    for dct in status_json_result:
        try:
            component = parse_component(dct)
        except UnknownComponentError:
            logger.warning(f"Dropping unknown component: {dct}")
            continue
        if isinstance(component, Phone):
            phone = component
        elif isinstance(component, Hardware):
            hardware = component
        elif isinstance(component, Sensor):
            sensors.append(component)
        elif isinstance(component, Recording):
            recording = component
        elif isinstance(component, NetworkDevice):
            pass  # no need to handle NetworkDevice updates here
        else:
            logger.warning(f"Unknown model class: {type(component).__name__}")
    sensors.sort(key=lambda s: (not s.connected, s.conn_type, s.sensor))
    if not phone:
        raise ValueError("Status data must include a 'Phone' component.")
    return cls(phone, hardware, sensors, recording)

matching_sensors 

matching_sensors(name: SensorName, connection: ConnectionType) -> Iterator[Sensor]

Find sensors matching specified criteria.

Parameters:

  • name (SensorName) –

    Sensor name to match, or ANY to match any name.

  • connection (ConnectionType) –

    Connection type to match, or ANY to match any type.

Yields:

  • Sensor ( Sensor ) –

    Sensors matching the criteria.

Source code in src/pupil_labs/realtime_api/models.py 
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
def matching_sensors(
    self, name: SensorName, connection: ConnectionType
) -> Iterator[Sensor]:
    """Find sensors matching specified criteria.

    Args:
        name: Sensor name to match, or ANY to match any name.
        connection: Connection type to match, or ANY to match any type.

    Yields:
        Sensor: Sensors matching the criteria.

    """
    for sensor in self.sensors:
        if name is not SensorName.ANY and sensor.sensor != name.value:
            continue
        if (
            connection is not ConnectionType.ANY
            and sensor.conn_type != connection.value
        ):
            continue
        yield sensor

update 

update(component: Component) -> None

Update Component.

Parameters:

  • component (Component) –

    Component to update.

Source code in src/pupil_labs/realtime_api/models.py 
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
def update(self, component: Component) -> None:
    """Update Component.

    Args:
        component: Component to update.

    """
    if isinstance(component, Phone):
        self.phone = component
    elif isinstance(component, Hardware):
        self.hardware = component
    elif isinstance(component, Recording):
        self.recording = component
    elif isinstance(component, Sensor):
        for idx, sensor in enumerate(self.sensors):
            if (
                sensor.sensor == component.sensor
                and sensor.conn_type == component.conn_type
            ):
                self.sensors[idx] = component
                break

Full Code Examples

Check the whole example code here

discover_devices.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
import asyncio
import contextlib

from pupil_labs.realtime_api.discovery import Network, discover_devices


async def main():
    async with Network() as network:
        print("Looking for the next best device...\n\t", end="")
        print(await network.wait_for_new_device(timeout_seconds=5))

        print("---")
        print("All devices after searching for additional 5 seconds:")
        await asyncio.sleep(5)
        print(network.devices)

    print("---")
    print("Starting new, indefinitive search... hit ctrl-c to stop.")
    # optionally set timeout_seconds argument to limit search duration
    async for device_info in discover_devices():
        print(f"\t{device_info}")


if __name__ == "__main__":
    with contextlib.suppress(KeyboardInterrupt):
        asyncio.run(main())
device_status_get_current.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
import asyncio

from pupil_labs.realtime_api import Device, Network


async def main():
    async with Network() as network:
        dev_info = await network.wait_for_new_device(timeout_seconds=5)
    if dev_info is None:
        print("No device could be found! Abort")
        return

    async with Device.from_discovered_device(dev_info) as device:
        status = await device.get_status()

        print(f"Device IP address: {status.phone.ip}")
        print(f"Battery level: {status.phone.battery_level} %")

        print(f"Connected glasses: SN {status.hardware.glasses_serial}")
        print(f"Connected scene camera: SN {status.hardware.world_camera_serial}")

        world = status.direct_world_sensor()
        print(f"World sensor: connected={world.connected} url={world.url}")

        gaze = status.direct_gaze_sensor()
        print(f"Gaze sensor: connected={gaze.connected} url={gaze.url}")


if __name__ == "__main__":
    asyncio.run(main())

device_status_update_via_callback.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
import asyncio

from pupil_labs.realtime_api import Device, Network, StatusUpdateNotifier


def print_component(component):
    print(component)


async def main():
    async with Network() as network:
        dev_info = await network.wait_for_new_device(timeout_seconds=5)
    if dev_info is None:
        print("No device could be found! Abort")
        return

    async with Device.from_discovered_device(dev_info) as device:
        duration = 20
        print(f"Starting auto-update for {duration} seconds")
        # callbacks can be awaitable, too
        notifier = StatusUpdateNotifier(device, callbacks=[print_component])
        await notifier.receive_updates_start()
        await asyncio.sleep(duration)
        print("Stopping auto-update")
        await notifier.receive_updates_stop()


if __name__ == "__main__":
    asyncio.run(main())