0
# Error Handling
1

2
Exception classes and error management for comprehensive BLE error handling and debugging. pygatt provides a hierarchical exception system to help applications handle different types of BLE-related errors appropriately.
3

4
## Capabilities
5

6
### Base Exception Classes
7

8
Core exception hierarchy for general BLE operations and connection management.
9

10
```python { .api }
11
class BLEError(Exception):
12
    """
13
    Base exception class for all pygatt operations.
14
    
15
    Raised for general BLE errors including:
16
    - Adapter initialization failures
17
    - Device operation errors
18
    - Protocol-level issues
19
    """
20

21
class NotConnectedError(BLEError):
22
    """
23
    Raised when operations are attempted on disconnected devices.
24
    
25
    Common scenarios:
26
    - Device physically disconnected
27
    - Connection lost due to range/interference
28
    - Device powered off
29
    - Connection timeout expired
30
    """
31

32
class NotificationTimeout(BLEError):
33
    """
34
    Raised when notification or indication operations timeout.
35
    
36
    Args:
37
        msg: Error message string
38
        gatttool_output: Raw gatttool output for debugging (GATTTool backend only)
39
    
40
    Attributes:
41
        gatttool_output: str - Original CLI output that may contain additional error details
42
    """
43
    def __init__(self, msg: str = None, gatttool_output: str = None):
44
        super().__init__(msg)
45
        self.gatttool_output = gatttool_output
46
```
47

48
**Usage Example:**
49

50
```python
51
import pygatt
52

53
adapter = pygatt.GATTToolBackend()
54

55
try:
56
    adapter.start()
57
    device = adapter.connect('01:23:45:67:89:ab')
58
    
59
    # This might raise NotConnectedError if device disconnects
60
    value = device.char_read('00002a19-0000-1000-8000-00805f9b34fb')
61
    
62
except pygatt.NotConnectedError:
63
    print("Device disconnected during operation")
64
    # Implement reconnection logic
65
    
66
except pygatt.BLEError as e:
67
    print(f"BLE operation failed: {e}")
68
    # Handle general BLE errors
69
    
70
except Exception as e:
71
    print(f"Unexpected error: {e}")
72
```
73

74
### BGAPI-Specific Exceptions
75

76
Specialized exceptions for BGAPI backend protocol and hardware errors.
77

78
```python { .api }
79
class BGAPIError(Exception):
80
    """
81
    BGAPI backend-specific errors.
82
    
83
    Raised for:
84
    - USB communication failures
85
    - BGAPI protocol errors
86
    - Hardware adapter issues
87
    - Serial port problems
88
    """
89

90
class ExpectedResponseTimeout(BGAPIError):
91
    """
92
    BGAPI command response timeout.
93
    
94
    Args:
95
        expected_packets: Description of expected BGAPI response packets
96
        timeout: Timeout value in seconds that was exceeded
97
    
98
    Raised when:
99
    - BGAPI commands don't receive expected responses
100
    - USB communication is interrupted
101
    - Adapter becomes unresponsive
102
    """
103
    def __init__(self, expected_packets, timeout):
104
        super().__init__(f"Timed out after {timeout}s waiting for {expected_packets}")
105
```
106

107
**Usage Example:**
108

109
```python
110
import pygatt
111

112
adapter = pygatt.BGAPIBackend()
113

114
try:
115
    adapter.start()
116
    
117
except pygatt.BGAPIError as e:
118
    print(f"BGAPI hardware error: {e}")
119
    # Check USB connection, try different port
120
    
121
except pygatt.ExpectedResponseTimeout:
122
    print("BGAPI adapter not responding")
123
    # Try adapter reset or reconnection
124

125
try:
126
    device = adapter.connect('01:23:45:67:89:ab', timeout=10)
127
    
128
except pygatt.ExpectedResponseTimeout:
129
    print("Connection timeout - device may be out of range")
130
    
131
except pygatt.BGAPIError:
132
    print("BGAPI connection error - check adapter status")
133
```
134

135
### Error Context and Debugging
136

137
Extract useful debugging information from exceptions for troubleshooting.
138

139
```python { .api }
140
# NotificationTimeout provides additional context
141
try:
142
    device.subscribe(uuid, callback=handler, wait_for_response=True)
143
except pygatt.NotificationTimeout as e:
144
    print(f"Subscription failed: {e}")
145
    if e.gatttool_output:
146
        print(f"GATTTool output: {e.gatttool_output}")
147
        # Analyze raw CLI output for specific error details
148
```
149

150
## Common Error Scenarios
151

152
### Connection Errors
153

154
Handle various connection failure modes with appropriate recovery strategies.
155

156
```python
157
import pygatt
158
import time
159

160
def robust_connect(adapter, address, max_retries=3):
161
    """
162
    Robust connection with retry logic and error handling.
163
    """
164
    for attempt in range(max_retries):
165
        try:
166
            device = adapter.connect(address, timeout=10)
167
            print(f"Connected successfully on attempt {attempt + 1}")
168
            return device
169
            
170
        except pygatt.NotConnectedError:
171
            print(f"Connection attempt {attempt + 1} failed - device unreachable")
172
            if attempt < max_retries - 1:
173
                time.sleep(2)  # Wait before retry
174
                
175
        except pygatt.BGAPIError as e:
176
            print(f"BGAPI error: {e}")
177
            # Hardware issue - may need adapter reset
178
            break
179
            
180
        except pygatt.BLEError as e:
181
            print(f"BLE error: {e}")
182
            # General BLE issue - retry may help
183
            if attempt < max_retries - 1:
184
                time.sleep(1)
185
    
186
    raise pygatt.NotConnectedError(f"Failed to connect after {max_retries} attempts")
187

188
# Usage
189
adapter = pygatt.BGAPIBackend()
190
adapter.start()
191

192
try:
193
    device = robust_connect(adapter, '01:23:45:67:89:ab')
194
except pygatt.NotConnectedError:
195
    print("Device permanently unreachable")
196
```
197

198
### Operation Timeouts
199

200
Handle timeout scenarios in characteristic operations and subscriptions.
201

202
```python
203
import pygatt
204

205
def safe_char_read(device, uuid, timeout=5, retries=2):
206
    """
207
    Safe characteristic read with timeout handling.
208
    """
209
    for attempt in range(retries + 1):
210
        try:
211
            if attempt > 0:
212
                print(f"Retry {attempt} for characteristic read")
213
                
214
            value = device.char_read(uuid)
215
            return value
216
            
217
        except pygatt.NotConnectedError:
218
            print("Device disconnected during read")
219
            raise  # Don't retry connection errors
220
            
221
        except pygatt.NotificationTimeout as e:
222
            print(f"Read timeout: {e}")
223
            if e.gatttool_output:
224
                print(f"Debug info: {e.gatttool_output}")
225
            
226
            if attempt < retries:
227
                time.sleep(1)
228
            else:
229
                raise
230
                
231
        except pygatt.BLEError as e:
232
            print(f"Read error: {e}")
233
            if attempt < retries:
234
                time.sleep(0.5)
235
            else:
236
                raise
237

238
# Usage
239
try:
240
    value = safe_char_read(device, '00002a19-0000-1000-8000-00805f9b34fb')
241
    print(f"Battery level: {value[0]}%")
242
except pygatt.BLEError:
243
    print("Unable to read battery level after retries")
244
```
245

246
### Subscription Management
247

248
Handle notification subscription errors and connection loss during streaming.
249

250
```python
251
import pygatt
252
import threading
253
import time
254

255
class RobustSubscription:
256
    def __init__(self, adapter, device_address, characteristic_uuid):
257
        self.adapter = adapter
258
        self.device_address = device_address
259
        self.characteristic_uuid = characteristic_uuid
260
        self.device = None
261
        self.running = False
262
        self.reconnect_thread = None
263
    
264
    def notification_handler(self, handle, value):
265
        print(f"Data received: {value.hex()}")
266
    
267
    def start_subscription(self):
268
        """
269
        Start subscription with automatic reconnection on errors.
270
        """
271
        self.running = True
272
        self.reconnect_thread = threading.Thread(target=self._maintain_subscription)
273
        self.reconnect_thread.start()
274
    
275
    def stop_subscription(self):
276
        self.running = False
277
        if self.reconnect_thread:
278
            self.reconnect_thread.join()
279
    
280
    def _maintain_subscription(self):
281
        while self.running:
282
            try:
283
                if not self.device:
284
                    print("Connecting to device...")
285
                    self.device = self.adapter.connect(self.device_address)
286
                
287
                print("Subscribing to notifications...")
288
                self.device.subscribe(self.characteristic_uuid, 
289
                                    callback=self.notification_handler)
290
                
291
                # Keep subscription alive
292
                while self.running:
293
                    time.sleep(1)
294
                    
295
            except pygatt.NotConnectedError:
296
                print("Device disconnected - will reconnect")
297
                self.device = None
298
                time.sleep(5)  # Wait before reconnect
299
                
300
            except pygatt.NotificationTimeout as e:
301
                print(f"Subscription timeout: {e}")
302
                if self.device:
303
                    try:
304
                        self.device.unsubscribe(self.characteristic_uuid)
305
                    except:
306
                        pass
307
                self.device = None
308
                time.sleep(2)
309
                
310
            except pygatt.BLEError as e:
311
                print(f"BLE error in subscription: {e}")
312
                self.device = None
313
                time.sleep(3)
314

315
# Usage
316
adapter = pygatt.GATTToolBackend()
317
adapter.start()
318

319
subscription = RobustSubscription(
320
    adapter, 
321
    '01:23:45:67:89:ab',
322
    'sensor-data-uuid'
323
)
324

325
try:
326
    subscription.start_subscription()
327
    time.sleep(60)  # Run for 1 minute
328
finally:
329
    subscription.stop_subscription()
330
    adapter.stop()
331
```
332

333
### Hardware-Specific Errors
334

335
Handle backend-specific hardware and system errors.
336

337
```python
338
import pygatt
339

340
def initialize_adapter(backend_type='auto', **kwargs):
341
    """
342
    Initialize adapter with fallback between backends.
343
    """
344
    if backend_type == 'auto':
345
        # Try BGAPI first, fallback to GATTTool
346
        try:
347
            adapter = pygatt.BGAPIBackend(**kwargs)
348
            adapter.start()
349
            print("Using BGAPI backend")
350
            return adapter
351
        except pygatt.BGAPIError:
352
            print("BGAPI not available, trying GATTTool")
353
            
354
        try:
355
            adapter = pygatt.GATTToolBackend(**kwargs)
356
            adapter.start()
357
            print("Using GATTTool backend")
358
            return adapter
359
        except pygatt.BLEError:
360
            raise pygatt.BLEError("No BLE backend available")
361
    
362
    elif backend_type == 'bgapi':
363
        adapter = pygatt.BGAPIBackend(**kwargs)
364
        try:
365
            adapter.start()
366
            return adapter
367
        except pygatt.BGAPIError as e:
368
            raise pygatt.BLEError(f"BGAPI initialization failed: {e}")
369
    
370
    elif backend_type == 'gatttool':
371
        adapter = pygatt.GATTToolBackend(**kwargs)
372
        try:
373
            adapter.start()
374
            return adapter
375
        except pygatt.BLEError as e:
376
            raise pygatt.BLEError(f"GATTTool initialization failed: {e}")
377

378
# Usage with error handling
379
try:
380
    adapter = initialize_adapter('auto')
381
except pygatt.BLEError as e:
382
    print(f"No BLE adapter available: {e}")
383
    print("Check USB connections or system Bluetooth status")
384
    exit(1)
385
```
386

387
## Logging and Debugging
388

389
Enable comprehensive logging to diagnose complex error scenarios.
390

391
```python
392
import logging
393
import pygatt
394

395
# Configure logging for debugging
396
logging.basicConfig(
397
    level=logging.DEBUG,
398
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
399
)
400

401
# Enable pygatt debug logging
402
logging.getLogger('pygatt').setLevel(logging.DEBUG)
403

404
# For GATTTool backend, enable CLI logging
405
adapter = pygatt.GATTToolBackend(gatttool_logfile='/tmp/gatttool_debug.log')
406

407
try:
408
    adapter.start()
409
    device = adapter.connect('01:23:45:67:89:ab')
410
    
411
except pygatt.BLEError:
412
    print("Check logs for detailed error information")
413
    # Examine /tmp/gatttool_debug.log for CLI interactions
414
```
415

416
## Error Recovery Patterns
417

418
### Automatic Retry with Backoff
419

420
```python
421
import time
422
import random
423

424
def exponential_backoff_retry(func, max_retries=5, base_delay=1):
425
    """
426
    Retry function with exponential backoff.
427
    """
428
    for attempt in range(max_retries):
429
        try:
430
            return func()
431
        except (pygatt.BLEError, pygatt.BGAPIError) as e:
432
            if attempt == max_retries - 1:
433
                raise e
434
            
435
            delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
436
            print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay:.1f}s")
437
            time.sleep(delay)
438
```
439

440
### Circuit Breaker Pattern
441

442
```python
443
import time
444

445
class BLECircuitBreaker:
446
    def __init__(self, failure_threshold=5, recovery_timeout=30):
447
        self.failure_threshold = failure_threshold
448
        self.recovery_timeout = recovery_timeout
449
        self.failure_count = 0
450
        self.last_failure_time = None
451
        self.state = 'closed'  # closed, open, half-open
452
    
453
    def call(self, func, *args, **kwargs):
454
        if self.state == 'open':
455
            if time.time() - self.last_failure_time > self.recovery_timeout:
456
                self.state = 'half-open'
457
            else:
458
                raise pygatt.BLEError("Circuit breaker is open")
459
        
460
        try:
461
            result = func(*args, **kwargs)
462
            if self.state == 'half-open':
463
                self.state = 'closed'
464
                self.failure_count = 0
465
            return result
466
            
467
        except (pygatt.BLEError, pygatt.BGAPIError) as e:
468
            self.failure_count += 1
469
            self.last_failure_time = time.time()
470
            
471
            if self.failure_count >= self.failure_threshold:
472
                self.state = 'open'
473
            
474
            raise e
475
```
476

477
This comprehensive error handling approach ensures robust BLE applications that can gracefully handle the various failure modes inherent in wireless communication.