mirrors/hdpv_ble
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: linting and formatting

Browse Source
This commit is contained in:
Richard Mann
2026-04-06 15:13:21 +10:00
parent af08d18d62
commit 652337e32c
6 changed files with 128 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
custom_components/hunterdouglas_powerview_ble/binary_sensor.py
View File

@@ -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__(

111
custom_components/hunterdouglas_powerview_ble/config_flow.py
View File

@@ -2,8 +2,8 @@
import asyncio
import base64
import struct
from dataclasses import dataclass
import struct
from typing import Any
import aiohttp
@@ -114,6 +114,9 @@ async def _fetch_key_and_shades_from_hub(
) as resp:
resp.raise_for_status()
result = await resp.json(content_type=None)
except (TimeoutError, aiohttp.ClientError) as ex:
last_error = ex
continue
responses = result.get("responses", [])
if len(responses) != 1 or "hex" not in responses[0]:
@@ -131,9 +134,6 @@ async def _fetch_key_and_shades_from_hub(
if len(key_data) != 16:
continue
return key_data, hub_shades
except (aiohttp.ClientError, asyncio.TimeoutError) as ex:
last_error = ex
continue
raise ValueError(f"No reachable shade returned a valid key: {last_error}")
@@ -205,6 +205,27 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
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:
@@ -220,40 +241,28 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return True
if method == "manual":
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
return self._validate_manual_key(user_input, errors)
if method != "hub":
return False
if method == "hub":
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
)
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
except aiohttp.ClientResponseError:
errors["hub_url"] = "hub_http_error"
except aiohttp.ClientConnectionError:
errors["hub_url"] = "hub_connection_error"
except (asyncio.TimeoutError, TimeoutError):
errors["hub_url"] = "hub_timeout"
except ValueError:
errors["hub_url"] = "hub_protocol_error"
return False
async def async_step_bluetooth(
self, discovery_info: BluetoothServiceInfoBleak
@@ -310,8 +319,9 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors: dict[str, str] = {}
if user_input is not None:
if await self._validate_homekey_input(user_input, errors):
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:
@@ -349,7 +359,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
break
if not self._hub_url:
self._hub_url = hub_url
except (aiohttp.ClientError, asyncio.TimeoutError, ValueError):
except (TimeoutError, aiohttp.ClientError, ValueError):
pass
def _hub_name_for(self, ble_name: str) -> str | None:
@@ -370,18 +380,14 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return await self.async_step_select_device()
return await self.async_step_homekey()
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:
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]
# Build entry info for every selected shade
entries: list[dict[str, Any]] = []
for address in addresses:
device = self._discovered_devices[address]
@@ -401,16 +407,28 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"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)
# Kick off auto-add flows for all but the last shade
await asyncio.gather(*(
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]
@@ -518,8 +536,9 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Configure homekey — collected before device selection."""
errors: dict[str, str] = {}
if user_input is not None:
if await self._validate_homekey_input(user_input, errors):
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(

9
custom_components/hunterdouglas_powerview_ble/coordinator.py
View File

@@ -20,7 +20,10 @@ 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."""
@@ -28,7 +31,9 @@ class PVCoordinator(PassiveBluetoothDataUpdateCoordinator):
self._mac = ble_device.address
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""
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")

2
custom_components/hunterdouglas_powerview_ble/cover.py
View File

@@ -36,7 +36,7 @@ async def async_setup_entry(
coordinator: PVCoordinator = config_entry.runtime_data
model: Final[str | None] = coordinator.dev_details.get("model")
entities: list[PowerViewCover] = []
if model in ["39"]:
if model == "39":
entities.append(PowerViewCoverTiltOnly(coordinator))
else:
entities.append(PowerViewCover(coordinator))

4
scripts/extract_gateway3_homekey.py
View File

@@ -61,7 +61,9 @@ def get_shade_key(hub: str, ble_name) -> bytes:
response: Final[bytes] = bytes.fromhex(responses[0]["hex"])
dec_resp: Final[dict[str, Any]] = decode_response(response)
if dec_resp["errorCode"] != 0:
raise ValueError(f"BLE errorCode={dec_resp['errorCode']} data={dec_resp['data'].hex()}")
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"]