diff --git a/homeassistant/components/shelly/ble_manufacturer_data.py b/homeassistant/components/shelly/ble_manufacturer_data.py deleted file mode 100644 index ed572869792..00000000000 --- a/homeassistant/components/shelly/ble_manufacturer_data.py +++ /dev/null @@ -1,105 +0,0 @@ -"""Bluetooth support for Shelly integration.""" - -from __future__ import annotations - -import logging - -_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): - if offset + 1 > len(data): - break - - 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 not isinstance(flags, int): - return False - - return bool(flags & FLAG_RPC_OVER_BLE_ENABLED) diff --git a/homeassistant/components/shelly/config_flow.py b/homeassistant/components/shelly/config_flow.py index 5fe6c2432d2..41f6ed3290a 100644 --- a/homeassistant/components/shelly/config_flow.py +++ b/homeassistant/components/shelly/config_flow.py @@ -7,6 +7,8 @@ from collections.abc import AsyncIterator, Mapping from contextlib import asynccontextmanager from typing import Any, Final +from aioshelly.ble.manufacturer_data import has_rpc_over_ble +from aioshelly.ble.provisioning import async_provision_wifi, async_scan_wifi_networks from aioshelly.block_device import BlockDevice from aioshelly.common import ConnectionOptions, get_info from aioshelly.const import BLOCK_GENERATIONS, DEFAULT_HTTP_PORT, RPC_GENERATIONS @@ -19,6 +21,7 @@ from aioshelly.exceptions import ( RpcCallError, ) from aioshelly.rpc_device import RpcDevice +from aioshelly.zeroconf import async_lookup_device_by_name from bleak.backends.device import BLEDevice import voluptuous as vol @@ -46,7 +49,6 @@ from homeassistant.helpers.selector import ( ) from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo -from .ble_manufacturer_data import has_rpc_over_ble from .ble_provisioning import ( ProvisioningState, async_get_provisioning_registry, @@ -63,7 +65,6 @@ from .const import ( BLEScannerMode, ) from .coordinator import ShellyConfigEntry, async_reconnect_soon -from .provision_wifi import async_provision_wifi, async_scan_wifi_networks from .utils import ( get_block_device_sleep_period, get_coap_context, @@ -76,7 +77,6 @@ from .utils import ( get_ws_context, mac_address_from_name, ) -from .zeroconf_helpers import async_lookup_device_by_name CONFIG_SCHEMA: Final = vol.Schema( { diff --git a/homeassistant/components/shelly/provision_wifi.py b/homeassistant/components/shelly/provision_wifi.py deleted file mode 100644 index e866d96b62a..00000000000 --- a/homeassistant/components/shelly/provision_wifi.py +++ /dev/null @@ -1,77 +0,0 @@ -"""WiFi provisioning via BLE for Shelly devices.""" - -from __future__ import annotations - -from typing import Any, cast - -from aioshelly.common import ConnectionOptions -from aioshelly.rpc_device import RpcDevice -from bleak.backends.device import BLEDevice - - -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 - - """ - options = ConnectionOptions(ble_device=ble_device) - device = await RpcDevice.create( - aiohttp_session=None, - ws_context=None, - ip_or_options=options, - ) - - try: - await device.initialize() - # 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", [])) - finally: - await device.shutdown() - - -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 - - """ - options = ConnectionOptions(ble_device=ble_device) - device = await RpcDevice.create( - aiohttp_session=None, - ws_context=None, - ip_or_options=options, - ) - - try: - await device.initialize() - await device.call_rpc( - "WiFi.SetConfig", - { - "config": { - "sta": { - "ssid": ssid, - "pass": password, - "enable": True, - } - } - }, - ) - finally: - await device.shutdown() diff --git a/homeassistant/components/shelly/zeroconf_helpers.py b/homeassistant/components/shelly/zeroconf_helpers.py deleted file mode 100644 index 4aa47734051..00000000000 --- a/homeassistant/components/shelly/zeroconf_helpers.py +++ /dev/null @@ -1,42 +0,0 @@ -"""Zeroconf helper functions for Shelly integration.""" - -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 await service_info.async_request(aiozc.zeroconf, 5000): - addresses = service_info.parsed_addresses(IPVersion.V4Only) - if addresses and service_info.port: - host = addresses[0] - port = service_info.port - _LOGGER.debug("Found device via active lookup at %s:%s", host, port) - return (host, port) - _LOGGER.debug("Active lookup found service but no IPv4 addresses or port") - else: - _LOGGER.debug("Active lookup did not find service") - - return None