ble-reticulum/src/RNS/Interfaces/linux_bluetooth_driver.py

"""
Linux Bluetooth Driver for BLE

This module implements the BLEDriverInterface abstraction for Linux using:
- bleak: BLE central operations (scanning, connecting, GATT client)
- bluezero: BLE peripheral operations (GATT server, advertising)
- D-Bus: Direct BlueZ API access for platform-specific workarounds

Platform-specific workarounds included:
1. BlueZ ServicesResolved race condition (Bleak 1.1.1 + bluezero)
2. LE-only connection via D-Bus ConnectDevice (BlueZ >= 5.49)
3. BLE Agent registration for automatic pairing
4. MTU negotiation via 3 fallback methods

USAGE EXAMPLE:
--------------

    from linux_bluetooth_driver import LinuxBluetoothDriver

    # Create driver instance (no Reticulum dependencies)
    driver = LinuxBluetoothDriver(
        discovery_interval=5.0,
        connection_timeout=10.0,
        min_rssi=-90,
        service_discovery_delay=1.5,
        max_peers=7,
        adapter_index=0  # hci0
    )

    # Set up callbacks
    def on_device_discovered(device):
        print(f"Discovered: {device.name} ({device.address}) RSSI: {device.rssi}")

    def on_device_connected(address):
        print(f"Connected: {address}")

    def on_data_received(address, data):
        print(f"Received {len(data)} bytes from {address}")

    def on_mtu_negotiated(address, mtu):
        print(f"MTU negotiated with {address}: {mtu}")

    driver.on_device_discovered = on_device_discovered
    driver.on_device_connected = on_device_connected
    driver.on_data_received = on_data_received
    driver.on_mtu_negotiated = on_mtu_negotiated

    # Start driver
    driver.start(
        service_uuid="37145b00-442d-4a94-917f-8f42c5da28e3",
        rx_char_uuid="37145b00-442d-4a94-917f-8f42c5da28e5",
        tx_char_uuid="37145b00-442d-4a94-917f-8f42c5da28e4",
        identity_char_uuid="37145b00-442d-4a94-917f-8f42c5da28e6"
    )

    # Set identity for peripheral mode
    driver.set_identity(b"\\x01\\x02\\x03...\\x10")  # 16 bytes

    # Start scanning (central mode)
    driver.start_scanning()

    # Start advertising (peripheral mode)
    driver.start_advertising("MyDevice", b"\\x01\\x02\\x03...\\x10")

    # Connect to a peer
    driver.connect("AA:BB:CC:DD:EE:FF")

    # Send data (automatically uses GATT write or notification)
    driver.send("AA:BB:CC:DD:EE:FF", b"Hello, peer!")

    # Stop driver
    driver.stop()

ARCHITECTURE:
-------------

The driver uses a dedicated asyncio event loop in a separate thread to handle
all BLE operations asynchronously. This allows the main thread to remain
responsive while BLE operations run in the background.

Thread Architecture:
- Main thread: User-facing API (start, stop, connect, send, etc.)
- Event loop thread: All async BLE operations (scanning, connecting, GATT ops)
- GATT server thread: Bluezero peripheral (blocking publish())

Cross-thread communication:
- Main → Event loop: asyncio.run_coroutine_threadsafe()
- Event loop → Main: Callbacks (on_device_discovered, on_data_received, etc.)
- GATT server → Main: Callbacks from bluezero write_callback

ROLE-AWARE send():
------------------

The send() method automatically determines whether to use GATT write (central)
or notification (peripheral) based on the connection type:

- Central connection (we connected to them): GATT write to RX characteristic
- Peripheral connection (they connected to us): Notification on TX characteristic

This abstraction simplifies the high-level interface logic by hiding the
BLE role complexity at the driver level.

DEPENDENCIES:
-------------

Required:
- bleak >= 0.22.0 (BLE central operations)
- dbus-fast >= 1.0.0 (D-Bus communication)

Optional (for peripheral mode):
- bluezero >= 0.9.1 (GATT server)
- dbus-python >= 1.2.18 (bluezero dependency)

Author: Reticulum BLE Interface Contributors
License: MIT
"""

from __future__ import annotations

import asyncio
import threading
import time
import logging
from typing import Optional, Callable, List, Dict
from dataclasses import dataclass

# Import the abstraction
try:
    from bluetooth_driver import BLEDriverInterface, BLEDevice, DriverState
except ImportError:
    import sys
    import os
    sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
    from bluetooth_driver import BLEDriverInterface, BLEDevice, DriverState

# Bleak (BLE central operations)
try:
    import bleak
    from bleak import BleakScanner, BleakClient
    from bleak.backends.bluezdbus.manager import BlueZManager
    HAS_BLEAK = True
except ImportError:
    HAS_BLEAK = False
    BleakScanner = None
    BleakClient = None

# Bluezero (BLE peripheral operations)
try:
    from bluezero import peripheral, adapter
    BLUEZERO_AVAILABLE = True
except ImportError:
    BLUEZERO_AVAILABLE = False

# BLE Agent for automatic pairing
try:
    from BLEAgent import register_agent, unregister_agent
    HAS_BLE_AGENT = True
except ImportError:
    try:
        from RNS.Interfaces.BLEAgent import register_agent, unregister_agent
        HAS_BLE_AGENT = True
    except ImportError:
        HAS_BLE_AGENT = False

# D-Bus for platform-specific operations
try:
    from dbus_fast.aio import MessageBus
    from dbus_fast import BusType, Variant
    HAS_DBUS = True
except ImportError:
    HAS_DBUS = False


# ============================================================================
# BlueZ ServicesResolved Race Condition Workaround
# ============================================================================
# Issue: When connecting to BlueZ-based GATT servers (like bluezero), BlueZ
#        sets ServicesResolved=True BEFORE services are fully exported to D-Bus
# Cause: BlueZ GATT database cache timing issue (bluez/bluez#1489)
# Impact: Bleak attempts to enumerate services before they're available,
#         causing -5 (EIO) error and immediate disconnect
# Fix: Poll D-Bus service map to verify services actually exist before proceeding
# Status: Works with bluezero; proper fix should be in BlueZ or Bleak upstream
# GitHub: https://github.com/hbldh/bleak/issues/1677
# ============================================================================

def apply_bluez_services_resolved_patch():
    """
    Apply monkey patch to fix BlueZ ServicesResolved race condition.

    This must be called before any BleakClient connections are made.
    """
    if not HAS_BLEAK:
        return False

    try:
        # Store original method
        _original_wait_for_services_discovery = BlueZManager._wait_for_services_discovery

        async def _patched_wait_for_services_discovery(self, device_path: str) -> None:
            """
            Patched version that waits for services to actually appear in D-Bus.

            Fixes race condition where ServicesResolved=True before services
            are fully exported to D-Bus (common when connecting to BlueZ peripherals).
            """
            # Call original wait for ServicesResolved property
            await _original_wait_for_services_discovery(self, device_path)

            # Additional verification: Poll until services actually appear in D-Bus
            max_attempts = 20  # 20 attempts * 100ms = 2 seconds max
            retry_delay = 0.1  # 100ms between attempts

            for attempt in range(max_attempts):
                # Check if services are actually present in the service map
                service_paths = self._service_map.get(device_path, set())

                if service_paths and len(service_paths) > 0:
                    # Services found! Verify at least one service has been fully loaded
                    # by checking if it exists in the properties dictionary
                    try:
                        first_service_path = next(iter(service_paths))
                        if first_service_path in self._properties:
                            # Success: Services are actually in D-Bus
                            logging.debug(f"BlueZ timing fix: Services verified in D-Bus after {attempt * retry_delay:.2f}s")
                            return
                    except (StopIteration, KeyError):
                        pass  # Service not ready yet

                # Services not ready yet, wait before next check
                if attempt < max_attempts - 1:  # Don't sleep on last attempt
                    await asyncio.sleep(retry_delay)

            # If we get here, services didn't appear within timeout
            # Log warning but don't raise - let get_services() handle it
            logging.warning(f"BlueZ timing fix: Services not found in D-Bus after {max_attempts * retry_delay}s, proceeding anyway")

        # Apply the patch
        BlueZManager._wait_for_services_discovery = _patched_wait_for_services_discovery
        logging.info("Applied Bleak BlueZ ServicesResolved timing patch for bluezero compatibility")
        return True

    except Exception as e:
        # If patching fails, log warning but don't prevent driver from loading
        logging.warning(f"Failed to apply Bleak BlueZ timing patch: {e}. Connections to bluezero peripherals may fail.")
        return False


@dataclass
class PeerConnection:
    """Tracks information about a connected peer."""
    address: str
    client: Optional[BleakClient] = None  # For central connections
    mtu: int = 23  # Negotiated MTU
    connection_type: str = "unknown"  # "central" or "peripheral"
    connected_at: float = 0.0
    peer_identity: Optional[bytes] = None  # 16-byte identity hash


class LinuxBluetoothDriver(BLEDriverInterface):
    """
    Linux implementation of BLE driver using bleak and bluezero.

    This driver provides:
    - Central mode: BLE scanning and connections via bleak
    - Peripheral mode: GATT server and advertising via bluezero
    - Platform workarounds for BlueZ quirks
    - Dedicated asyncio event loop in separate thread
    - Role-aware send() that automatically uses GATT write or notification

    Architecture:
    - Main thread: User-facing API (start, stop, send, etc.)
    - Event loop thread: All async BLE operations
    - Cross-thread communication via run_coroutine_threadsafe
    """

    def __init__(
        self,
        discovery_interval: float = 5.0,
        connection_timeout: float = 10.0,
        min_rssi: int = -90,
        service_discovery_delay: float = 1.5,
        max_peers: int = 7,
        adapter_index: int = 0,
        agent_capability: str = "NoInputNoOutput"
    ):
        """
        Initialize Linux BLE driver.

        Args:
            discovery_interval: Seconds between discovery scans (default: 5.0)
            connection_timeout: Connection timeout in seconds (default: 10.0)
            min_rssi: Minimum RSSI for connection attempts (default: -90 dBm)
            service_discovery_delay: Delay after connection for bluezero D-Bus registration (default: 1.5s)
            max_peers: Maximum simultaneous connections (default: 7)
            adapter_index: Bluetooth adapter index (0 = hci0, 1 = hci1, etc.)
            agent_capability: BLE pairing agent capability (default: "NoInputNoOutput" for Just Works pairing)
        """
        # Validate dependencies
        if not HAS_BLEAK:
            raise ImportError("bleak library required for Linux BLE driver. Install with: pip install bleak>=0.22.0")

        # Configuration
        self.discovery_interval = discovery_interval
        self.connection_timeout = connection_timeout
        self.min_rssi = min_rssi
        self.service_discovery_delay = service_discovery_delay
        self.max_peers = max_peers
        self.adapter_index = adapter_index
        self.adapter_path = f"/org/bluez/hci{adapter_index}"
        self.agent_capability = agent_capability

        # Service UUIDs (set by start())
        self.service_uuid: Optional[str] = None
        self.rx_char_uuid: Optional[str] = None
        self.tx_char_uuid: Optional[str] = None
        self.identity_char_uuid: Optional[str] = None

        # State
        self._state = DriverState.IDLE
        self._running = False
        self._scanning = False
        self._advertising = False

        # Connected peers
        self._peers: Dict[str, PeerConnection] = {}  # address -> PeerConnection
        self._peers_lock = threading.RLock()

        # Local identity (for peripheral mode)
        self._local_identity: Optional[bytes] = None

        # Local adapter address (for connection direction preference)
        self.local_address: Optional[str] = None

        # Power mode
        self.power_mode = "balanced"  # "aggressive", "balanced", "saver"

        # Event loop management
        self.loop: Optional[asyncio.AbstractEventLoop] = None
        self.loop_thread: Optional[threading.Thread] = None

        # Peripheral mode (bluezero)
        self.gatt_server: Optional['BluezeroGATTServer'] = None
        self.ble_agent = None

        # BlueZ version detection
        self.bluez_version: Optional[tuple] = None
        self.has_connect_device = None  # None = unknown, True/False = tested

        # Logging
        self.log_prefix = "LinuxBLEDriver"

        # Apply BlueZ timing patch
        apply_bluez_services_resolved_patch()

        # Detect BlueZ version
        self._detect_bluez_version()

    def _log(self, message: str, level: str = "INFO"):
        """Log message with appropriate level."""
        log_func = getattr(logging, level.lower(), logging.info)
        log_func(f"{self.log_prefix} {message}")

    # ========================================================================
    # Lifecycle & Configuration
    # ========================================================================

    def start(self, service_uuid: str, rx_char_uuid: str, tx_char_uuid: str, identity_char_uuid: str):
        """
        Initialize the driver and start the BLE stack.

        This creates the dedicated event loop thread and initializes the GATT server.
        """
        if self._running:
            self._log("Driver already running", "WARNING")
            return

        self._log("Starting Linux BLE driver...")

        # Store UUIDs
        self.service_uuid = service_uuid
        self.rx_char_uuid = rx_char_uuid
        self.tx_char_uuid = tx_char_uuid
        self.identity_char_uuid = identity_char_uuid

        # Start event loop thread
        self.loop_thread = threading.Thread(target=self._run_event_loop, daemon=True, name="BLE-EventLoop")
        self.loop_thread.start()

        # Wait for event loop to be ready
        timeout = 5.0
        start_time = time.time()
        while self.loop is None and (time.time() - start_time) < timeout:
            time.sleep(0.1)

        if self.loop is None:
            raise RuntimeError("Failed to start event loop within timeout")

        # Get local adapter address
        future = asyncio.run_coroutine_threadsafe(self._get_local_adapter_address(), self.loop)
        try:
            self.local_address = future.result(timeout=5.0)
            if self.local_address:
                self._log(f"Local adapter address: {self.local_address}")
        except Exception as e:
            self._log(f"Could not get local adapter address: {e}", "WARNING")

        # Initialize GATT server for peripheral mode (if bluezero available)
        if BLUEZERO_AVAILABLE:
            try:
                self.gatt_server = BluezeroGATTServer(
                    driver=self,
                    service_uuid=service_uuid,
                    rx_char_uuid=rx_char_uuid,
                    tx_char_uuid=tx_char_uuid,
                    identity_char_uuid=identity_char_uuid,
                    adapter_index=self.adapter_index,
                    agent_capability=self.agent_capability
                )
                self._log("GATT server initialized")
            except Exception as e:
                self._log(f"Failed to initialize GATT server: {e}", "WARNING")
                self.gatt_server = None
        else:
            self._log("Bluezero not available, peripheral mode disabled", "WARNING")

        self._running = True
        self._state = DriverState.IDLE
        self._log("Driver started successfully")

    def stop(self):
        """Stop all BLE activity and release resources."""
        if not self._running:
            return

        self._log("Stopping Linux BLE driver...")
        self._running = False

        # Stop scanning
        if self._scanning:
            self.stop_scanning()

        # Stop advertising
        if self._advertising:
            self.stop_advertising()

        # Disconnect all peers
        with self._peers_lock:
            for address in list(self._peers.keys()):
                try:
                    self.disconnect(address)
                except Exception as e:
                    self._log(f"Error disconnecting {address}: {e}", "WARNING")

        # Stop GATT server
        if self.gatt_server:
            try:
                self.gatt_server.stop()
            except Exception as e:
                self._log(f"Error stopping GATT server: {e}", "WARNING")

        # Stop event loop
        if self.loop and self.loop.is_running():
            self.loop.call_soon_threadsafe(self.loop.stop)

        # Wait for thread to exit
        if self.loop_thread and self.loop_thread.is_alive():
            self.loop_thread.join(timeout=5.0)

        self._state = DriverState.IDLE
        self._log("Driver stopped")

    def set_identity(self, identity_bytes: bytes):
        """Set the local identity for the GATT server."""
        if not isinstance(identity_bytes, bytes):
            raise TypeError(f"identity_bytes must be bytes, got {type(identity_bytes)}")

        if len(identity_bytes) != 16:
            raise ValueError(f"identity_bytes must be 16 bytes, got {len(identity_bytes)}")

        self._local_identity = identity_bytes

        if self.gatt_server:
            self.gatt_server.set_identity(identity_bytes)

        self._log(f"Local identity set: {identity_bytes.hex()}")

    # ========================================================================
    # State & Properties
    # ========================================================================

    @property
    def state(self) -> DriverState:
        """Return current driver state."""
        return self._state

    @property
    def connected_peers(self) -> List[str]:
        """Return list of connected peer addresses."""
        with self._peers_lock:
            return list(self._peers.keys())

    # ========================================================================
    # Scanning (Central Mode)
    # ========================================================================

    def start_scanning(self):
        """Start scanning for BLE devices."""
        if not self._running:
            self._log("Cannot start scanning: driver not running", "ERROR")
            return

        if self._scanning:
            self._log("Already scanning", "DEBUG")
            return

        self._log("Starting BLE scanning...")
        self._scanning = True
        self._state = DriverState.SCANNING

        # Start scan loop in event loop
        asyncio.run_coroutine_threadsafe(self._scan_loop(), self.loop)

    def stop_scanning(self):
        """Stop scanning for BLE devices."""
        if not self._scanning:
            return

        self._log("Stopping BLE scanning...")
        self._scanning = False

        if not self._advertising:
            self._state = DriverState.IDLE

    async def _scan_loop(self):
        """Main scanning loop (runs in event loop thread)."""
        self._log("Scan loop started", "DEBUG")

        while self._scanning and self._running:
            try:
                await self._perform_scan()

                # Sleep based on power mode
                if self.power_mode == "aggressive":
                    sleep_time = 1.0
                elif self.power_mode == "saver":
                    # Skip scanning if we have connected peers
                    with self._peers_lock:
                        if len(self._peers) > 0:
                            sleep_time = 60.0
                        else:
                            sleep_time = 30.0
                else:  # balanced
                    sleep_time = self.discovery_interval

                await asyncio.sleep(sleep_time)

            except Exception as e:
                self._log(f"Error in scan loop: {e}", "ERROR")
                await asyncio.sleep(5.0)  # Back off on errors

        self._log("Scan loop stopped", "DEBUG")

    async def _perform_scan(self):
        """Perform a single BLE scan."""
        discovered_devices = []

        def detection_callback(device, advertisement_data):
            """Called for each discovered device."""
            discovered_devices.append((device, advertisement_data))

        # Scan duration based on power mode
        if self.power_mode == "aggressive":
            scan_time = 2.0
        elif self.power_mode == "saver":
            scan_time = 0.5
        else:  # balanced
            scan_time = 1.0

        scanner = BleakScanner(detection_callback=detection_callback)

        try:
            await scanner.start()
            await asyncio.sleep(scan_time)
            await scanner.stop()
        except Exception as e:
            error_msg = str(e)

            # Check for adapter power issues
            if "No powered Bluetooth adapters" in error_msg or "Not Powered" in error_msg:
                self._log("Bluetooth adapter is not powered!", "ERROR")
                if self.on_error:
                    self.on_error("error", "Bluetooth adapter not powered. Run 'bluetoothctl power on'", e)
                return
            else:
                raise

        # Process discovered devices
        for device, adv_data in discovered_devices:
            # Check if device advertises our service UUID
            if self.service_uuid and self.service_uuid.lower() in [uuid.lower() for uuid in adv_data.service_uuids]:
                # Check RSSI threshold
                if adv_data.rssi < self.min_rssi:
                    continue

                # Create BLEDevice and notify callback
                ble_device = BLEDevice(
                    address=device.address,
                    name=device.name or "Unknown",
                    rssi=adv_data.rssi,
                    service_uuids=list(adv_data.service_uuids),
                    manufacturer_data=dict(adv_data.manufacturer_data) if hasattr(adv_data, 'manufacturer_data') else {}
                )

                if self.on_device_discovered:
                    try:
                        self.on_device_discovered(ble_device)
                    except Exception as e:
                        self._log(f"Error in device discovered callback: {e}", "ERROR")

    # ========================================================================
    # Advertising (Peripheral Mode)
    # ========================================================================

    def start_advertising(self, device_name: str, identity: bytes):
        """Start advertising as a BLE peripheral."""
        if not self._running:
            self._log("Cannot start advertising: driver not running", "ERROR")
            return

        if not self.gatt_server:
            self._log("Cannot start advertising: GATT server not available", "ERROR")
            if self.on_error:
                self.on_error("error", "GATT server not available (bluezero not installed?)", None)
            return

        if self._advertising:
            self._log("Already advertising", "DEBUG")
            return

        self._log(f"Starting BLE advertising as '{device_name}'...")

        # Set identity
        self.set_identity(identity)

        # Start GATT server
        try:
            self.gatt_server.start(device_name)
            self._advertising = True
            self._state = DriverState.ADVERTISING
            self._log("Advertising started")
        except Exception as e:
            self._log(f"Failed to start advertising: {e}", "ERROR")
            if self.on_error:
                self.on_error("error", f"Failed to start advertising: {e}", e)

    def stop_advertising(self):
        """Stop advertising."""
        if not self._advertising:
            return

        self._log("Stopping BLE advertising...")

        if self.gatt_server:
            try:
                self.gatt_server.stop()
            except Exception as e:
                self._log(f"Error stopping GATT server: {e}", "WARNING")

        self._advertising = False

        if not self._scanning:
            self._state = DriverState.IDLE

    # ========================================================================
    # Connection Management (Central Mode)
    # ========================================================================

    def connect(self, address: str):
        """Connect to a peer device (central role)."""
        if not self._running:
            self._log("Cannot connect: driver not running", "ERROR")
            return

        # Check if already connected
        with self._peers_lock:
            if address in self._peers:
                self._log(f"Already connected to {address}", "DEBUG")
                return

        # Check max peers
        with self._peers_lock:
            if len(self._peers) >= self.max_peers:
                self._log(f"Cannot connect to {address}: max peers ({self.max_peers}) reached", "WARNING")
                return

        # Start connection in event loop
        asyncio.run_coroutine_threadsafe(self._connect_to_peer(address), self.loop)

    def disconnect(self, address: str):
        """Disconnect from a peer device."""
        with self._peers_lock:
            if address not in self._peers:
                self._log(f"Not connected to {address}", "DEBUG")
                return

            peer = self._peers[address]

        # Disconnect based on connection type
        if peer.connection_type == "central" and peer.client:
            # Central connection: disconnect client
            future = asyncio.run_coroutine_threadsafe(peer.client.disconnect(), self.loop)
            try:
                future.result(timeout=5.0)
            except Exception as e:
                self._log(f"Error disconnecting from {address}: {e}", "WARNING")

        # For peripheral connections, client disconnects from us (we can't force disconnect)

        # Clean up
        with self._peers_lock:
            if address in self._peers:
                del self._peers[address]

        if self.on_device_disconnected:
            try:
                self.on_device_disconnected(address)
            except Exception as e:
                self._log(f"Error in device disconnected callback: {e}", "ERROR")

        self._log(f"Disconnected from {address}")

    async def _connect_to_peer(self, address: str):
        """Connect to a peer (runs in event loop thread)."""
        self._log(f"Connecting to {address}...", "DEBUG")

        try:
            # Create disconnection callback
            def disconnected_callback(client_obj):
                """Called when device disconnects."""
                self._log(f"Device {address} disconnected unexpectedly", "WARNING")

                # Clean up
                with self._peers_lock:
                    if address in self._peers:
                        del self._peers[address]

                if self.on_device_disconnected:
                    try:
                        self.on_device_disconnected(address)
                    except Exception as e:
                        self._log(f"Error in device disconnected callback: {e}", "ERROR")

            # Try LE-specific connection if BlueZ >= 5.49
            le_connection_attempted = False
            if self.bluez_version and self.bluez_version >= (5, 49) and self.has_connect_device is None:
                try:
                    await self._connect_via_dbus_le(address)
                    le_connection_attempted = True
                    self._log(f"LE-specific connection initiated for {address}", "DEBUG")
                except Exception as e:
                    self._log(f"ConnectDevice() unavailable, falling back to standard connection", "DEBUG")
                    self.has_connect_device = False

            # Create BleakClient
            client = BleakClient(address, disconnected_callback=disconnected_callback, timeout=self.connection_timeout)

            # Connect
            if not le_connection_attempted:
                await client.connect(timeout=self.connection_timeout)
            else:
                # If ConnectDevice was used, check if already connected
                if not client.is_connected:
                    await client.connect(timeout=self.connection_timeout)

            if not client.is_connected:
                raise RuntimeError("Connection failed")

            # Service discovery delay (for bluezero D-Bus registration)
            if self.service_discovery_delay > 0:
                self._log(f"Waiting {self.service_discovery_delay}s for service discovery...", "DEBUG")
                await asyncio.sleep(self.service_discovery_delay)

            # Discover services
            services = list(client.services) if client.services else []

            # Fallback: force discovery if services empty
            if not services:
                self._log("Services property empty, forcing discovery...", "DEBUG")
                services_collection = await client.get_services()
                services = list(services_collection)

            # Find Reticulum service
            reticulum_service = None
            for svc in services:
                if svc.uuid.lower() == self.service_uuid.lower():
                    reticulum_service = svc
                    break

            if not reticulum_service:
                raise RuntimeError(f"Reticulum service {self.service_uuid} not found")

            # Read identity characteristic
            peer_identity = None
            for char in reticulum_service.characteristics:
                if char.uuid.lower() == self.identity_char_uuid.lower():
                    identity_value = await client.read_gatt_char(char)
                    if len(identity_value) == 16:
                        peer_identity = bytes(identity_value)
                        self._log(f"Read identity from {address}: {peer_identity.hex()}", "DEBUG")
                    break

            if not peer_identity:
                raise RuntimeError("Could not read peer identity")

            # Negotiate MTU
            mtu = await self._negotiate_mtu(client)
            self._log(f"Negotiated MTU {mtu} with {address}", "DEBUG")

            # Store connection
            peer_conn = PeerConnection(
                address=address,
                client=client,
                mtu=mtu,
                connection_type="central",
                connected_at=time.time(),
                peer_identity=peer_identity
            )

            with self._peers_lock:
                self._peers[address] = peer_conn

            # Set up notifications
            await client.start_notify(
                self.tx_char_uuid,
                lambda sender, data: self._handle_notification(address, data)
            )

            # Send identity handshake (if we have local identity)
            if self._local_identity:
                try:
                    await client.write_gatt_char(
                        self.rx_char_uuid,
                        self._local_identity,
                        response=True
                    )
                    self._log(f"Sent identity handshake to {address}", "DEBUG")
                except Exception as e:
                    self._log(f"Failed to send identity handshake: {e}", "WARNING")

            # Notify callback with peer identity
            if self.on_device_connected:
                try:
                    self.on_device_connected(address, peer_identity)
                except Exception as e:
                    self._log(f"Error in device connected callback: {e}", "ERROR")

            # Notify MTU callback
            if self.on_mtu_negotiated:
                try:
                    self.on_mtu_negotiated(address, mtu)
                except Exception as e:
                    self._log(f"Error in MTU negotiated callback: {e}", "ERROR")

            self._log(f"Connected to {address} (MTU: {mtu})")

        except asyncio.TimeoutError:
            self._log(f"Connection timeout to {address}", "WARNING")
            if self.on_error:
                self.on_error("warning", f"Connection timeout to {address}", None)
        except Exception as e:
            self._log(f"Connection failed to {address}: {e}", "ERROR")
            if self.on_error:
                self.on_error("error", f"Connection failed to {address}: {e}", e)

    async def _connect_via_dbus_le(self, peer_address: str) -> bool:
        """
        Connect using D-Bus ConnectDevice() with explicit LE type.

        This forces BLE connection instead of BR/EDR on dual-mode devices.
        Requires BlueZ >= 5.49 with experimental mode (-E flag).
        """
        if not HAS_DBUS:
            raise ImportError("dbus_fast not available")

        self._log(f"Attempting LE-specific connection via ConnectDevice() to {peer_address}", "DEBUG")

        bus = await MessageBus(bus_type=BusType.SYSTEM).connect()

        # Get adapter interface
        introspection = await bus.introspect('org.bluez', self.adapter_path)
        adapter_obj = bus.get_proxy_object('org.bluez', self.adapter_path, introspection)
        adapter_iface = adapter_obj.get_interface('org.bluez.Adapter1')

        # Call ConnectDevice with LE parameters
        params = {
            "Address": Variant("s", peer_address),
            "AddressType": Variant("s", "public")  # Force LE public address
        }

        await adapter_iface.call_connect_device(params)

        self._log(f"ConnectDevice() succeeded for {peer_address}", "DEBUG")
        self.has_connect_device = True
        return True

    async def _negotiate_mtu(self, client: BleakClient) -> int:
        """
        Negotiate MTU using 3 fallback methods.

        Returns negotiated MTU size.
        """
        mtu = None

        # Method 1: Try direct MTU property access (BlueZ 5.62+)
        if hasattr(client, '_backend') and hasattr(client, 'services') and client.services:
            try:
                for char in client.services.characteristics.values():
                    if hasattr(char, 'obj') and len(char.obj) > 1:
                        char_props = char.obj[1]
                        if isinstance(char_props, dict) and "MTU" in char_props:
                            mtu = char_props["MTU"]
                            self._log(f"Read MTU {mtu} from characteristic property", "DEBUG")
                            break
            except Exception as e:
                self._log(f"Could not read MTU from characteristic properties: {e}", "DEBUG")

        # Method 2: Try _acquire_mtu() for older BlueZ versions
        if mtu is None and hasattr(client, '_backend') and hasattr(client._backend, '_acquire_mtu'):
            try:
                await client._backend._acquire_mtu()
                mtu = client.mtu_size
                self._log(f"Acquired MTU {mtu} via _acquire_mtu()", "DEBUG")
            except Exception as e:
                self._log(f"Failed to acquire MTU via _acquire_mtu(): {e}", "DEBUG")

        # Method 3: Fallback to client.mtu_size
        if mtu is None:
            try:
                mtu = client.mtu_size
                self._log(f"Using fallback MTU {mtu} from client.mtu_size", "DEBUG")
            except Exception as e:
                self._log(f"Could not get MTU, using default 23: {e}", "WARNING")
                mtu = 23

        return mtu

    def _handle_notification(self, address: str, data: bytes):
        """Handle incoming notification from peer."""
        if self.on_data_received:
            try:
                self.on_data_received(address, data)
            except Exception as e:
                self._log(f"Error in data received callback: {e}", "ERROR")

    # ========================================================================
    # Data Transmission
    # ========================================================================

    def send(self, address: str, data: bytes):
        """
        Send data to a connected peer.

        Automatically chooses GATT write (central) or notification (peripheral).
        """
        with self._peers_lock:
            if address not in self._peers:
                raise RuntimeError(f"Not connected to {address}")

            peer = self._peers[address]

        if peer.connection_type == "central":
            # We connected to them: use GATT write
            future = asyncio.run_coroutine_threadsafe(
                peer.client.write_gatt_char(self.rx_char_uuid, data, response=False),
                self.loop
            )
            try:
                future.result(timeout=5.0)
            except Exception as e:
                self._log(f"Error sending data to {address}: {e}", "ERROR")
                raise

        elif peer.connection_type == "peripheral":
            # They connected to us: use notification
            if self.gatt_server:
                try:
                    self.gatt_server.send_notification(address, data)
                except Exception as e:
                    self._log(f"Error sending notification to {address}: {e}", "ERROR")
                    raise
            else:
                raise RuntimeError("GATT server not available for peripheral connection")

        else:
            raise RuntimeError(f"Unknown connection type: {peer.connection_type}")

    # ========================================================================
    # GATT Characteristic Operations
    # ========================================================================

    def read_characteristic(self, address: str, char_uuid: str) -> bytes:
        """Read a GATT characteristic value."""
        with self._peers_lock:
            if address not in self._peers:
                raise RuntimeError(f"Not connected to {address}")

            peer = self._peers[address]

        if peer.connection_type != "central" or not peer.client:
            raise RuntimeError("Can only read characteristics in central mode")

        future = asyncio.run_coroutine_threadsafe(
            peer.client.read_gatt_char(char_uuid),
            self.loop
        )

        try:
            result = future.result(timeout=5.0)
            return bytes(result)
        except Exception as e:
            self._log(f"Error reading characteristic {char_uuid} from {address}: {type(e).__name__}: {e}", "ERROR")
            raise

    def write_characteristic(self, address: str, char_uuid: str, data: bytes):
        """Write a value to a GATT characteristic."""
        with self._peers_lock:
            if address not in self._peers:
                raise RuntimeError(f"Not connected to {address}")

            peer = self._peers[address]

        if peer.connection_type != "central" or not peer.client:
            raise RuntimeError("Can only write characteristics in central mode")

        future = asyncio.run_coroutine_threadsafe(
            peer.client.write_gatt_char(char_uuid, data, response=True),
            self.loop
        )

        try:
            future.result(timeout=5.0)
        except Exception as e:
            self._log(f"Error writing characteristic {char_uuid} to {address}: {e}", "ERROR")
            raise

    def start_notify(self, address: str, char_uuid: str, callback: Callable[[bytes], None]):
        """Subscribe to notifications from a GATT characteristic."""
        with self._peers_lock:
            if address not in self._peers:
                raise RuntimeError(f"Not connected to {address}")

            peer = self._peers[address]

        if peer.connection_type != "central" or not peer.client:
            raise RuntimeError("Can only subscribe to notifications in central mode")

        def notification_handler(sender, data):
            """Wrapper to call user callback."""
            try:
                callback(bytes(data))
            except Exception as e:
                self._log(f"Error in notification callback: {e}", "ERROR")

        future = asyncio.run_coroutine_threadsafe(
            peer.client.start_notify(char_uuid, notification_handler),
            self.loop
        )

        try:
            future.result(timeout=5.0)
        except Exception as e:
            self._log(f"Error starting notifications for {char_uuid} from {address}: {e}", "ERROR")
            raise

    # ========================================================================
    # Configuration & Queries
    # ========================================================================

    def get_local_address(self) -> str:
        """Return local Bluetooth adapter MAC address."""
        return self.local_address or "00:00:00:00:00:00"

    def get_peer_role(self, address: str) -> Optional[str]:
        """Return the connection role ('central' or 'peripheral') for a peer."""
        with self._peers_lock:
            if address in self._peers:
                return self._peers[address].connection_type
            return None

    def set_service_discovery_delay(self, seconds: float):
        """Set delay between connection and service discovery."""
        self.service_discovery_delay = seconds
        self._log(f"Service discovery delay set to {seconds}s")

    def set_power_mode(self, mode: str):
        """Set power mode for scanning."""
        if mode not in ["aggressive", "balanced", "saver"]:
            raise ValueError(f"Invalid power mode: {mode}")

        self.power_mode = mode
        self._log(f"Power mode set to {mode}")

    # ========================================================================
    # Event Loop Management
    # ========================================================================

    def _run_event_loop(self):
        """Run asyncio event loop in separate thread."""
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)
        self._log("Event loop thread started", "DEBUG")
        self.loop.run_forever()
        self._log("Event loop thread stopped", "DEBUG")

    # ========================================================================
    # Platform Detection
    # ========================================================================

    async def _get_local_adapter_address(self) -> Optional[str]:
        """Get local Bluetooth adapter MAC address via D-Bus."""
        if not HAS_DBUS:
            return None

        try:
            from bleak.backends.bluezdbus import defs

            bus = await MessageBus(bus_type=BusType.SYSTEM).connect()

            # Try specified adapter
            try:
                introspection = await bus.introspect('org.bluez', self.adapter_path)
                obj = bus.get_proxy_object('org.bluez', self.adapter_path, introspection)
                adapter = obj.get_interface(defs.ADAPTER_INTERFACE)
                properties_interface = obj.get_interface('org.freedesktop.DBus.Properties')
                address = await properties_interface.call_get(defs.ADAPTER_INTERFACE, 'Address')

                # Extract value from Variant
                if hasattr(address, 'value'):
                    address = address.value

                self._log(f"Local adapter address: {address}", "DEBUG")
                return address

            except Exception as e:
                self._log(f"Could not get adapter address via D-Bus: {e}", "DEBUG")
                return None

        except Exception as e:
            self._log(f"D-Bus adapter address retrieval failed: {e}", "DEBUG")
            return None

    def _detect_bluez_version(self):
        """Detect BlueZ version from bluetoothctl."""
        try:
            import subprocess
            result = subprocess.run(
                ['bluetoothctl', '--version'],
                capture_output=True,
                text=True,
                timeout=5
            )
            version_str = result.stdout.strip().split()[-1]
            self.bluez_version = tuple(map(int, version_str.split('.')))
            self._log(f"Detected BlueZ version {version_str}")
        except Exception as e:
            self._log(f"Could not detect BlueZ version: {e}", "DEBUG")
            self.bluez_version = None


# ============================================================================
# Bluezero GATT Server (Peripheral Mode)
# ============================================================================

class BluezeroGATTServer:
    """
    GATT server implementation using bluezero.

    This handles peripheral mode operations:
    - Creating GATT service and characteristics
    - Accepting connections from centrals
    - Receiving data via RX characteristic (centrals write to us)
    - Sending data via TX characteristic (we notify centrals)
    """

    def __init__(
        self,
        driver: LinuxBluetoothDriver,
        service_uuid: str,
        rx_char_uuid: str,
        tx_char_uuid: str,
        identity_char_uuid: str,
        adapter_index: int = 0,
        agent_capability: str = "NoInputNoOutput"
    ):
        """Initialize GATT server."""
        if not BLUEZERO_AVAILABLE:
            raise ImportError("bluezero library required for GATT server")

        self.driver = driver
        self.service_uuid = service_uuid
        self.rx_char_uuid = rx_char_uuid
        self.tx_char_uuid = tx_char_uuid
        self.identity_char_uuid = identity_char_uuid
        self.adapter_index = adapter_index
        self.agent_capability = agent_capability

        # bluezero objects
        self.peripheral_obj = None
        self.tx_characteristic = None
        self.identity_characteristic = None

        # State
        self.running = False

        # Identity
        self.identity_bytes: Optional[bytes] = None

        # BLE agent
        self.ble_agent = None

        # Thread
        self.server_thread: Optional[threading.Thread] = None
        self.stop_event = threading.Event()
        self.started_event = threading.Event()

        # Connected centrals (address -> info dict)
        self.connected_centrals: Dict[str, dict] = {}
        self.centrals_lock = threading.RLock()

    def _log(self, message: str, level: str = "INFO"):
        """Log message."""
        self.driver._log(f"GATTServer: {message}", level)

    def set_identity(self, identity_bytes: bytes):
        """Set the identity value for the Identity characteristic."""
        if len(identity_bytes) != 16:
            raise ValueError("Identity must be 16 bytes")

        self.identity_bytes = identity_bytes
        # Proactively update the characteristic value if it already exists
        if self.identity_characteristic:
            self.identity_characteristic.set_value(list(self.identity_bytes))

        self._log(f"Identity set: {identity_bytes.hex()}")

    def start(self, device_name: str):
        """Start GATT server and advertising."""
        if self.running:
            self._log("Server already running", "WARNING")
            return

        # Ensure identity is set before starting
        if not self.identity_bytes:
            raise RuntimeError("Identity must be set before starting GATT server. Call set_identity() first.")

        self._log(f"Starting GATT server with device name '{device_name}'...")

        # Reset events
        self.stop_event.clear()
        self.started_event.clear()

        # Start server thread
        self.server_thread = threading.Thread(
            target=self._run_server_thread,
            args=(device_name,),
            daemon=True,
            name="bluezero-gatt-server"
        )
        self.server_thread.start()

        # Wait for server to start
        started = self.started_event.wait(timeout=10.0)

        if not started or not self.running:
            raise RuntimeError("GATT server failed to start within timeout")

        self._log("GATT server started and advertising")

    def stop(self):
        """Stop GATT server and advertising."""
        if not self.running:
            return

        self._log("Stopping GATT server...")

        # Signal server thread to stop
        self.stop_event.set()
        self.running = False

        # Wait for thread to exit
        if self.server_thread and self.server_thread.is_alive():
            self.server_thread.join(timeout=5.0)

        # Unregister agent
        if self.ble_agent and HAS_BLE_AGENT:
            try:
                unregister_agent(self.ble_agent)
                self._log("BLE agent unregistered", "DEBUG")
            except Exception as e:
                self._log(f"Error unregistering agent: {e}", "DEBUG")
            self.ble_agent = None

        with self.centrals_lock:
            self.connected_centrals.clear()

        self._log("GATT server stopped")

    def _run_server_thread(self, device_name: str):
        """Run GATT server in separate thread."""
        try:
            self._log("Server thread starting...", "DEBUG")

            # Register BLE agent for automatic pairing
            if HAS_BLE_AGENT:
                try:
                    self.ble_agent = register_agent(self.agent_capability)
                    self._log(f"BLE agent registered with capability: {self.agent_capability}")
                except Exception as e:
                    self._log(f"Failed to register BLE agent: {e}", "WARNING")
                    self.ble_agent = None

            # Suppress bluezero logging
            logging.getLogger('bluezero').setLevel(logging.WARNING)
            logging.getLogger('bluezero.GATT').setLevel(logging.WARNING)
            logging.getLogger('bluezero.localGATT').setLevel(logging.WARNING)
            logging.getLogger('bluezero.adapter').setLevel(logging.WARNING)
            logging.getLogger('bluezero.peripheral').setLevel(logging.WARNING)

            # Get adapter
            adapters = adapter.list_adapters()
            if not adapters:
                self._log("No Bluetooth adapters found!", "ERROR")
                self.started_event.set()
                return

            if self.adapter_index >= len(adapters):
                self._log(f"Adapter index {self.adapter_index} out of range (only {len(adapters)} adapters)", "ERROR")
                self.started_event.set()
                return

            local_adapter = adapter.Adapter(adapters[self.adapter_index])
            adapter_address = local_adapter.address
            self._log(f"Using adapter: {adapter_address}", "DEBUG")

            # Create peripheral
            self.peripheral_obj = peripheral.Peripheral(
                adapter_address,
                local_name=device_name
            )

            # Add service
            self.peripheral_obj.add_service(
                srv_id=1,
                uuid=self.service_uuid,
                primary=True
            )
            self._log(f"Added service: {self.service_uuid}", "DEBUG")

            # Add RX characteristic (centrals write to us)
            self.peripheral_obj.add_characteristic(
                srv_id=1,
                chr_id=1,
                uuid=self.rx_char_uuid,
                value=[],
                notifying=False,
                flags=['write', 'write-without-response'],
                write_callback=self._handle_write_rx
            )
            self._log(f"Added RX characteristic: {self.rx_char_uuid}", "DEBUG")

            # Add TX characteristic (we notify centrals)
            self.peripheral_obj.add_characteristic(
                srv_id=1,
                chr_id=2,
                uuid=self.tx_char_uuid,
                value=[],
                notifying=True,
                flags=['read', 'notify']
            )
            self._log(f"Added TX characteristic: {self.tx_char_uuid}", "DEBUG")

            # Add Identity characteristic (centrals read our identity)
            self.peripheral_obj.add_characteristic(
                srv_id=1,
                chr_id=3,
                uuid=self.identity_char_uuid,
                value=[0]*16,  # Initialize with 16-byte placeholder
                notifying=False,
                flags=['read'],
                read_callback=self._handle_read_identity
            )
            self.identity_characteristic = self.peripheral_obj.characteristics[-1]
            self._log(f"Added Identity characteristic: {self.identity_char_uuid}", "DEBUG")

            # Set the identity value (guaranteed to be available by start() precondition)
            self.identity_characteristic.set_value(list(self.identity_bytes))
            self._log(f"Identity characteristic set to: {self.identity_bytes.hex()}")

            # Save TX characteristic reference
            if len(self.peripheral_obj.characteristics) >= 2:
                self.tx_characteristic = self.peripheral_obj.characteristics[1]  # chr_id=2
                self._log("Saved TX characteristic reference", "DEBUG")
            else:
                self._log(f"ERROR: TX characteristic not found!", "ERROR")
                self.started_event.set()
                return

            self._log("GATT server configured successfully")

            # Signal ready
            self.running = True
            self.started_event.set()

            # Publish (blocks until stopped)
            self._log("Publishing (blocking call)...", "DEBUG")
            self.peripheral_obj.publish()

        except Exception as e:
            self._log(f"Server thread error: {type(e).__name__}: {e}", "ERROR")
            import traceback
            traceback.print_exc()
            self.started_event.set()
        finally:
            self.running = False
            self._log("Server thread exiting", "DEBUG")

    def _handle_write_rx(self, value, options):
        """Handle write to RX characteristic (bluezero callback)."""
        # Convert to bytes
        if isinstance(value, list):
            data = bytes(value)
        elif isinstance(value, bytes):
            data = value
        else:
            data = bytes(value)

        # Extract central address and MTU
        central_address = options.get("device", "unknown")
        if central_address and central_address != "unknown":
            central_address = central_address.split("/")[-1].replace("_", ":")

        mtu = options.get("mtu", None)

        self._log(f"Received {len(data)} bytes from {central_address} (MTU: {mtu})", "DEBUG")

        # Track central connection
        with self.centrals_lock:
            if central_address not in self.connected_centrals:
                self._handle_central_connected(central_address, mtu)
            elif mtu is not None:
                # Update MTU
                old_mtu = self.connected_centrals[central_address].get("mtu", "unknown")
                if old_mtu != mtu:
                    self.connected_centrals[central_address]["mtu"] = mtu
                    self._log(f"Updated MTU for {central_address}: {old_mtu} -> {mtu}", "DEBUG")

                    # Notify callback
                    if self.driver.on_mtu_negotiated:
                        try:
                            self.driver.on_mtu_negotiated(central_address, mtu)
                        except Exception as e:
                            self._log(f"Error in MTU negotiated callback: {e}", "ERROR")

        # Pass data to driver callback
        if self.driver.on_data_received:
            try:
                self.driver.on_data_received(central_address, data)
            except Exception as e:
                self._log(f"Error in data received callback: {e}", "ERROR")

        return value  # bluezero expects value to be returned

    def _handle_read_identity(self, options):
        """Handle read of Identity characteristic (bluezero callback)."""
        central_address = options.get("device", "unknown")
        if central_address and central_address != "unknown":
            central_address = central_address.split("/")[-1].replace("_", ":")

        if self.identity_bytes is None:
            self._log(f"Identity read from {central_address}: not available", "WARNING")
            return []

        identity_list = list(self.identity_bytes)
        self._log(f"Identity read from {central_address}: {len(identity_list)} bytes", "DEBUG")
        return identity_list

    def _handle_central_connected(self, central_address: str, mtu: Optional[int]):
        """Handle new central connection."""
        if central_address in self.connected_centrals:
            self._log(f"Central {central_address} already connected", "WARNING")
            return

        effective_mtu = mtu if mtu is not None else 185

        self.connected_centrals[central_address] = {
            "address": central_address,
            "connected_at": time.time(),
            "mtu": effective_mtu
        }

        # Add to driver's peer list
        peer_conn = PeerConnection(
            address=central_address,
            client=None,  # No client for peripheral connections
            mtu=effective_mtu,
            connection_type="peripheral",
            connected_at=time.time()
        )

        with self.driver._peers_lock:
            self.driver._peers[central_address] = peer_conn

        self._log(f"Central connected: {central_address} (MTU: {effective_mtu})")

        # Notify callback (identity not available yet for peripheral connections)
        if self.driver.on_device_connected:
            try:
                self.driver.on_device_connected(central_address, None)
            except Exception as e:
                self._log(f"Error in device connected callback: {e}", "ERROR")

        # Notify MTU callback
        if self.driver.on_mtu_negotiated:
            try:
                self.driver.on_mtu_negotiated(central_address, effective_mtu)
            except Exception as e:
                self._log(f"Error in MTU negotiated callback: {e}", "ERROR")

    def send_notification(self, central_address: str, data: bytes):
        """Send notification to a connected central."""
        if not self.running or not self.tx_characteristic:
            raise RuntimeError("GATT server not running")

        with self.centrals_lock:
            if central_address not in self.connected_centrals:
                raise RuntimeError(f"Central {central_address} not connected")

        # Convert to list for bluezero
        if isinstance(data, bytes):
            value = list(data)
        else:
            value = data

        # Update characteristic value (bluezero automatically sends notification)
        self.tx_characteristic.set_value(value)

        self._log(f"Sent notification: {len(data)} bytes to {central_address}", "DEBUG")


# ============================================================================
# Module Exports
# ============================================================================

__all__ = [
    'LinuxBluetoothDriver',
    'apply_bluez_services_resolved_patch',
]