tessl/pypi-pygatt

Python Bluetooth LE (Low Energy) and GATT Library providing cross-platform BLE operations with multiple backends

—

Pending

Overview

Eval results

Files

Error Handling

Name: tessl/pypi-pygatt
Author: tessl

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.

Capabilities

Base Exception Classes

Core exception hierarchy for general BLE operations and connection management.

class BLEError(Exception):
    """
    Base exception class for all pygatt operations.
    
    Raised for general BLE errors including:
    - Adapter initialization failures
    - Device operation errors
    - Protocol-level issues
    """

class NotConnectedError(BLEError):
    """
    Raised when operations are attempted on disconnected devices.
    
    Common scenarios:
    - Device physically disconnected
    - Connection lost due to range/interference
    - Device powered off
    - Connection timeout expired
    """

class NotificationTimeout(BLEError):
    """
    Raised when notification or indication operations timeout.
    
    Args:
        msg: Error message string
        gatttool_output: Raw gatttool output for debugging (GATTTool backend only)
    
    Attributes:
        gatttool_output: str - Original CLI output that may contain additional error details
    """
    def __init__(self, msg: str = None, gatttool_output: str = None):
        super().__init__(msg)
        self.gatttool_output = gatttool_output

Usage Example:

import pygatt

adapter = pygatt.GATTToolBackend()

try:
    adapter.start()
    device = adapter.connect('01:23:45:67:89:ab')
    
    # This might raise NotConnectedError if device disconnects
    value = device.char_read('00002a19-0000-1000-8000-00805f9b34fb')
    
except pygatt.NotConnectedError:
    print("Device disconnected during operation")
    # Implement reconnection logic
    
except pygatt.BLEError as e:
    print(f"BLE operation failed: {e}")
    # Handle general BLE errors
    
except Exception as e:
    print(f"Unexpected error: {e}")

BGAPI-Specific Exceptions

Specialized exceptions for BGAPI backend protocol and hardware errors.

class BGAPIError(Exception):
    """
    BGAPI backend-specific errors.
    
    Raised for:
    - USB communication failures
    - BGAPI protocol errors
    - Hardware adapter issues
    - Serial port problems
    """

class ExpectedResponseTimeout(BGAPIError):
    """
    BGAPI command response timeout.
    
    Args:
        expected_packets: Description of expected BGAPI response packets
        timeout: Timeout value in seconds that was exceeded
    
    Raised when:
    - BGAPI commands don't receive expected responses
    - USB communication is interrupted
    - Adapter becomes unresponsive
    """
    def __init__(self, expected_packets, timeout):
        super().__init__(f"Timed out after {timeout}s waiting for {expected_packets}")

Usage Example:

import pygatt

adapter = pygatt.BGAPIBackend()

try:
    adapter.start()
    
except pygatt.BGAPIError as e:
    print(f"BGAPI hardware error: {e}")
    # Check USB connection, try different port
    
except pygatt.ExpectedResponseTimeout:
    print("BGAPI adapter not responding")
    # Try adapter reset or reconnection

try:
    device = adapter.connect('01:23:45:67:89:ab', timeout=10)
    
except pygatt.ExpectedResponseTimeout:
    print("Connection timeout - device may be out of range")
    
except pygatt.BGAPIError:
    print("BGAPI connection error - check adapter status")

Error Context and Debugging

Extract useful debugging information from exceptions for troubleshooting.

# NotificationTimeout provides additional context
try:
    device.subscribe(uuid, callback=handler, wait_for_response=True)
except pygatt.NotificationTimeout as e:
    print(f"Subscription failed: {e}")
    if e.gatttool_output:
        print(f"GATTTool output: {e.gatttool_output}")
        # Analyze raw CLI output for specific error details

Common Error Scenarios

Connection Errors

Handle various connection failure modes with appropriate recovery strategies.

import pygatt
import time

def robust_connect(adapter, address, max_retries=3):
    """
    Robust connection with retry logic and error handling.
    """
    for attempt in range(max_retries):
        try:
            device = adapter.connect(address, timeout=10)
            print(f"Connected successfully on attempt {attempt + 1}")
            return device
            
        except pygatt.NotConnectedError:
            print(f"Connection attempt {attempt + 1} failed - device unreachable")
            if attempt < max_retries - 1:
                time.sleep(2)  # Wait before retry
                
        except pygatt.BGAPIError as e:
            print(f"BGAPI error: {e}")
            # Hardware issue - may need adapter reset
            break
            
        except pygatt.BLEError as e:
            print(f"BLE error: {e}")
            # General BLE issue - retry may help
            if attempt < max_retries - 1:
                time.sleep(1)
    
    raise pygatt.NotConnectedError(f"Failed to connect after {max_retries} attempts")

# Usage
adapter = pygatt.BGAPIBackend()
adapter.start()

try:
    device = robust_connect(adapter, '01:23:45:67:89:ab')
except pygatt.NotConnectedError:
    print("Device permanently unreachable")

Operation Timeouts

Handle timeout scenarios in characteristic operations and subscriptions.

import pygatt

def safe_char_read(device, uuid, timeout=5, retries=2):
    """
    Safe characteristic read with timeout handling.
    """
    for attempt in range(retries + 1):
        try:
            if attempt > 0:
                print(f"Retry {attempt} for characteristic read")
                
            value = device.char_read(uuid)
            return value
            
        except pygatt.NotConnectedError:
            print("Device disconnected during read")
            raise  # Don't retry connection errors
            
        except pygatt.NotificationTimeout as e:
            print(f"Read timeout: {e}")
            if e.gatttool_output:
                print(f"Debug info: {e.gatttool_output}")
            
            if attempt < retries:
                time.sleep(1)
            else:
                raise
                
        except pygatt.BLEError as e:
            print(f"Read error: {e}")
            if attempt < retries:
                time.sleep(0.5)
            else:
                raise

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

Subscription Management

Handle notification subscription errors and connection loss during streaming.

import pygatt
import threading
import time

class RobustSubscription:
    def __init__(self, adapter, device_address, characteristic_uuid):
        self.adapter = adapter
        self.device_address = device_address
        self.characteristic_uuid = characteristic_uuid
        self.device = None
        self.running = False
        self.reconnect_thread = None
    
    def notification_handler(self, handle, value):
        print(f"Data received: {value.hex()}")
    
    def start_subscription(self):
        """
        Start subscription with automatic reconnection on errors.
        """
        self.running = True
        self.reconnect_thread = threading.Thread(target=self._maintain_subscription)
        self.reconnect_thread.start()
    
    def stop_subscription(self):
        self.running = False
        if self.reconnect_thread:
            self.reconnect_thread.join()
    
    def _maintain_subscription(self):
        while self.running:
            try:
                if not self.device:
                    print("Connecting to device...")
                    self.device = self.adapter.connect(self.device_address)
                
                print("Subscribing to notifications...")
                self.device.subscribe(self.characteristic_uuid, 
                                    callback=self.notification_handler)
                
                # Keep subscription alive
                while self.running:
                    time.sleep(1)
                    
            except pygatt.NotConnectedError:
                print("Device disconnected - will reconnect")
                self.device = None
                time.sleep(5)  # Wait before reconnect
                
            except pygatt.NotificationTimeout as e:
                print(f"Subscription timeout: {e}")
                if self.device:
                    try:
                        self.device.unsubscribe(self.characteristic_uuid)
                    except:
                        pass
                self.device = None
                time.sleep(2)
                
            except pygatt.BLEError as e:
                print(f"BLE error in subscription: {e}")
                self.device = None
                time.sleep(3)

# Usage
adapter = pygatt.GATTToolBackend()
adapter.start()

subscription = RobustSubscription(
    adapter, 
    '01:23:45:67:89:ab',
    'sensor-data-uuid'
)

try:
    subscription.start_subscription()
    time.sleep(60)  # Run for 1 minute
finally:
    subscription.stop_subscription()
    adapter.stop()

Hardware-Specific Errors

Handle backend-specific hardware and system errors.

import pygatt

def initialize_adapter(backend_type='auto', **kwargs):
    """
    Initialize adapter with fallback between backends.
    """
    if backend_type == 'auto':
        # Try BGAPI first, fallback to GATTTool
        try:
            adapter = pygatt.BGAPIBackend(**kwargs)
            adapter.start()
            print("Using BGAPI backend")
            return adapter
        except pygatt.BGAPIError:
            print("BGAPI not available, trying GATTTool")
            
        try:
            adapter = pygatt.GATTToolBackend(**kwargs)
            adapter.start()
            print("Using GATTTool backend")
            return adapter
        except pygatt.BLEError:
            raise pygatt.BLEError("No BLE backend available")
    
    elif backend_type == 'bgapi':
        adapter = pygatt.BGAPIBackend(**kwargs)
        try:
            adapter.start()
            return adapter
        except pygatt.BGAPIError as e:
            raise pygatt.BLEError(f"BGAPI initialization failed: {e}")
    
    elif backend_type == 'gatttool':
        adapter = pygatt.GATTToolBackend(**kwargs)
        try:
            adapter.start()
            return adapter
        except pygatt.BLEError as e:
            raise pygatt.BLEError(f"GATTTool initialization failed: {e}")

# Usage with error handling
try:
    adapter = initialize_adapter('auto')
except pygatt.BLEError as e:
    print(f"No BLE adapter available: {e}")
    print("Check USB connections or system Bluetooth status")
    exit(1)

Logging and Debugging

Enable comprehensive logging to diagnose complex error scenarios.

import logging
import pygatt

# Configure logging for debugging
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

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

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

try:
    adapter.start()
    device = adapter.connect('01:23:45:67:89:ab')
    
except pygatt.BLEError:
    print("Check logs for detailed error information")
    # Examine /tmp/gatttool_debug.log for CLI interactions

Error Recovery Patterns

Automatic Retry with Backoff

import time
import random

def exponential_backoff_retry(func, max_retries=5, base_delay=1):
    """
    Retry function with exponential backoff.
    """
    for attempt in range(max_retries):
        try:
            return func()
        except (pygatt.BLEError, pygatt.BGAPIError) as e:
            if attempt == max_retries - 1:
                raise e
            
            delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
            print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay:.1f}s")
            time.sleep(delay)

Circuit Breaker Pattern

import time

class BLECircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=30):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = 'closed'  # closed, open, half-open
    
    def call(self, func, *args, **kwargs):
        if self.state == 'open':
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = 'half-open'
            else:
                raise pygatt.BLEError("Circuit breaker is open")
        
        try:
            result = func(*args, **kwargs)
            if self.state == 'half-open':
                self.state = 'closed'
                self.failure_count = 0
            return result
            
        except (pygatt.BLEError, pygatt.BGAPIError) as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            
            if self.failure_count >= self.failure_threshold:
                self.state = 'open'
            
            raise e

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

Install with Tessl CLI