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