Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions custom_components/hunterdouglas_powerview_ble/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,10 @@

PLATFORMS: list[Platform] = [
Platform.BINARY_SENSOR,
Platform.BUTTON,
Platform.COVER,
Platform.NUMBER,
Platform.SENSOR,
Platform.BUTTON,
]

type ConfigEntryType = ConfigEntry[PVCoordinator]
Expand All @@ -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:
Expand Down
59 changes: 43 additions & 16 deletions custom_components/hunterdouglas_powerview_ble/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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()
Expand All @@ -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."""
Expand All @@ -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."""
Expand Down Expand Up @@ -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")
Expand All @@ -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
Expand Down Expand Up @@ -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,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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__(
Expand Down
Loading