Add helpers for BLE WiFi scanning and provisioning (#976)

Author: bdraco

aioshelly/ble/manufacturer_data.py

@@ -0,0 +1,103 @@
+"""Shelly BLE manufacturer data parsing."""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING
+
+LOGGER = logging.getLogger(__name__)
+
+ALLTERCO_MFID = 0x0BA9
+
+# Block types in manufacturer data
+BLOCK_TYPE_FLAGS = 0x01
+BLOCK_TYPE_MAC = 0x0A
+BLOCK_TYPE_MODEL = 0x0B
+
+# Shelly bitfield flags (block type 0x01)
+FLAG_DISCOVERABLE = 1 << 0
+FLAG_AUTH_ENABLED = 1 << 1
+FLAG_RPC_OVER_BLE_ENABLED = 1 << 2
+FLAG_BUZZER_ENABLED = 1 << 3
+FLAG_IN_PAIRING_MODE = 1 << 4
+
+
+def parse_shelly_manufacturer_data(
+    manufacturer_data: dict[int, bytes],
+) -> dict[str, int | str] | None:
+    """Parse Shelly manufacturer data from BLE advertisement.
+
+    Args:
+        manufacturer_data: Manufacturer data from BLE advertisement
+
+    Returns:
+        Dict with parsed data (flags, mac, model) or None if invalid
+
+    """
+    if ALLTERCO_MFID not in manufacturer_data:
+        return None
+
+    data = manufacturer_data[ALLTERCO_MFID]
+    if len(data) < 1:
+        return None
+
+    result: dict[str, int | str] = {}
+    offset = 0
+
+    # Parse blocks
+    while offset < len(data):
+        block_type = data[offset]
+        offset += 1
+
+        if block_type == BLOCK_TYPE_FLAGS:
+            # 2 bytes of flags
+            if offset + 2 > len(data):
+                break
+            flags = int.from_bytes(data[offset : offset + 2], byteorder="little")
+            result["flags"] = flags
+            offset += 2
+
+        elif block_type == BLOCK_TYPE_MAC:
+            # 6 bytes MAC address
+            if offset + 6 > len(data):
+                break
+            mac_bytes = data[offset : offset + 6]
+            # Format as standard MAC address
+            result["mac"] = ":".join(f"{b:02X}" for b in mac_bytes)
+            offset += 6
+
+        elif block_type == BLOCK_TYPE_MODEL:
+            # 2 bytes model ID
+            if offset + 2 > len(data):
+                break
+            model_id = int.from_bytes(data[offset : offset + 2], byteorder="little")
+            result["model_id"] = model_id
+            offset += 2
+
+        else:
+            # Unknown block type - can't continue parsing
+            LOGGER.debug("Unknown block type in manufacturer data: 0x%02X", block_type)
+            break
+
+    return result if result else None
+
+
+def has_rpc_over_ble(manufacturer_data: dict[int, bytes]) -> bool:
+    """Check if device has RPC-over-BLE enabled.
+
+    Args:
+        manufacturer_data: Manufacturer data from BLE advertisement
+
+    Returns:
+        True if RPC-over-BLE is enabled
+
+    """
+    parsed = parse_shelly_manufacturer_data(manufacturer_data)
+    if not parsed or "flags" not in parsed:
+        return False
+
+    flags = parsed["flags"]
+    if TYPE_CHECKING:
+        assert isinstance(flags, int)
+
+    return bool(flags & FLAG_RPC_OVER_BLE_ENABLED)

aioshelly/ble/provisioning.py

@@ -0,0 +1,90 @@
+"""WiFi provisioning via BLE for Shelly devices."""
+
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+from typing import TYPE_CHECKING, Any, cast
+
+from ..common import ConnectionOptions
+from ..rpc_device import RpcDevice
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncIterator
+
+    from bleak.backends.device import BLEDevice
+
+
+@asynccontextmanager
+async def ble_rpc_device(ble_device: BLEDevice) -> AsyncIterator[RpcDevice]:
+    """Create and manage BLE RPC device connection.
+
+    Args:
+        ble_device: BLE device to connect to
+
+    Yields:
+        Initialized RPC device
+
+    Raises:
+        DeviceConnectionError: If connection to device fails
+
+    """
+    options = ConnectionOptions(ble_device=ble_device)
+    device = await RpcDevice.create(
+        aiohttp_session=None,
+        ws_context=None,
+        ip_or_options=options,
+    )
+
+    try:
+        await device.initialize()
+        yield device
+    finally:
+        await device.shutdown()
+
+
+async def async_scan_wifi_networks(ble_device: BLEDevice) -> list[dict[str, Any]]:
+    """Scan for WiFi networks via BLE.
+
+    Args:
+        ble_device: BLE device to connect to
+
+    Returns:
+        List of WiFi networks with ssid, rssi, auth fields
+
+    Raises:
+        DeviceConnectionError: If connection to device fails
+        RpcCallError: If RPC call fails
+
+    """
+    async with ble_rpc_device(ble_device) as device:
+        # WiFi scan can take up to 20 seconds - use 30s timeout to be safe
+        scan_result = await device.call_rpc("WiFi.Scan", timeout=30)
+        return cast(list[dict[str, Any]], scan_result.get("results", []))
+
+
+async def async_provision_wifi(ble_device: BLEDevice, ssid: str, password: str) -> None:
+    """Provision WiFi credentials to device via BLE.
+
+    Args:
+        ble_device: BLE device to connect to
+        ssid: WiFi network SSID
+        password: WiFi network password
+
+    Raises:
+        DeviceConnectionError: If connection to device fails
+        RpcCallError: If RPC call fails
+
+    """
+    async with ble_rpc_device(ble_device) as device:
+        await device.call_rpc(
+            "WiFi.SetConfig",
+            {
+                "config": {
+                    "sta": {
+                        "ssid": ssid,
+                        "pass": password,
+                        "enable": True,
+                    }
+                }
+            },
+        )

aioshelly/zeroconf.py

@@ -0,0 +1,43 @@
+"""Zeroconf helper functions for Shelly devices."""
+
+from __future__ import annotations
+
+import logging
+
+from zeroconf import IPVersion
+from zeroconf.asyncio import AsyncServiceInfo, AsyncZeroconf
+
+LOGGER = logging.getLogger(__name__)
+
+
+async def async_lookup_device_by_name(
+    aiozc: AsyncZeroconf, device_name: str
+) -> tuple[str, int] | None:
+    """Look up a Shelly device by name via zeroconf.
+
+    Args:
+        aiozc: AsyncZeroconf instance
+        device_name: Device name (e.g., "ShellyPlugUS-C049EF8873E8")
+
+    Returns:
+        Tuple of (host, port) if found, None otherwise
+
+    """
+    service_name = f"{device_name}._http._tcp.local."
+
+    LOGGER.debug("Active lookup for: %s", service_name)
+    service_info = AsyncServiceInfo("_http._tcp.local.", service_name)
+
+    if not await service_info.async_request(aiozc.zeroconf, 5000):
+        LOGGER.debug("Active lookup did not find service")
+        return None
+
+    addresses = service_info.parsed_addresses(IPVersion.V4Only)
+    if not addresses or not service_info.port:
+        LOGGER.debug("Active lookup found service but no IPv4 addresses or port")
+        return None
+
+    host = addresses[0]
+    port = service_info.port
+    LOGGER.debug("Found device via active lookup at %s:%s", host, port)
+    return (host, port)

pyproject.toml

@@ -28,6 +28,7 @@ dependencies = [
   "habluetooth>=3.42.0",
   "orjson>=3.8.1",
   "yarl",
+  "zeroconf>=0.148.0"
 ]
 
 [project.optional-dependencies]