17
17
* Adafruit's BLE library: https://github.com/adafruit/Adafruit_CircuitPython_BLE
18
18
"""
19
19
20
+ from __future__ import annotations
21
+
20
22
import struct
21
23
import time
22
24
25
+ try :
26
+ from typing import Generator , Union , Dict , Optional , Any
27
+ except ImportError :
28
+ pass
29
+
23
30
from adafruit_ble .services import Service
24
31
from adafruit_ble .uuid import VendorUUID
25
32
from adafruit_ble .characteristics .stream import StreamIn , StreamOut
29
36
30
37
31
38
class _NotificationAttribute :
32
- def __init__ (self , attribute_id , * , max_length = False ):
39
+ def __init__ (self , attribute_id : int , * , max_length : bool = False ) -> None :
33
40
self ._id = attribute_id
34
41
self ._max_length = max_length
35
42
36
- def __get__ (self , notification , cls ) :
43
+ def __get__ (self , notification : Notification , cls : Any ) -> str :
37
44
if self ._id in notification ._attribute_cache :
38
45
return notification ._attribute_cache [self ._id ]
39
46
@@ -104,14 +111,14 @@ class Notification:
104
111
105
112
def __init__ (
106
113
self ,
107
- notification_id ,
108
- event_flags ,
109
- category_id ,
110
- category_count ,
114
+ notification_id : int ,
115
+ event_flags : int ,
116
+ category_id : int ,
117
+ category_count : int ,
111
118
* ,
112
- control_point ,
113
- data_source
114
- ):
119
+ control_point : StreamIn ,
120
+ data_source : StreamOut ,
121
+ ) -> None :
115
122
self .id = notification_id # pylint: disable=invalid-name
116
123
"""Integer id of the notification."""
117
124
@@ -136,28 +143,28 @@ def __init__(
136
143
137
144
self .update (event_flags , category_id , category_count )
138
145
139
- self ._attribute_cache = {}
146
+ self ._attribute_cache : Dict [ int , str ] = {}
140
147
141
148
self .control_point = control_point
142
149
self .data_source = data_source
143
150
144
- def send_positive_action (self ):
151
+ def send_positive_action (self ) -> None :
145
152
"""Sends positive action on this notification. For example, to accept an IncomingCall."""
146
153
cmd = 2 # ANCS_CMD_PERFORM_NOTIFICATION_ACTION,
147
154
uid = self .id
148
155
action_id = 0 # ANCS_ACTION_POSITIVE
149
156
buffer = struct .pack ("<BIB" , cmd , uid , action_id )
150
157
self .control_point .write (buffer )
151
158
152
- def send_negative_action (self ):
159
+ def send_negative_action (self ) -> None :
153
160
"""Sends negative action on this notification. For example, to decline an IncomingCall."""
154
161
cmd = 2 # ANCS_CMD_PERFORM_NOTIFICATION_ACTION,
155
162
uid = self .id
156
163
action_id = 1 # ANCS_ACTION_NEGATIVE
157
164
buffer = struct .pack ("<BIB" , cmd , uid , action_id )
158
165
self .control_point .write (buffer )
159
166
160
- def update (self , event_flags , category_id , category_count ) :
167
+ def update (self , event_flags : int , category_id : int , category_count : int ) -> None :
161
168
"""Update the notification and clear the attribute cache."""
162
169
self .category_id = category_id
163
170
@@ -171,7 +178,7 @@ def update(self, event_flags, category_id, category_count):
171
178
172
179
self ._attribute_cache = {}
173
180
174
- def __str__ (self ):
181
+ def __str__ (self ) -> str :
175
182
# pylint: disable=too-many-branches
176
183
flags = []
177
184
category = None
@@ -223,11 +230,11 @@ class AppleNotificationCenterService(Service):
223
230
uuid = VendorUUID ("9FBF120D-6301-42D9-8C58-25E699A21DBD" ), buffer_size = 8 * 100
224
231
)
225
232
226
- def __init__ (self , service = None ):
233
+ def __init__ (self , service : Service = None ) -> None :
227
234
super ().__init__ (service = service )
228
- self ._active_notifications = {}
235
+ self ._active_notifications : Dict [ tuple , Notification ] = {}
229
236
230
- def _update (self ):
237
+ def _update (self ) -> Generator [ Union [ Notification , None ], None , None ] :
231
238
# Pylint is incorrectly inferring the type of self.notification_source so disable no-member.
232
239
while self .notification_source .in_waiting > 7 : # pylint: disable=no-member
233
240
buffer = self .notification_source .read (8 ) # pylint: disable=no-member
@@ -254,7 +261,9 @@ def _update(self):
254
261
del self ._active_notifications [nid ]
255
262
yield None
256
263
257
- def wait_for_new_notifications (self , timeout = None ):
264
+ def wait_for_new_notifications (
265
+ self , timeout : Optional [float ] = None
266
+ ) -> Generator [Union [Notification , None ], None , None ]:
258
267
"""Waits for new notifications and yields them. Returns on timeout, update, disconnect or
259
268
clear."""
260
269
start_time = time .monotonic ()
@@ -267,7 +276,7 @@ def wait_for_new_notifications(self, timeout=None):
267
276
yield new_notification
268
277
269
278
@property
270
- def active_notifications (self ):
279
+ def active_notifications (self ) -> dict :
271
280
"""A dictionary of active notifications keyed by id."""
272
281
for _ in self ._update ():
273
282
pass
0 commit comments