9
9
building on the native `_bleio` module.
10
10
11
11
"""
12
+
13
+ from __future__ import annotations
14
+
12
15
# pylint: disable=wrong-import-position
16
+
13
17
import sys
14
18
15
19
if sys .implementation .name == "circuitpython" and sys .implementation .version [0 ] <= 4 :
23
27
from .services import Service
24
28
from .advertising import Advertisement
25
29
30
+ try :
31
+ from typing import Optional , Iterator , Union , Tuple , NoReturn , TYPE_CHECKING
32
+ from typing_extensions import Literal
33
+
34
+ if TYPE_CHECKING :
35
+ from circuitpython_typing import ReadableBuffer
36
+ from adafruit_ble .uuid import UUID
37
+ from adafruit_ble .characteristics import Characteristic
38
+
39
+ except ImportError :
40
+ pass
41
+
26
42
__version__ = "0.0.0-auto.0"
27
43
__repo__ = "https://github.com/adafruit/Adafruit_CircuitPython_BLE.git"
28
44
@@ -36,14 +52,14 @@ class BLEConnection:
36
52
37
53
"""
38
54
39
- def __init__ (self , bleio_connection ) :
55
+ def __init__ (self , bleio_connection : _bleio . Connection ) -> None :
40
56
self ._bleio_connection = bleio_connection
41
57
# _bleio.Service objects representing services found during discovery.
42
58
self ._discovered_bleio_services = {}
43
59
# Service objects that wrap remote services.
44
60
self ._constructed_services = {}
45
61
46
- def _discover_remote (self , uuid ) :
62
+ def _discover_remote (self , uuid : UUID ) -> Optional [ _bleio . Service ] :
47
63
remote_service = None
48
64
if uuid in self ._discovered_bleio_services :
49
65
remote_service = self ._discovered_bleio_services [uuid ]
@@ -56,7 +72,7 @@ def _discover_remote(self, uuid):
56
72
self ._discovered_bleio_services [uuid ] = remote_service
57
73
return remote_service
58
74
59
- def __contains__ (self , key ) :
75
+ def __contains__ (self , key : Union [ UUID , Service ]) -> bool :
60
76
"""
61
77
Allows easy testing for a particular Service class or a particular UUID
62
78
associated with this connection.
@@ -74,7 +90,7 @@ def __contains__(self, key):
74
90
uuid = key .uuid
75
91
return self ._discover_remote (uuid ) is not None
76
92
77
- def __getitem__ (self , key ) :
93
+ def __getitem__ (self , key : Union [ UUID , Service ]) -> Optional [ Service ] :
78
94
"""Return the Service for the given Service class or uuid, if any."""
79
95
uuid = key
80
96
maybe_service = False
@@ -96,17 +112,17 @@ def __getitem__(self, key):
96
112
raise KeyError ("{!r} object has no service {}" .format (self , key ))
97
113
98
114
@property
99
- def connected (self ):
115
+ def connected (self ) -> bool :
100
116
"""True if the connection to the peer is still active."""
101
117
return self ._bleio_connection .connected
102
118
103
119
@property
104
- def paired (self ):
120
+ def paired (self ) -> bool :
105
121
"""True if the paired to the peer."""
106
122
return self ._bleio_connection .paired
107
123
108
124
@property
109
- def connection_interval (self ):
125
+ def connection_interval (self ) -> float :
110
126
"""Time between transmissions in milliseconds. Will be multiple of 1.25ms. Lower numbers
111
127
increase speed and decrease latency but increase power consumption.
112
128
@@ -118,14 +134,14 @@ def connection_interval(self):
118
134
return self ._bleio_connection .connection_interval
119
135
120
136
@connection_interval .setter
121
- def connection_interval (self , value ) :
137
+ def connection_interval (self , value : float ) -> None :
122
138
self ._bleio_connection .connection_interval = value
123
139
124
- def pair (self , * , bond = True ):
140
+ def pair (self , * , bond : bool = True ) -> None :
125
141
"""Pair to the peer to increase security of the connection."""
126
142
return self ._bleio_connection .pair (bond = bond )
127
143
128
- def disconnect (self ):
144
+ def disconnect (self ) -> None :
129
145
"""Disconnect from peer."""
130
146
self ._bleio_connection .disconnect ()
131
147
@@ -138,7 +154,7 @@ class BLERadio:
138
154
139
155
It uses this library's `Advertisement` classes and the `BLEConnection` class."""
140
156
141
- def __init__ (self , adapter = None ):
157
+ def __init__ (self , adapter : Optional [ _bleio . Adapter ] = None ) -> None :
142
158
"""If no adapter is supplied, use the built-in `_bleio.adapter`.
143
159
If no built-in adapter is available, raise `RuntimeError`.
144
160
"""
@@ -149,8 +165,12 @@ def __init__(self, adapter=None):
149
165
self ._connection_cache = {}
150
166
151
167
def start_advertising (
152
- self , advertisement , scan_response = None , interval = 0.1 , timeout = None
153
- ):
168
+ self ,
169
+ advertisement : Advertisement ,
170
+ scan_response : Optional [ReadableBuffer ] = None ,
171
+ interval : float = 0.1 ,
172
+ timeout : Optional [int ] = None ,
173
+ ) -> None :
154
174
"""
155
175
Starts advertising the given advertisement.
156
176
@@ -195,21 +215,21 @@ def start_advertising(
195
215
timeout = 0 if timeout is None else timeout ,
196
216
)
197
217
198
- def stop_advertising (self ):
218
+ def stop_advertising (self ) -> None :
199
219
"""Stops advertising."""
200
220
self ._adapter .stop_advertising ()
201
221
202
222
def start_scan (
203
223
self ,
204
- * advertisement_types ,
205
- buffer_size = 512 ,
206
- extended = False ,
207
- timeout = None ,
208
- interval = 0.1 ,
209
- window = 0.1 ,
210
- minimum_rssi = - 80 ,
211
- active = True
212
- ):
224
+ * advertisement_types : Advertisement ,
225
+ buffer_size : int = 512 ,
226
+ extended : bool = False ,
227
+ timeout : Optional [ float ] = None ,
228
+ interval : float = 0.1 ,
229
+ window : float = 0.1 ,
230
+ minimum_rssi : int = - 80 ,
231
+ active : bool = True ,
232
+ ) -> Iterator [ Advertisement ] :
213
233
"""
214
234
Starts scanning. Returns an iterator of advertisement objects of the types given in
215
235
advertisement_types. The iterator will block until an advertisement is heard or the scan
@@ -269,14 +289,16 @@ def start_scan(
269
289
if advertisement :
270
290
yield advertisement
271
291
272
- def stop_scan (self ):
292
+ def stop_scan (self ) -> None :
273
293
"""Stops any active scan.
274
294
275
295
The scan results iterator will return any buffered results and then raise StopIteration
276
296
once empty."""
277
297
self ._adapter .stop_scan ()
278
298
279
- def connect (self , peer , * , timeout = 4.0 ):
299
+ def connect (
300
+ self , peer : Union [Advertisement , _bleio .Address ], * , timeout : float = 4.0
301
+ ) -> BLEConnection :
280
302
"""
281
303
Initiates a `BLEConnection` to the peer that advertised the given advertisement.
282
304
@@ -293,12 +315,12 @@ def connect(self, peer, *, timeout=4.0):
293
315
return self ._connection_cache [connection ]
294
316
295
317
@property
296
- def connected (self ):
318
+ def connected (self ) -> bool :
297
319
"""True if any peers are connected."""
298
320
return self ._adapter .connected
299
321
300
322
@property
301
- def connections (self ):
323
+ def connections (self ) -> Tuple [ Optional [ BLEConnection ], ...] :
302
324
"""A tuple of active `BLEConnection` objects."""
303
325
self ._clean_connection_cache ()
304
326
connections = self ._adapter .connections
@@ -311,36 +333,36 @@ def connections(self):
311
333
return tuple (wrapped_connections )
312
334
313
335
@property
314
- def name (self ):
336
+ def name (self ) -> str :
315
337
"""The name for this device. Used in advertisements and
316
338
as the Device Name in the Generic Access Service, available to a connected peer.
317
339
"""
318
340
return self ._adapter .name
319
341
320
342
@name .setter
321
- def name (self , value ) :
343
+ def name (self , value : str ) -> None :
322
344
self ._adapter .name = value
323
345
324
346
@property
325
- def tx_power (self ):
347
+ def tx_power (self ) -> Literal [ 0 ] :
326
348
"""Transmit power, in dBm."""
327
349
return 0
328
350
329
351
@tx_power .setter
330
- def tx_power (self , value ):
352
+ def tx_power (self , value ) -> NoReturn :
331
353
raise NotImplementedError ("setting tx_power not yet implemented" )
332
354
333
355
@property
334
- def address_bytes (self ):
356
+ def address_bytes (self ) -> bytes :
335
357
"""The device address, as a ``bytes()`` object of length 6."""
336
358
return self ._adapter .address .address_bytes
337
359
338
360
@property
339
- def advertising (self ):
361
+ def advertising (self ) -> bool :
340
362
"""The advertising state"""
341
363
return self ._adapter .advertising # pylint: disable=no-member
342
364
343
- def _clean_connection_cache (self ):
365
+ def _clean_connection_cache (self ) -> None :
344
366
"""Remove cached connections that have disconnected."""
345
367
for k , connection in list (self ._connection_cache .items ()):
346
368
if not connection .connected :
0 commit comments