Exception Handling and Error Management

class BleakError(Exception):
    """
    Base exception for all Bleak operations.
    
    All Bleak-specific exceptions inherit from this class, allowing
    for unified exception handling.
    """

class BleakCharacteristicNotFoundError(BleakError):
    """
    Exception raised when a characteristic cannot be found.
    
    Raised when attempting to access a characteristic that doesn't exist
    or is not accessible on the connected device.
    """
    
    char_specifier: Union[int, str, uuid.UUID]
    """The characteristic specifier that was not found."""
    
    def __init__(self, char_specifier: Union[int, str, uuid.UUID]) -> None:
        """
        Initialize characteristic not found error.
        
        Args:
            char_specifier: Handle or UUID of the characteristic that was not found
        """

class BleakDeviceNotFoundError(BleakError):
    """
    Exception raised when a device cannot be found for connection operations.
    
    This occurs when the OS Bluetooth stack has never seen the device,
    or when the device was removed and forgotten from the OS.
    """
    
    identifier: str
    """The device identifier (address or UUID) that was not found."""
    
    def __init__(self, identifier: str, *args: object) -> None:
        """
        Initialize device not found error.
        
        Args:
            identifier: Device identifier (Bluetooth address or UUID) that was not found
            *args: Additional exception arguments
        """

class BleakDBusError(BleakError):
    """
    Specialized exception for D-Bus errors in the BlueZ backend.
    
    Provides detailed information about D-Bus communication failures
    and includes mappings for common protocol error codes.
    """
    
    def __init__(self, dbus_error: str, error_body: list[Any]):
        """
        Initialize D-Bus error.
        
        Args:
            dbus_error: D-Bus error name (e.g., 'org.freedesktop.DBus.Error.UnknownObject')
            error_body: List containing error description or details
        """
    
    @property
    def dbus_error(self) -> str:
        """Gets the D-Bus error name."""
    
    @property 
    def dbus_error_details(self) -> Optional[str]:
        """
        Gets optional D-Bus error details with enhanced descriptions.
        
        Automatically enhances ATT error codes with human-readable descriptions.
        """
    
    def __str__(self) -> str:
        """Format error with name and details."""

# Bluetooth controller error codes
CONTROLLER_ERROR_CODES: dict[int, str] = {
    0x00: "Success",
    0x01: "Unknown HCI Command",
    0x02: "Unknown Connection Identifier", 
    0x03: "Hardware Failure",
    0x04: "Page Timeout",
    0x05: "Authentication Failure",
    0x06: "PIN or Key Missing",
    0x07: "Memory Capacity Exceeded",
    0x08: "Connection Timeout",
    0x09: "Connection Limit Exceeded",
    0x0A: "Synchronous Connection Limit To A Device Exceeded",
    0x0B: "Connection Already Exists",
    0x0C: "Command Disallowed",
    0x0D: "Connection Rejected due to Limited Resources",
    0x0E: "Connection Rejected Due To Security Reasons",
    0x0F: "Connection Rejected due to Unacceptable BD_ADDR",
    0x10: "Connection Accept Timeout Exceeded",
    0x11: "Unsupported Feature or Parameter Value",
    0x12: "Invalid HCI Command Parameters",
    0x13: "Remote User Terminated Connection",
    0x14: "Remote Device Terminated Connection due to Low Resources",
    0x15: "Remote Device Terminated Connection due to Power Off",
    0x16: "Connection Terminated By Local Host",
    0x17: "Repeated Attempts",
    0x18: "Pairing Not Allowed",
    0x19: "Unknown LMP PDU",
    0x1A: "Unsupported Remote Feature / Unsupported LMP Feature",
    0x1B: "SCO Offset Rejected",
    0x1C: "SCO Interval Rejected",
    0x1D: "SCO Air Mode Rejected",
    0x1E: "Invalid LMP Parameters / Invalid LL Parameters",
    0x1F: "Unspecified Error",
    0x20: "Unsupported LMP Parameter Value / Unsupported LL Parameter Value",
    0x21: "Role Change Not Allowed",
    0x22: "LMP Response Timeout / LL Response Timeout",
    0x23: "LMP Error Transaction Collision / LL Procedure Collision",
    0x24: "LMP PDU Not Allowed",
    0x25: "Encryption Mode Not Acceptable",
    0x26: "Link Key cannot be Changed",
    0x27: "Requested QoS Not Supported",
    0x28: "Instant Passed",
    0x29: "Pairing With Unit Key Not Supported",
    0x2A: "Different Transaction Collision",
    0x2B: "Reserved for future use",
    0x2C: "QoS Unacceptable Parameter",
    0x2D: "QoS Rejected",
    0x2E: "Channel Classification Not Supported",
    0x2F: "Insufficient Security",
    0x30: "Parameter Out Of Mandatory Range",
    0x31: "Reserved for future use",
    0x32: "Role Switch Pending",
    0x33: "Reserved for future use",
    0x34: "Reserved Slot Violation",
    0x35: "Role Switch Failed",
    0x36: "Extended Inquiry Response Too Large",
    0x37: "Secure Simple Pairing Not Supported By Host",
    0x38: "Host Busy - Pairing",
    0x39: "Connection Rejected due to No Suitable Channel Found",
    0x3A: "Controller Busy",
    0x3B: "Unacceptable Connection Parameters",
    0x3C: "Advertising Timeout",
    0x3D: "Connection Terminated due to MIC Failure",
    0x3E: "Connection Failed to be Established / Synchronization Timeout",
    0x3F: "MAC Connection Failed",
    0x40: "Coarse Clock Adjustment Rejected but Will Try to Adjust Using Clock",
    0x41: "Type0 Submap Not Defined",
    0x42: "Unknown Advertising Identifier",
    0x43: "Limit Reached",
    0x44: "Operation Cancelled by Host",
    0x45: "Packet Too Long",
}

# GATT protocol error codes  
PROTOCOL_ERROR_CODES: dict[int, str] = {
    0x01: "Invalid Handle",
    0x02: "Read Not Permitted",
    0x03: "Write Not Permitted",
    0x04: "Invalid PDU",
    0x05: "Insufficient Authentication",
    0x06: "Request Not Supported",
    0x07: "Invalid Offset",
    0x08: "Insufficient Authorization",
    0x09: "Prepare Queue Full",
    0x0A: "Attribute Not Found",
    0x0B: "Attribute Not Long",
    0x0C: "Insufficient Encryption Key Size",
    0x0D: "Invalid Attribute Value Length",
    0x0E: "Unlikely Error",
    0x0F: "Insufficient Encryption",
    0x10: "Unsupported Group Type",
    0x11: "Insufficient Resource",
    0x12: "Database Out Of Sync",
    0x13: "Value Not Allowed",
    0xFC: "Write Request Rejected",
    0xFD: "Client Characteristic Configuration Descriptor Improperly Configured",
    0xFE: "Procedure Already in Progress",
    0xFF: "Out of Range",
}

import asyncio
from bleak import BleakClient, BleakError, BleakCharacteristicNotFoundError

async def safe_characteristic_read():
    address = "00:11:22:33:44:55"  # Replace with actual device address
    char_uuid = "00002a19-0000-1000-8000-00805f9b34fb"  # Battery Level
    
    try:
        async with BleakClient(address) as client:
            # Attempt to read characteristic
            data = await client.read_gatt_char(char_uuid)
            battery_level = int.from_bytes(data, byteorder='little')
            print(f"Battery level: {battery_level}%")
            
    except BleakCharacteristicNotFoundError as e:
        print(f"Characteristic not found: {e.char_specifier}")
        print("Device may not support battery level reporting")
        
    except BleakError as e:
        print(f"BLE operation failed: {e}")
        
    except Exception as e:
        print(f"Unexpected error: {e}")

asyncio.run(safe_characteristic_read())

import asyncio
from bleak import BleakClient, BleakDeviceNotFoundError, BleakError

async def safe_connection():
    address = "00:11:22:33:44:55"  # Replace with actual device address
    
    try:
        client = BleakClient(address, timeout=10.0)
        await client.connect()
        
        print(f"Connected to {client.name}")
        print(f"Address: {client.address}")
        print(f"Connected: {client.is_connected}")
        
        # Perform operations...
        
        await client.disconnect()
        
    except BleakDeviceNotFoundError as e:
        print(f"Device not found: {e.identifier}")
        print("Make sure the device is powered on and in range")
        
    except asyncio.TimeoutError:
        print("Connection timed out")
        print("Device may be out of range or not advertising")
        
    except BleakError as e:
        print(f"Connection failed: {e}")
        
    except Exception as e:
        print(f"Unexpected error during connection: {e}")

asyncio.run(safe_connection())

import asyncio
import platform
from bleak import BleakClient, BleakDBusError, BleakError

async def platform_specific_handling():
    address = "00:11:22:33:44:55"  # Replace with actual device address
    
    try:
        async with BleakClient(address) as client:
            # Attempt some operation
            services = client.services
            print(f"Discovered {len(services.services)} services")
            
    except BleakDBusError as e:
        # Linux/BlueZ specific error handling
        print(f"D-Bus error: {e.dbus_error}")
        if e.dbus_error_details:
            print(f"Details: {e.dbus_error_details}")
            
        # Handle specific D-Bus errors
        if "org.freedesktop.DBus.Error.UnknownObject" in e.dbus_error:
            print("Device object not found in BlueZ")
        elif "org.bluez.Error.NotReady" in e.dbus_error:
            print("Bluetooth adapter not ready")
            
    except BleakError as e:
        print(f"General BLE error: {e}")
        
        # Platform-specific advice
        if platform.system() == "Darwin":  # macOS
            print("Note: On macOS, device addresses are UUIDs, not MAC addresses")
        elif platform.system() == "Windows":
            print("Note: On Windows, ensure device is paired if required")
        elif platform.system() == "Linux":
            print("Note: On Linux, ensure BlueZ is running and user has permissions")

asyncio.run(platform_specific_handling())

import asyncio
from bleak import BleakClient, BleakError, BleakCharacteristicNotFoundError

def notification_handler(characteristic, data):
    print(f"Notification: {data.hex()}")

async def safe_notifications():
    address = "00:11:22:33:44:55"  # Replace with actual device address
    notify_char = "12345678-1234-5678-9012-123456789abc"  # Replace with actual UUID
    
    try:
        async with BleakClient(address) as client:
            # Check if characteristic exists and supports notifications
            char = client.services.get_characteristic(notify_char)
            if not char:
                raise BleakCharacteristicNotFoundError(notify_char)
                
            if "notify" not in char.properties:
                print(f"Characteristic {notify_char} does not support notifications")
                return
            
            # Start notifications
            await client.start_notify(notify_char, notification_handler)
            print("Notifications started successfully")
            
            # Keep connection alive
            await asyncio.sleep(30)
            
            # Stop notifications (automatic on disconnect)
            await client.stop_notify(notify_char)
            print("Notifications stopped")
            
    except BleakCharacteristicNotFoundError as e:
        print(f"Notification characteristic not found: {e.char_specifier}")
        
    except BleakError as e:
        print(f"Notification setup failed: {e}")
        
        # Check for common notification issues
        error_str = str(e).lower()
        if "not connected" in error_str:
            print("Ensure device is connected before starting notifications")
        elif "cccd" in error_str or "descriptor" in error_str:
            print("Device may not support notifications on this characteristic")

asyncio.run(safe_notifications())

import asyncio
from bleak import BleakClient, BleakError, BleakCharacteristicNotFoundError

async def safe_write_operations():
    address = "00:11:22:33:44:55"  # Replace with actual device address
    write_char = "12345678-1234-5678-9012-123456789abc"  # Replace with actual UUID
    
    try:
        async with BleakClient(address) as client:
            # Check if characteristic supports writing
            char = client.services.get_characteristic(write_char)
            if not char:
                raise BleakCharacteristicNotFoundError(write_char)
            
            # Check write capabilities
            can_write = "write" in char.properties
            can_write_no_response = "write-without-response" in char.properties
            
            if not (can_write or can_write_no_response):
                print(f"Characteristic {write_char} does not support writing")
                return
            
            data = b"Hello, BLE device!"
            
            # Attempt write with response if supported
            if can_write:
                try:
                    await client.write_gatt_char(write_char, data, response=True)
                    print("Write with response successful")
                except BleakError as e:
                    print(f"Write with response failed: {e}")
                    
                    # Try write without response as fallback
                    if can_write_no_response:
                        await client.write_gatt_char(write_char, data, response=False)
                        print("Fallback write without response successful")
            else:
                # Only write without response available
                await client.write_gatt_char(write_char, data, response=False)
                print("Write without response successful")
                
    except BleakCharacteristicNotFoundError as e:
        print(f"Write characteristic not found: {e.char_specifier}")
        
    except BleakError as e:
        print(f"Write operation failed: {e}")
        
        # Handle common write errors
        error_str = str(e).lower()
        if "mtu" in error_str or "size" in error_str:
            print("Data may be too large for device MTU")
            print(f"Try reducing data size (current: {len(data)} bytes)")
        elif "authentication" in error_str or "permission" in error_str:
            print("Write may require device pairing or authentication")

asyncio.run(safe_write_operations())

import asyncio
from bleak import BleakClient, BleakError

async def safe_service_discovery():
    address = "00:11:22:33:44:55"  # Replace with actual device address
    
    try:
        async with BleakClient(address) as client:
            print(f"Connected to {client.name}")
            
            # Access services (triggers service discovery if not done)
            try:
                services = client.services
                print(f"Discovered {len(services.services)} services")
                
                # Verify service discovery completed successfully
                if not services.services:
                    print("Warning: No services discovered - device may not be fully connected")
                    
                # List discovered services
                for service in services:
                    print(f"Service: {service.uuid} - {service.description}")
                    
            except BleakError as e:
                print(f"Service discovery failed: {e}")
                
                # Common service discovery issues
                error_str = str(e).lower()
                if "not performed" in error_str:
                    print("Service discovery was not completed during connection")
                elif "timeout" in error_str:
                    print("Service discovery timed out - device may be slow to respond")
                    
    except BleakError as e:
        print(f"Connection or service discovery error: {e}")

asyncio.run(safe_service_discovery())

# Error code mappings (partial - full mappings in bleak.exc)
CONTROLLER_ERROR_CODES: dict[int, str]
PROTOCOL_ERROR_CODES: dict[int, str]

# Exception hierarchy
class BleakError(Exception): ...
class BleakCharacteristicNotFoundError(BleakError): ...
class BleakDeviceNotFoundError(BleakError): ...  
class BleakDBusError(BleakError): ...