diff --git a/custom_components/hunterdouglas_powerview_ble/__init__.py b/custom_components/hunterdouglas_powerview_ble/__init__.py index dd594f2..c3a6d2f 100644 --- a/custom_components/hunterdouglas_powerview_ble/__init__.py +++ b/custom_components/hunterdouglas_powerview_ble/__init__.py @@ -18,9 +18,10 @@ PLATFORMS: list[Platform] = [ Platform.BINARY_SENSOR, + Platform.BUTTON, Platform.COVER, + Platform.NUMBER, Platform.SENSOR, - Platform.BUTTON, ] type ConfigEntryType = ConfigEntry[PVCoordinator] @@ -42,7 +43,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntryType) -> bool f"Could not find PowerView device ({entry.unique_id}) via Bluetooth" ) - coordinator = PVCoordinator(hass, ble_device, entry.data.copy()) + coordinator = PVCoordinator(hass, ble_device, entry.data.copy(), entry.title) try: await coordinator.query_dev_info() except BleakError as err: diff --git a/custom_components/hunterdouglas_powerview_ble/api.py b/custom_components/hunterdouglas_powerview_ble/api.py index 5a6cc97..65242cb 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 @@ -58,6 +58,31 @@ 62: "Venetian, Tilt Anywhere", } +class ShadeCapability(NamedTuple): + """Capability flags for a shade type.""" + + has_tilt: bool = False + tilt_only: bool = False + + +SHADE_CAPABILITIES: Final[dict[int, ShadeCapability]] = { + # tilt anywhere (position + tilt) + 51: ShadeCapability(has_tilt=True), + 62: ShadeCapability(has_tilt=True), + # tilt only (no position movement) + 39: ShadeCapability(has_tilt=True, tilt_only=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 @@ -97,18 +122,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 +142,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 +155,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.""" @@ -227,7 +253,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 +262,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 @@ -355,6 +381,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..7749300 100644 --- a/custom_components/hunterdouglas_powerview_ble/binary_sensor.py +++ b/custom_components/hunterdouglas_powerview_ble/binary_sensor.py @@ -40,7 +40,9 @@ async def async_setup_entry( ) -class PVBinarySensor(PassiveBluetoothCoordinatorEntity[PVCoordinator], BinarySensorEntity): # type: ignore[reportIncompatibleMethodOverride] +class PVBinarySensor( + PassiveBluetoothCoordinatorEntity[PVCoordinator], BinarySensorEntity +): # type: ignore[reportIncompatibleMethodOverride] """The generic PV binary sensor implementation.""" def __init__( diff --git a/custom_components/hunterdouglas_powerview_ble/config_flow.py b/custom_components/hunterdouglas_powerview_ble/config_flow.py index 7a38b35..6c88d50 100644 --- a/custom_components/hunterdouglas_powerview_ble/config_flow.py +++ b/custom_components/hunterdouglas_powerview_ble/config_flow.py @@ -1,8 +1,13 @@ """Config flow for BLE Battery Management System integration.""" +import asyncio +import base64 +import contextlib from dataclasses import dataclass +import struct from typing import Any +import aiohttp import voluptuous as vol from homeassistant import config_entries @@ -12,21 +17,163 @@ ) 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 .api import UUID_COV_SERVICE as UUID -from .const import DOMAIN, LOGGER, MFCT_ID +from .const import CONF_HOME_KEY, DOMAIN, LOGGER, MFCT_ID + + +def _needs_encryption(manufacturer_data_hex: str) -> bool: + """Return True if the BLE advertisement indicates encryption (home_id != 0).""" + data = bytearray.fromhex(manufacturer_data_hex) + if len(data) < 2: + return False + home_id = int.from_bytes(data[0:2], byteorder="little") + return home_id != 0 + + +@dataclass +class HubShadeInfo: + """Shade metadata from the PowerView hub.""" + + name: str # Human-readable name (decoded from base64) + ble_name: str # BLE advertisement name, e.g. "DUE:94ED" + + +async def _fetch_shades_from_hub( + hass: HomeAssistant, hub_url: str +) -> list[HubShadeInfo]: + """Fetch shade list with human-readable names from a PowerView G3 hub. + + 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: + return [] + + hub_shades: list[HubShadeInfo] = [] + for shade in shades: + 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 + hub_shades.append(HubShadeInfo(name=name, ble_name=ble_name)) + return hub_shades + + +async def _fetch_key_and_shades_from_hub( + hass: HomeAssistant, hub_url: str +) -> tuple[bytes, list[HubShadeInfo]]: + """Fetch 16-byte homekey and shade list from a PowerView G3 hub. + + Returns (key, shade_list). The key is network-wide so any reachable shade + returns the same value. The shade list contains human-readable names that + can be used to label BLE-discovered devices. + + Raises ValueError on protocol/key errors. + Raises aiohttp.ClientError on network errors. + Raises asyncio.TimeoutError on timeout. + """ + hub_shades = await _fetch_shades_from_hub(hass, hub_url) + if not hub_shades: + raise ValueError("No shades found on the hub") + + session = async_get_clientsession(hass) + timeout = aiohttp.ClientTimeout(total=10) + + # GetShadeKey BLE request: sid=251, cid=18, seqId=1, data_len=0 + request_frame = struct.pack(" None: self._discovered_device: ConfigFlow.DiscoveredDevice | None = None self._discovered_devices: dict[str, ConfigFlow.DiscoveredDevice] = {} + self._manufacturer_data_hex: str = "" + self._device_name: str = "" + self._home_key: str = "" + self._hub_url: str = "" + self._hub_shades: list[HubShadeInfo] = [] + + def _create_entry(self) -> ConfigFlowResult: + """Create the config entry with collected data.""" + data: dict[str, str] = { + "manufacturer_data": self._manufacturer_data_hex, + CONF_HOME_KEY: self._home_key, + } + if self._hub_url: + data["hub_url"] = self._hub_url + return self.async_create_entry( + title=self._device_name, + data=data, + ) + + def _validate_manual_key( + self, user_input: dict[str, Any], errors: dict[str, str] + ) -> bool: + """Validate a manually entered hex key and store it. + + 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). + On skip, self._home_key is set to "". + """ + method = user_input.get("key_method", "skip") + + if method == "skip": + self._home_key = "" + return True + + if method == "manual": + return self._validate_manual_key(user_input, errors) + + if method != "hub": + return False + + hub_url = user_input.get("hub_url", "").rstrip("/") + _HUB_ERROR_MAP: dict[type[Exception], str] = { + aiohttp.ClientResponseError: "hub_http_error", + aiohttp.ClientConnectionError: "hub_connection_error", + TimeoutError: "hub_timeout", + ValueError: "hub_protocol_error", + } + try: + key, hub_shades = await _fetch_key_and_shades_from_hub(self.hass, hub_url) + except tuple(_HUB_ERROR_MAP) as ex: + errors["hub_url"] = _HUB_ERROR_MAP[type(ex)] + return False + + self._home_key = key.hex() + self._hub_url = hub_url + self._hub_shades = hub_shades + return True async def async_step_bluetooth( self, discovery_info: BluetoothServiceInfoBleak @@ -64,14 +288,17 @@ async def async_step_bluetooth_confirm( 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() - }, + self._manufacturer_data_hex = ( + self._discovered_device.discovery_info.manufacturer_data[MFCT_ID].hex() ) + self._device_name = self._discovered_device.name + + # Unencrypted shades can skip the homekey step entirely + if not _needs_encryption(self._manufacturer_data_hex): + await self._resolve_friendly_name() + return self._create_entry() + + return await self.async_step_homekey_bluetooth() self._set_confirm_only() @@ -80,29 +307,153 @@ async def async_step_bluetooth_confirm( description_placeholders={"name": self._discovered_device.name}, ) + async def async_step_homekey_bluetooth( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Configure homekey for a shade discovered via Bluetooth.""" + # Reuse an existing key if another shade was already configured + existing = self._existing_home_key() + if existing and user_input is None: + self._home_key = existing + await self._resolve_friendly_name() + return self._create_entry() + + errors: dict[str, str] = {} + + if user_input is not None and await self._validate_homekey_input( + user_input, errors + ): + # Use hub name for the entry title if available + friendly = self._hub_name_for(self._device_name) + if friendly: + self._device_name = friendly + return self._create_entry() + + return self.async_show_form( + step_id="homekey_bluetooth", + data_schema=_HOMEKEY_SCHEMA, + errors=errors, + description_placeholders={ + "name": self._device_name, + "hub_url_example": "http://powerview-g3.local", + }, + ) + + def _existing_entry_value(self, key: str) -> str: + """Return the first non-empty value for *key* across configured entries.""" + for entry in self._async_current_entries(): + if value := entry.data.get(key, ""): + return value + return "" + + def _existing_home_key(self) -> str: + """Return the home_key from any already-configured entry, or ''.""" + return self._existing_entry_value(CONF_HOME_KEY) + + async def _resolve_friendly_name(self) -> None: + """Try to resolve BLE device name to hub friendly name.""" + hub_url = self._hub_url or self._existing_entry_value("hub_url") + if not hub_url: + return + try: + shades = await _fetch_shades_from_hub(self.hass, hub_url) + for hs in shades: + if hs.ble_name == self._device_name: + self._device_name = hs.name + break + if not self._hub_url: + self._hub_url = hub_url + except (TimeoutError, aiohttp.ClientError, ValueError): + pass + + def _hub_name_for(self, ble_name: str) -> str | None: + """Return the human-readable hub name for a BLE name, or None.""" + for hs in self._hub_shades: + if hs.ble_name == ble_name: + return hs.name + return None + async def async_step_user( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: - """Handle the user step to pick discovered device.""" + """Handle the user step — reuse existing key or offer a menu.""" LOGGER.debug("user step") + existing = self._existing_home_key() + if existing: + self._home_key = existing + self._hub_url = self._hub_url or self._existing_entry_value("hub_url") + if self._hub_url and not self._hub_shades: + with contextlib.suppress( + TimeoutError, aiohttp.ClientError, ValueError + ): + self._hub_shades = await _fetch_shades_from_hub( + self.hass, self._hub_url + ) + return self.async_show_menu( + step_id="user", + menu_options=["select_device", "manual"], + ) + return await self.async_step_homekey() - 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] + def _build_selected_entries( + self, user_input: dict[str, Any] + ) -> list[dict[str, Any]]: + """Build config entry data for each selected shade address.""" + addresses: list[str] = user_input[CONF_ADDRESS] + if isinstance(addresses, str): + addresses = [addresses] - self.context["title_placeholders"] = {"name": self._discovered_device.name} + entries: list[dict[str, Any]] = [] + for address in addresses: + device = self._discovered_devices[address] + ble_name = device.name + name = self._hub_name_for(ble_name) or ble_name + mfct_hex = device.discovery_info.manufacturer_data[MFCT_ID].hex() + entry_data: dict[str, str] = { + "manufacturer_data": mfct_hex, + CONF_HOME_KEY: self._home_key, + } + if self._hub_url: + entry_data["hub_url"] = self._hub_url + entries.append( + { + "address": address, + "name": name, + "data": entry_data, + } + ) + return entries + + async def async_step_select_device( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Select one or more BLE-discovered shades, or fall through to manual.""" + LOGGER.debug("select_device step") + + if user_input is not None: + entries = self._build_selected_entries(user_input) - return self.async_create_entry( - title=self._discovered_device.name, - data={ - "manufacturer_data": self._discovered_device.discovery_info.manufacturer_data[ - MFCT_ID - ].hex() - }, + # Kick off auto-add flows for all but the last shade + await asyncio.gather( + *( + self.hass.config_entries.flow.async_init( + DOMAIN, + context={"source": "auto_add"}, + data=info, + ) + for info in entries[:-1] + ) ) + # Create the final entry normally (ends this flow) + last = entries[-1] + await self.async_set_unique_id(last["address"], raise_on_progress=False) + self._abort_if_unique_id_configured() + self._device_name = last["name"] + self._manufacturer_data_hex = last["data"]["manufacturer_data"] + self.context["title_placeholders"] = {"name": self._device_name} + return self._create_entry() + current_addresses = self._async_current_ids() for discovery_info in async_discovered_service_info(self.hass, False): address = discovery_info.address @@ -120,19 +471,96 @@ async def async_step_user( ) if not self._discovered_devices: - return self.async_abort(reason="no_devices_found") + return await self.async_step_manual() titles: list[SelectOptionDict] = [] for address, discovery in self._discovered_devices.items(): - titles.append({"value": address, "label": discovery.name}) + hub_name = self._hub_name_for(discovery.name) + label = f"{hub_name} ({discovery.name})" if hub_name else discovery.name + titles.append({"value": address, "label": label}) return self.async_show_form( - step_id="user", + step_id="select_device", data_schema=vol.Schema( { vol.Required(CONF_ADDRESS): SelectSelector( - SelectSelectorConfig(options=titles) + SelectSelectorConfig(options=titles, multiple=True) ) } ), ) + + async def async_step_auto_add( + self, discovery_info: dict[str, Any] + ) -> ConfigFlowResult: + """Handle a shade queued from multi-select for individual setup.""" + await self.async_set_unique_id(discovery_info["address"]) + self._abort_if_unique_id_configured() + + self._device_name = discovery_info["name"] + self._manufacturer_data_hex = discovery_info["data"]["manufacturer_data"] + self._home_key = discovery_info["data"].get(CONF_HOME_KEY, "") + self._hub_url = discovery_info["data"].get("hub_url", "") + + self.context["title_placeholders"] = {"name": self._device_name} + return await self.async_step_auto_add_confirm() + + async def async_step_auto_add_confirm( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Confirm adding a shade discovered via multi-select.""" + if user_input is not None: + return self._create_entry() + + self._set_confirm_only() + return self.async_show_form( + step_id="auto_add_confirm", + description_placeholders={"name": self._device_name}, + ) + + async def async_step_manual( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Handle manual entry of a BLE device address and name.""" + if user_input is not None: + address = user_input[CONF_ADDRESS].upper().strip() + self._device_name = user_input["ble_name"].strip() + await self.async_set_unique_id(address, raise_on_progress=False) + self._abort_if_unique_id_configured() + self.context["title_placeholders"] = {"name": self._device_name} + self._manufacturer_data_hex = "" + return self._create_entry() + + return self.async_show_form( + step_id="manual", + data_schema=vol.Schema( + { + vol.Required(CONF_ADDRESS): TextSelector( + TextSelectorConfig(type=TextSelectorType.TEXT) + ), + vol.Required("ble_name"): TextSelector( + TextSelectorConfig(type=TextSelectorType.TEXT) + ), + } + ), + ) + + async def async_step_homekey( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Configure homekey — collected before device selection.""" + errors: dict[str, str] = {} + + if user_input is not None and await self._validate_homekey_input( + user_input, errors + ): + return await self.async_step_select_device() + + return self.async_show_form( + step_id="homekey", + data_schema=_HOMEKEY_SCHEMA, + errors=errors, + description_placeholders={ + "hub_url_example": "http://powerview-g3.local", + }, + ) diff --git a/custom_components/hunterdouglas_powerview_ble/const.py b/custom_components/hunterdouglas_powerview_ble/const.py index d723594..0e02520 100644 --- a/custom_components/hunterdouglas_powerview_ble/const.py +++ b/custom_components/hunterdouglas_powerview_ble/const.py @@ -8,10 +8,7 @@ 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" # 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..10aabb8 100644 --- a/custom_components/hunterdouglas_powerview_ble/coordinator.py +++ b/custom_components/hunterdouglas_powerview_ble/coordinator.py @@ -12,27 +12,37 @@ from homeassistant.core import HomeAssistant, callback 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.""" 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._manuf_dat = data.get("manufacturer_data") self.dev_details: dict[str, str] = {} + self.velocity: int = 0 LOGGER.debug( "Initializing coordinator for %s (%s)", - ble_device.name, + self._friendly_name, ble_device.address, ) super().__init__( @@ -42,6 +52,18 @@ def __init__( bluetooth.BluetoothScanningMode.ACTIVE, ) + @property + def type_id(self) -> int | None: + """Return the shade type ID from manufacturer data.""" + if self._manuf_dat: + return int(bytes.fromhex(self._manuf_dat)[2]) + return None + + @property + def shade_capabilities(self) -> ShadeCapability: + """Return the shade capabilities based on type ID.""" + return get_shade_capabilities(self.type_id) + async def query_dev_info(self) -> None: """Receive detailed information from device.""" LOGGER.debug("%s: querying device info", self.name) @@ -50,24 +72,23 @@ async def query_dev_info(self) -> None: @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 + 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"), @@ -93,10 +114,8 @@ 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.api.set_ble_device(service_info.device) self.data = {ATTR_RSSI: service_info.rssi} if change == bluetooth.BluetoothChange.ADVERTISEMENT: self.data.update( diff --git a/custom_components/hunterdouglas_powerview_ble/cover.py b/custom_components/hunterdouglas_powerview_ble/cover.py index 4c25bf5..3d463d7 100644 --- a/custom_components/hunterdouglas_powerview_ble/cover.py +++ b/custom_components/hunterdouglas_powerview_ble/cover.py @@ -18,11 +18,11 @@ ) 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 .api import CLOSED_POSITION, OPEN_POSITION -from .const import DOMAIN, HOME_KEY, LOGGER +from .const import DOMAIN, LOGGER from .coordinator import PVCoordinator @@ -31,15 +31,17 @@ async def async_setup_entry( config_entry: ConfigEntry, async_add_entities: AddEntitiesCallback, ) -> None: - """Set up the demo cover platform.""" + """Set up the 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)) + caps = coordinator.shade_capabilities + + if caps.tilt_only: + entities: list[PowerViewCover] = [PowerViewCoverTiltOnly(coordinator)] + elif caps.has_tilt: + entities = [PowerViewCoverTilt(coordinator)] else: - entities.append(PowerViewCover(coordinator)) + entities = [PowerViewCover(coordinator)] async_add_entities(entities) @@ -62,7 +64,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,11 +75,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 - @property def is_opening(self) -> bool | None: # type: ignore[reportIncompatibleVariableOverride] """Return if the cover is opening or not.""" @@ -107,7 +104,7 @@ def is_closed(self) -> bool: # type: ignore[reportIncompatibleVariableOverride] 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 + self._coord.data.get("home_id") and not self._coord.api.has_key ) or self._coord.data.get("battery_charging"): return CoverEntityFeature(0) @@ -133,7 +130,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( @@ -153,7 +153,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 +166,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) @@ -227,7 +227,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: diff --git a/custom_components/hunterdouglas_powerview_ble/manifest.json b/custom_components/hunterdouglas_powerview_ble/manifest.json index eeab944..47a475c 100644 --- a/custom_components/hunterdouglas_powerview_ble/manifest.json +++ b/custom_components/hunterdouglas_powerview_ble/manifest.json @@ -16,5 +16,5 @@ "issue_tracker": "https://github.com/patman15/hdpv_ble/issues", "loggers": ["hunterdouglas_powerview_ble"], "requirements": ["cryptography>=43.0.0"], - "version": "0.23" + "version": "0.24" } diff --git a/custom_components/hunterdouglas_powerview_ble/number.py b/custom_components/hunterdouglas_powerview_ble/number.py new file mode 100644 index 0000000..1d76d95 --- /dev/null +++ b/custom_components/hunterdouglas_powerview_ble/number.py @@ -0,0 +1,69 @@ +"""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 +from .const import DOMAIN, LOGGER +from .coordinator import PVCoordinator + + +async def async_setup_entry( + _hass: HomeAssistant, + config_entry: ConfigEntryType, + async_add_entities: AddEntitiesCallback, +) -> None: + """Set up the velocity number entity.""" + + coordinator: PVCoordinator = config_entry.runtime_data + async_add_entities([PowerViewVelocity(coordinator)]) + + +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/strings.json b/custom_components/hunterdouglas_powerview_ble/strings.json index 19601be..5e3e3e7 100644 --- a/custom_components/hunterdouglas_powerview_ble/strings.json +++ b/custom_components/hunterdouglas_powerview_ble/strings.json @@ -2,10 +2,73 @@ "config": { "flow_title": "Setup {name}", "step": { + "user": { + "title": "Add PowerView Shade", + "menu_options": { + "select_device": "Select from discovered shades", + "manual": "Enter device details manually" + }, + "menu_option_descriptions": { + "select_device": "Choose from shades detected via Bluetooth nearby.", + "manual": "Enter a Bluetooth MAC address and device name directly, for example if a shade is out of range of discovery." + } + }, "bluetooth_confirm": { "description": "[%key:component::bluetooth::config::step::bluetooth_confirm::description%]" + }, + "homekey": { + "title": "Configure HomeKey", + "description": "All shades on a PowerView network share the same encryption key. The recommended method is to fetch it from your G3 hub. You can also enter the key manually, or skip if your shades are unencrypted.", + "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...)" + } + }, + "homekey_bluetooth": { + "title": "Configure HomeKey for {name}", + "description": "This shade uses encryption. The recommended method is to fetch the key from your G3 hub. You can also enter it manually, or skip (controls will be disabled until a key is configured).", + "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...)" + } + }, + "auto_add_confirm": { + "description": "Do you want to set up {name}?" + }, + "select_device": { + "title": "Select Shades", + "description": "Select the PowerView shades to add via Bluetooth.", + "data": { + "address": "Shades" + } + }, + "manual": { + "title": "Enter Device Details", + "description": "Enter the device details manually.", + "data": { + "address": "Bluetooth MAC address (e.g. AA:BB:CC:DD:EE:FF)", + "ble_name": "BLE device name (e.g. DUE:94ED)" + } } }, + "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%]", diff --git a/custom_components/hunterdouglas_powerview_ble/translations/en.json b/custom_components/hunterdouglas_powerview_ble/translations/en.json index 528686d..7830431 100644 --- a/custom_components/hunterdouglas_powerview_ble/translations/en.json +++ b/custom_components/hunterdouglas_powerview_ble/translations/en.json @@ -1,15 +1,75 @@ { "config": { - "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": { + "user": { + "title": "Add PowerView Shade", + "menu_options": { + "select_device": "Select from discovered shades", + "manual": "Enter device details manually" + }, + "menu_option_descriptions": { + "select_device": "Choose from shades detected via Bluetooth nearby.", + "manual": "Enter a Bluetooth MAC address and device name directly, for example if a shade is out of range of discovery." + } + }, "bluetooth_confirm": { "description": "Do you want to set up {name}?" + }, + "homekey": { + "title": "Configure HomeKey", + "description": "All shades on a PowerView network share the same encryption key. The recommended method is to fetch it from your G3 hub. You can also enter the key manually, or skip if your shades are unencrypted.", + "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...)" + } + }, + "homekey_bluetooth": { + "title": "Configure HomeKey for {name}", + "description": "This shade uses encryption. The recommended method is to fetch the key from your G3 hub. You can also enter it manually, or skip (controls will be disabled until a key is configured).", + "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...)" + } + }, + "select_device": { + "title": "Select Shades", + "description": "Select the PowerView shades to add via Bluetooth.", + "data": { + "address": "Shades" + } + }, + "manual": { + "title": "Enter Device Details", + "description": "Enter the device details manually.", + "data": { + "address": "Bluetooth MAC address (e.g. AA:BB:CC:DD:EE:FF)", + "ble_name": "BLE device name (e.g. DUE:94ED)" + } } + }, + "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" } } } 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']}'")