"""BLE Manager for Dangerous Pi. Handles Bluetooth Low Energy functionality including: - GATT server with full PM3/WiFi/System/Update services - Notifications for updates, backups, and battery alerts - Uses the Pi Zero 2 W built-in Bluetooth via bless library """ import asyncio import json import logging from dataclasses import dataclass from datetime import datetime from enum import Enum from typing import Optional, Dict, Any, List from .. import config # Try to import bless for GATT server try: from ..ble.bluez_adapter import BlueZGATTAdapter, get_ble_adapter BLESS_AVAILABLE = True except ImportError: BLESS_AVAILABLE = False # Fallback to dbus for basic operations try: import dbus import dbus.mainloop.glib from gi.repository import GLib DBUS_AVAILABLE = True except ImportError: DBUS_AVAILABLE = False logger = logging.getLogger(__name__) class NotificationType(str, Enum): """BLE notification type enum.""" UPDATE_AVAILABLE = "update_available" UPDATE_COMPLETE = "update_complete" BACKUP_COMPLETE = "backup_complete" BATTERY_WARNING = "battery_warning" BATTERY_CRITICAL = "battery_critical" SHUTDOWN_INITIATED = "shutdown_initiated" PM3_STATUS = "pm3_status" CONFIG_REQUIRED = "config_required" @dataclass class BLENotification: """BLE notification data.""" type: NotificationType message: str data: Dict[str, Any] timestamp: str class BLEManager: """Manages BLE functionality via Pi Zero 2 W Bluetooth. Provides both: - Full GATT server with PM3/WiFi/System/Update services (via bless) - Simple notifications for system events """ def __init__(self): """Initialize the BLE manager.""" self._enabled = config.BLE_ENABLED self._device_name = config.BLE_DEVICE_NAME self._is_available = False self._is_advertising = False self._gatt_server_running = False self._connected_devices: List[str] = [] self._notification_queue: asyncio.Queue = asyncio.Queue() self._service_uuid = "12345678-1234-5678-1234-56789abcdef0" # Custom service UUID self._char_uuid = "12345678-1234-5678-1234-56789abcdef1" # Notification characteristic self._bus = None self._adapter = None self._gatt_adapter: Optional[BlueZGATTAdapter] = None async def initialize(self) -> bool: """Initialize BLE adapter and check availability. Returns: True if initialization successful, False otherwise """ if not self._enabled: logger.info("BLE is disabled in configuration") return False # Check if Bluetooth adapter is available first has_adapter = await self._check_bluetooth_adapter() if not has_adapter: logger.warning("No Bluetooth adapter found") return False self._is_available = True # Try to initialize GATT server with bless (preferred) if BLESS_AVAILABLE: try: self._gatt_adapter = get_ble_adapter() self._gatt_adapter.device_name = self._device_name logger.info("BLE GATT adapter initialized (bless): %s", self._device_name) except Exception as e: logger.warning("Failed to initialize bless GATT adapter: %s", e) self._gatt_adapter = None else: logger.info("bless library not available, using basic BLE mode") logger.info("BLE initialized: %s", self._device_name) return True async def start_advertising(self): """Start BLE advertising and GATT server.""" if not self._is_available: logger.warning("BLE not available, cannot start advertising") return try: # Start full GATT server if available (preferred) if self._gatt_adapter: success = await self._gatt_adapter.start() if success: self._gatt_server_running = True self._is_advertising = True logger.info("BLE GATT server started: %s", self._device_name) return else: logger.warning("Failed to start GATT server, falling back to basic mode") # Fallback to basic advertising via bluetoothctl await self._set_device_name(self._device_name) await self._set_discoverable(True) self._is_advertising = True logger.info("BLE advertising started (basic mode): %s", self._device_name) except Exception as e: logger.error("Failed to start BLE advertising: %s", e) self._is_advertising = False async def stop_advertising(self): """Stop BLE advertising and GATT server.""" if not self._is_advertising: return try: # Stop GATT server if running if self._gatt_adapter and self._gatt_server_running: await self._gatt_adapter.stop() self._gatt_server_running = False logger.info("BLE GATT server stopped") else: # Stop basic advertising await self._set_discoverable(False) self._is_advertising = False logger.info("BLE advertising stopped") except Exception as e: logger.error("Failed to stop BLE advertising: %s", e) async def send_notification( self, notification_type: NotificationType, message: str, data: Optional[Dict[str, Any]] = None ): """Send a BLE notification to connected devices. Args: notification_type: Type of notification message: Human-readable message data: Optional additional data """ if not self._is_available: return notification = BLENotification( type=notification_type, message=message, data=data or {}, timestamp=datetime.utcnow().isoformat() ) await self._notification_queue.put(notification) # Process notification - always broadcast if GATT server is running if self._connected_devices or self._gatt_server_running: await self._broadcast_notification(notification) else: logger.debug("BLE notification queued (no devices connected): %s", message) async def get_status(self) -> Dict[str, Any]: """Get BLE manager status. Returns: Dictionary with BLE status information """ return { "enabled": self._enabled, "available": self._is_available, "advertising": self._is_advertising, "gatt_server_running": self._gatt_server_running, "gatt_available": BLESS_AVAILABLE, "connected_devices": len(self._connected_devices), "device_name": self._device_name, "queued_notifications": self._notification_queue.qsize() } async def _check_bluetooth_adapter(self) -> bool: """Check if Bluetooth adapter is available, unblocking and powering on if needed. Returns: True if adapter is available, False otherwise """ try: # Use bluetoothctl to check for adapter process = await asyncio.create_subprocess_exec( "bluetoothctl", "list", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) stdout, stderr = await process.communicate() if b"Controller" not in stdout: return False # Unblock Bluetooth if soft-blocked by rfkill await self._run_cmd("rfkill", "unblock", "bluetooth") # Power on the adapter via bluetoothctl out = await self._run_cmd("bluetoothctl", "power", "on") if b"succeeded" in out.lower() or b"yes" in out.lower(): logger.info("Bluetooth adapter powered on") else: logger.warning("Bluetooth power on response: %s", out.decode(errors='replace').strip()) return True except FileNotFoundError: logger.warning("bluetoothctl not found - BlueZ not installed") return False except Exception as e: logger.error("Error checking Bluetooth adapter: %s", e) return False async def _run_cmd(self, *args: str) -> bytes: """Run a command and return stdout.""" try: process = await asyncio.create_subprocess_exec( *args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) stdout, _ = await process.communicate() return stdout except FileNotFoundError: return b"" async def _set_device_name(self, name: str): """Set the Bluetooth device name. Args: name: Device name to set """ try: process = await asyncio.create_subprocess_exec( "bluetoothctl", "system-alias", name, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) await process.communicate() except Exception as e: logger.error("Failed to set device name: %s", e) async def _set_discoverable(self, enabled: bool): """Set Bluetooth discoverable state. Args: enabled: True to enable discoverable, False to disable """ try: command = "on" if enabled else "off" process = await asyncio.create_subprocess_exec( "bluetoothctl", "discoverable", command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) await process.communicate() except Exception as e: logger.error("Failed to set discoverable: %s", e) async def _broadcast_notification(self, notification: BLENotification): """Broadcast notification to all connected devices. Args: notification: Notification to broadcast """ # Convert notification to JSON payload = { "type": notification.type.value, "message": notification.message, "data": notification.data, "timestamp": notification.timestamp } payload_bytes = json.dumps(payload).encode('utf-8') # Use GATT server if available if self._gatt_adapter and self._gatt_server_running: try: # Send via system notification characteristic from ..ble.characteristics import SystemCharacteristicUUIDs await self._gatt_adapter.send_notification( SystemCharacteristicUUIDs.INFO, payload_bytes ) logger.debug("BLE notification sent via GATT: %s", notification.type.value) return except Exception as e: logger.warning("Failed to send GATT notification: %s", e) # Fallback: log the notification logger.info("BLE Notification: %s - %s", notification.type.value, notification.message) def is_available(self) -> bool: """Check if BLE is available. Returns: True if BLE is available, False otherwise """ return self._is_available def is_advertising(self) -> bool: """Check if BLE is currently advertising. Returns: True if advertising, False otherwise """ return self._is_advertising def get_connected_devices(self) -> List[str]: """Get list of connected device addresses. Returns: List of connected device MAC addresses """ return self._connected_devices.copy() # Global BLE manager instance _ble_manager: Optional[BLEManager] = None def get_ble_manager() -> BLEManager: """Get the global BLE manager instance.""" global _ble_manager if _ble_manager is None: _ble_manager = BLEManager() return _ble_manager