Compare commits
2 Commits
improve-co
...
shade-capa
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
acb2c5ff52 | ||
|
|
7a2bf2193a |
@@ -18,9 +18,10 @@ from .coordinator import PVCoordinator
|
|||||||
|
|
||||||
PLATFORMS: list[Platform] = [
|
PLATFORMS: list[Platform] = [
|
||||||
Platform.BINARY_SENSOR,
|
Platform.BINARY_SENSOR,
|
||||||
Platform.COVER,
|
|
||||||
Platform.SENSOR,
|
|
||||||
Platform.BUTTON,
|
Platform.BUTTON,
|
||||||
|
Platform.COVER,
|
||||||
|
Platform.NUMBER,
|
||||||
|
Platform.SENSOR,
|
||||||
]
|
]
|
||||||
|
|
||||||
type ConfigEntryType = ConfigEntry[PVCoordinator]
|
type ConfigEntryType = ConfigEntry[PVCoordinator]
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ import asyncio
|
|||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from enum import Enum
|
from enum import Enum
|
||||||
import time
|
import time
|
||||||
from typing import Final
|
from typing import Final, NamedTuple
|
||||||
|
|
||||||
from bleak import BleakClient
|
from bleak import BleakClient
|
||||||
from bleak.backends.device import BLEDevice
|
from bleak.backends.device import BLEDevice
|
||||||
@@ -58,6 +58,31 @@ SHADE_TYPE: Final[dict[int, str]] = {
|
|||||||
62: "Venetian, Tilt Anywhere",
|
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
|
OPEN_POSITION: Final[int] = 100
|
||||||
CLOSED_POSITION: Final[int] = 0
|
CLOSED_POSITION: Final[int] = 0
|
||||||
|
|
||||||
@@ -237,20 +262,20 @@ class PowerViewBLE:
|
|||||||
disconnect,
|
disconnect,
|
||||||
)
|
)
|
||||||
|
|
||||||
async def open(self) -> None:
|
async def open(self, velocity: int = 0x0) -> None:
|
||||||
"""Fully open cover."""
|
"""Fully open cover."""
|
||||||
LOGGER.debug("%s open", self.name)
|
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:
|
async def stop(self) -> None:
|
||||||
"""Stop device movement."""
|
"""Stop device movement."""
|
||||||
LOGGER.debug("%s stop", self.name)
|
LOGGER.debug("%s stop", self.name)
|
||||||
await self._cmd((ShadeCmd.STOP, b""))
|
await self._cmd((ShadeCmd.STOP, b""))
|
||||||
|
|
||||||
async def close(self) -> None:
|
async def close(self, velocity: int = 0x0) -> None:
|
||||||
"""Fully close cover."""
|
"""Fully close cover."""
|
||||||
LOGGER.debug("%s close", self.name)
|
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
|
# uint8_t scene#, uint8_t unknown
|
||||||
# open: scene 2
|
# open: scene 2
|
||||||
|
|||||||
@@ -12,7 +12,7 @@ from homeassistant.components.bluetooth.passive_update_coordinator import (
|
|||||||
from homeassistant.core import HomeAssistant, callback
|
from homeassistant.core import HomeAssistant, callback
|
||||||
from homeassistant.helpers.device_registry import CONNECTION_BLUETOOTH, DeviceInfo
|
from homeassistant.helpers.device_registry import CONNECTION_BLUETOOTH, DeviceInfo
|
||||||
|
|
||||||
from .api import SHADE_TYPE, PowerViewBLE
|
from .api import SHADE_TYPE, PowerViewBLE, ShadeCapability, get_shade_capabilities
|
||||||
from .const import ATTR_RSSI, CONF_HOME_KEY, DOMAIN, LOGGER
|
from .const import ATTR_RSSI, CONF_HOME_KEY, DOMAIN, LOGGER
|
||||||
|
|
||||||
|
|
||||||
@@ -38,6 +38,7 @@ class PVCoordinator(PassiveBluetoothDataUpdateCoordinator):
|
|||||||
self.data: dict[str, int | float | bool] = {}
|
self.data: dict[str, int | float | bool] = {}
|
||||||
self._manuf_dat = data.get("manufacturer_data")
|
self._manuf_dat = data.get("manufacturer_data")
|
||||||
self.dev_details: dict[str, str] = {}
|
self.dev_details: dict[str, str] = {}
|
||||||
|
self.velocity: int = 0
|
||||||
|
|
||||||
LOGGER.debug(
|
LOGGER.debug(
|
||||||
"Initializing coordinator for %s (%s)",
|
"Initializing coordinator for %s (%s)",
|
||||||
@@ -51,6 +52,18 @@ class PVCoordinator(PassiveBluetoothDataUpdateCoordinator):
|
|||||||
bluetooth.BluetoothScanningMode.ACTIVE,
|
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:
|
async def query_dev_info(self) -> None:
|
||||||
"""Receive detailed information from device."""
|
"""Receive detailed information from device."""
|
||||||
LOGGER.debug("%s: querying device info", self.name)
|
LOGGER.debug("%s: querying device info", self.name)
|
||||||
@@ -70,12 +83,12 @@ class PVCoordinator(PassiveBluetoothDataUpdateCoordinator):
|
|||||||
configuration_url=None,
|
configuration_url=None,
|
||||||
manufacturer="Hunter Douglas",
|
manufacturer="Hunter Douglas",
|
||||||
model=(
|
model=(
|
||||||
str(SHADE_TYPE.get(int(bytes.fromhex(self._manuf_dat)[2]), "unknown"))
|
str(SHADE_TYPE.get(self.type_id, "unknown"))
|
||||||
if self._manuf_dat
|
if self.type_id is not None
|
||||||
else None
|
else None
|
||||||
),
|
),
|
||||||
model_id=(
|
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"),
|
serial_number=self.dev_details.get("serial_nr"),
|
||||||
sw_version=self.dev_details.get("sw_rev"),
|
sw_version=self.dev_details.get("sw_rev"),
|
||||||
|
|||||||
@@ -18,7 +18,7 @@ from homeassistant.components.cover import (
|
|||||||
)
|
)
|
||||||
from homeassistant.config_entries import ConfigEntry
|
from homeassistant.config_entries import ConfigEntry
|
||||||
from homeassistant.core import HomeAssistant
|
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 homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||||
|
|
||||||
from .api import CLOSED_POSITION, OPEN_POSITION
|
from .api import CLOSED_POSITION, OPEN_POSITION
|
||||||
@@ -31,15 +31,17 @@ async def async_setup_entry(
|
|||||||
config_entry: ConfigEntry,
|
config_entry: ConfigEntry,
|
||||||
async_add_entities: AddEntitiesCallback,
|
async_add_entities: AddEntitiesCallback,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Set up the demo cover platform."""
|
"""Set up the cover platform."""
|
||||||
|
|
||||||
coordinator: PVCoordinator = config_entry.runtime_data
|
coordinator: PVCoordinator = config_entry.runtime_data
|
||||||
model: Final[str | None] = coordinator.dev_details.get("model")
|
caps = coordinator.shade_capabilities
|
||||||
entities: list[PowerViewCover] = []
|
|
||||||
if model == "39":
|
if caps.tilt_only:
|
||||||
entities.append(PowerViewCoverTiltOnly(coordinator))
|
entities: list[PowerViewCover] = [PowerViewCoverTiltOnly(coordinator)]
|
||||||
|
elif caps.has_tilt:
|
||||||
|
entities = [PowerViewCoverTilt(coordinator)]
|
||||||
else:
|
else:
|
||||||
entities.append(PowerViewCover(coordinator))
|
entities = [PowerViewCover(coordinator)]
|
||||||
|
|
||||||
async_add_entities(entities)
|
async_add_entities(entities)
|
||||||
|
|
||||||
@@ -73,11 +75,6 @@ class PowerViewCover(PassiveBluetoothCoordinatorEntity[PVCoordinator], CoverEnti
|
|||||||
)
|
)
|
||||||
super().__init__(coordinator)
|
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
|
@property
|
||||||
def is_opening(self) -> bool | None: # type: ignore[reportIncompatibleVariableOverride]
|
def is_opening(self) -> bool | None: # type: ignore[reportIncompatibleVariableOverride]
|
||||||
"""Return if the cover is opening or not."""
|
"""Return if the cover is opening or not."""
|
||||||
@@ -133,7 +130,10 @@ class PowerViewCover(PassiveBluetoothCoordinatorEntity[PVCoordinator], CoverEnti
|
|||||||
return
|
return
|
||||||
self._target_position = round(target_position)
|
self._target_position = round(target_position)
|
||||||
try:
|
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()
|
self.async_write_ha_state()
|
||||||
except BleakError as err:
|
except BleakError as err:
|
||||||
LOGGER.error(
|
LOGGER.error(
|
||||||
@@ -153,7 +153,7 @@ class PowerViewCover(PassiveBluetoothCoordinatorEntity[PVCoordinator], CoverEnti
|
|||||||
return
|
return
|
||||||
try:
|
try:
|
||||||
self._target_position = OPEN_POSITION
|
self._target_position = OPEN_POSITION
|
||||||
await self._coord.api.open()
|
await self._coord.api.open(velocity=self._coord.velocity)
|
||||||
self.async_write_ha_state()
|
self.async_write_ha_state()
|
||||||
except BleakError as err:
|
except BleakError as err:
|
||||||
LOGGER.error("Failed to open cover '%s': %s", self.name, err)
|
LOGGER.error("Failed to open cover '%s': %s", self.name, err)
|
||||||
@@ -166,7 +166,7 @@ class PowerViewCover(PassiveBluetoothCoordinatorEntity[PVCoordinator], CoverEnti
|
|||||||
return
|
return
|
||||||
try:
|
try:
|
||||||
self._target_position = CLOSED_POSITION
|
self._target_position = CLOSED_POSITION
|
||||||
await self._coord.api.close()
|
await self._coord.api.close(velocity=self._coord.velocity)
|
||||||
self.async_write_ha_state()
|
self.async_write_ha_state()
|
||||||
except BleakError as err:
|
except BleakError as err:
|
||||||
LOGGER.error("Failed to close cover '%s': %s", self.name, err)
|
LOGGER.error("Failed to close cover '%s': %s", self.name, err)
|
||||||
@@ -227,7 +227,9 @@ class PowerViewCoverTilt(PowerViewCover):
|
|||||||
|
|
||||||
try:
|
try:
|
||||||
await self._coord.api.set_position(
|
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()
|
self.async_write_ha_state()
|
||||||
except BleakError as err:
|
except BleakError as err:
|
||||||
|
|||||||
69
custom_components/hunterdouglas_powerview_ble/number.py
Normal file
69
custom_components/hunterdouglas_powerview_ble/number.py
Normal file
@@ -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()
|
||||||
Reference in New Issue
Block a user