diff --git a/custom_components/hunterdouglas_powerview_ble/__init__.py b/custom_components/hunterdouglas_powerview_ble/__init__.py index dd594f2..83e3e47 100644 --- a/custom_components/hunterdouglas_powerview_ble/__init__.py +++ b/custom_components/hunterdouglas_powerview_ble/__init__.py @@ -4,74 +4,283 @@ @license: Apache-2.0 license """ +import base64 +from collections.abc import Callable + +import aiohttp from bleak.backends.device import BLEDevice from bleak.exc import BleakError -from homeassistant.components.bluetooth import async_ble_device_from_address +from homeassistant.components import bluetooth +from homeassistant.components.bluetooth import ( + BluetoothCallbackMatcher, + BluetoothScanningMode, + BluetoothServiceInfoBleak, + async_ble_device_from_address, + async_discovered_service_info, +) from homeassistant.config_entries import ConfigEntry from homeassistant.const import Platform -from homeassistant.core import HomeAssistant -from homeassistant.exceptions import ConfigEntryError, ConfigEntryNotReady +from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers import device_registry as dr +from homeassistant.helpers.aiohttp_client import async_get_clientsession +from homeassistant.helpers.dispatcher import ( + async_dispatcher_connect, + async_dispatcher_send, +) +from homeassistant.helpers.entity_platform import AddEntitiesCallback -from .const import LOGGER +from .api import UUID_COV_SERVICE as UUID +from .const import ( + CONF_FRIENDLY_NAMES, + CONF_HUB_URL, + DOMAIN, + LOGGER, + MFCT_ID, + SIGNAL_NEW_SHADE, +) from .coordinator import PVCoordinator PLATFORMS: list[Platform] = [ Platform.BINARY_SENSOR, + Platform.BUTTON, Platform.COVER, + Platform.NUMBER, Platform.SENSOR, - Platform.BUTTON, ] -type ConfigEntryType = ConfigEntry[PVCoordinator] +type HubRuntimeData = dict[str, PVCoordinator] +type ConfigEntryType = ConfigEntry[HubRuntimeData] +type AddEntitiesFn = Callable[[PVCoordinator, AddEntitiesCallback], None] -async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntryType) -> bool: - """Set up BT Battery Management System from a config entry.""" - LOGGER.debug("Setup of %s", repr(entry)) - if entry.unique_id is None: - raise ConfigEntryError("Missing unique ID for device.") +def async_setup_shade_platform( + hass: HomeAssistant, + config_entry: ConfigEntryType, + async_add_entities: AddEntitiesCallback, + add_fn: AddEntitiesFn, +) -> None: + """Set up a platform for all current and future shades.""" + for coordinator in config_entry.runtime_data.values(): + add_fn(coordinator, async_add_entities) - ble_device: BLEDevice | None = async_ble_device_from_address( - hass=hass, address=entry.unique_id, connectable=True + @callback + def _async_new_shade(coordinator: PVCoordinator) -> None: + add_fn(coordinator, async_add_entities) + + config_entry.async_on_unload( + async_dispatcher_connect( + hass, + SIGNAL_NEW_SHADE.format(entry_id=config_entry.entry_id), + _async_new_shade, + ) ) + +async def _fetch_shade_names(hass: HomeAssistant, hub_url: str) -> dict[str, str]: + """Fetch the shades' friendly names from the hub, keyed by BLE advert name. + + Returns empty dict on failure. + """ + session = async_get_clientsession(hass) + timeout = aiohttp.ClientTimeout(total=10) + try: + async with session.get(f"{hub_url}/home/shades", timeout=timeout) as resp: + resp.raise_for_status() + shades = await resp.json(content_type=None) + except (TimeoutError, aiohttp.ClientError, ValueError): + return {} + + names: dict[str, str] = {} + for shade in shades or []: + ble_name = shade.get("bleName", "") + if not ble_name: + continue + name_b64 = shade.get("name", "") + try: + name = base64.b64decode(name_b64).decode("utf-8") if name_b64 else ble_name + except Exception: # noqa: BLE001 + name = ble_name + names[ble_name] = name + return names + + +def _persist_cache_entry( + hass: HomeAssistant, + entry: ConfigEntryType, + key: str, + address: str, + value: object, +) -> None: + """Read-modify-write a per-address value into a cache stored on entry.data. + + HA replaces entry.data atomically, so concurrent setup tasks each pull + the latest snapshot before merging — the no-op guard prevents writes + that would just notify listeners with unchanged data. + """ + cache = dict(entry.data.get(key, {})) + if cache.get(address) == value: + return + cache[address] = value + hass.config_entries.async_update_entry(entry, data={**entry.data, key: cache}) + + +def _resolve_friendly_name( + hass: HomeAssistant, + entry: ConfigEntryType, + service_info: BluetoothServiceInfoBleak, + hub_name: str | None, +) -> str: + """Resolve a shade's friendly name (Shelly-style) and refresh the cache. + + Hub data wins; otherwise fall back to the cached value from a prior + successful resolution; otherwise fall back to the BLE advert name. + """ + address = service_info.address + cached_names: dict[str, str] = entry.data.get(CONF_FRIENDLY_NAMES, {}) + if hub_name is not None: + friendly_name = hub_name + elif address in cached_names: + friendly_name = cached_names[address] + else: + friendly_name = service_info.name or address + + _persist_cache_entry(hass, entry, CONF_FRIENDLY_NAMES, address, friendly_name) + return friendly_name + + +async def _async_setup_shade( + hass: HomeAssistant, + entry: ConfigEntryType, + service_info: BluetoothServiceInfoBleak, + shade_names: dict[str, str], +) -> None: + """Create a coordinator for a newly discovered shade.""" + address = service_info.address + + if address in entry.runtime_data: + return + + ble_device: BLEDevice | None = async_ble_device_from_address( + hass=hass, address=address, connectable=True + ) if not ble_device: - raise ConfigEntryNotReady( - f"Could not find PowerView device ({entry.unique_id}) via Bluetooth" - ) + LOGGER.debug("BLE device %s not connectable, skipping", address) + return + + friendly_name = _resolve_friendly_name( + hass, entry, service_info, shade_names.get(service_info.name) + ) + + coordinator = PVCoordinator(hass, ble_device, entry.data.copy(), friendly_name) + + entry.runtime_data[address] = coordinator + entry.async_on_unload(coordinator.async_start()) - coordinator = PVCoordinator(hass, ble_device, entry.data.copy()) + # Populate dev_details before entity dispatch so the device registers with + # firmware/serial on first creation — the HA registry doesn't re-read + # DeviceInfo later. Failures are retried from the advertisement handler. try: await coordinator.query_dev_info() - except BleakError as err: - raise ConfigEntryNotReady("Unable to query device info.") from err + except (BleakError, TimeoutError): + LOGGER.debug( + "Initial device info query failed for %s (%s); will retry via adverts", + friendly_name, + address, + ) + + async_dispatcher_send( + hass, + SIGNAL_NEW_SHADE.format(entry_id=entry.entry_id), + coordinator, + ) + + +async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntryType) -> bool: + """Set up PowerView Home from a config entry.""" + LOGGER.debug("Setup of %s", repr(entry)) + + entry.runtime_data = {} - entry.runtime_data = coordinator + # Resolve shade friendly names from hub if available + hub_url = entry.data.get(CONF_HUB_URL, "") + shade_names: dict[str, str] = {} + if hub_url: + shade_names = await _fetch_shade_names(hass, hub_url) + + # Forward platforms first so dispatched entities have their setup ready await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) - entry.async_on_unload(coordinator.async_start()) - return True + # Kick off shade setup for already-discovered BLE devices (non-blocking) + for service_info in async_discovered_service_info(hass, connectable=True): + if ( + MFCT_ID in service_info.manufacturer_data + and UUID in service_info.service_uuids + ): + hass.async_create_task( + _async_setup_shade(hass, entry, service_info, shade_names) + ) -async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntryType) -> bool: - """Unload a config entry.""" - unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS) + # Register for future BLE discoveries + def _async_discovered_device( + service_info: BluetoothServiceInfoBleak, + change: bluetooth.BluetoothChange, + ) -> None: + if service_info.address not in entry.runtime_data: + hass.async_create_task( + _async_setup_shade(hass, entry, service_info, shade_names) + ) - LOGGER.debug("Unloaded config entry: %s, ok? %s!", entry.unique_id, str(unload_ok)) - return unload_ok + entry.async_on_unload( + bluetooth.async_register_callback( + hass, + _async_discovered_device, + BluetoothCallbackMatcher( + service_uuid=UUID, + manufacturer_id=MFCT_ID, + ), + BluetoothScanningMode.ACTIVE, + ) + ) + + return True -async def async_migrate_entry( - _hass: HomeAssistant, config_entry: ConfigEntryType +async def async_remove_config_entry_device( + hass: HomeAssistant, + entry: ConfigEntryType, + device_entry: dr.DeviceEntry, ) -> bool: - """Migrate old entry.""" + """Allow user-driven device removal; purge cached state for that address.""" + addresses = {ident[1] for ident in device_entry.identifiers if ident[0] == DOMAIN} + if not addresses: + return True + + new_data = dict(entry.data) + cache = dict(new_data.get(CONF_FRIENDLY_NAMES, {})) + for addr in addresses: + cache.pop(addr, None) + new_data[CONF_FRIENDLY_NAMES] = cache + hass.config_entries.async_update_entry(entry, data=new_data) - if config_entry.version > 1: - # This means the user has downgraded from a future version - LOGGER.debug("Cannot downgrade from version %s", config_entry.version) - return False + for addr in addresses: + coord = entry.runtime_data.pop(addr, None) + if coord is not None: + # _async_stop is a parent-class (DataUpdateCoordinator) convention; + # entry.async_on_unload only fires on full entry unload, so we + # invoke it directly here for per-device removal. + coord._async_stop() # noqa: SLF001 + + return True - LOGGER.debug("Migrating from version %s", config_entry.version) - return False +async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntryType) -> bool: + """Unload a config entry.""" + unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS) + + if unload_ok: + entry.runtime_data.clear() + + LOGGER.debug("Unloaded config entry: %s, ok? %s!", entry.unique_id, str(unload_ok)) + return unload_ok diff --git a/custom_components/hunterdouglas_powerview_ble/api.py b/custom_components/hunterdouglas_powerview_ble/api.py index 5a6cc97..703e1e0 100644 --- a/custom_components/hunterdouglas_powerview_ble/api.py +++ b/custom_components/hunterdouglas_powerview_ble/api.py @@ -4,7 +4,7 @@ from dataclasses import dataclass from enum import Enum import time -from typing import Final +from typing import Final, NamedTuple from bleak import BleakClient from bleak.backends.device import BLEDevice @@ -40,30 +40,111 @@ 6: "Duette", 10: "Duette and Applause SkyLift", 19: "Provenance Woven Wood", + 26: "Skyline Panel, Left Stack", + 27: "Skyline Panel, Right Stack", + 28: "Skyline Panel, Split Stack", 31: "Vignette", 32: "Vignette", 42: "M25T Roller Blind", 49: "AC Roller", 52: "Banded Shades", 53: "Sonnette", + 57: "Carole Roman Shades", + 69: "Curtain, Left Stack", + 70: "Curtain, Right Stack", + 71: "Curtain, Split Stack", 84: "Vignette", - # top down bottom up + # top down (single rail, inverted position) + 7: "Top Down", + # top down bottom up (dual rail) 8: "Duette, Top Down Bottom Up", 9: "Duette DuoLite, Top Down Bottom Up", 33: "Duette Architella, Top Down Bottom Up", - 39: "Parkland", 47: "Pleated, Top Down Bottom Up", - # top down, tilt anywhere + # tilt only (no position movement) + 39: "Parkland", + 40: "Everwood Alternative Wood Blinds", + 66: "Palm Beach Shutters", + # tilt on closed + 18: "Bottom Up, Tilt on Closed 90°", + 23: "Silhouette", + 43: "Facette", + 44: "Twist", + 72: "Silhouette", + # tilt anywhere (position + tilt) 51: "Venetian, Tilt Anywhere", + 54: "Vertical Slats, Left Stack", + 55: "Vertical Slats, Right Stack", + 56: "Vertical Slats, Split Stack", 62: "Venetian, Tilt Anywhere", + 103: "Designer Banded, Tilt Anywhere", + # duolite (dual overlapping fabrics) + 38: "Dual Overlapped, Tilt 90°", + 65: "Dual Overlapped", + 79: "Duolite Lift", + 95: "Dual Overlapped Illuminated", +} + + +class ShadeCapability(NamedTuple): + """Capability flags for a shade type.""" + + has_tilt: bool = False + tilt_only: bool = False + is_tilt_on_closed: bool = False # tilt only available when fully closed + is_top_down: bool = False # position logic is inverted (SkyLift style) + is_tdbu: bool = False # dual-rail Top Down Bottom Up (needs two entities) + is_duolite: bool = False # dual-fabric sheer+opaque (needs three entities) + + +SHADE_CAPABILITIES: Final[dict[int, ShadeCapability]] = { + # tilt anywhere (position + tilt) + 51: ShadeCapability(has_tilt=True), + 54: ShadeCapability(has_tilt=True), + 55: ShadeCapability(has_tilt=True), + 56: ShadeCapability(has_tilt=True), + 62: ShadeCapability(has_tilt=True), + 103: ShadeCapability(has_tilt=True), + # tilt only (no position movement) + 39: ShadeCapability(has_tilt=True, tilt_only=True), + 40: ShadeCapability(has_tilt=True, tilt_only=True), + 66: ShadeCapability(has_tilt=True, tilt_only=True), + # tilt on closed (tilt only available at fully closed position) + 18: ShadeCapability(has_tilt=True, is_tilt_on_closed=True), + 23: ShadeCapability(has_tilt=True, is_tilt_on_closed=True), + 43: ShadeCapability(has_tilt=True, is_tilt_on_closed=True), + 44: ShadeCapability(has_tilt=True, is_tilt_on_closed=True), + 72: ShadeCapability(has_tilt=True, is_tilt_on_closed=True), + # top-down only (single rail, inverted position) + 7: ShadeCapability(is_top_down=True), + 10: ShadeCapability(is_top_down=True), + # dual-rail top-down/bottom-up (two independent rails → two entities) + 8: ShadeCapability(is_tdbu=True), + 33: ShadeCapability(is_tdbu=True), + 47: ShadeCapability(is_tdbu=True), + # duolite (dual overlapping fabrics → three entities) + 9: ShadeCapability(is_tdbu=True, is_duolite=True), + 38: ShadeCapability(is_duolite=True), + 65: ShadeCapability(is_duolite=True), + 79: ShadeCapability(is_duolite=True), + 95: ShadeCapability(is_duolite=True), } +_DEFAULT_CAPABILITY: Final[ShadeCapability] = ShadeCapability() + + +def get_shade_capabilities(type_id: int | None) -> ShadeCapability: + """Return shade capabilities for a given type_id.""" + if type_id is None: + return _DEFAULT_CAPABILITY + return SHADE_CAPABILITIES.get(type_id, _DEFAULT_CAPABILITY) + + OPEN_POSITION: Final[int] = 100 CLOSED_POSITION: Final[int] = 0 POWER_LEVELS: Final[dict[int, int]] = { - 4: 100, # 4 is hardwired - 3: 100, # 3 = 100% to 51% power remaining + 3: 100, # 3 = 100% to 51% power remaining (also reported by hardwired) 2: 50, # 2 = 50% to 21% power remaining 1: 20, # 1 = 20% or less power remaining 0: 0, # 0 = No power remaining @@ -77,6 +158,7 @@ class ShadeCmd(Enum): STOP = 0xB8F7 ACTIVATE_SCENE = 0xBAF7 IDENTIFY = 0x11F7 + POWER_STATUS = 0xDEFF @dataclass @@ -97,18 +179,10 @@ class PowerViewBLE: def __init__(self, ble_device: BLEDevice, home_key: bytes = b"") -> None: """Initialize device API via Bluetooth.""" - self._ble_device: Final[BLEDevice] = ble_device + self._ble_device: BLEDevice = ble_device self.name: Final[str] = self._ble_device.name or "unknown" self._seqcnt: int = 1 - self._client: BleakClient = BleakClient( - self._ble_device, - disconnected_callback=self._on_disconnect, - services=[ - UUID_COV_SERVICE, - UUID_DEV_SERVICE, - # self.UUID_BAT_SERVICE, - ], - ) + self._client: BleakClient = BleakClient(self._ble_device) self._data_event = asyncio.Event() self._data: bytes = b"" self._info: PVDeviceInfo = PVDeviceInfo() @@ -125,6 +199,10 @@ async def _wait_event(self) -> None: await self._data_event.wait() self._data_event.clear() + def set_ble_device(self, ble_device: BLEDevice) -> None: + """Update the BLE device reference (e.g. when proxy details change).""" + self._ble_device = ble_device + @property def encrypted(self) -> bool: """Return whether communication with this shade is encrypted.""" @@ -134,6 +212,11 @@ def encrypted(self) -> bool: def encrypted(self, value: bool) -> None: self._is_encrypted = value + @property + def has_key(self) -> bool: + """Return True if a valid homekey was provided.""" + return self._cipher is not None + @property def info(self) -> PVDeviceInfo: """Return device information, e.g. SW version.""" @@ -144,6 +227,28 @@ def is_connected(self) -> bool: """Return whether remote device is connected.""" return self._client.is_connected + async def _transact(self, cmd: tuple[ShadeCmd, bytes]) -> int: + # Assumes _cmd_lock is held and _connect() has run. Writes a framed + # (optionally encrypted) request and waits for the device's reply; + # caller inspects/validates self._data afterwards. Returns the seq + # number used so callers can verify the echo. + tx_data: bytes = bytes( + int.to_bytes(cmd[0].value, 2, byteorder="little") + + bytes([self._seqcnt, len(cmd[1])]) + + cmd[1] + ) + LOGGER.debug("sending cmd: %s", tx_data.hex(" ")) + if self._cipher is not None and self._is_encrypted: + enc: AEADEncryptionContext = self._cipher.encryptor() + tx_data = enc.update(tx_data) + enc.finalize() + LOGGER.debug(" encrypted: %s", tx_data.hex(" ")) + self._data_event.clear() + await self._client.write_gatt_char(UUID_TX, tx_data, False) + seq = self._seqcnt + self._seqcnt += 1 + await asyncio.wait_for(self._wait_event(), timeout=TIMEOUT) + return seq + # general cmd: uint16_t cmd, uint8_t seqID, uint8_t data_len async def _cmd(self, cmd: tuple[ShadeCmd, bytes], disconnect: bool = True) -> None: self._cmd_next = cmd @@ -155,23 +260,9 @@ async def _cmd(self, cmd: tuple[ShadeCmd, bytes], disconnect: bool = True) -> No try: await self._connect() cmd_run: tuple[ShadeCmd, bytes] = self._cmd_next - tx_data: bytes = bytes( - int.to_bytes(cmd_run[0].value, 2, byteorder="little") - + bytes([self._seqcnt, len(cmd_run[1])]) - + cmd_run[1] - ) - LOGGER.debug("sending cmd: %s", tx_data.hex(" ")) - if self._cipher is not None and self._is_encrypted: - enc: AEADEncryptionContext = self._cipher.encryptor() - tx_data = enc.update(tx_data) + enc.finalize() - LOGGER.debug(" encrypted: %s", tx_data.hex(" ")) - self._data_event.clear() - await self._client.write_gatt_char(UUID_TX, tx_data, False) - self._seqcnt += 1 - LOGGER.debug("waiting for response") try: - await asyncio.wait_for(self._wait_event(), timeout=TIMEOUT) - self._verify_response(self._data, self._seqcnt - 1, cmd_run[0]) + seq = await self._transact(cmd_run) + self._verify_ack_reply(self._data, seq, cmd_run[0]) except TimeoutError as ex: raise TimeoutError("Device did not send confirmation.") from ex finally: @@ -181,28 +272,51 @@ async def _cmd(self, cmd: tuple[ShadeCmd, bytes], disconnect: bool = True) -> No LOGGER.error("Error: %s - %s", type(ex).__name__, ex) raise + async def _query(self, cmd: tuple[ShadeCmd, bytes]) -> bytes: + """Send a read-type opcode and return its payload bytes.""" + async with self._cmd_lock: + await self._connect() + try: + try: + seq = await self._transact(cmd) + except TimeoutError as ex: + raise TimeoutError("Device did not send response.") from ex + if not self._verify_header(self._data, seq, cmd[0]): + raise BleakError("Malformed query response header") + length = int(self._data[3]) + return bytes(self._data[4 : 4 + length]) + finally: + await self._client.disconnect() + @staticmethod - def dec_manufacturer_data(data: bytearray) -> list[tuple[str, float]]: + def dec_manufacturer_data(data: bytearray) -> dict[str, float | int | bool]: """Decode manufacturer data from BLE advertisement V2.""" if len(data) != 9: LOGGER.debug("not a V2 record!") - return [] - pos: Final[int] = int.from_bytes(data[3:5], byteorder="little") + return {} + # data[3] lower 2 bits are status flags; pos is in bits 2-7 of data[3] + # and bits 0-3 of data[4]. Read flags before extracting position so + # the masking below doesn't accidentally overwrite them. + flags: Final[int] = data[3] & 0x3 + # Mask pos2 bits (upper nibble of data[4]) out before forming the + # 10-bit position value, otherwise a non-zero top-rail position on + # TDBU shades contaminates the bottom-rail reading. + pos: Final[int] = ((data[4] & 0x0F) << 6) | ((data[3] >> 2) & 0x3F) pos2: Final[int] = (int(data[5]) << 4) + (int(data[4]) >> 4) - return [ - (ATTR_CURRENT_POSITION, ((pos >> 2) / 10)), - ("position2", pos2 >> 2), - ("position3", int(data[6])), - (ATTR_CURRENT_TILT_POSITION, int(data[7])), - ("home_id", int.from_bytes(data[0:2], byteorder="little")), - ("type_id", int(data[2])), - ("is_opening", bool(pos & 0x3 == 0x2)), - ("is_closing", bool(pos & 0x3 == 0x1)), - ("battery_charging", bool(pos & 0x3 == 0x3)), # observed - ("battery_level", POWER_LEVELS[(data[8] >> 6)]), # cannot hit 4 - ("resetMode", bool(data[8] & 0x1)), - ("resetClock", bool(data[8] & 0x2)), - ] + return { + ATTR_CURRENT_POSITION: pos / 10, + "position2": pos2 >> 2, + "position3": int(data[6]), + ATTR_CURRENT_TILT_POSITION: int(data[7]), + "home_id": int.from_bytes(data[0:2], byteorder="little"), + "type_id": int(data[2]), + "is_opening": bool(flags == 0x2), + "is_closing": bool(flags == 0x1), + "battery_charging": bool(flags == 0x3), # observed + "battery_level": POWER_LEVELS[(data[8] >> 6)], + "reset_mode": bool(data[8] & 0x1), + "reset_clock": bool(data[8] & 0x2), + } # position cmd: uint16_t pos1, uint16_t pos2, uint16_t pos3, uint16_t tilt, uint8_t velocity async def set_position( @@ -227,7 +341,7 @@ async def set_position( await self._cmd( ( ShadeCmd.SET_POSITION, - int.to_bytes(pos1*100, 2, byteorder="little") + int.to_bytes(pos1 * 100, 2, byteorder="little") + int.to_bytes(pos2, 2, byteorder="little") + int.to_bytes(pos3, 2, byteorder="little") + int.to_bytes(tilt, 2, byteorder="little") @@ -236,20 +350,20 @@ async def set_position( disconnect, ) - async def open(self) -> None: + async def open(self, velocity: int = 0x0) -> None: """Fully open cover.""" LOGGER.debug("%s open", self.name) - await self.set_position(OPEN_POSITION, disconnect=False) + await self.set_position(OPEN_POSITION, velocity=velocity, disconnect=False) async def stop(self) -> None: """Stop device movement.""" LOGGER.debug("%s stop", self.name) await self._cmd((ShadeCmd.STOP, b"")) - async def close(self) -> None: + async def close(self, velocity: int = 0x0) -> None: """Fully close cover.""" LOGGER.debug("%s close", self.name) - await self.set_position(CLOSED_POSITION, disconnect=False) + await self.set_position(CLOSED_POSITION, velocity=velocity, disconnect=False) # uint8_t scene#, uint8_t unknown # open: scene 2 @@ -269,8 +383,8 @@ async def identify(self, beeps: int = 0x3) -> None: LOGGER.debug("%s identify (%i)", self.name, beeps) await self._cmd((ShadeCmd.IDENTIFY, bytes([min(beeps, 0xFF)]))) - def _verify_response(self, data: bytes, seq_nr: int, cmd: ShadeCmd) -> bool: - """Verify shade response data.""" + def _verify_header(self, data: bytes, seq_nr: int, cmd: ShadeCmd) -> bool: + """Verify common header fields (length, echoed opcode, seq match).""" if len(data) < 4: LOGGER.error("Response message too short") return False @@ -282,6 +396,12 @@ def _verify_response(self, data: bytes, seq_nr: int, cmd: ShadeCmd) -> bool: "Response sequence id %i wrong, expected %d", int(data[2]), seq_nr ) return False + return True + + def _verify_ack_reply(self, data: bytes, seq_nr: int, cmd: ShadeCmd) -> bool: + """Verify an ack-only reply (1-byte status payload, 0 == success).""" + if not self._verify_header(data, seq_nr, cmd): + return False if int(data[3]) != 1: LOGGER.error("Wrong response data length") return False @@ -321,6 +441,17 @@ async def query_dev_info(self) -> dict[str, str]: LOGGER.debug("%s device data: %s", self.name, data) return data.copy() + async def query_power_status(self) -> bytes: + """Return the raw 0xFFDE power-status reply, uninterpreted. + + Deliberately returns bytes rather than a decoded power source. The + encoding is not established: byte 0 was previously read as a power-type + enum, but every sample behind that reading came from hardwired shades, + and acting on it misclassified battery shades as hardwired. Until a + confirmed battery-shade sample exists, this is reported, never acted on. + """ + return await self._query((ShadeCmd.POWER_STATUS, b"")) + def _on_disconnect(self, client: BleakClient) -> None: """Disconnect callback function.""" @@ -355,6 +486,7 @@ async def _connect(self) -> None: self._ble_device, self.name, disconnected_callback=self._on_disconnect, + ble_device_callback=lambda: self._ble_device, services=[ UUID_COV_SERVICE, UUID_DEV_SERVICE, diff --git a/custom_components/hunterdouglas_powerview_ble/binary_sensor.py b/custom_components/hunterdouglas_powerview_ble/binary_sensor.py index dda9127..b5d2553 100644 --- a/custom_components/hunterdouglas_powerview_ble/binary_sensor.py +++ b/custom_components/hunterdouglas_powerview_ble/binary_sensor.py @@ -8,12 +8,12 @@ from homeassistant.components.bluetooth.passive_update_coordinator import ( PassiveBluetoothCoordinatorEntity, ) -from homeassistant.const import ATTR_BATTERY_CHARGING +from homeassistant.const import ATTR_BATTERY_CHARGING, EntityCategory from homeassistant.core import HomeAssistant from homeassistant.helpers.device_registry import format_mac from homeassistant.helpers.entity_platform import AddEntitiesCallback -from . import ConfigEntryType +from . import ConfigEntryType, async_setup_shade_platform from .const import DOMAIN from .coordinator import PVCoordinator @@ -22,25 +22,51 @@ key=ATTR_BATTERY_CHARGING, translation_key=ATTR_BATTERY_CHARGING, device_class=BinarySensorDeviceClass.BATTERY_CHARGING, - ) + ), + # Diagnostic flags from byte 8 of the BLE advertisement. They surface when + # a battery wand has been removed (charging) and the shade has lost its + # RTC / scheduling state — see the "shade unresponsive after charging" + # known issue. Disabled by default to keep the device page clean for + # users who aren't debugging this scenario. + BinarySensorEntityDescription( + key="reset_clock", + translation_key="reset_clock", + device_class=BinarySensorDeviceClass.PROBLEM, + entity_registry_enabled_default=False, + entity_category=EntityCategory.DIAGNOSTIC, + ), + BinarySensorEntityDescription( + key="reset_mode", + translation_key="reset_mode", + device_class=BinarySensorDeviceClass.PROBLEM, + entity_registry_enabled_default=False, + entity_category=EntityCategory.DIAGNOSTIC, + ), ] +def _add_entities( + coordinator: PVCoordinator, async_add_entities: AddEntitiesCallback +) -> None: + """Create binary sensor entities for a single shade coordinator.""" + mac = format_mac(coordinator.address) + async_add_entities( + PVBinarySensor(coordinator, descr, mac) for descr in BINARY_SENSOR_TYPES + ) + + async def async_setup_entry( - _hass: HomeAssistant, + hass: HomeAssistant, config_entry: ConfigEntryType, async_add_entities: AddEntitiesCallback, ) -> None: """Add sensors for passed config_entry in Home Assistant.""" - - coord: PVCoordinator = config_entry.runtime_data - for descr in BINARY_SENSOR_TYPES: - async_add_entities( - [PVBinarySensor(coord, descr, format_mac(config_entry.unique_id))] - ) + async_setup_shade_platform(hass, config_entry, async_add_entities, _add_entities) -class PVBinarySensor(PassiveBluetoothCoordinatorEntity[PVCoordinator], BinarySensorEntity): # type: ignore[reportIncompatibleMethodOverride] +class PVBinarySensor( + PassiveBluetoothCoordinatorEntity[PVCoordinator], BinarySensorEntity +): # type: ignore[reportIncompatibleMethodOverride] """The generic PV binary sensor implementation.""" def __init__( @@ -59,4 +85,6 @@ def __init__( @property def is_on(self) -> bool | None: # type: ignore[reportIncompatibleVariableOverride] """Handle updated data from the coordinator.""" + if not self.coordinator.data_available: + return None return bool(self.coordinator.data.get(self.entity_description.key)) diff --git a/custom_components/hunterdouglas_powerview_ble/button.py b/custom_components/hunterdouglas_powerview_ble/button.py index bf7649b..b5ee2da 100644 --- a/custom_components/hunterdouglas_powerview_ble/button.py +++ b/custom_components/hunterdouglas_powerview_ble/button.py @@ -10,12 +10,12 @@ ButtonEntity, ButtonEntityDescription, ) -from homeassistant.config_entries import ConfigEntry from homeassistant.const import EntityCategory from homeassistant.core import HomeAssistant -from homeassistant.helpers.device_registry import DeviceInfo, format_mac +from homeassistant.helpers.device_registry import format_mac from homeassistant.helpers.entity_platform import AddEntitiesCallback +from . import ConfigEntryType, async_setup_shade_platform from .const import DOMAIN, LOGGER from .coordinator import PVCoordinator @@ -28,23 +28,26 @@ ] +def _add_entities( + coordinator: PVCoordinator, async_add_entities: AddEntitiesCallback +) -> None: + """Create button entities for a single shade coordinator.""" + async_add_entities([PowerViewButton(coordinator, descr) for descr in BUTTONS_SHADE]) + + async def async_setup_entry( - _hass: HomeAssistant, - config_entry: ConfigEntry, + hass: HomeAssistant, + config_entry: ConfigEntryType, async_add_entities: AddEntitiesCallback, ) -> None: - """Set up the demo cover platform.""" - - coordinator: PVCoordinator = config_entry.runtime_data - for descr in BUTTONS_SHADE: - async_add_entities([PowerViewButton(coordinator, descr)]) + """Set up the button platform.""" + async_setup_shade_platform(hass, config_entry, async_add_entities, _add_entities) class PowerViewButton(PassiveBluetoothCoordinatorEntity[PVCoordinator], ButtonEntity): # type: ignore[reportIncompatibleVariableOverride] """Representation of a powerview shade.""" _attr_has_entity_name = True - _attr_device_class = ButtonDeviceClass.IDENTIFY def __init__( self, @@ -60,11 +63,6 @@ def __init__( ) super().__init__(coordinator) - @property - def device_info(self) -> DeviceInfo: # type: ignore[reportIncompatibleVariableOverride] - """Return the device_info of the device.""" - return self._coord.device_info - async def async_press(self) -> None: """Handle the button press.""" LOGGER.debug("identify cover") diff --git a/custom_components/hunterdouglas_powerview_ble/config_flow.py b/custom_components/hunterdouglas_powerview_ble/config_flow.py index 7a38b35..bc70473 100644 --- a/custom_components/hunterdouglas_powerview_ble/config_flow.py +++ b/custom_components/hunterdouglas_powerview_ble/config_flow.py @@ -1,45 +1,316 @@ -"""Config flow for BLE Battery Management System integration.""" +"""Config flow for Hunter Douglas PowerView BLE integration.""" -from dataclasses import dataclass +import asyncio +import hashlib +import struct from typing import Any +import aiohttp import voluptuous as vol from homeassistant import config_entries -from homeassistant.components.bluetooth import ( - BluetoothServiceInfoBleak, - async_discovered_service_info, -) +from homeassistant.components.bluetooth import BluetoothServiceInfoBleak from homeassistant.config_entries import ConfigFlowResult -from homeassistant.const import CONF_ADDRESS +from homeassistant.core import HomeAssistant +from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.selector import ( SelectOptionDict, SelectSelector, SelectSelectorConfig, + TextSelector, + TextSelectorConfig, + TextSelectorType, ) +from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo -from .api import UUID_COV_SERVICE as UUID -from .const import DOMAIN, LOGGER, MFCT_ID +from .const import CONF_HOME_KEY, CONF_HUB_URL, DOMAIN, LOGGER, MFCT_ID +_DEFAULT_HUB_URL = "http://powerview-g3.local" -class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): - """Handle a config flow for BT Battery Management System.""" - VERSION = 1 - MINOR_VERSION = 0 +def _hub_unique_id(home_key: str) -> str: + """Derive a stable unique ID for a hub entry from the home key.""" + if home_key: + digest = hashlib.sha256(home_key.encode()).hexdigest()[:16] + return f"pvhome_{digest}" + return "pvhome_unencrypted" + + +def _parse_key_response(ble_name: str, result: dict) -> bytes | None: # noqa: PLR0911 + """Parse a shade exec response and return the 16-byte key, or None.""" + if result.get("err"): + err_msg = (result.get("responses") or [{}])[0].get("errMsg", "unknown") + LOGGER.warning( + "Shade %s: hub BLE command failed (err=%s: %s)", + ble_name, + result["err"], + err_msg, + ) + return None + + responses = result.get("responses", []) + if len(responses) != 1 or "hex" not in responses[0]: + LOGGER.warning( + "Shade %s returned unexpected response structure: %s", + ble_name, + result, + ) + return None + + response_bytes = bytes.fromhex(responses[0]["hex"]) + if len(response_bytes) < 5: + LOGGER.warning( + "Shade %s response too short (%d bytes)", ble_name, len(response_bytes) + ) + return None + _s, _c, _q, length = struct.unpack(" bytes: + """Fetch 16-byte homekey from a PowerView G3 hub. + + Tries each shade on the hub until one returns a valid key. + The key is network-wide so any reachable shade returns the same value. + + The hub must establish a BLE connection to each shade before it can proxy + the key request. On the first pass that connection is often not yet open, + so the hub returns "command timed out" (err=8). We retry the whole shade + list up to _KEY_FETCH_PASSES times, pausing between passes to let the hub + complete the BLE connections the earlier requests kicked off. + + Raises ValueError on protocol/key errors. + Raises aiohttp.ClientError on network errors. + Raises asyncio.TimeoutError on timeout. + """ + session = async_get_clientsession(hass) + timeout = aiohttp.ClientTimeout(total=10) + + async with session.get(f"{hub_url}/home/shades", timeout=timeout) as resp: + resp.raise_for_status() + shades = await resp.json(content_type=None) + + if not shades: + raise ValueError("No shades found on the hub") - @dataclass - class DiscoveredDevice: - """A discovered bluetooth device.""" + # Sort by signal strength (strongest first) — a stronger signal means the + # hub is more likely to have an active BLE connection to that shade. + shades.sort(key=lambda s: s.get("signalStrength", -100), reverse=True) + ble_names = [s["bleName"] for s in shades if s.get("bleName")] + if not ble_names: + raise ValueError("No BLE-capable shades found on the hub") - name: str - discovery_info: BluetoothServiceInfoBleak + # GetShadeKey BLE request: sid=251, cid=18, seqId=1, data_len=0 + request_frame = struct.pack(" vol.Schema: + """Build the homekey form schema, defaulting the hub URL field. + + When a gateway has been discovered via zeroconf, ``hub_url_default`` is the + discovered URL so the manual-entry fallback comes pre-filled. + """ + return vol.Schema( + { + vol.Required("key_method", default="hub"): SelectSelector( + SelectSelectorConfig( + options=[ + SelectOptionDict( + value="hub", + label="Fetch automatically from PowerView hub", + ), + SelectOptionDict( + value="manual", + label="Enter key manually (32 hex characters)", + ), + SelectOptionDict( + value="skip", + label=( + "Skip (no key — controls disabled for " + "encrypted shades)" + ), + ), + ] + ) + ), + vol.Optional(CONF_HUB_URL, default=hub_url_default): TextSelector( + TextSelectorConfig(type=TextSelectorType.URL) + ), + vol.Optional("home_key", default=""): TextSelector( + TextSelectorConfig(type=TextSelectorType.TEXT) + ), + } + ) + + +class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): + """Handle a config flow for Hunter Douglas PowerView BLE.""" + + VERSION = 2 + MINOR_VERSION = 1 def __init__(self) -> None: """Initialize the config flow.""" + self._home_key: str = "" + self._hub_url: str = "" + + def _existing_hub_entry(self) -> config_entries.ConfigEntry | None: + """Return the existing hub (v2+) config entry, if any.""" + for entry in self._async_current_entries(): + if entry.version >= 2: + return entry + return None - self._discovered_device: ConfigFlow.DiscoveredDevice | None = None - self._discovered_devices: dict[str, ConfigFlow.DiscoveredDevice] = {} + async def _create_entry(self) -> ConfigFlowResult: + """Set the stable unique ID and create the hub config entry.""" + await self.async_set_unique_id( + _hub_unique_id(self._home_key), raise_on_progress=False + ) + self._abort_if_unique_id_configured() + data: dict[str, str] = {CONF_HOME_KEY: self._home_key} + if self._hub_url: + data[CONF_HUB_URL] = self._hub_url + return self.async_create_entry(title="PowerView Home", data=data) + + def _show_homekey_form( + self, step_id: str, errors: dict[str, str], **placeholders: str + ) -> ConfigFlowResult: + """Render the key-source form, pre-filling the hub URL when discovered.""" + return self.async_show_form( + step_id=step_id, + data_schema=_homekey_schema(self._hub_url or _DEFAULT_HUB_URL), + errors=errors, + description_placeholders={ + "hub_url_example": _DEFAULT_HUB_URL, + **placeholders, + }, + ) + + def _validate_manual_key( + self, user_input: dict[str, Any], errors: dict[str, str] + ) -> bool: + """Validate a manually entered hex key. + + Returns True on success, False on validation error. + """ + raw = user_input.get("home_key", "").strip() + if "\\x" in raw: + raw = raw.replace("\\x", "") + if len(raw) != 32: + errors["home_key"] = "invalid_key_length" + return False + try: + bytes.fromhex(raw) + except ValueError: + errors["home_key"] = "invalid_key_format" + return False + self._home_key = raw.lower() + return True + + async def _validate_homekey_input( + self, user_input: dict[str, Any], errors: dict[str, str] + ) -> bool: + """Parse and validate homekey user_input, populating self state. + + Returns True on success, False on validation error (errors dict is populated). + """ + method = user_input.get("key_method", "skip") + + if method == "skip": + self._home_key = "" + return True + + if method == "manual": + # Still capture a hub URL if the user provided one — the integration + # uses it to fetch the shades' friendly names even when the homekey + # itself was supplied manually. + hub_url = user_input.get(CONF_HUB_URL, "").rstrip("/") + if hub_url: + self._hub_url = hub_url + return self._validate_manual_key(user_input, errors) + + if method != "hub": + return False + + hub_url = user_input.get(CONF_HUB_URL, "").rstrip("/") + try: + key = await _fetch_key_from_hub(self.hass, hub_url) + except tuple(_HUB_ERROR_MAP) as ex: + LOGGER.warning("Hub key fetch failed: %s", ex) + errors[CONF_HUB_URL] = _HUB_ERROR_MAP[type(ex)] + return False + + self._home_key = key.hex() + self._hub_url = hub_url + return True async def async_step_bluetooth( self, discovery_info: BluetoothServiceInfoBleak @@ -47,92 +318,92 @@ async def async_step_bluetooth( """Handle a flow initialized by Bluetooth discovery.""" LOGGER.debug("Bluetooth device detected: %s", discovery_info) - await self.async_set_unique_id(discovery_info.address) + # Derive a home-wide unique ID from the home_id embedded in the BLE + # advertisement (bytes 0-1 of the manufacturer payload). All shades on + # the same network share the same home_id, so HA deduplicates every + # subsequent shade discovery into this single flow via + # "already_in_progress" rather than spawning one notification per shade. + mfr_data = bytearray(discovery_info.manufacturer_data.get(MFCT_ID, b"")) + if len(mfr_data) >= 2: + home_id = int.from_bytes(mfr_data[0:2], byteorder="little") + unique_id = f"pvhome_{home_id}" + else: + unique_id = DOMAIN + + await self.async_set_unique_id(unique_id) self._abort_if_unique_id_configured() - self._discovered_device = ConfigFlow.DiscoveredDevice( - discovery_info.name, discovery_info - ) - self.context["title_placeholders"] = {"name": self._discovered_device.name} - return await self.async_step_bluetooth_confirm() + # If a hub entry already exists (unique_id may differ), shades are + # auto-discovered internally — nothing more for the user to do. + if self._existing_hub_entry(): + return self.async_abort(reason="already_configured") - async def async_step_bluetooth_confirm( - self, user_input: dict[str, Any] | None = None + # No hub entry yet — redirect to user setup + return await self.async_step_user() + + async def async_step_zeroconf( + self, discovery_info: ZeroconfServiceInfo ) -> ConfigFlowResult: - """Confirm bluetooth device discovery.""" - assert self._discovered_device is not None - LOGGER.debug("confirm step for %s", self._discovered_device.name) - - if user_input is not None: - return self.async_create_entry( - title=self._discovered_device.name, - data={ - "manufacturer_data": self._discovered_device.discovery_info.manufacturer_data[ - MFCT_ID - ].hex() - }, - ) + """Handle a flow initialized by zeroconf discovery of a G3 gateway.""" + LOGGER.debug("Zeroconf gateway detected: %s", discovery_info) - self._set_confirm_only() + host = discovery_info.hostname.rstrip(".") or str(discovery_info.ip_address) + self._hub_url = f"http://{host}" - return self.async_show_form( - step_id="bluetooth_confirm", - description_placeholders={"name": self._discovered_device.name}, - ) + # Dedup repeated mDNS announcements while a flow is in progress. + await self.async_set_unique_id(f"gateway_{host}") - async def async_step_user( + # If the single hub entry already exists, just record the (possibly + # changed) gateway URL so metadata/key fetches keep working after a + # DHCP/hostname change, then abort — there is nothing for the user to do. + entry = self._existing_hub_entry() + if entry: + if entry.data.get(CONF_HUB_URL) != self._hub_url: + self.hass.config_entries.async_update_entry( + entry, + data={**entry.data, CONF_HUB_URL: self._hub_url}, + ) + self.hass.config_entries.async_schedule_reload(entry.entry_id) + return self.async_abort(reason="already_configured") + + self.context["title_placeholders"] = {"name": host} + return await self.async_step_zeroconf_confirm() + + async def async_step_zeroconf_confirm( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: - """Handle the user step to pick discovered device.""" - LOGGER.debug("user step") + """Confirm a discovered gateway and choose how to obtain the key. - if user_input is not None: - address = user_input[CONF_ADDRESS] - await self.async_set_unique_id(address, raise_on_progress=False) - self._abort_if_unique_id_configured() - self._discovered_device = self._discovered_devices[address] - - self.context["title_placeholders"] = {"name": self._discovered_device.name} - - return self.async_create_entry( - title=self._discovered_device.name, - data={ - "manufacturer_data": self._discovered_device.discovery_info.manufacturer_data[ - MFCT_ID - ].hex() - }, - ) + Offers the same options as the manual user step (fetch from the hub, + enter the key manually, or skip) with the discovered URL pre-filled, so + the user can bypass the automatic hub fetch if they prefer. + """ + errors: dict[str, str] = {} - current_addresses = self._async_current_ids() - for discovery_info in async_discovered_service_info(self.hass, False): - address = discovery_info.address - if address in current_addresses or address in self._discovered_devices: - continue + if user_input is not None and await self._validate_homekey_input( + user_input, errors + ): + return await self._create_entry() - if MFCT_ID not in discovery_info.manufacturer_data: - continue + return self._show_homekey_form( + "zeroconf_confirm", errors, name=self._hub_url + ) - if UUID not in discovery_info.service_uuids: - continue + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Handle the user step — create a hub entry.""" + LOGGER.debug("user step") - self._discovered_devices[address] = ConfigFlow.DiscoveredDevice( - discovery_info.name, discovery_info - ) + # Only one hub entry allowed (per key, but for simplicity one total) + if self._existing_hub_entry(): + return self.async_abort(reason="single_instance_allowed") - if not self._discovered_devices: - return self.async_abort(reason="no_devices_found") + errors: dict[str, str] = {} - titles: list[SelectOptionDict] = [] - for address, discovery in self._discovered_devices.items(): - titles.append({"value": address, "label": discovery.name}) + if user_input is not None and await self._validate_homekey_input( + user_input, errors + ): + return await self._create_entry() - return self.async_show_form( - step_id="user", - data_schema=vol.Schema( - { - vol.Required(CONF_ADDRESS): SelectSelector( - SelectSelectorConfig(options=titles) - ) - } - ), - ) + return self._show_homekey_form("user", errors) diff --git a/custom_components/hunterdouglas_powerview_ble/const.py b/custom_components/hunterdouglas_powerview_ble/const.py index d723594..b01ffe6 100644 --- a/custom_components/hunterdouglas_powerview_ble/const.py +++ b/custom_components/hunterdouglas_powerview_ble/const.py @@ -8,10 +8,12 @@ MFCT_ID: Final[int] = 2073 TIMEOUT: Final[int] = 5 -# put the key here, needs to be 16 bytes long, e.g. -# HOME_KEY: Final[bytes] = b"\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f" -HOME_KEY: Final[bytes] = b"" +CONF_HOME_KEY: Final[str] = "home_key" +CONF_HUB_URL: Final[str] = "hub_url" +CONF_FRIENDLY_NAMES: Final[str] = "friendly_names" +# dispatcher signal for newly discovered shades (format with entry_id) +SIGNAL_NEW_SHADE: Final[str] = f"{DOMAIN}_new_shade_{{entry_id}}" # attributes (do not change) ATTR_RSSI: Final[str] = "rssi" diff --git a/custom_components/hunterdouglas_powerview_ble/coordinator.py b/custom_components/hunterdouglas_powerview_ble/coordinator.py index 68883cc..0d82994 100644 --- a/custom_components/hunterdouglas_powerview_ble/coordinator.py +++ b/custom_components/hunterdouglas_powerview_ble/coordinator.py @@ -1,8 +1,11 @@ """Home Assistant coordinator for Hunter Douglas PowerView (BLE) integration.""" -from typing import Any +import asyncio +import time +from typing import Any, Final from bleak.backends.device import BLEDevice +from bleak.exc import BleakError from homeassistant.components import bluetooth from homeassistant.components.bluetooth.const import DOMAIN as BLUETOOTH_DOMAIN @@ -10,29 +13,52 @@ PassiveBluetoothDataUpdateCoordinator, ) from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers import device_registry as dr from homeassistant.helpers.device_registry import CONNECTION_BLUETOOTH, DeviceInfo -from .api import SHADE_TYPE, PowerViewBLE -from .const import ATTR_RSSI, DOMAIN, HOME_KEY, LOGGER +from .api import SHADE_TYPE, PowerViewBLE, ShadeCapability, get_shade_capabilities +from .const import ATTR_RSSI, CONF_HOME_KEY, DOMAIN, LOGGER class PVCoordinator(PassiveBluetoothDataUpdateCoordinator): """Update coordinator for a battery management system.""" + # Firmware isn't in the advert — pull it via GATT. Retry every minute + # while empty, then re-check daily so firmware upgrades eventually surface. + _DEV_INFO_REFRESH_S: Final[float] = 24 * 3600 + _DEV_INFO_RETRY_S: Final[float] = 60 + + # Position/battery come only from V2 adverts. Past this many seconds without + # a fresh V2 record we stop reporting the retained sample as current, so an + # out-of-range shade doesn't keep showing a stale position. + _STALE_AFTER_S: Final[float] = 300.0 + def __init__( - self, hass: HomeAssistant, ble_device: BLEDevice, data: dict[str, Any] + self, + hass: HomeAssistant, + ble_device: BLEDevice, + data: dict[str, Any], + friendly_name: str | None = None, ) -> None: """Initialize BMS data coordinator.""" assert ble_device.name is not None - self._mac = ble_device.address - self.api = PowerViewBLE(ble_device, HOME_KEY) + self._friendly_name = friendly_name or ble_device.name + home_key_hex: str = data.get(CONF_HOME_KEY, "") + home_key: bytes = ( + bytes.fromhex(home_key_hex) if len(home_key_hex) == 32 else b"" + ) + self.api = PowerViewBLE(ble_device, home_key) self.data: dict[str, int | float | bool] = {} + self._last_v2_at: float = 0.0 self._manuf_dat = data.get("manufacturer_data") self.dev_details: dict[str, str] = {} + self.velocity: int = 0 + self._last_dev_info_at: float = 0.0 + self._dev_info_task: asyncio.Task[None] | None = None LOGGER.debug( "Initializing coordinator for %s (%s)", - ble_device.name, + self._friendly_name, ble_device.address, ) super().__init__( @@ -42,33 +68,103 @@ def __init__( bluetooth.BluetoothScanningMode.ACTIVE, ) + @property + def friendly_name(self) -> str: + """Return the shade's resolved friendly name.""" + return self._friendly_name + + @property + def type_id(self) -> int | None: + """Return the shade type ID from manufacturer data or live BLE data.""" + if self._manuf_dat: + return int(bytes.fromhex(self._manuf_dat)[2]) + live = self.data.get("type_id") + return int(live) if live is not None else None + + @property + def shade_capabilities(self) -> ShadeCapability: + """Return the shade capabilities based on type ID.""" + return get_shade_capabilities(self.type_id) + + @property + def data_available(self) -> bool: + """Whether the last V2 advertisement is recent enough to trust. + + Position, tilt and battery are only carried in V2 adverts. An + out-of-range shade stops sending them while HA may still consider the + device present, so past ``_STALE_AFTER_S`` we no longer report the + retained sample as current. + """ + return ( + self._last_v2_at != 0.0 + and time.monotonic() - self._last_v2_at < self._STALE_AFTER_S + ) + async def query_dev_info(self) -> None: - """Receive detailed information from device.""" + """Fetch device info over GATT and push into the device registry. + + The HA device registry does not re-read DeviceInfo when the underlying + data changes, so we update it explicitly when new details arrive. + """ LOGGER.debug("%s: querying device info", self.name) - self.dev_details.update(await self.api.query_dev_info()) + self._last_dev_info_at = time.monotonic() + new = await self.api.query_dev_info() + if new and new != self.dev_details: + self.dev_details.update(new) + self._push_device_registry_update() + + def _push_device_registry_update(self) -> None: + reg = dr.async_get(self.hass) + device = reg.async_get_device(identifiers={(DOMAIN, self.address)}) + if device is None: + return + reg.async_update_device( + device.id, + sw_version=self.dev_details.get("sw_rev"), + hw_version=self.dev_details.get("hw_rev"), + serial_number=self.dev_details.get("serial_nr"), + ) + + async def _refresh_dev_info_safe(self) -> None: + try: + await self.query_dev_info() + except (BleakError, TimeoutError) as ex: + LOGGER.debug("%s: dev_info refresh failed: %s", self.name, ex) + + def _maybe_refresh_dev_info(self) -> None: + """Schedule a background dev_info refresh if stale.""" + if self._dev_info_task is not None and not self._dev_info_task.done(): + return + interval = ( + self._DEV_INFO_REFRESH_S + if self.dev_details.get("sw_rev") + else self._DEV_INFO_RETRY_S + ) + if time.monotonic() - self._last_dev_info_at < interval: + return + self._dev_info_task = self.hass.async_create_background_task( + self._refresh_dev_info_safe(), name=f"pvble_dev_info_{self.address}" + ) @property def device_info(self) -> DeviceInfo: """Return detailed device information for GUI.""" - LOGGER.debug("%s: device_info, %s", self.name, self.dev_details) + LOGGER.debug("%s: device_info, %s", self._friendly_name, self.dev_details) return DeviceInfo( identifiers={ - (DOMAIN, self.name), + (DOMAIN, self.address), (BLUETOOTH_DOMAIN, self.address), }, connections={(CONNECTION_BLUETOOTH, self.address)}, - name=self.name, + name=self._friendly_name, configuration_url=None, - # properties used in GUI: manufacturer="Hunter Douglas", model=( - str(SHADE_TYPE.get(int(bytes.fromhex(self._manuf_dat)[2]), "unknown")) - if self._manuf_dat + str(SHADE_TYPE.get(self.type_id, "unknown")) + if self.type_id is not None else None ), - model_id=( - str(bytes.fromhex(self._manuf_dat)[2]) if self._manuf_dat else None - ), + model_id=(str(self.type_id) if self.type_id is not None else None), serial_number=self.dev_details.get("serial_nr"), sw_version=self.dev_details.get("sw_rev"), hw_version=self.dev_details.get("hw_rev"), @@ -77,11 +173,15 @@ def device_info(self) -> DeviceInfo: @property def device_present(self) -> bool: """Check if a device is present.""" - return bluetooth.async_address_present(self.hass, self._mac, connectable=True) + return bluetooth.async_address_present( + self.hass, self.address, connectable=True + ) def _async_stop(self) -> None: """Shutdown coordinator and any connection.""" LOGGER.debug("%s: shutting down BMS device", self.name) + if self._dev_info_task is not None and not self._dev_info_task.done(): + self._dev_info_task.cancel() self.hass.async_create_task(self.api.disconnect()) super()._async_stop() @@ -93,18 +193,27 @@ def _async_handle_bluetooth_event( ) -> None: """Handle a Bluetooth event.""" - # if not self.dev_details: - # self.hass.async_create_task(self._get_device_info()) - LOGGER.debug("BLE event %s: %s", change, service_info.manufacturer_data) - self.data = {ATTR_RSSI: service_info.rssi} + self.api.set_ble_device(service_info.device) + + # Merge onto the retained sample rather than replacing it: an advert + # without a valid V2 record (wrong length / legacy format / 2073 + # absent) decodes to {} and must not wipe the last-known position, + # tilt and battery state. + new_data: dict[str, int | float | bool] = dict(self.data) + new_data[ATTR_RSSI] = service_info.rssi if change == bluetooth.BluetoothChange.ADVERTISEMENT: - self.data.update( - self.api.dec_manufacturer_data( - bytearray(service_info.manufacturer_data.get(2073, b"")) - ) + decoded = self.api.dec_manufacturer_data( + bytearray(service_info.manufacturer_data.get(2073, b"")) ) - self.api.encrypted = bool(self.data.get("home_id")) - + if decoded: + new_data.update(decoded) + self.api.encrypted = bool(decoded.get("home_id")) + self._last_v2_at = time.monotonic() + self._maybe_refresh_dev_info() + + if new_data == self.data: + return + self.data = new_data LOGGER.debug("data sample %s", self.data) super()._async_handle_bluetooth_event(service_info, change) diff --git a/custom_components/hunterdouglas_powerview_ble/cover.py b/custom_components/hunterdouglas_powerview_ble/cover.py index 4c25bf5..313a0d0 100644 --- a/custom_components/hunterdouglas_powerview_ble/cover.py +++ b/custom_components/hunterdouglas_powerview_ble/cover.py @@ -16,34 +16,45 @@ CoverEntity, CoverEntityFeature, ) -from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant -from homeassistant.helpers.device_registry import DeviceInfo, format_mac +from homeassistant.helpers.device_registry import format_mac from homeassistant.helpers.entity_platform import AddEntitiesCallback +from . import ConfigEntryType, async_setup_shade_platform from .api import CLOSED_POSITION, OPEN_POSITION -from .const import DOMAIN, HOME_KEY, LOGGER +from .const import DOMAIN, LOGGER from .coordinator import PVCoordinator -async def async_setup_entry( - _hass: HomeAssistant, - config_entry: ConfigEntry, - async_add_entities: AddEntitiesCallback, +def _add_entities( + coordinator: PVCoordinator, async_add_entities: AddEntitiesCallback ) -> None: - """Set up the demo cover platform.""" - - coordinator: PVCoordinator = config_entry.runtime_data - model: Final[str|None] = coordinator.dev_details.get("model") - entities: list[PowerViewCover] = [] - if model in ["39"]: - entities.append(PowerViewCoverTiltOnly(coordinator)) + """Create cover entities for a single shade coordinator.""" + caps = coordinator.shade_capabilities + + if caps.tilt_only: + entities: list[PowerViewCover] = [PowerViewCoverTiltOnly(coordinator)] + elif caps.is_tilt_on_closed: + entities = [PowerViewCoverTiltOnClosed(coordinator)] + elif caps.has_tilt: + entities = [PowerViewCoverTilt(coordinator)] + elif caps.is_top_down: + entities = [PowerViewCoverTopDown(coordinator)] else: - entities.append(PowerViewCover(coordinator)) + entities = [PowerViewCover(coordinator)] async_add_entities(entities) +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntryType, + async_add_entities: AddEntitiesCallback, +) -> None: + """Set up the cover platform.""" + async_setup_shade_platform(hass, config_entry, async_add_entities, _add_entities) + + class PowerViewCover(PassiveBluetoothCoordinatorEntity[PVCoordinator], CoverEntity): # type: ignore[reportIncompatibleVariableOverride] """Representation of a PowerView shade with Up/Down functionality only.""" @@ -62,7 +73,7 @@ def __init__( ) -> None: """Initialize the shade.""" LOGGER.debug("%s: init() PowerViewCover", coordinator.name) - self._attr_name = CoverDeviceClass.SHADE + self._attr_name = None self._coord: PVCoordinator = coordinator self._attr_device_info = self._coord.device_info self._target_position: int | None = round( @@ -73,14 +84,11 @@ def __init__( ) super().__init__(coordinator) - @property - def device_info(self) -> DeviceInfo: # type: ignore[reportIncompatibleVariableOverride] - """Return the device_info of the device.""" - return self._coord.device_info - @property def is_opening(self) -> bool | None: # type: ignore[reportIncompatibleVariableOverride] """Return if the cover is opening or not.""" + if not self._coord.data_available: + return None return bool(self._coord.data.get("is_opening")) or ( isinstance(self._target_position, int) and isinstance(self.current_cover_position, int) @@ -91,6 +99,8 @@ def is_opening(self) -> bool | None: # type: ignore[reportIncompatibleVariableO @property def is_closing(self) -> bool | None: # type: ignore[reportIncompatibleVariableOverride] """Return if the cover is closing or not.""" + if not self._coord.data_available: + return None return bool(self._coord.data.get("is_closing")) or ( isinstance(self._target_position, int) and isinstance(self.current_cover_position, int) @@ -106,9 +116,7 @@ def is_closed(self) -> bool: # type: ignore[reportIncompatibleVariableOverride] @property def supported_features(self) -> CoverEntityFeature: # type: ignore[reportIncompatibleVariableOverride] """Flag supported features, disable control if encryption is needed.""" - if ( - self._coord.data.get("home_id") and len(HOME_KEY) != 16 - ) or self._coord.data.get("battery_charging"): + if self._coord.data.get("home_id") and not self._coord.api.has_key: return CoverEntityFeature(0) return super().supported_features @@ -119,8 +127,7 @@ def current_cover_position(self) -> int | None: # type: ignore[reportIncompatib None is unknown, 0 is closed, 100 is fully open. """ - pos: Final = self._coord.data.get(ATTR_CURRENT_POSITION) - return round(pos) if pos is not None else None + return self._fresh_position(ATTR_CURRENT_POSITION) async def async_set_cover_position(self, **kwargs: Any) -> None: """Move the cover to a specific position.""" @@ -133,7 +140,10 @@ async def async_set_cover_position(self, **kwargs: Any) -> None: return self._target_position = round(target_position) try: - await self._coord.api.set_position(round(target_position)) + await self._coord.api.set_position( + round(target_position), + velocity=self._coord.velocity, + ) self.async_write_ha_state() except BleakError as err: LOGGER.error( @@ -146,6 +156,13 @@ async def async_set_cover_position(self, **kwargs: Any) -> None: def _reset_target_position(self) -> None: self._target_position = None + def _fresh_position(self, key: str) -> int | None: + """Return the rounded ``key`` position, or None if stale/absent.""" + if not self._coord.data_available: + return None + pos = self._coord.data.get(key) + return round(pos) if pos is not None else None + async def async_open_cover(self, **kwargs: Any) -> None: """Open the cover.""" LOGGER.debug("open cover") @@ -153,7 +170,7 @@ async def async_open_cover(self, **kwargs: Any) -> None: return try: self._target_position = OPEN_POSITION - await self._coord.api.open() + await self._coord.api.open(velocity=self._coord.velocity) self.async_write_ha_state() except BleakError as err: LOGGER.error("Failed to open cover '%s': %s", self.name, err) @@ -166,7 +183,7 @@ async def async_close_cover(self, **kwargs: Any) -> None: return try: self._target_position = CLOSED_POSITION - await self._coord.api.close() + await self._coord.api.close(velocity=self._coord.velocity) self.async_write_ha_state() except BleakError as err: LOGGER.error("Failed to close cover '%s': %s", self.name, err) @@ -211,8 +228,7 @@ def current_cover_tilt_position(self) -> int | None: # type: ignore[reportIncom None is unknown """ - pos: Final = self._coord.data.get(ATTR_CURRENT_TILT_POSITION) - return round(pos) if pos is not None else None + return self._fresh_position(ATTR_CURRENT_TILT_POSITION) async def async_set_cover_tilt_position(self, **kwargs: Any) -> None: """Move the tilt to a specific position.""" @@ -227,7 +243,9 @@ async def async_set_cover_tilt_position(self, **kwargs: Any) -> None: try: await self._coord.api.set_position( - self.current_cover_position, tilt=target_position + self.current_cover_position, + tilt=target_position, + velocity=self._coord.velocity, ) self.async_write_ha_state() except BleakError as err: @@ -240,7 +258,7 @@ async def async_set_cover_tilt_position(self, **kwargs: Any) -> None: async def async_stop_cover_tilt(self, **kwargs: Any) -> None: """Stop the cover.""" - await self.async_stop_cover(kwargs=kwargs) + await self.async_stop_cover(**kwargs) async def async_open_cover_tilt(self, **kwargs: Any) -> None: """Open the cover tilt.""" @@ -255,6 +273,111 @@ async def async_close_cover_tilt(self, **kwargs: Any) -> None: await self.async_set_cover_tilt_position(**_kwargs) +class PowerViewCoverTiltOnClosed(PowerViewCoverTilt): + """Representation of a PowerView shade whose tilt is only available when closed. + + Examples: Bottom Up 90° (type 18), Twist (type 44). + + If a tilt command arrives while the shade is open, the shade is closed first + so the tilt mechanism is engaged before the command is sent. + """ + + def __init__(self, coordinator: PVCoordinator) -> None: + """Initialize the shade.""" + LOGGER.debug("%s: init() PowerViewCoverTiltOnClosed", coordinator.name) + super().__init__(coordinator) + + async def async_set_cover_tilt_position(self, **kwargs: Any) -> None: + """Move the tilt to a specific position, closing first if needed.""" + if self.current_cover_position != CLOSED_POSITION: + LOGGER.debug("tilt-on-closed: closing shade before tilting") + try: + self._target_position = CLOSED_POSITION + await self._coord.api.close(velocity=self._coord.velocity) + self.async_write_ha_state() + except BleakError as err: + LOGGER.error( + "Failed to close cover '%s' before tilt: %s", self.name, err + ) + self._reset_target_position() + return + await super().async_set_cover_tilt_position(**kwargs) + + +class PowerViewCoverTopDown(PowerViewCover): + """Representation of a top-down PowerView shade. + + The device position axis is inverted: device 0 = open (fabric retracted), + device 100 = closed (fabric fully extended). We translate at the boundary + so HA's standard 0=closed / 100=open convention is preserved. + """ + + @property + def current_cover_position(self) -> int | None: # type: ignore[reportIncompatibleVariableOverride] + """Return current position, inverting the device axis.""" + pos = self._fresh_position(ATTR_CURRENT_POSITION) + return OPEN_POSITION - pos if pos is not None else None + + async def async_set_cover_position(self, **kwargs: Any) -> None: + """Move the cover to a specific position, inverting for the device.""" + target_position: Final = kwargs.get(ATTR_POSITION) + if target_position is not None: + inverted = OPEN_POSITION - round(target_position) + LOGGER.debug( + "set top-down cover to position %f (device %i)", + target_position, + inverted, + ) + if self.current_cover_position == round(target_position) and not ( + self.is_closing or self.is_opening + ): + return + self._target_position = round(target_position) + try: + await self._coord.api.set_position( + inverted, + velocity=self._coord.velocity, + ) + self.async_write_ha_state() + except BleakError as err: + LOGGER.error( + "Failed to move cover '%s' to %f%%: %s", + self.name, + target_position, + err, + ) + + async def async_open_cover(self, **kwargs: Any) -> None: + """Open the cover (send device position 0).""" + LOGGER.debug("open top-down cover") + if self.current_cover_position == OPEN_POSITION: + return + try: + self._target_position = OPEN_POSITION + await self._coord.api.set_position( + CLOSED_POSITION, velocity=self._coord.velocity + ) + self.async_write_ha_state() + except BleakError as err: + LOGGER.error("Failed to open cover '%s': %s", self.name, err) + self._reset_target_position() + + async def async_close_cover(self, **kwargs: Any) -> None: + """Close the cover (send device position 100).""" + LOGGER.debug("close top-down cover") + if self.current_cover_position == CLOSED_POSITION: + return + try: + self._target_position = CLOSED_POSITION + await self._coord.api.set_position( + OPEN_POSITION, velocity=self._coord.velocity + ) + self.async_write_ha_state() + except BleakError as err: + LOGGER.error("Failed to close cover '%s': %s", self.name, err) + self._reset_target_position() + + class PowerViewCoverTiltOnly(PowerViewCoverTilt): """Representation of a PowerView shade with additional tilt functionality.""" diff --git a/custom_components/hunterdouglas_powerview_ble/diagnostics.py b/custom_components/hunterdouglas_powerview_ble/diagnostics.py new file mode 100644 index 0000000..7dea69d --- /dev/null +++ b/custom_components/hunterdouglas_powerview_ble/diagnostics.py @@ -0,0 +1,273 @@ +"""Diagnostics for the Hunter Douglas PowerView (BLE) integration. + +Dumps every signal the integration holds for each shade, decoded as little as +possible: the advertisement bytes, the shade's replies to its GATT queries, the +hub's record, plus device, capability and connectivity state. It is meant to be +attached to a bug report, whatever the bug is — position and tilt, range and +connectivity, encryption and control failures, the reset/reinit flags, or a +shade's power source. + +Raw values are reported *alongside* the integration's interpretation of them, +never instead of it, because the two have disagreed before: byte 0 of the 0xFFDE +reply was taken for a power-source enum when it is really the protocol's status +code. It reads 0 on every successful reply, so the integration concluded that +every shade was mains-powered and stripped the battery sensors off the ones that +were not. A decoded field can be wrong; the bytes it came from cannot. +""" + +import asyncio +from typing import Any, Final + +import aiohttp +from bleak.exc import BleakError + +from homeassistant.components import bluetooth +from homeassistant.components.bluetooth import BluetoothServiceInfoBleak +from homeassistant.core import HomeAssistant +from homeassistant.helpers.aiohttp_client import async_get_clientsession +from homeassistant.helpers.device_registry import DeviceEntry +from homeassistant.helpers.redact import async_redact_data + +from . import ConfigEntryType +from .api import ShadeCmd +from .const import CONF_HOME_KEY, CONF_HUB_URL, DOMAIN, MFCT_ID +from .coordinator import PVCoordinator + +# home_key is the AES key that drives the shades, home_id identifies the +# PowerView network, and a serial number identifies the owner's hardware. None +# is needed to diagnose anything, and this file gets attached to public issues. +# Each key is listed in both spellings: the shade reports snake_case over GATT +# and the hub reports camelCase in its JSON, so redacting one spelling alone +# silently leaks the same value from the other source. +TO_REDACT: Final[set[str]] = { + CONF_HOME_KEY, + "homeId", + "homeKey", + "home_id", + "serialNumber", + "serial_nr", + "serial_number", +} + +# Every GATT query costs a BLE connection. Opening one per shade at once is the +# adapter contention that made the old startup-time queries unreliable, so keep +# the fan-out bounded even though a user has to click a button to get here. +_MAX_PARALLEL_QUERIES: Final[int] = 3 +_QUERY_TIMEOUT_S: Final[float] = 20.0 +_HUB_TIMEOUT_S: Final[float] = 10.0 + +_NOTE: Final[str] = ( + "Raw, uninterpreted values for every shade, for attaching to a GitHub " + "issue. Some things the integration cannot know and must be stated in the " + "issue text — notably whether a shade runs on a battery or is wired to " + "mains, which it does not detect." +) + + +async def _async_hub_shades( + hass: HomeAssistant, hub_url: str +) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]: + """Return the hub's status and its raw /home/shades records, by BLE name. + + Records pass through verbatim. Establishing what the hub actually sends is + half the value here, so nothing is mapped, renamed or filtered. + """ + if not hub_url: + return {"configured": False}, {} + + session = async_get_clientsession(hass) + timeout = aiohttp.ClientTimeout(total=_HUB_TIMEOUT_S) + try: + async with session.get(f"{hub_url}/home/shades", timeout=timeout) as resp: + resp.raise_for_status() + shades = await resp.json(content_type=None) + except (TimeoutError, aiohttp.ClientError, ValueError) as ex: + # A .local hub URL resolves over mDNS, which is flaky from inside a + # container at boot; the integration swallows that failure silently, so + # surface it here. + return { + "configured": True, + "url": hub_url, + "reachable": False, + "error": f"{type(ex).__name__}: {ex}", + }, {} + + records: dict[str, dict[str, Any]] = { + shade["bleName"]: shade for shade in shades or [] if shade.get("bleName") + } + status: dict[str, Any] = { + "configured": True, + "url": hub_url, + "reachable": True, + "shade_count": len(records), + # Hub records are joined to shades by an exact name match, so a shade + # whose advertised name differs from the hub's bleName silently loses + # its hub data. List the names to make such a near-miss visible. + "ble_names": sorted(records), + } + return status, records + + +def _device(coord: PVCoordinator) -> dict[str, Any]: + """Return the shade's identity: type, model and firmware revisions.""" + return { + "type_id": coord.type_id, + **async_redact_data(dict(coord.dev_details), TO_REDACT), + } + + +def _capabilities(coord: PVCoordinator) -> dict[str, bool]: + """Return what the integration believes this shade type can do. + + Derived from type_id alone, so a shade whose entities look wrong (a missing + tilt, a phantom second rail) is usually a wrong or unmapped type_id here. + """ + caps = coord.shade_capabilities + return { + "has_tilt": caps.has_tilt, + "tilt_only": caps.tilt_only, + "is_tilt_on_closed": caps.is_tilt_on_closed, + "is_top_down": caps.is_top_down, + "is_tdbu": caps.is_tdbu, + "is_duolite": caps.is_duolite, + } + + +def _connectivity( + coord: PVCoordinator, service_info: BluetoothServiceInfoBleak | None +) -> dict[str, Any]: + """Return radio and encryption state, for range and control failures.""" + return { + "present": coord.device_present, + "rssi": service_info.rssi if service_info else None, + # Position, tilt and battery ride only on V2 adverts; once those stop + # arriving the retained sample is stale and the entities read unknown. + "advert_stale": not coord.data_available, + "encrypted": coord.api.encrypted, + "home_key_configured": coord.api.has_key, + } + + +def _advertisement( + coord: PVCoordinator, service_info: BluetoothServiceInfoBleak | None +) -> dict[str, Any]: + """Return the last advertisement: raw bytes and the decoded sample.""" + if service_info is None: + return {"error": "no advertisement seen for this shade"} + + raw = bytes(service_info.manufacturer_data.get(MFCT_ID, b"")) + return { + "raw": raw.hex(" "), + "length": len(raw), + "is_v2_record": len(raw) == 9, + # The top two bits of byte 8 drive the battery sensor via a + # 3/2/1/0 -> 100/50/20/0 % table. Report the code itself: hardwired + # shades also sit at 3, so the decoded 100% below means little on its + # own. + "power_level_code": raw[8] >> 6 if len(raw) == 9 else None, + "decoded": async_redact_data(dict(coord.data), TO_REDACT), + "velocity": coord.velocity, + } + + +async def _async_queries( + coord: PVCoordinator, sem: asyncio.Semaphore +) -> dict[str, Any]: + """Read the shade's GATT queries, reported byte-for-byte. + + Failure is data too, not an error to raise: a shade that cannot be reached + still belongs in the report, so the reason is recorded and the dump goes on. + Keyed by opcode so further queries can be added without reshaping the file. + """ + if coord.api.encrypted and not coord.api.has_key: + reason = {"error": "shade is encrypted and no home key is configured"} + return {ShadeCmd.POWER_STATUS.name: reason} + + async with sem: + try: + payload = await asyncio.wait_for( + coord.api.query_power_status(), timeout=_QUERY_TIMEOUT_S + ) + result: dict[str, Any] = { + "opcode": f"0x{ShadeCmd.POWER_STATUS.value:04X}", + "raw": payload.hex(" "), + "length": len(payload), + "bytes": {f"b{idx}": value for idx, value in enumerate(payload)}, + } + except (BleakError, TimeoutError) as ex: + result = { + "opcode": f"0x{ShadeCmd.POWER_STATUS.value:04X}", + "error": f"{type(ex).__name__}: {ex}", + } + + return {ShadeCmd.POWER_STATUS.name: result} + + +async def _async_shade( + hass: HomeAssistant, + coord: PVCoordinator, + hub_records: dict[str, dict[str, Any]], + sem: asyncio.Semaphore, +) -> dict[str, Any]: + """Collect every signal the integration holds for one shade.""" + service_info = bluetooth.async_last_service_info( + hass, coord.address, connectable=True + ) + ble_name = coord.api.name + record = hub_records.get(ble_name) + return { + "name": coord.friendly_name, + "ble_name": ble_name, + "address": coord.address, + "device": _device(coord), + "capabilities": _capabilities(coord), + "connectivity": _connectivity(coord, service_info), + "advertisement": _advertisement(coord, service_info), + "queries": await _async_queries(coord, sem), + "hub_matched": record is not None, + "hub_record": async_redact_data(record, TO_REDACT) if record else None, + } + + +async def async_get_config_entry_diagnostics( + hass: HomeAssistant, entry: ConfigEntryType +) -> dict[str, Any]: + """Return diagnostics for every shade in the config entry.""" + hub_status, hub_records = await _async_hub_shades( + hass, entry.data.get(CONF_HUB_URL, "") + ) + sem = asyncio.Semaphore(_MAX_PARALLEL_QUERIES) + shades = await asyncio.gather( + *( + _async_shade(hass, coord, hub_records, sem) + for coord in entry.runtime_data.values() + ) + ) + return { + "note": _NOTE, + "config_entry": async_redact_data(dict(entry.data), TO_REDACT), + "hub": hub_status, + "shades": shades, + } + + +async def async_get_device_diagnostics( + hass: HomeAssistant, entry: ConfigEntryType, device: DeviceEntry +) -> dict[str, Any]: + """Return diagnostics for a single shade.""" + addresses = {ident[1] for ident in device.identifiers if ident[0] == DOMAIN} + coord = next( + (coord for addr, coord in entry.runtime_data.items() if addr in addresses), + None, + ) + if coord is None: + return {"error": "no coordinator is running for this device"} + + hub_status, hub_records = await _async_hub_shades( + hass, entry.data.get(CONF_HUB_URL, "") + ) + return { + "note": _NOTE, + "hub": hub_status, + "shade": await _async_shade(hass, coord, hub_records, asyncio.Semaphore(1)), + } diff --git a/custom_components/hunterdouglas_powerview_ble/manifest.json b/custom_components/hunterdouglas_powerview_ble/manifest.json index eeab944..2f5d9de 100644 --- a/custom_components/hunterdouglas_powerview_ble/manifest.json +++ b/custom_components/hunterdouglas_powerview_ble/manifest.json @@ -11,10 +11,14 @@ "config_flow": true, "dependencies": ["bluetooth_adapters"], "documentation": "https://github.com/patman15/hdpv_ble", - "integration_type": "device", + "integration_type": "hub", "iot_class": "local_polling", "issue_tracker": "https://github.com/patman15/hdpv_ble/issues", "loggers": ["hunterdouglas_powerview_ble"], "requirements": ["cryptography>=43.0.0"], - "version": "0.23" + "version": "0.24", + "zeroconf": [ + { "type": "_powerview-g3._tcp.local." }, + { "type": "_PowerView-G3._tcp.local." } + ] } diff --git a/custom_components/hunterdouglas_powerview_ble/number.py b/custom_components/hunterdouglas_powerview_ble/number.py new file mode 100644 index 0000000..8e24995 --- /dev/null +++ b/custom_components/hunterdouglas_powerview_ble/number.py @@ -0,0 +1,72 @@ +"""Hunter Douglas PowerView velocity control.""" + +from homeassistant.components.bluetooth.passive_update_coordinator import ( + PassiveBluetoothCoordinatorEntity, +) +from homeassistant.components.number import NumberMode, RestoreNumber +from homeassistant.const import EntityCategory +from homeassistant.core import HomeAssistant +from homeassistant.helpers.device_registry import format_mac +from homeassistant.helpers.entity_platform import AddEntitiesCallback + +from . import ConfigEntryType, async_setup_shade_platform +from .const import DOMAIN, LOGGER +from .coordinator import PVCoordinator + + +def _add_entities( + coordinator: PVCoordinator, async_add_entities: AddEntitiesCallback +) -> None: + """Create velocity number entity for a single shade coordinator.""" + async_add_entities([PowerViewVelocity(coordinator)]) + + +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntryType, + async_add_entities: AddEntitiesCallback, +) -> None: + """Set up the velocity number entity.""" + async_setup_shade_platform(hass, config_entry, async_add_entities, _add_entities) + + +class PowerViewVelocity( + PassiveBluetoothCoordinatorEntity[PVCoordinator], RestoreNumber +): # type: ignore[reportIncompatibleVariableOverride] + """Number entity to control shade movement velocity.""" + + _attr_has_entity_name = True + _attr_name = "Velocity" + _attr_icon = "mdi:speedometer" + _attr_mode = NumberMode.SLIDER + _attr_native_min_value = 0 + _attr_native_max_value = 100 + _attr_native_step = 1 + _attr_entity_category = EntityCategory.CONFIG + + def __init__(self, coordinator: PVCoordinator) -> None: + """Initialize the velocity entity.""" + self._coord = coordinator + self._attr_device_info = self._coord.device_info + self._attr_unique_id = f"{DOMAIN}_{format_mac(self._coord.address)}_velocity" + super().__init__(coordinator) + + @property + def native_value(self) -> int: + """Return the current velocity value.""" + return self._coord.velocity + + async def async_added_to_hass(self) -> None: + """Restore last known velocity on startup.""" + await super().async_added_to_hass() + last_data = await self.async_get_last_number_data() + if last_data and last_data.native_value is not None: + self._coord.velocity = int(last_data.native_value) + LOGGER.debug( + "%s: restored velocity to %s", self._coord.name, self._coord.velocity + ) + + async def async_set_native_value(self, value: float) -> None: + """Set the velocity value.""" + self._coord.velocity = int(value) + self.async_write_ha_state() diff --git a/custom_components/hunterdouglas_powerview_ble/sensor.py b/custom_components/hunterdouglas_powerview_ble/sensor.py index d19333d..07512bf 100644 --- a/custom_components/hunterdouglas_powerview_ble/sensor.py +++ b/custom_components/hunterdouglas_powerview_ble/sensor.py @@ -15,7 +15,7 @@ from homeassistant.helpers.device_registry import format_mac from homeassistant.helpers.entity_platform import AddEntitiesCallback -from . import ConfigEntryType +from . import ConfigEntryType, async_setup_shade_platform from .const import ATTR_RSSI, DOMAIN from .coordinator import PVCoordinator @@ -39,18 +39,25 @@ ] +def _add_entities( + coordinator: PVCoordinator, async_add_entities: AddEntitiesCallback +) -> None: + """Create sensor entities for a single shade coordinator.""" + async_add_entities( + [ + PVSensor(coordinator, descr, format_mac(coordinator.address)) + for descr in SENSOR_TYPES + ] + ) + + async def async_setup_entry( - _hass: HomeAssistant, + hass: HomeAssistant, config_entry: ConfigEntryType, async_add_entities: AddEntitiesCallback, ) -> None: """Add sensors for passed config_entry in Home Assistant.""" - - pv_dev: PVCoordinator = config_entry.runtime_data - for descr in SENSOR_TYPES: - async_add_entities( - [PVSensor(pv_dev, descr, format_mac(config_entry.unique_id))] - ) + async_setup_shade_platform(hass, config_entry, async_add_entities, _add_entities) class PVSensor(PassiveBluetoothCoordinatorEntity[PVCoordinator], SensorEntity): # type: ignore[reportIncompatibleMethodOverride] @@ -69,5 +76,14 @@ def __init__( @property def native_value(self) -> int | float | None: # type: ignore[reportIncompatibleVariableOverride] - """Return the sensor value.""" + """Return the sensor value. + + RSSI refreshes on every advert, so it is always reported. V2-derived + values (e.g. battery) read as unknown once the advert data is stale. + """ + if ( + self.entity_description.key != ATTR_RSSI + and not self.coordinator.data_available + ): + return None return self.coordinator.data.get(self.entity_description.key) diff --git a/custom_components/hunterdouglas_powerview_ble/strings.json b/custom_components/hunterdouglas_powerview_ble/strings.json index 19601be..7f6ff51 100644 --- a/custom_components/hunterdouglas_powerview_ble/strings.json +++ b/custom_components/hunterdouglas_powerview_ble/strings.json @@ -1,15 +1,55 @@ { "config": { - "flow_title": "Setup {name}", + "flow_title": "PowerView Home Setup", "step": { - "bluetooth_confirm": { - "description": "[%key:component::bluetooth::config::step::bluetooth_confirm::description%]" + "user": { + "title": "Set up PowerView Home", + "description": "Configure your PowerView encryption key. All shades on your network will be discovered automatically via Bluetooth.", + "data": { + "key_method": "Key source", + "hub_url": "PowerView hub URL", + "home_key": "HomeKey (32 hex characters or \\xNN format)" + }, + "data_description": { + "hub_url": "Base URL of your PowerView G3 hub, e.g. {hub_url_example}", + "home_key": "32-character hex string (e.g. 0102030405060708090a0b0c0d0e0f10) or \\x escaped format (e.g. \\x01\\x02...)" + } + }, + "zeroconf_confirm": { + "title": "PowerView gateway discovered", + "description": "A PowerView G3 gateway was found at {name}. Choose how to configure your encryption key — fetch it automatically from the gateway, enter it manually, or skip. All shades on your network will be discovered automatically via Bluetooth.", + "data": { + "key_method": "Key source", + "hub_url": "PowerView hub URL", + "home_key": "HomeKey (32 hex characters or \\xNN format)" + }, + "data_description": { + "hub_url": "Base URL of your PowerView G3 hub, e.g. {hub_url_example}", + "home_key": "32-character hex string (e.g. 0102030405060708090a0b0c0d0e0f10) or \\x escaped format (e.g. \\x01\\x02...)" + } } }, + "error": { + "invalid_key_format": "HomeKey must be a valid hexadecimal string", + "invalid_key_length": "HomeKey must be exactly 32 hex characters (16 bytes)", + "hub_connection_error": "Cannot connect to the PowerView hub", + "hub_http_error": "Hub returned an HTTP error", + "hub_timeout": "Connection to hub timed out", + "hub_protocol_error": "Hub returned an unexpected response" + }, "abort": { "already_configured": "[%key:common::config_flow::abort::already_configured_device%]", - "no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]", - "not_supported": "Device not supported" + "single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]" + } + }, + "entity": { + "binary_sensor": { + "reset_clock": { + "name": "Clock reset required" + }, + "reset_mode": { + "name": "Mode reset required" + } } } } diff --git a/custom_components/hunterdouglas_powerview_ble/translations/en.json b/custom_components/hunterdouglas_powerview_ble/translations/en.json index 528686d..4f51729 100644 --- a/custom_components/hunterdouglas_powerview_ble/translations/en.json +++ b/custom_components/hunterdouglas_powerview_ble/translations/en.json @@ -1,14 +1,54 @@ { "config": { + "flow_title": "PowerView Home Setup", + "step": { + "user": { + "title": "Set up PowerView Home", + "description": "Configure your PowerView encryption key. All shades on your network will be discovered automatically via Bluetooth.", + "data": { + "key_method": "Key source", + "hub_url": "PowerView hub URL", + "home_key": "HomeKey (32 hex characters or \\xNN format)" + }, + "data_description": { + "hub_url": "Base URL of your PowerView G3 hub, e.g. {hub_url_example}", + "home_key": "32-character hex string (e.g. 0102030405060708090a0b0c0d0e0f10) or \\x escaped format (e.g. \\x01\\x02...)" + } + }, + "zeroconf_confirm": { + "title": "PowerView gateway discovered", + "description": "A PowerView G3 gateway was found at {name}. Choose how to configure your encryption key — fetch it automatically from the gateway, enter it manually, or skip. All shades on your network will be discovered automatically via Bluetooth.", + "data": { + "key_method": "Key source", + "hub_url": "PowerView hub URL", + "home_key": "HomeKey (32 hex characters or \\xNN format)" + }, + "data_description": { + "hub_url": "Base URL of your PowerView G3 hub, e.g. {hub_url_example}", + "home_key": "32-character hex string (e.g. 0102030405060708090a0b0c0d0e0f10) or \\x escaped format (e.g. \\x01\\x02...)" + } + } + }, + "error": { + "invalid_key_format": "HomeKey must be a valid hexadecimal string", + "invalid_key_length": "HomeKey must be exactly 32 hex characters (16 bytes)", + "hub_connection_error": "Cannot connect to the PowerView hub", + "hub_http_error": "Hub returned an HTTP error", + "hub_timeout": "Connection to hub timed out", + "hub_protocol_error": "Hub returned an unexpected response" + }, "abort": { "already_configured": "Device is already configured", - "no_devices_found": "No devices found on the network", - "not_supported": "Device not supported" - }, - "flow_title": "Setup {name}", - "step": { - "bluetooth_confirm": { - "description": "Do you want to set up {name}?" + "single_instance_allowed": "Already configured. Only a single configuration possible." + } + }, + "entity": { + "binary_sensor": { + "reset_clock": { + "name": "Clock reset required" + }, + "reset_mode": { + "name": "Mode reset required" } } } diff --git a/scripts/extract_gateway3_homekey.py b/scripts/extract_gateway3_homekey.py index 0058c3d..a90ecc5 100644 --- a/scripts/extract_gateway3_homekey.py +++ b/scripts/extract_gateway3_homekey.py @@ -55,12 +55,15 @@ def get_shade_key(hub: str, ble_name) -> bytes: raise result: dict = json.loads(shades_exec_resp.content) - if result.get("err") != 0 or len(result.get("responses", [])) != 1: - raise OSError("Error when attempting GetShadeKey") - response: Final[bytes] = bytes.fromhex(result["responses"][0]["hex"]) + responses = result.get("responses", []) + if len(responses) != 1 or "hex" not in responses[0]: + raise OSError(f"Error when attempting GetShadeKey: {result}") + response: Final[bytes] = bytes.fromhex(responses[0]["hex"]) dec_resp: Final[dict[str, Any]] = decode_response(response) if dec_resp["errorCode"] != 0: - raise ValueError("BLE errorCode is not 0") + raise ValueError( + f"BLE errorCode={dec_resp['errorCode']} data={dec_resp['data'].hex()}" + ) if len(dec_resp["data"]) != 16: raise ValueError("Expected 16 byte homekey") return dec_resp["data"] @@ -79,9 +82,23 @@ def main(hub: str) -> int: shades = json.loads(shades_resp.content) print(f"Found {len(shades)} shades, interrogating") + network_key: bytes | None = None for shade in shades: name: str = base64.b64decode(shade["name"]).decode("utf-8") - key: bytes = get_shade_key(hub, shade["bleName"]) + try: + key: bytes = get_shade_key(hub, shade["bleName"]) + network_key = key + except (OSError, ValueError) as ex: + if network_key is not None: + key = network_key + print(f"Shade '{name}':") + print(f"\tBLE name: '{shade['bleName']}'") + print(f"\tHomeKey: {key.hex()} (shade unreachable, using network key)") + else: + print(f"Shade '{name}':") + print(f"\tBLE name: '{shade['bleName']}'") + print(f"\tHomeKey: ERROR - {ex}") + continue print(f"Shade '{name}':") print(f"\tBLE name: '{shade['bleName']}'") diff --git a/scripts/shade_report.py b/scripts/shade_report.py new file mode 100644 index 0000000..4a80466 --- /dev/null +++ b/scripts/shade_report.py @@ -0,0 +1,663 @@ +"""Standalone debug report for PowerView BLE shade(s). + +Talks directly to the shade over BLE (AES-CTR + GATT), with no coupling +to the Home Assistant integration code. Fetches the shade's homekey +from a G3 Gateway, scans for the BLE advertisement, then dumps: + 1. BLE advertisement (raw + decoded) + 2. Standard GATT device-info characteristics + 3. Decoded read-only PowerView queries: + - 0xFFDD sel=4 product info (firm update count, sw_rev) + - 0xFFDD sel=5 extended product info (serial, fw/sw/hw_rev, type) + - 0xFFDE power status (hardwired vs battery, level) + +Dependencies: bleak, bleak-retry-connector, cryptography, requests (no HA) + +SAFETY: only the read-only opcodes above are ever sent. No set/move/ +scene/rekey/factory-reset opcodes are implemented here. + +Usage: + python scripts/shade_report.py --ble-name PV-XXXXXX + python scripts/shade_report.py --all + python scripts/shade_report.py --all --hub http://hub.local -v +""" + +from __future__ import annotations + +from pathlib import Path +import sys + +# Allow "python scripts/shade_report.py" (the documented invocation) to resolve +# the sibling package import below: we prepend the project root to sys.path so +# `scripts.extract_gateway3_homekey` is importable regardless of CWD. +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import argparse +import asyncio +import base64 +import contextlib +import json +import logging +import struct +from typing import Any, Self + +from bleak import BleakClient, BleakScanner +from bleak.backends.device import BLEDevice +from bleak.backends.scanner import AdvertisementData +from bleak.uuids import normalize_uuid_str +from bleak_retry_connector import establish_connection +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes +import requests + +from scripts.extract_gateway3_homekey import HUB, get_shade_key + +# --- PowerView BLE protocol --------------------------------------- +MFCT_ID = 2073 # Hunter Douglas BLE manufacturer identifier +UUID_TX = "cafe1001-c0ff-ee01-8000-a110ca7ab1e0" +UUID_COV_SERVICE = normalize_uuid_str("fdc1") +UUID_DEV_SERVICE = normalize_uuid_str("180a") + +# Standard GATT Device Information Service characteristics +GATT_DEV_INFO: list[tuple[str, str]] = [ + ("manufacturer", "2a29"), + ("model", "2a24"), + ("serial_nr", "2a25"), + ("hw_rev", "2a27"), + ("fw_rev", "2a26"), + ("sw_rev", "2a28"), +] + +# Read-only protocol queries ONLY. Never add opcodes that mutate state +# (e.g. 0x01F7 move, 0xFA5A set-scene, 0xFB02 rekey, 0xFFDF set-power- +# type, 0xFFEE factory-reset). +# +# Decoded queries (fw_rev=22, Duette type 6, hardwired): +# 0xFFDD → product-info opcode gated on a 1-byte selector. sel=4 +# returns a 14-byte page (firm update count u32 LE at bytes +# 2-5, firm current version u16 LE at bytes 6-7 matching +# GATT sw_rev). sel=5 returns a 28-byte extended page with +# every field confirmed against GATT Device Information: +# serial (bytes 2-9, little-endian), fw_rev (10-11), sw_rev +# (14-15), hw_rev (18-21 u32 LE), build/mfg id (22-25), and +# type_id (26) matching the advertisement. Decoded in +# annotate_query. 0xF1DD returns the same sel=4 page and +# a truncated 22-byte sel=5 (no trailing 6 bytes); not +# queried separately since its data is a strict subset. +# The emulator's ret_valFFDD (29 B, labelled "HW +# diagnostics") is actually this extended product info with +# dummy data, off-by-one vs real firmware. +# 0xFFDE → 8-byte payload. Byte-by-byte findings on hardwired Duette +# (type 6, fw_rev=22), verified across four shades and ~10 +# targeted motion/idle experiments on one unit, then corrected +# against a six-shade dump cross-checked with the G3 hub: +# b0 = 0x00 status, 0 = success. NOT the power type. +# Every reply in this protocol leads with a +# status code — a rejected 0xFFDE returns a +# 1-byte 0x04 (invalid length) in this same +# position, and _verify_ack_reply treats +# payload[0] == 0 as success. It was read as +# a power-source enum once, which made every +# shade that answered look hardwired (0 being +# the hardwired code) and silently stripped +# the battery sensors off battery shades. A +# constant cannot classify anything; the only +# reason it looked right is that every sample +# behind it came from a hardwired shade. +# b1 = 0x01 power type, and it agrees with the hub: the +# G3 reports powerType=1 for these same six +# shades, all of them confirmed mains-powered. +# The battery and rechargeable codes are still +# unknown — no such sample exists yet. +# b2 = 0x64 (100) strong hypothesis: battery/power-level +# %, pegged at 100 on wall power. A +# battery shade should report its real % +# here. NOT a duplicate of the advert's +# 2-bit level field — this is the full +# byte-resolution value. +# b3 live / noisy byte — read-to-read drift +# of 3-9 counts with no physical change. +# Too jittery to be temperature; best +# guess is radio or ADC telemetry. Not +# practically useful without a burst- +# capture characterization. +# b4 = 0x07 constant on all hardwired shades seen. +# Likely capability or power-config flags. +# b5 persistent per-shade state. Observed +# to change exactly once across ~10 +# experiments on one unit (227 -> 205 +# during the first commanded move of a +# session). Each idle shade holds its +# own stable value (114/127/139/205 seen +# at position=100%). Triggers RULED OUT: +# every move, cold-start after 15-min +# idle, short-period timer, distance- +# gated motion snapshot, top & bottom +# end-stop calibration. Remaining +# candidates: boot-triggered state, +# long-period (hours+) timer, or hub- +# initiated refresh. +# b6 type_id matches advertisement byte 2 (confirmed). +# b7 = 0x00 reserved / padding on all observations. +# All b1-b7 findings are fw_rev=22 hardwired-only; re-verify +# on battery / rechargeable hardware before relying on them. +# 0xFFDE remains the only reliable source for battery-vs- +# hardwired detection (the advert's 2-bit level field caps +# at 3 = "100% OR hardwired" and cannot disambiguate). +# (cmd, payload, label) — each entry is a query that returns real +# decoded data on fw_rev=22. 0xFFDD's two selectors are complementary: +# sel=4 carries firm update count + firm current version; sel=5 carries +# serial / fw_rev / sw_rev / hw_rev / build id / type_id. 0xF1DD is a +# strict subset of 0xFFDD (same sel=4 bytes; sel=5 drops the trailing 6) +# so it's not queried separately. +GET_QUERIES: list[tuple[int, bytes, str]] = [ + (0xFFDD, b"\x04", "product info (sel=4)"), + (0xFFDD, b"\x05", "ext product info (sel=5)"), + (0xFFDE, b"", "power status"), +] + +# Advert level field is 2 bits, so only values 0-3 are observable on the wire. +POWER_LEVELS = {3: 100, 2: 50, 1: 20, 0: 0} + +# PowerView BLE error-response shape (see api.py _verify_response): when a +# query's response has data_len=1, byte 4 is an error code rather than real +# payload. Expand this dict as users report codes from different firmwares. +PV_ERROR_CODES: dict[int, str] = { + 0x04: "invalid length (hypothesis)", +} + +# 0xFFDE byte 1 — power type (byte 0 is the reply's status code, not the power +# type). 1 = hardwired is confirmed: six shades at fw_rev=22 all report b1=0x01, +# the G3 hub independently reports powerType=1 for the same six, and all six are +# known to be mains-powered. Every other code is deliberately absent: guessing +# at them is what produced the last bug, so an unseen value prints as unknown. +POWER_TYPE_LABELS: dict[int, str] = { + 1: "hardwired (confirmed)", +} + +# Known-good response lengths per opcode. Used only by annotate_query +# to decide whether a 1-byte reply should be interpreted as an error +# code (i.e. "we expected a longer payload and got a single byte — that +# byte is the error"). 0xFFDD has two valid shapes because its two +# queried selectors return different lengths; 0xF1DD's 14- and 22-byte +# shapes are retained for completeness even though the script no longer +# queries that opcode directly. Emulator's 0xFFDD dummy is 29 B but +# real fw_rev=22 returns 28 B at sel=5 — treat emulator as off-by-one. +EXPECTED_QUERY_LEN: dict[int, set[int]] = { + 0xF1DD: {14, 22}, + 0xFFDD: {14, 28}, + 0xFFDE: {8}, +} + +SCAN_TIMEOUT = 30.0 +CMD_TIMEOUT = 30.0 +GATEWAY_TIMEOUT = 30 + + +# --- Gateway shade list -------------------------------------------- +def fetch_shade_list(hub: str) -> list[dict[str, Any]]: + """Return the list of shades registered with the G3 gateway.""" + resp = requests.get(f"{hub}/home/shades", timeout=GATEWAY_TIMEOUT) + resp.raise_for_status() + return json.loads(resp.content) + + +# --- BLE advertisement decode -------------------------------------- +def decode_adv(manuf: bytes) -> dict[str, Any] | None: + """Decode the 9-byte V2 advertisement payload.""" + if len(manuf) != 9: + return None + flags = manuf[3] & 0x3 + pos = ((manuf[4] & 0x0F) << 6) | ((manuf[3] >> 2) & 0x3F) + pos2 = (manuf[5] << 4) + (manuf[4] >> 4) + level_bits = (manuf[8] >> 6) & 0x3 + return { + "home_id": int.from_bytes(manuf[0:2], "little"), + "type_id": manuf[2], + "position": pos / 10, + "position2": pos2 >> 2, + "position3": manuf[6], + "tilt": manuf[7], + "opening": flags == 0x2, + "closing": flags == 0x1, + "charging_flag": flags == 0x3, + "level_bits": level_bits, + "level_pct": POWER_LEVELS.get(level_bits, "?"), + } + + +# --- BLE scanner helper -------------------------------------------- +async def find_shades( + names: set[str], timeout: float +) -> dict[str, tuple[BLEDevice, AdvertisementData]]: + """Scan once, return the first advertisement seen for each named shade.""" + loop = asyncio.get_running_loop() + results: dict[str, tuple[BLEDevice, AdvertisementData]] = {} + done = asyncio.Event() + + def detected(dev: BLEDevice, adv: AdvertisementData) -> None: + if dev.name in names and dev.name not in results: + results[dev.name] = (dev, adv) + if len(results) == len(names): + loop.call_soon_threadsafe(done.set) + + scanner = BleakScanner(detection_callback=detected) + await scanner.start() + try: + await asyncio.wait_for(done.wait(), timeout=timeout) + except TimeoutError: + pass + finally: + await scanner.stop() + return results + + +# --- Minimal PowerView BLE client ---------------------------------- +class PowerViewClient: + """Talks the PowerView BLE protocol. + + Wire format (TX and RX, both encrypted with AES-128-CTR, IV = 16 zero + bytes, key = the shade's homekey): + + byte 0..1 command (little-endian uint16) + byte 2 sequence (TX: incremented; RX: echoed from TX) + byte 3 payload length + byte 4.. payload + + Both write and notify happen on the same UUID_TX characteristic. + """ + + def __init__(self, device: BLEDevice, home_key: bytes) -> None: + """Store connection parameters; actual connect happens in __aenter__.""" + self.device = device + self.client = BleakClient(device) + self.home_key = home_key + self.seq = 1 + self._rx: bytes = b"" + self._rx_event = asyncio.Event() + + async def __aenter__(self) -> Self: + """Connect over BLE and subscribe to the TX characteristic.""" + # establish_connection (bleak_retry_connector) retries automatically on + # transient WinRT errors and weak-signal cancellations, which plain + # BleakClient.connect() surfaces as TimeoutError. The services= filter + # tells the stack which GATT services to enumerate, speeding discovery + # on marginal links. Mirrors the integration's connect path in api.py. + self.client = await establish_connection( + BleakClient, + self.device, + self.device.name or "shade", + max_attempts=3, + services=[UUID_COV_SERVICE, UUID_DEV_SERVICE], + ) + await self.client.start_notify(UUID_TX, self._on_notify) + return self + + async def __aexit__(self, *_: object) -> None: + """Disconnect best-effort; errors here are not actionable.""" + with contextlib.suppress(Exception): + await self.client.disconnect() + + def _aes(self) -> Cipher[modes.CTR]: + return Cipher(algorithms.AES(self.home_key), modes.CTR(bytes(16))) + + def _on_notify(self, _sender: object, data: bytearray) -> None: + dec = self._aes().decryptor() + self._rx = dec.update(bytes(data)) + dec.finalize() + self._rx_event.set() + + async def read_gatt(self, uuid: str) -> bytes: + """Read a standard GATT characteristic by 16-bit UUID.""" + return bytes(await self.client.read_gatt_char(normalize_uuid_str(uuid))) + + async def query(self, cmd: int, data: bytes = b"") -> bytes: + """Send a read-only command, return the response payload.""" + # cmd is packed BIG-endian so the wire bytes appear in the order the + # shade expects: wire byte 0 = high byte of opcode, byte 1 = low byte. + # GET_QUERIES uses the emulator's case-label naming (0xF1DD, 0xFFDD, + # 0xFFDE), where byte 0 = 0xF1/0xFF and byte 1 = 0xDD/0xDE. The + # integration enum (api.py) uses pre-byte-swapped values with + # little-endian packing to reach the same wire bytes; here we keep + # the emulator naming and byte-swap at pack time instead. + tx = struct.pack(">HBB", cmd, self.seq, len(data)) + data + enc = self._aes().encryptor() + tx_enc = enc.update(tx) + enc.finalize() + self.seq = (self.seq + 1) & 0xFF + self._rx_event.clear() + await self.client.write_gatt_char(UUID_TX, tx_enc, response=False) + await asyncio.wait_for(self._rx_event.wait(), timeout=CMD_TIMEOUT) + if len(self._rx) < 4: + raise ValueError(f"Response too short: {self._rx.hex(' ')}") + payload_len = self._rx[3] + return self._rx[4 : 4 + payload_len] + + +# --- Report rendering ---------------------------------------------- +def _section(title: str) -> None: + print(f"\n {title}") + print(f" {'-' * len(title)}") + + +def annotate_query(cmd: int, payload: bytes) -> str | None: + """Return a one-line decode of a query response, or None if unknown. + + Printed beneath the raw hex so reporters see both. Everything + marked "hypothesis" still needs cross-shade confirmation — that's + the whole point of this report. + """ + expected = EXPECTED_QUERY_LEN.get(cmd, set()) + # Shade returned a 1-byte reply where we know a longer one is valid: + # standard PowerView error-response shape — byte is an error code. + if expected and len(payload) == 1 and 1 not in expected: + err = payload[0] + note = PV_ERROR_CODES.get(err, "unknown error code") + good = ", ".join(str(n) for n in sorted(expected)) + return ( + f"likely error code 0x{err:02X} ({note}); " + f"known good lengths: {good} B" + ) + # Two-byte `8c XX` rejection: byte 0 is a fixed reject code, byte 1 + # echoes the bad input byte (observed for selector=0 and selector=3 + # on fw_rev=22, where the selector is the first byte of the query + # payload). Decoding this helps distinguish "bad selector" from + # other short error shapes. + if len(payload) == 2 and payload[0] == 0x8C: + return ( + f"rejection (0x8C) — selector 0x{payload[1]:02X} " + f"({payload[1]}) not supported by this opcode" + ) + # 14-byte product-info page, returned by 0xF1DD and 0xFFDD when the + # payload's first byte is 0x04. The two opcodes return the same + # bytes here (sel=5 is where they diverge — see below). Confirmed + # on fw_rev=22: byte 1 echoes the selector (0x04), bytes 2-5 are a + # u32 counter (=1 on a freshly-imaged shade), and bytes 6-7 LE are + # the firmware version (matches GATT sw_rev exactly — the emulator + # also embeds SW_VERSION there in its ret_valF1DD dummy). Bytes + # 8-13 are zero on our samples; meaning unknown. + if cmd in (0xF1DD, 0xFFDD) and len(payload) == 14 and payload[1] == 0x04: + counter = int.from_bytes(payload[2:6], "little") + firm_current = int.from_bytes(payload[6:8], "little") + indent = " " + return ( + "14-byte product info (selector=0x04). Fields:\n" + f"{indent}byte 1 = 0x{payload[1]:02X} → echoed selector\n" + f"{indent}bytes 2-5 = {payload[2:6].hex(' ')} → " + f"counter={counter} (u32 LE; hypothesis — firm update count?)\n" + f"{indent}bytes 6-7 = {payload[6:8].hex(' ')} → firm current " + f"version={firm_current} (confirmed; matches GATT sw_rev)\n" + f"{indent}bytes 8-13 = {payload[8:14].hex(' ')} → unknown " + f"(zero on all samples)" + ) + # Extended product info (selector=0x05). 0xF1DD returns a 22-byte + # subset; 0xFFDD returns 28 bytes (same first 22 plus 4-byte build/ + # mfg counter + type_id + model byte). Every field below is + # confirmed against GATT Device Information on fw_rev=22: + # bytes 2-9 = 8-byte serial LE (matches GATT 0x2a25 hex-reversed) + # bytes 10-11 = u16 LE fw_rev (matches GATT 0x2a26) + # bytes 14-15 = u16 LE sw_rev (matches GATT 0x2a28) + # bytes 18-21 = u32 LE hw_rev (matches GATT 0x2a27) + # The FFDD-only tail is still provisional — byte 26 matches the + # advert's type_id; bytes 22-25 look like another 32-bit identifier + # (1015780 on the "Study" shade), and byte 27 is a model/family byte. + if ( + cmd in (0xF1DD, 0xFFDD) + and len(payload) in (22, 28) + and payload[1] == 0x05 + ): + serial_hex = payload[2:10][::-1].hex().upper() + fw_rev = int.from_bytes(payload[10:12], "little") + sw_rev = int.from_bytes(payload[14:16], "little") + hw_rev = int.from_bytes(payload[18:22], "little") + indent = " " + lines = [ + f"{len(payload)}-byte extended product info (selector=0x05). " + "Fields (all confirmed vs GATT):", + f"{indent}byte 1 = 0x{payload[1]:02X} → echoed selector", + f"{indent}bytes 2-9 = {payload[2:10].hex(' ')} → " + f"serial={serial_hex} (LE; matches GATT 0x2a25)", + f"{indent}bytes 10-11 = {payload[10:12].hex(' ')} → " + f"fw_rev={fw_rev} (matches GATT 0x2a26)", + f"{indent}bytes 12-13 = {payload[12:14].hex(' ')} → " + "reserved (zero on samples)", + f"{indent}bytes 14-15 = {payload[14:16].hex(' ')} → " + f"sw_rev={sw_rev} (matches GATT 0x2a28)", + f"{indent}bytes 16-17 = {payload[16:18].hex(' ')} → " + "reserved (zero on samples)", + f"{indent}bytes 18-21 = {payload[18:22].hex(' ')} → " + f"hw_rev={hw_rev} (matches GATT 0x2a27)", + ] + if len(payload) == 28: + build_id = int.from_bytes(payload[22:26], "little") + lines.extend( + [ + f"{indent}bytes 22-25 = {payload[22:26].hex(' ')} → " + f"build/mfg id={build_id} (u32 LE; hypothesis)", + f"{indent}byte 26 = 0x{payload[26]:02X} → " + f"type_id={payload[26]} (matches advert)", + f"{indent}byte 27 = 0x{payload[27]:02X} → " + "model/family byte (hypothesis)", + ] + ) + return "\n".join(lines) + if cmd == 0xFFDE and len(payload) == 8: + power_type = payload[1] + label = POWER_TYPE_LABELS.get(power_type, f"unknown ({power_type})") + # Per-byte interpretations as of 2026-04-21, corrected once the hub + # cross-check landed (see module-level comment). b0, b1 and b6 are + # confirmed; b2/b4/b7 are strong hypotheses based on hardwired shades + # at fw_rev=22; b3 is a live noisy byte; b5 is persistent per-shade + # state with an unknown refresh trigger. + # 11-space indent on continuation lines aligns with the "→ " + # prefix added by _print_queries (9 spaces + arrow + space). + indent = " " + return ( + f"byte 0 = 0x{payload[0]:02X} → status (0 = success; NOT the power " + f"type — reading it as one is what hid the battery sensors)\n" + f"{indent}byte 1 = 0x{payload[1]:02X} → power_type={power_type} " + f"({label}); same encoding as the hub's powerType\n" + f"{indent}byte 2 = 0x{payload[2]:02X} ({payload[2]}) → " + f"battery/power level % (hypothesis — pegged at 100 on hardwired)\n" + f"{indent}byte 3 = 0x{payload[3]:02X} → live/noisy byte " + f"(drifts between reads; radio or ADC telemetry)\n" + f"{indent}byte 4 = 0x{payload[4]:02X} → capability/config flags " + f"(const 0x07 on hardwired; hypothesis)\n" + f"{indent}byte 5 = 0x{payload[5]:02X} → persistent per-shade state " + f"(rarely updates; refresh trigger unknown)\n" + f"{indent}byte 6 = 0x{payload[6]:02X} → type_id={payload[6]} " + f"(cross-check with advert)\n" + f"{indent}byte 7 = 0x{payload[7]:02X} → reserved/padding" + ) + return None + + +def _print_advertisement(device: BLEDevice, adv: AdvertisementData) -> None: + """Print section 1: the BLE advertisement (raw + decoded).""" + print(f" address: {device.address}") + print(f" rssi: {adv.rssi} dBm") + manuf = adv.manufacturer_data.get(MFCT_ID, b"") + print(f" raw: {manuf.hex(' ') if manuf else '(none)'}") + decoded = decode_adv(manuf) + if not decoded: + return + print(f" home_id: 0x{decoded['home_id']:04x}") + print(f" type_id: {decoded['type_id']}") + print( + f" position: {decoded['position']} " + f"pos2={decoded['position2']} pos3={decoded['position3']} " + f"tilt={decoded['tilt']}" + ) + print( + f" flags: opening={decoded['opening']} " + f"closing={decoded['closing']} " + f"charging_flag={decoded['charging_flag']}" + ) + print( + f" level: bits={decoded['level_bits']} (→ {decoded['level_pct']}%)" + ) + + +async def _print_gatt_info(api: PowerViewClient) -> None: + """Print section 2: standard GATT device-info characteristics.""" + for key, uuid in GATT_DEV_INFO: + try: + value = (await api.read_gatt(uuid)).decode("utf-8", errors="replace") + print(f" {key:13} {value}") + except Exception as ex: # noqa: BLE001 + print(f" {key:13} ") + + +async def _print_queries(api: PowerViewClient) -> None: + """Print section 3: read-only protocol queries.""" + for cmd, req, label in GET_QUERIES: + try: + resp = await api.query(cmd, req) + print( + f" 0x{cmd:04X} {label:25} " + f"({len(resp):2} B): {resp.hex(' ')}" + ) + note = annotate_query(cmd, resp) + if note: + print(f" → {note}") + except Exception as ex: # noqa: BLE001 + print(f" 0x{cmd:04X} {label:25} FAILED — {ex}") + break + + +async def report_shade( + friendly_name: str, + ble_name: str, + home_key: bytes, + device: BLEDevice | None, + adv: AdvertisementData | None, +) -> None: + """Print the full report for a single shade: advertisement + GATT + queries.""" + print(f"\n=== '{friendly_name}' (BLE: {ble_name}) ===") + + _section("1. BLE advertisement") + if device is None or adv is None: + print(" not seen on air within scan window — skipping BLE queries") + return + _print_advertisement(device, adv) + + async with PowerViewClient(device, home_key) as api: + _section("2. GATT device info") + await _print_gatt_info(api) + + _section("3. Protocol queries (read-only)") + await _print_queries(api) + + +# --- Main ---------------------------------------------------------- +PER_SHADE_TIMEOUT = 45.0 # hard cap on total time spent per shade's report + + +async def run( + hub: str, + targets: list[tuple[str, str]] | None, + all_shades: bool, + scan_timeout: float, + verbose: bool, +) -> int: + """Fetch the shade list, homekey, then report on each target.""" + if verbose: + logging.basicConfig(level=logging.DEBUG) + + # Always fetch the hub's full shade list: we need it regardless of + # whether targets came from --all or --ble-name, because the homekey + # fetch should use whichever shade the hub sees strongest, not + # necessarily one the user happens to be targeting locally. + print(f"Fetching shade list from {hub}...") + try: + shades = fetch_shade_list(hub) + except Exception as ex: # noqa: BLE001 + print(f" failed: {ex}") + return 1 + print(f" found {len(shades)} shade(s)") + + if all_shades: + targets = [ + (base64.b64decode(s["name"]).decode("utf-8"), s["bleName"]) + for s in shades + ] + assert targets is not None + + # Homekey is network-wide — fetch once from any shade the hub can + # reach. Try the hub's strongest-signal shades first to avoid + # burning the 10s timeout on one that's currently unreachable. + key_candidates = sorted( + shades, key=lambda s: s.get("signalStrength", -100), reverse=True + ) + print(f"\nFetching homekey from {hub} (one-time for the home)...") + home_key: bytes | None = None + for s in key_candidates: + ble = s["bleName"] + sig = s.get("signalStrength", "?") + try: + home_key = get_shade_key(hub, ble) + print(f" got it via {ble} (hub signal {sig}): {home_key.hex()}") + break + except Exception as ex: # noqa: BLE001 + print(f" {ble} (hub signal {sig}): {ex} — trying next shade...") + if home_key is None: + print("Could not fetch homekey from any shade. Aborting.") + return 1 + + name_set = {ble for _, ble in targets} + print( + f"\nScanning for {len(name_set)} shade(s) " + f"(up to {scan_timeout:.0f}s; tap a hardwired shade if not seen)..." + ) + seen = await find_shades(name_set, scan_timeout) + print(f" {len(seen)}/{len(name_set)} visible on air") + + for friendly, ble in targets: + dev, adv = seen.get(ble, (None, None)) + # Per-shade timeout so one unreachable shade can't stall the whole run. + try: + await asyncio.wait_for( + report_shade(friendly, ble, home_key, dev, adv), + timeout=PER_SHADE_TIMEOUT, + ) + except TimeoutError: + print(f" (shade {ble} exceeded {PER_SHADE_TIMEOUT:.0f}s — skipping)") + return 0 + + +def main() -> int: + """Parse CLI args and dispatch to the async report runner.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--hub", default=HUB, help=f"Gateway URL (default: {HUB})") + group = parser.add_mutually_exclusive_group(required=True) + group.add_argument( + "--ble-name", + action="append", + help="BLE name of a shade (repeatable), e.g. 'PV-XXXXXX'", + ) + group.add_argument( + "--all", action="store_true", help="Report on every shade known to the gateway" + ) + parser.add_argument( + "--scan-timeout", + type=float, + default=SCAN_TIMEOUT, + help=f"BLE scan timeout in seconds (default: {SCAN_TIMEOUT})", + ) + parser.add_argument( + "-v", "--verbose", action="store_true", help="Enable debug logging" + ) + args = parser.parse_args() + + targets: list[tuple[str, str]] | None = None + if args.ble_name: + targets = [(ble, ble) for ble in args.ble_name] + return asyncio.run( + run( + args.hub, + targets, + args.all, + args.scan_timeout, + args.verbose, + ) + ) + + +if __name__ == "__main__": + sys.exit(main())