g3pylib

Glasses3 Python Library

This library is still in alpha. It's not yet feature complete and there are no guarantees that the API will be stable.

A Python client for Glasses3. It handles all the low level details of communication with the Glasses3 websocket API and exposes a Python API which includes all the endpoints in the websocket API of the Glasses3 as well as some extra convenience methods. It also implements streaming with RTSP and service discovery with Zeroconf.

Note

This library should be platform independent in theory but the development has been done mainly in Windows environments and it's only briefly tested on other platforms.

Installation

For the moment we only support Python 3.10.

To install this package, clone it and use either

flit install

or

pip install .

To run examples or tests you need some extra dependencies which are installed by default with the flit command. If you are using pip the extra dependencies can be installed alongside the library with

pip install '.[test, examples, example-app]'

on windows: pip install ".[test, examples, example-app]"

Documentation

The library documentation can be found here and there is also a developer guide for the glasses API in PDF format which can be downloaded here.

Environment

The tests and examples load the glasses hostname, which by default is the serial number, from the .env file in the project root folder. See example content below:

G3_HOSTNAME=tg03b-080200045321

You can also specify this variable directly in your environment.

Examples

The example folder contains a few smaller examples showcasing different use cases of the library as well as a larger controller application with a simple GUI.

Run the example app with python examples/g3pycontroller.

Note, the examples use OpenCV, and there is a known issue with OpenCV and PyAv. If you experience freezes when displaying video frames in the samples, please check out the workarounds in the OpenCV repo. Thanks @edavalosanaya for mentioning this in https://github.com/tobiipro/g3pylib/issues/83.

Contributing

More information on how to contribute to this project can be found in CONTRIBUTING.md.

It contains developer guidelines as well as information on how to run tests and enable logging.

g3pylib is a python wrapper around the Glasses3 web API which lets you control Glasses3 devices.

API Endpoints

All endpoints in the g3pylib module corresponding to an endpoint in the Glasses3 web API are undocumented and placed first in each module. The following naming convention is used to translate web API endpoint names to g3pylib API endpoint names:

  • Properties: example_property -> get_example_property/set_example_property
  • Actions: example_action -> example_action
  • Signals: example_signal -> subscribe_to_example_signal

The web API endpoints can be browsed in the Glasses3 Example web client accessed via http://your-g3-address.

Useful information

In any code examples, g3 will be a connected instance of Glasses3.

The default hostname of a Glasses3 device is its serial number.

  1"""
  2.. include:: ../../README.md
  3
  4g3pylib is a python wrapper around the Glasses3 web API which lets you control Glasses3 devices.
  5
  6## API Endpoints
  7All endpoints in the `g3pylib` module corresponding to an endpoint in the Glasses3 web API are undocumented and placed first in each module.
  8The following naming convention is used to translate web API endpoint names to `g3pylib
  9` API endpoint names:
 10 - Properties: example_property -> get_example_property/set_example_property
 11 - Actions: example_action -> example_action
 12 - Signals: example_signal -> subscribe_to_example_signal
 13
 14The web API endpoints can be browsed in the Glasses3 Example web client accessed via http://*your-g3-address*.
 15
 16## Useful information
 17In any code examples, `g3` will be a connected instance of `Glasses3`.
 18
 19The default hostname of a Glasses3 device is its serial number.
 20"""
 21from __future__ import annotations
 22
 23import logging
 24from contextlib import asynccontextmanager
 25from types import TracebackType
 26from typing import Any, AsyncIterator, Coroutine, Generator, Optional, Tuple, Type, cast
 27
 28
 29import g3pylib.websocket
 30from g3pylib._utils import APIComponent
 31from g3pylib.calibrate import Calibrate
 32from g3pylib.exceptions import FeatureNotAvailableError
 33from g3pylib.g3typing import URI, LoggerLike
 34from g3pylib.recorder import Recorder
 35from g3pylib.recordings import Recordings
 36from g3pylib.rudimentary import Rudimentary
 37from g3pylib.settings import Settings
 38from g3pylib.streams import Streams
 39from g3pylib.system import System
 40from g3pylib.websocket import G3WebSocketClientProtocol
 41from g3pylib.zeroconf import DEFAULT_WEBSOCKET_PATH, G3Service, G3ServiceDiscovery
 42
 43__version__ = "0.3.1-alpha"
 44
 45DEFAULT_RTSP_LIVE_PATH = "/live/all"
 46DEFAULT_RTSP_PORT = 8554
 47DEFAULT_HTTP_PORT = 80
 48
 49_logger = logging.getLogger(__name__)
 50
 51
 52class Glasses3(APIComponent):
 53    """
 54    Represents a Glasses3 device.
 55
 56    Holds the API components and a WebSocket connection to a Glasses3 device.
 57    The `stream_rtsp` context can be used for live stream functionality.
 58
 59    For the recommended way to create a connected instance of Glasses3, see `connect_to_glasses`.
 60    """
 61
 62    def __init__(
 63        self,
 64        connection: G3WebSocketClientProtocol,
 65        rtsp_url: Optional[str],
 66        http_url: Optional[str],
 67        logger: Optional[LoggerLike] = None,
 68    ) -> None:
 69        self.logger: LoggerLike = (
 70            logging.getLogger(__name__) if logger is None else logger
 71        )
 72        self._rtsp_url = rtsp_url
 73        self._http_url = http_url
 74        self._connection: G3WebSocketClientProtocol = connection
 75        self._recorder: Optional[Recorder] = None
 76        self._recordings: Optional[Recordings] = None
 77        self._rudimentary: Optional[Rudimentary] = None
 78        self._system: Optional[System] = None
 79        self._calibrate: Optional[Calibrate] = None
 80        self._settings: Optional[Settings] = None
 81
 82    @property
 83    def calibrate(self) -> Calibrate:
 84        if self._calibrate is None:
 85            self._calibrate = Calibrate(self._connection, URI("/calibrate"))
 86        return self._calibrate
 87
 88    @property
 89    def recorder(self) -> Recorder:
 90        if self._recorder is None:
 91            self._recorder = Recorder(self._connection, URI("/recorder"))
 92        return self._recorder
 93
 94    @property
 95    def recordings(self) -> Recordings:
 96        if self._recordings is None:
 97            self._recordings = Recordings(
 98                self._connection, URI("/recordings"), self._http_url
 99            )
100        return self._recordings
101
102    @property
103    def rudimentary(self) -> Rudimentary:
104        if self._rudimentary is None:
105            self._rudimentary = Rudimentary(self._connection, URI("/rudimentary"))
106        return self._rudimentary
107
108    @property
109    def system(self) -> System:
110        if self._system is None:
111            self._system = System(self._connection, URI("/system"))
112        return self._system
113
114    @property
115    def settings(self) -> Settings:
116        if self._settings is None:
117            self._settings = Settings(self._connection, URI("/settings"))
118        return self._settings
119
120    @property
121    def rtsp_url(self) -> Optional[str]:
122        """The RTSP URL used for live stream."""
123        return self._rtsp_url
124
125    @asynccontextmanager
126    async def stream_rtsp(
127        self,
128        scene_camera: bool = True,
129        audio: bool = False,
130        eye_cameras: bool = False,
131        gaze: bool = False,
132        sync: bool = False,
133        imu: bool = False,
134        events: bool = False,
135    ) -> AsyncIterator[Streams]:
136        """Set up an RTSP connection in the form of a Streams object with the Stream properties indicated by the arguments.
137
138        The Stream objects can be used to demux/decode their stream. For example, `stream_rtsp()` can be used as follows:
139        ```python
140        async with connect_to_glasses(g3_hostname) as g3:
141            async with g3.stream_rtsp() as streams:
142                async with streams.scene_camera.decode() as decoded_stream:
143                    for _ in range(500):
144                        frame, _timestamp = await decoded_stream.get()
145                        image = frame.to_ndarray(format="bgr24")
146                        cv2.imshow("Video", image)
147                        cv2.waitKey(1)
148        ```
149
150        *Alpha version note:* Only the scene_camera, eye_camera and gaze attributes are implemented so far.
151        """
152        if self.rtsp_url is None:
153            raise FeatureNotAvailableError(
154                "This Glasses3 object was initialized without a proper RTSP url."
155            )
156        async with Streams.connect(
157            self.rtsp_url,
158            scene_camera=scene_camera,
159            audio=audio,
160            eye_cameras=eye_cameras,
161            gaze=gaze,
162            sync=sync,
163            imu=imu,
164            events=events,
165        ) as streams:
166            await streams.play()
167            yield streams
168
169    async def close(self) -> None:
170        """Close down the underlying websocket connection to the Glasses3 device."""
171        await self._connection.close()
172
173
174class connect_to_glasses:
175    """This class contains a set of classmethods which are used to connect to a pair of glasses.
176
177    The preferred way to use this class is as an async context manager like this:
178
179    ```python
180    async with connect_to_glasses.with_hostname(glasses_serial_number) as g3:
181        # Here you can call the glasses
182        await g3.get_name()
183    # Here the connection is closed
184    ```
185
186    It does however also support usage without a `with`-block like this:
187
188    ```python
189    g3 = await connect_to_glasses.with_hostname(glasses_serial_number)
190    # Here you can call the glasses
191    await g3.get_name()
192    # You have to remember to close the connection like this:
193    g3.close()
194    # And here g3 still exists but it is unusable
195    ```
196    """
197
198    def __init__(
199        self,
200        url_generator: Coroutine[Any, Any, Tuple[str, Optional[str], Optional[str]]],
201    ) -> None:
202        """You should probably not use this constructor unless you need to generate the URLs to your glasses in a very specific way.
203        The regular use cases are covered in the alternative constructors below: `with_url`, `with_zeroconf`, `with_hostname` and `with_service`.
204
205        If you want to use this constructor you need to supply a couroutine which returns a tuple that contains two URLs.
206        The first URL should point to the websocket and the second URL should point to the RTSP endpoint.
207        """
208        self.url_generator = url_generator
209
210    @staticmethod
211    async def _urls_from_zeroconf(
212        using_ip: bool, timeout: float
213    ) -> Tuple[str, Optional[str], Optional[str]]:
214        async with G3ServiceDiscovery.listen() as service_discovery:
215            service = await service_discovery.wait_for_single_service(
216                service_discovery.events,
217                timeout,
218            )
219        return await connect_to_glasses._urls_from_service(service, using_ip)
220
221    @staticmethod
222    async def _urls_from_service(
223        service: G3Service, using_ip: bool
224    ) -> Tuple[str, Optional[str], Optional[str]]:
225        return (
226            service.ws_url(using_ip),
227            service.rtsp_url(using_ip),
228            service.http_url(using_ip),
229        )
230
231    @staticmethod
232    async def _urls_from_hostname(
233        hostname: str, using_zeroconf: bool, using_ip: bool
234    ) -> Tuple[str, Optional[str], Optional[str]]:
235        if not using_zeroconf:
236            return (
237                f"ws://{hostname}{DEFAULT_WEBSOCKET_PATH}",
238                f"rtsp://{hostname}:{DEFAULT_RTSP_PORT}{DEFAULT_RTSP_LIVE_PATH}",
239                f"http://{hostname}:{DEFAULT_HTTP_PORT}",
240            )
241        else:
242            service = await G3ServiceDiscovery.request_service(hostname)
243            return await connect_to_glasses._urls_from_service(service, using_ip)
244
245    @classmethod
246    def with_zeroconf(
247        cls, using_ip: bool = True, timeout: float = 3000
248    ) -> connect_to_glasses:
249        """Connects by listening for available glasses on the network using zeroconf.
250        Connects to the first pair of glasses that answers so if there are multiple glasses on the
251        network the behavior is undefined.
252
253        If `using_ip` is set to True (default) we will generate the the URL used for connection with the ip.
254        If it's set to False we will use the hostname, which will depend on DNS working as it should.
255
256        `timeout` defines the time in milliseconds before `asyncio.TimeoutError` is raised.
257        """
258        return cls(cls._urls_from_zeroconf(using_ip, timeout))
259
260    @classmethod
261    def with_hostname(
262        cls, hostname: str, using_zeroconf: bool = False, using_ip: bool = True
263    ) -> connect_to_glasses:
264        """Connects to the pair of glasses with the given hostname.
265
266        If `using_zeroconf` is set to False (default) we will not depend on zeroconf
267        for fetching details on how to generate the URL and instead use detault values for the URL components specified
268        in the [developer guide](https://www.tobiipro.com/product-listing/tobii-pro-glasses3-api/#ResourcesSpecifications).
269        If it's set to True, all URL components are fetched with zeroconf.
270
271        `using_ip` specifies if the ip or the hostname should be used in the URL used for connecting when zeroconf is used.
272        If the hostname is used, it depends on DNS working as it should."""
273        return cls(cls._urls_from_hostname(hostname, using_zeroconf, using_ip))
274
275    @classmethod
276    def with_service(
277        cls, service: G3Service, using_ip: bool = True
278    ) -> connect_to_glasses:
279        """Connects to the pair of glasses referred to by the given service.
280
281        `using_ip` specifies if the ip or the hostname should be used in the URL used for connecting.
282        If the hostname is used, it depends on DNS working as it should.
283        """
284        return cls(cls._urls_from_service(service, using_ip))
285
286    @classmethod
287    def with_url(
288        cls, ws_url: str, rtsp_url: Optional[str] = None, http_url: Optional[str] = None
289    ):
290        """Connects to the pair of glasses at the specified URL. `ws_url` should
291        be a websocket URL (starting with `ws://`) and `rtsp_url` should be an RTSP
292        url (starting with `rtsp://` or `rtspt://`)."""
293
294        async def urls():
295            return (ws_url, rtsp_url, http_url)
296
297        return cls(urls())
298
299    def __await__(self) -> Generator[Any, None, Glasses3]:
300        return self.__await_impl__().__await__()
301
302    async def __await_impl__(self) -> Glasses3:
303        ws_url, rtsp_url, http_url = await self.url_generator
304        _logger.info(
305            f"Attempting connection to websocket {ws_url}, RTSP {rtsp_url} and HTTP {http_url}"
306        )
307        connection = cast(
308            G3WebSocketClientProtocol, await g3pylib.websocket.connect(ws_url)
309        )
310        connection.start_receiver_task()
311        self.connection = connection
312        return Glasses3(connection, rtsp_url, http_url)
313
314    async def __aenter__(self) -> Glasses3:
315        return await self
316
317    async def __aexit__(
318        self,
319        exception_type: Optional[Type[BaseException]],
320        exception_value: Optional[BaseException],
321        traceback: Optional[TracebackType],
322    ) -> None:
323        await self.connection.close()
class Glasses3(g3pylib._utils.APIComponent): 
 53class Glasses3(APIComponent):
 54    """
 55    Represents a Glasses3 device.
 56
 57    Holds the API components and a WebSocket connection to a Glasses3 device.
 58    The `stream_rtsp` context can be used for live stream functionality.
 59
 60    For the recommended way to create a connected instance of Glasses3, see `connect_to_glasses`.
 61    """
 62
 63    def __init__(
 64        self,
 65        connection: G3WebSocketClientProtocol,
 66        rtsp_url: Optional[str],
 67        http_url: Optional[str],
 68        logger: Optional[LoggerLike] = None,
 69    ) -> None:
 70        self.logger: LoggerLike = (
 71            logging.getLogger(__name__) if logger is None else logger
 72        )
 73        self._rtsp_url = rtsp_url
 74        self._http_url = http_url
 75        self._connection: G3WebSocketClientProtocol = connection
 76        self._recorder: Optional[Recorder] = None
 77        self._recordings: Optional[Recordings] = None
 78        self._rudimentary: Optional[Rudimentary] = None
 79        self._system: Optional[System] = None
 80        self._calibrate: Optional[Calibrate] = None
 81        self._settings: Optional[Settings] = None
 82
 83    @property
 84    def calibrate(self) -> Calibrate:
 85        if self._calibrate is None:
 86            self._calibrate = Calibrate(self._connection, URI("/calibrate"))
 87        return self._calibrate
 88
 89    @property
 90    def recorder(self) -> Recorder:
 91        if self._recorder is None:
 92            self._recorder = Recorder(self._connection, URI("/recorder"))
 93        return self._recorder
 94
 95    @property
 96    def recordings(self) -> Recordings:
 97        if self._recordings is None:
 98            self._recordings = Recordings(
 99                self._connection, URI("/recordings"), self._http_url
100            )
101        return self._recordings
102
103    @property
104    def rudimentary(self) -> Rudimentary:
105        if self._rudimentary is None:
106            self._rudimentary = Rudimentary(self._connection, URI("/rudimentary"))
107        return self._rudimentary
108
109    @property
110    def system(self) -> System:
111        if self._system is None:
112            self._system = System(self._connection, URI("/system"))
113        return self._system
114
115    @property
116    def settings(self) -> Settings:
117        if self._settings is None:
118            self._settings = Settings(self._connection, URI("/settings"))
119        return self._settings
120
121    @property
122    def rtsp_url(self) -> Optional[str]:
123        """The RTSP URL used for live stream."""
124        return self._rtsp_url
125
126    @asynccontextmanager
127    async def stream_rtsp(
128        self,
129        scene_camera: bool = True,
130        audio: bool = False,
131        eye_cameras: bool = False,
132        gaze: bool = False,
133        sync: bool = False,
134        imu: bool = False,
135        events: bool = False,
136    ) -> AsyncIterator[Streams]:
137        """Set up an RTSP connection in the form of a Streams object with the Stream properties indicated by the arguments.
138
139        The Stream objects can be used to demux/decode their stream. For example, `stream_rtsp()` can be used as follows:
140        ```python
141        async with connect_to_glasses(g3_hostname) as g3:
142            async with g3.stream_rtsp() as streams:
143                async with streams.scene_camera.decode() as decoded_stream:
144                    for _ in range(500):
145                        frame, _timestamp = await decoded_stream.get()
146                        image = frame.to_ndarray(format="bgr24")
147                        cv2.imshow("Video", image)
148                        cv2.waitKey(1)
149        ```
150
151        *Alpha version note:* Only the scene_camera, eye_camera and gaze attributes are implemented so far.
152        """
153        if self.rtsp_url is None:
154            raise FeatureNotAvailableError(
155                "This Glasses3 object was initialized without a proper RTSP url."
156            )
157        async with Streams.connect(
158            self.rtsp_url,
159            scene_camera=scene_camera,
160            audio=audio,
161            eye_cameras=eye_cameras,
162            gaze=gaze,
163            sync=sync,
164            imu=imu,
165            events=events,
166        ) as streams:
167            await streams.play()
168            yield streams
169
170    async def close(self) -> None:
171        """Close down the underlying websocket connection to the Glasses3 device."""
172        await self._connection.close()

Represents a Glasses3 device.

Holds the API components and a WebSocket connection to a Glasses3 device. The stream_rtsp context can be used for live stream functionality.

For the recommended way to create a connected instance of Glasses3, see connect_to_glasses.

Glasses3( connection: g3pylib.websocket.G3WebSocketClientProtocol, rtsp_url: Optional[str], http_url: Optional[str], logger: Union[logging.Logger, logging.LoggerAdapter, NoneType] = None) 
63    def __init__(
64        self,
65        connection: G3WebSocketClientProtocol,
66        rtsp_url: Optional[str],
67        http_url: Optional[str],
68        logger: Optional[LoggerLike] = None,
69    ) -> None:
70        self.logger: LoggerLike = (
71            logging.getLogger(__name__) if logger is None else logger
72        )
73        self._rtsp_url = rtsp_url
74        self._http_url = http_url
75        self._connection: G3WebSocketClientProtocol = connection
76        self._recorder: Optional[Recorder] = None
77        self._recordings: Optional[Recordings] = None
78        self._rudimentary: Optional[Rudimentary] = None
79        self._system: Optional[System] = None
80        self._calibrate: Optional[Calibrate] = None
81        self._settings: Optional[Settings] = None
rtsp_url: Optional[str]

The RTSP URL used for live stream.

@asynccontextmanager
def stream_rtsp( self, scene_camera: bool = True, audio: bool = False, eye_cameras: bool = False, gaze: bool = False, sync: bool = False, imu: bool = False, events: bool = False) -> AsyncIterator[g3pylib.streams.Streams]: 
126    @asynccontextmanager
127    async def stream_rtsp(
128        self,
129        scene_camera: bool = True,
130        audio: bool = False,
131        eye_cameras: bool = False,
132        gaze: bool = False,
133        sync: bool = False,
134        imu: bool = False,
135        events: bool = False,
136    ) -> AsyncIterator[Streams]:
137        """Set up an RTSP connection in the form of a Streams object with the Stream properties indicated by the arguments.
138
139        The Stream objects can be used to demux/decode their stream. For example, `stream_rtsp()` can be used as follows:
140        ```python
141        async with connect_to_glasses(g3_hostname) as g3:
142            async with g3.stream_rtsp() as streams:
143                async with streams.scene_camera.decode() as decoded_stream:
144                    for _ in range(500):
145                        frame, _timestamp = await decoded_stream.get()
146                        image = frame.to_ndarray(format="bgr24")
147                        cv2.imshow("Video", image)
148                        cv2.waitKey(1)
149        ```
150
151        *Alpha version note:* Only the scene_camera, eye_camera and gaze attributes are implemented so far.
152        """
153        if self.rtsp_url is None:
154            raise FeatureNotAvailableError(
155                "This Glasses3 object was initialized without a proper RTSP url."
156            )
157        async with Streams.connect(
158            self.rtsp_url,
159            scene_camera=scene_camera,
160            audio=audio,
161            eye_cameras=eye_cameras,
162            gaze=gaze,
163            sync=sync,
164            imu=imu,
165            events=events,
166        ) as streams:
167            await streams.play()
168            yield streams

Set up an RTSP connection in the form of a Streams object with the Stream properties indicated by the arguments.

The Stream objects can be used to demux/decode their stream. For example, stream_rtsp() can be used as follows:

async with connect_to_glasses(g3_hostname) as g3:
    async with g3.stream_rtsp() as streams:
        async with streams.scene_camera.decode() as decoded_stream:
            for _ in range(500):
                frame, _timestamp = await decoded_stream.get()
                image = frame.to_ndarray(format="bgr24")
                cv2.imshow("Video", image)
                cv2.waitKey(1)

Alpha version note: Only the scene_camera, eye_camera and gaze attributes are implemented so far.

async def close(self) -> None: 
170    async def close(self) -> None:
171        """Close down the underlying websocket connection to the Glasses3 device."""
172        await self._connection.close()

Close down the underlying websocket connection to the Glasses3 device.

Inherited Members
g3pylib._utils.APIComponent
generate_endpoint_uri
class connect_to_glasses: 
175class connect_to_glasses:
176    """This class contains a set of classmethods which are used to connect to a pair of glasses.
177
178    The preferred way to use this class is as an async context manager like this:
179
180    ```python
181    async with connect_to_glasses.with_hostname(glasses_serial_number) as g3:
182        # Here you can call the glasses
183        await g3.get_name()
184    # Here the connection is closed
185    ```
186
187    It does however also support usage without a `with`-block like this:
188
189    ```python
190    g3 = await connect_to_glasses.with_hostname(glasses_serial_number)
191    # Here you can call the glasses
192    await g3.get_name()
193    # You have to remember to close the connection like this:
194    g3.close()
195    # And here g3 still exists but it is unusable
196    ```
197    """
198
199    def __init__(
200        self,
201        url_generator: Coroutine[Any, Any, Tuple[str, Optional[str], Optional[str]]],
202    ) -> None:
203        """You should probably not use this constructor unless you need to generate the URLs to your glasses in a very specific way.
204        The regular use cases are covered in the alternative constructors below: `with_url`, `with_zeroconf`, `with_hostname` and `with_service`.
205
206        If you want to use this constructor you need to supply a couroutine which returns a tuple that contains two URLs.
207        The first URL should point to the websocket and the second URL should point to the RTSP endpoint.
208        """
209        self.url_generator = url_generator
210
211    @staticmethod
212    async def _urls_from_zeroconf(
213        using_ip: bool, timeout: float
214    ) -> Tuple[str, Optional[str], Optional[str]]:
215        async with G3ServiceDiscovery.listen() as service_discovery:
216            service = await service_discovery.wait_for_single_service(
217                service_discovery.events,
218                timeout,
219            )
220        return await connect_to_glasses._urls_from_service(service, using_ip)
221
222    @staticmethod
223    async def _urls_from_service(
224        service: G3Service, using_ip: bool
225    ) -> Tuple[str, Optional[str], Optional[str]]:
226        return (
227            service.ws_url(using_ip),
228            service.rtsp_url(using_ip),
229            service.http_url(using_ip),
230        )
231
232    @staticmethod
233    async def _urls_from_hostname(
234        hostname: str, using_zeroconf: bool, using_ip: bool
235    ) -> Tuple[str, Optional[str], Optional[str]]:
236        if not using_zeroconf:
237            return (
238                f"ws://{hostname}{DEFAULT_WEBSOCKET_PATH}",
239                f"rtsp://{hostname}:{DEFAULT_RTSP_PORT}{DEFAULT_RTSP_LIVE_PATH}",
240                f"http://{hostname}:{DEFAULT_HTTP_PORT}",
241            )
242        else:
243            service = await G3ServiceDiscovery.request_service(hostname)
244            return await connect_to_glasses._urls_from_service(service, using_ip)
245
246    @classmethod
247    def with_zeroconf(
248        cls, using_ip: bool = True, timeout: float = 3000
249    ) -> connect_to_glasses:
250        """Connects by listening for available glasses on the network using zeroconf.
251        Connects to the first pair of glasses that answers so if there are multiple glasses on the
252        network the behavior is undefined.
253
254        If `using_ip` is set to True (default) we will generate the the URL used for connection with the ip.
255        If it's set to False we will use the hostname, which will depend on DNS working as it should.
256
257        `timeout` defines the time in milliseconds before `asyncio.TimeoutError` is raised.
258        """
259        return cls(cls._urls_from_zeroconf(using_ip, timeout))
260
261    @classmethod
262    def with_hostname(
263        cls, hostname: str, using_zeroconf: bool = False, using_ip: bool = True
264    ) -> connect_to_glasses:
265        """Connects to the pair of glasses with the given hostname.
266
267        If `using_zeroconf` is set to False (default) we will not depend on zeroconf
268        for fetching details on how to generate the URL and instead use detault values for the URL components specified
269        in the [developer guide](https://www.tobiipro.com/product-listing/tobii-pro-glasses3-api/#ResourcesSpecifications).
270        If it's set to True, all URL components are fetched with zeroconf.
271
272        `using_ip` specifies if the ip or the hostname should be used in the URL used for connecting when zeroconf is used.
273        If the hostname is used, it depends on DNS working as it should."""
274        return cls(cls._urls_from_hostname(hostname, using_zeroconf, using_ip))
275
276    @classmethod
277    def with_service(
278        cls, service: G3Service, using_ip: bool = True
279    ) -> connect_to_glasses:
280        """Connects to the pair of glasses referred to by the given service.
281
282        `using_ip` specifies if the ip or the hostname should be used in the URL used for connecting.
283        If the hostname is used, it depends on DNS working as it should.
284        """
285        return cls(cls._urls_from_service(service, using_ip))
286
287    @classmethod
288    def with_url(
289        cls, ws_url: str, rtsp_url: Optional[str] = None, http_url: Optional[str] = None
290    ):
291        """Connects to the pair of glasses at the specified URL. `ws_url` should
292        be a websocket URL (starting with `ws://`) and `rtsp_url` should be an RTSP
293        url (starting with `rtsp://` or `rtspt://`)."""
294
295        async def urls():
296            return (ws_url, rtsp_url, http_url)
297
298        return cls(urls())
299
300    def __await__(self) -> Generator[Any, None, Glasses3]:
301        return self.__await_impl__().__await__()
302
303    async def __await_impl__(self) -> Glasses3:
304        ws_url, rtsp_url, http_url = await self.url_generator
305        _logger.info(
306            f"Attempting connection to websocket {ws_url}, RTSP {rtsp_url} and HTTP {http_url}"
307        )
308        connection = cast(
309            G3WebSocketClientProtocol, await g3pylib.websocket.connect(ws_url)
310        )
311        connection.start_receiver_task()
312        self.connection = connection
313        return Glasses3(connection, rtsp_url, http_url)
314
315    async def __aenter__(self) -> Glasses3:
316        return await self
317
318    async def __aexit__(
319        self,
320        exception_type: Optional[Type[BaseException]],
321        exception_value: Optional[BaseException],
322        traceback: Optional[TracebackType],
323    ) -> None:
324        await self.connection.close()

This class contains a set of classmethods which are used to connect to a pair of glasses.

The preferred way to use this class is as an async context manager like this:

async with connect_to_glasses.with_hostname(glasses_serial_number) as g3:
    # Here you can call the glasses
    await g3.get_name()
# Here the connection is closed

It does however also support usage without a with-block like this:

g3 = await connect_to_glasses.with_hostname(glasses_serial_number)
# Here you can call the glasses
await g3.get_name()
# You have to remember to close the connection like this:
g3.close()
# And here g3 still exists but it is unusable
connect_to_glasses( url_generator: Coroutine[Any, Any, Tuple[str, Optional[str], Optional[str]]]) 
199    def __init__(
200        self,
201        url_generator: Coroutine[Any, Any, Tuple[str, Optional[str], Optional[str]]],
202    ) -> None:
203        """You should probably not use this constructor unless you need to generate the URLs to your glasses in a very specific way.
204        The regular use cases are covered in the alternative constructors below: `with_url`, `with_zeroconf`, `with_hostname` and `with_service`.
205
206        If you want to use this constructor you need to supply a couroutine which returns a tuple that contains two URLs.
207        The first URL should point to the websocket and the second URL should point to the RTSP endpoint.
208        """
209        self.url_generator = url_generator

You should probably not use this constructor unless you need to generate the URLs to your glasses in a very specific way. The regular use cases are covered in the alternative constructors below: with_url, with_zeroconf, with_hostname and with_service.

If you want to use this constructor you need to supply a couroutine which returns a tuple that contains two URLs. The first URL should point to the websocket and the second URL should point to the RTSP endpoint.

@classmethod
def with_zeroconf( cls, using_ip: bool = True, timeout: float = 3000) -> g3pylib.connect_to_glasses: 
246    @classmethod
247    def with_zeroconf(
248        cls, using_ip: bool = True, timeout: float = 3000
249    ) -> connect_to_glasses:
250        """Connects by listening for available glasses on the network using zeroconf.
251        Connects to the first pair of glasses that answers so if there are multiple glasses on the
252        network the behavior is undefined.
253
254        If `using_ip` is set to True (default) we will generate the the URL used for connection with the ip.
255        If it's set to False we will use the hostname, which will depend on DNS working as it should.
256
257        `timeout` defines the time in milliseconds before `asyncio.TimeoutError` is raised.
258        """
259        return cls(cls._urls_from_zeroconf(using_ip, timeout))

Connects by listening for available glasses on the network using zeroconf. Connects to the first pair of glasses that answers so if there are multiple glasses on the network the behavior is undefined.

If using_ip is set to True (default) we will generate the the URL used for connection with the ip. If it's set to False we will use the hostname, which will depend on DNS working as it should.

timeout defines the time in milliseconds before asyncio.TimeoutError is raised.

@classmethod
def with_hostname( cls, hostname: str, using_zeroconf: bool = False, using_ip: bool = True) -> g3pylib.connect_to_glasses: 
261    @classmethod
262    def with_hostname(
263        cls, hostname: str, using_zeroconf: bool = False, using_ip: bool = True
264    ) -> connect_to_glasses:
265        """Connects to the pair of glasses with the given hostname.
266
267        If `using_zeroconf` is set to False (default) we will not depend on zeroconf
268        for fetching details on how to generate the URL and instead use detault values for the URL components specified
269        in the [developer guide](https://www.tobiipro.com/product-listing/tobii-pro-glasses3-api/#ResourcesSpecifications).
270        If it's set to True, all URL components are fetched with zeroconf.
271
272        `using_ip` specifies if the ip or the hostname should be used in the URL used for connecting when zeroconf is used.
273        If the hostname is used, it depends on DNS working as it should."""
274        return cls(cls._urls_from_hostname(hostname, using_zeroconf, using_ip))

Connects to the pair of glasses with the given hostname.

If using_zeroconf is set to False (default) we will not depend on zeroconf for fetching details on how to generate the URL and instead use detault values for the URL components specified in the developer guide. If it's set to True, all URL components are fetched with zeroconf.

using_ip specifies if the ip or the hostname should be used in the URL used for connecting when zeroconf is used. If the hostname is used, it depends on DNS working as it should.

@classmethod
def with_service( cls, service: g3pylib.zeroconf.G3Service, using_ip: bool = True) -> g3pylib.connect_to_glasses: 
276    @classmethod
277    def with_service(
278        cls, service: G3Service, using_ip: bool = True
279    ) -> connect_to_glasses:
280        """Connects to the pair of glasses referred to by the given service.
281
282        `using_ip` specifies if the ip or the hostname should be used in the URL used for connecting.
283        If the hostname is used, it depends on DNS working as it should.
284        """
285        return cls(cls._urls_from_service(service, using_ip))

Connects to the pair of glasses referred to by the given service.

using_ip specifies if the ip or the hostname should be used in the URL used for connecting. If the hostname is used, it depends on DNS working as it should.

@classmethod
def with_url( cls, ws_url: str, rtsp_url: Optional[str] = None, http_url: Optional[str] = None): 
287    @classmethod
288    def with_url(
289        cls, ws_url: str, rtsp_url: Optional[str] = None, http_url: Optional[str] = None
290    ):
291        """Connects to the pair of glasses at the specified URL. `ws_url` should
292        be a websocket URL (starting with `ws://`) and `rtsp_url` should be an RTSP
293        url (starting with `rtsp://` or `rtspt://`)."""
294
295        async def urls():
296            return (ws_url, rtsp_url, http_url)
297
298        return cls(urls())

Connects to the pair of glasses at the specified URL. ws_url should be a websocket URL (starting with ws://) and rtsp_url should be an RTSP url (starting with rtsp:// or rtspt://).