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

first usable version

Browse Source
This commit is contained in:
patman15
2024-08-07 21:28:51 +02:00
parent 00692b78d7
commit 37f78606a8
16 changed files with 584 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
custom_components/hunterdouglas_powerview_ble/__init__.py
View File

@@ -9,10 +9,11 @@ from homeassistant.exceptions import ConfigEntryError, ConfigEntryNotReady
from .const import DOMAIN, LOGGER
from .coordinator import PVCoordinator
PLATFORMS: list[Platform] = [Platform.COVER]
PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR, Platform.COVER, Platform.SENSOR]
type ConfigEntryType = ConfigEntry[PVCoordinator]
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))
@@ -29,18 +30,13 @@ 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)
# Query the device the first time, initialise coordinator.data
try:
await coordinator.async_config_entry_first_refresh()
except ConfigEntryNotReady:
# Ignore, e.g. timeouts, to gracefully handle connection issues
LOGGER.warning("Failed to initialize PowerView device %s, continuing", ble_device.name)
coordinator = PVCoordinator(hass, ble_device, entry.data.copy())
# Insert the coordinator in the global registry
hass.data.setdefault(DOMAIN, {})
entry.runtime_data = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(coordinator.async_start())
return True
@@ -52,7 +48,9 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntryType) -> boo
return unload_ok
async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntryType) -> bool:
async def async_migrate_entry(
_hass: HomeAssistant, config_entry: ConfigEntryType
) -> bool:
"""Migrate old entry."""
if config_entry.version > 1:

209
custom_components/hunterdouglas_powerview_ble/api.py
View File

@@ -1,48 +1,105 @@
"""Hunter Douglas PowerView BLE API."""
import asyncio
from dataclasses import dataclass
from enum import Enum
import time
from bleak import BleakClient
from bleak.backends.device import BLEDevice
from bleak.exc import BleakError
from bleak.uuids import normalize_uuid_str
from .const import LOGGER
from homeassistant.components.cover import ATTR_CURRENT_POSITION
from .const import LOGGER, TIMEOUT, UUID
ATTR_ACTIVITY = "activity"
class shade_cmd(Enum):
SHADE_TYPE: dict[int, str] = {
1: "Designer Roller",
4: "Roman",
5: "Bottom Up",
6: "Duette",
10: "Duette and Applause SkyLift",
19: "Provenance Woven Wood",
31: "Vignette",
32: "Vignette",
42: "M25T Roller Blind",
49: "AC Roller",
52: "Banded Shades",
53: "Sonnette",
84: "Vignette",
}
OPEN_POSITION = 100
CLOSED_POSITION = 0
POWER_LEVELS: dict[int, int] = {
4: 100, # 4 is hardwired
3: 100, # 3 = 100% to 51% power remaining
2: 50, # 2 = 50% to 21% power remaining
1: 20, # 1 = 20% or less power remaining
0: 0, # 0 = No power remaining
}
class ShadeCmd(Enum):
"""The PowerView cover commands."""
set_position = 0x01F7
stop = 0xB8F7
activate_scene = 0xBAF7
SET_POSITION = 0x01F7
STOP = 0xB8F7
ACTIVATE_SCENE = 0xBAF7
@dataclass
class DeviceInfo:
"""Dataclass holding available PowerView device information."""
manufacturer: str = ""
model: str = ""
serial_nr: str = ""
hw_rev: str = ""
fw_rev: str = ""
sw_rev: str = ""
battery_level: int = 0
class PowerViewBLE:
"""Class to handle connection to PowerView remote device."""
UUID_SERVICE = "0000fdc1-0000-1000-8000-00805f9b34fb"
UUID_TX = "cafe1001-c0ff-ee01-8000-A110CA7AB1E0"
UUID_COV_SERVICE = UUID
UUID_TX = "cafe1001-c0ff-ee01-8000-a110ca7ab1e0"
UUID_DEV_SERVICE = normalize_uuid_str("180a")
UUID_BAT_SERVICE = normalize_uuid_str("180f")
def __init__(self, ble_device: BLEDevice) -> None:
"""Initialize device API via Bluetooth."""
self._ble_device: BLEDevice = ble_device
self.name = self._ble_device.name
self.seqcnt: int = 0
self.seqcnt: int = 1
self._client: BleakClient | None = None
self._data_event = asyncio.Event()
self._data: bytearray
self._info: DeviceInfo = DeviceInfo()
async def _wait_event(self) -> None:
await self._data_event.wait()
self._data_event.clear()
@property
def info(self) -> DeviceInfo:
"""Return device information, e.g. SW version."""
return self._info
@property
def is_connected(self) -> bool:
"""Return whether remote device is connected."""
return self._client is not None and self._client.is_connected
# general cmd: uint16_t cmd, uint8_t seqID, uint8_t data_len
async def _cmd(self, cmd: shade_cmd, data: bytearray) -> None:
async def _cmd(self, cmd: ShadeCmd, data: bytearray) -> None:
try:
await self._connect()
assert self._client is not None, "missing BT client"
@@ -53,50 +110,127 @@ class PowerViewBLE:
)
+ data
)
self.seqcnt += 1
self._data_event.clear()
LOGGER.debug("sending cmd: %s", tx_data)
await self._client.write_gatt_char(self.UUID_TX, tx_data, False)
finally:
await self._disconnect()
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)
except TimeoutError as ex:
raise TimeoutError("device operation timed out") from ex
finally:
await self._client.disconnect()
except Exception as ex:
LOGGER.debug("Error: %s - %s", type(ex).__name__, ex)
raise
@staticmethod
def dec_manufacturer_data(data: bytearray) -> list[tuple[str, float]]:
"""Decode manufacturer data from BLE advertisement V2."""
if len(data) != 9:
LOGGER.debug("not a V2 record!")
return []
pos = int.from_bytes(data[3:5], byteorder="little")
return [
(ATTR_CURRENT_POSITION, ((pos >> 2) / 10)),
("home_id", int.from_bytes(data[0:2], byteorder="little")),
("type_id", int.from_bytes(data[2:3])),
("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)),
]
# position cmd: uint16_t pos1, uint16_t pos2, uint16_t pos3, uint16_t tilt, uint8_t velocity
async def set_position(self, value: int) -> None:
"""Set position of device."""
LOGGER.debug("%s setting position to %i", self.name, value)
await self._cmd(
shade_cmd.set_position,
ShadeCmd.SET_POSITION,
bytearray(
int.to_bytes(value * 100, 2, byteorder="little")
+ bytes([0x00, 0x80, 0x00, 0x80, 0x00, 0x80, 0x0])
),
)
# TODO: currently HA / Bleak communication is too slow to make that function reasonable
# def stop(self) -> None:
# """Stop device movement."""
# LOGGER.debug("%s stop")
async def stop(self) -> None:
"""Stop device movement."""
LOGGER.debug("%s stop", self.name)
await self._cmd(ShadeCmd.STOP, bytearray(b""))
# uint8_t scene#, uint8_t unknown
# open: scene 2
# close: scene 3
async def activate_scene(self, idx: int) -> None:
"""Stop device movement."""
"""Activate stored scene."""
LOGGER.debug("%s set scene #%i", self.name, idx)
await self._cmd(
shade_cmd.activate_scene,
bytearray(
int.to_bytes(idx, 1, byteorder="little")
+ bytes([0xA2])
),
ShadeCmd.ACTIVATE_SCENE,
bytearray(int.to_bytes(idx, 1, byteorder="little") + bytes([0xA2])),
)
def _verify_response(self, data: bytearray, seq_nr: int, cmd: ShadeCmd) -> bool:
"""Verify shade response data."""
if len(data) < 4:
LOGGER.warning("Message too short")
return False
if int.from_bytes(data[0:2], byteorder="little") != cmd.value & 0xFFEF:
LOGGER.warning("Response to wrong command")
return False
if int(data[2]) != seq_nr:
LOGGER.warning("Wrong sequence number")
return False
if int(data[3]) != 1:
LOGGER.warning("Wrong response data length")
return False
if int(data[4] != 0):
LOGGER.warning("Return code type error")
return False
return True
async def query_dev_info(self) -> dict[str, str]:
"""Return detailed device information."""
data: dict[str, str] = {}
uuids: dict[str, str] = {
"manufacturer": "2a29",
"model": "2a24",
"serial_nr": "2a25",
"hw_rev": "2a27",
"fw_rev": "2a26",
"sw_rev": "2a28",
}
try:
await self._connect()
assert self._client is not None
for key, uuid in uuids.items():
LOGGER.debug("querying %s(%s)", key, uuid)
data[key] = (
(await self._client.read_gatt_char(normalize_uuid_str(uuid)))
.copy()
.decode("UTF-8")
)
except BleakError as ex:
LOGGER.debug("%s error: %s", self.name, ex)
return {}
finally:
await self._disconnect()
LOGGER.debug("%s device data: %s", self.name, data)
return data
def _on_disconnect(self, client: BleakClient) -> None:
"""Disconnect callback function."""
LOGGER.debug("Disconnected from %s", client.address)
def _notification_handler(self, sender, data: bytearray) -> None:
def _notification_handler(self, _sender, data: bytearray) -> None:
LOGGER.debug("%s received BLE data: %s", self.name, data)
self._data = data
self._data_event.set()
async def _connect(self) -> None:
@@ -104,25 +238,32 @@ class PowerViewBLE:
if not self.is_connected:
LOGGER.debug("Connecting %s", self.name)
self._client = BleakClient(
self._ble_device,
disconnected_callback=self._on_disconnect,
services=[self.UUID_SERVICE],
)
await self._client.connect()
start = time.time()
if not isinstance(self._client, BleakClient):
self._client = BleakClient(
self._ble_device,
disconnected_callback=self._on_disconnect,
services=[
self.UUID_COV_SERVICE,
# self.UUID_DEV_SERVICE,
# self.UUID_BAT_SERVICE,
],
)
await self._client.connect() # dangerous_use_bleak_cache = True
LOGGER.debug("\tconnect took %i", time.time() - start)
await self._client.start_notify(self.UUID_TX, self._notification_handler)
# await self._query_dev_info()
else:
LOGGER.debug("%s already connected", self.name)
async def _disconnect(self) -> None:
"""Disconnect the device and stop notifications."""
if self._client and self.is_connected:
if self._client is not None and self.is_connected:
LOGGER.debug("Disconnecting device %s", self.name)
try:
self._data_event.clear()
await self._client.disconnect()
except BleakError:
LOGGER.warning("Disconnect failed!")
self._client = None

62
custom_components/hunterdouglas_powerview_ble/binary_sensor.py Normal file
View File

@@ -0,0 +1,62 @@
"""Support for Hunter Douglas PowerView binary sensors."""
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
BinarySensorEntityDescription,
)
from homeassistant.components.bluetooth.passive_update_coordinator import (
PassiveBluetoothCoordinatorEntity,
)
from homeassistant.const import ATTR_BATTERY_CHARGING
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
from .coordinator import PVCoordinator
BINARY_SENSOR_TYPES: list[BinarySensorEntityDescription] = [
BinarySensorEntityDescription(
key=ATTR_BATTERY_CHARGING,
translation_key=ATTR_BATTERY_CHARGING,
device_class=BinarySensorDeviceClass.BATTERY_CHARGING,
)
]
async def async_setup_entry(
_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))]
)
class PVBinarySensor(PassiveBluetoothCoordinatorEntity[PVCoordinator], BinarySensorEntity): # type: ignore[reportIncompatibleMethodOverride]
"""The generic PV binary sensor implementation."""
def __init__(
self,
coord: PVCoordinator,
descr: BinarySensorEntityDescription,
unique_id: str,
) -> None:
"""Intialize PV binary sensor."""
self._attr_unique_id = f"{DOMAIN}-{unique_id}-{descr.key}"
self._attr_device_info = coord.device_info
self._attr_has_entity_name = True
self.entity_description = descr
super().__init__(coord)
@property
def is_on(self) -> bool | None: # type: ignore[reportIncompatibleVariableOverride]
"""Handle updated data from the coordinator."""
return bool(self.coordinator.data.get(self.entity_description.key))

4
custom_components/hunterdouglas_powerview_ble/config_flow.py
View File

@@ -63,7 +63,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
if user_input is not None:
return self.async_create_entry(
title=self._discovered_device.name,
data={},
data={"manufacturer_data": self._discovered_device.discovery_info.manufacturer_data[MFCT_ID].hex()},
)
self._set_confirm_only()
@@ -89,7 +89,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_create_entry(
title=self._discovered_device.name,
data={},
data={"manufacturer_data": self._discovered_device.discovery_info.manufacturer_data[MFCT_ID].hex()},
)
current_addresses = self._async_current_ids()

7
custom_components/hunterdouglas_powerview_ble/const.py
View File

@@ -2,6 +2,8 @@
import logging
from bleak.uuids import normalize_uuid_str
# from homeassistant.const import ( # noqa: F401
# ATTR_BATTERY_CHARGING,
# ATTR_BATTERY_LEVEL,
@@ -12,10 +14,9 @@ import logging
DOMAIN = "hunterdouglas_powerview_ble"
LOGGER = logging.getLogger(__package__)
UPDATE_INTERVAL = 30 # in seconds
UUID = "0000fdc1-0000-1000-8000-00805f9b34fb"
UUID = normalize_uuid_str("fdc1")
MFCT_ID = 2073
TIMEOUT = 15
# attributes (do not change)
ATTR_RSSI = "rssi"

113
custom_components/hunterdouglas_powerview_ble/coordinator.py
View File

@@ -1,90 +1,97 @@
"""Home Assistant coordinator for Hunter Douglas PowerView (BLE) integration."""
from datetime import timedelta
from typing import Any
from bleak.backends.device import BLEDevice
from bleak.exc import BleakError
from homeassistant.components import bluetooth
from homeassistant.components.bluetooth import DOMAIN as BLUETOOTH_DOMAIN
from homeassistant.const import ATTR_NAME
from homeassistant.core import HomeAssistant
from homeassistant.components.bluetooth.passive_update_coordinator import (
PassiveBluetoothDataUpdateCoordinator,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import CONNECTION_BLUETOOTH, DeviceInfo
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .api import PowerViewBLE
from .const import ATTR_RSSI, DOMAIN, LOGGER, UPDATE_INTERVAL
from .api import SHADE_TYPE, PowerViewBLE
from .const import ATTR_RSSI, DOMAIN, LOGGER
class PVCoordinator(DataUpdateCoordinator[dict[str, Any]]):
class PVCoordinator(PassiveBluetoothDataUpdateCoordinator):
"""Update coordinator for a battery management system."""
def __init__(
self,
hass: HomeAssistant,
ble_device: BLEDevice,
self, hass: HomeAssistant, ble_device: BLEDevice, data: dict[str, Any]
) -> None:
"""Initialize BMS data coordinator."""
assert ble_device.name is not None
super().__init__(
hass=hass,
logger=LOGGER,
name=ble_device.name,
update_interval=timedelta(seconds=UPDATE_INTERVAL),
always_update=False, # only update when sensor value has changed
)
self._mac = ble_device.address
self.api = PowerViewBLE(ble_device)
self.data: dict[str, int | float | bool] = {}
self._manuf_dat = data.get("manufacturer_data")
self.dev_details: dict[str, str] = {}
LOGGER.debug(
"Initializing coordinator for %s (%s)",
ble_device.name,
ble_device.address,
)
if service_info := bluetooth.async_last_service_info(
self.hass, address=self._mac, connectable=True
):
LOGGER.debug("device data: %s", service_info.as_dict())
super().__init__(
hass,
LOGGER,
ble_device.address,
bluetooth.BluetoothScanningMode.ACTIVE,
)
self.device_info = DeviceInfo(
async def _get_device_info(self) -> None:
self.dev_details = await self.api.query_dev_info()
@property
def device_info(self) -> DeviceInfo:
"""Return detailed device information for GUI."""
LOGGER.debug("device_info, %s", self.dev_details)
return DeviceInfo(
identifiers={
(DOMAIN, ble_device.name),
(BLUETOOTH_DOMAIN, ble_device.address),
(DOMAIN, self.name),
(BLUETOOTH_DOMAIN, self.address),
},
connections={(CONNECTION_BLUETOOTH, ble_device.address)},
name=ble_device.name,
connections={(CONNECTION_BLUETOOTH, self.address)},
name=self.name,
configuration_url=None,
# properties used in GUI:
manufacturer="Hunter Douglas",
model="shade",
model=(
str(SHADE_TYPE.get(int(bytes.fromhex(self._manuf_dat)[2]), "unknown"))
if self._manuf_dat
else None
),
model_id=(
str(bytes.fromhex(self._manuf_dat)[2]) if self._manuf_dat 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"),
)
@property
def address(self) -> str:
"""Return MAC address of remote device."""
return self._mac
@callback
def _async_handle_bluetooth_event(
self,
service_info: bluetooth.BluetoothServiceInfoBleak,
change: bluetooth.BluetoothChange,
) -> None:
"""Handle a Bluetooth event."""
async def _async_update_data(self) -> dict[str, Any]:
"""Return the latest data from the device."""
LOGGER.debug("%s data update", self.device_info.get(ATTR_NAME))
# if not self.dev_details:
# self.hass.async_create_task(self._get_device_info())
try:
info = {} # await self._device.async_update()
except TimeoutError:
LOGGER.debug("Device communication timeout")
raise
except BleakError as err:
raise UpdateFailed(
f"device communicating failed: {err!s} ({type(err).__name__})"
) from err
if (
service_info := bluetooth.async_last_service_info(
self.hass, address=self._mac, connectable=True
LOGGER.debug("BLE event %s: %s", change, service_info.manufacturer_data)
self.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""))
)
)
) is not None:
info.update({ATTR_RSSI: service_info.rssi})
LOGGER.debug("data sample %s", info)
return info
LOGGER.debug("data sample %s", self.data)
super()._async_handle_bluetooth_event(service_info, change)

77
custom_components/hunterdouglas_powerview_ble/cover.py
View File

@@ -2,6 +2,9 @@
from typing import Any
from homeassistant.components.bluetooth.passive_update_coordinator import (
PassiveBluetoothCoordinatorEntity,
)
from homeassistant.components.cover import (
ATTR_CURRENT_POSITION,
ATTR_POSITION,
@@ -10,17 +13,17 @@ from homeassistant.components.cover import (
CoverEntityFeature,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_IDENTIFIERS, ATTR_NAME
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo, format_mac
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .api import CLOSED_POSITION, OPEN_POSITION
from .const import DOMAIN, LOGGER
from .coordinator import PVCoordinator
async def async_setup_entry(
hass: HomeAssistant,
_hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
@@ -30,7 +33,7 @@ async def async_setup_entry(
async_add_entities([PowerViewCover(coordinator)])
class PowerViewCover(CoverEntity):
class PowerViewCover(PassiveBluetoothCoordinatorEntity[PVCoordinator], CoverEntity): # type: ignore[reportIncompatibleVariableOverride]
"""Representation of a powerview shade."""
_attr_has_entity_name = True
@@ -39,9 +42,8 @@ class PowerViewCover(CoverEntity):
CoverEntityFeature.OPEN
| CoverEntityFeature.CLOSE
| CoverEntityFeature.SET_POSITION
# | CoverEntityFeature.STOP
| CoverEntityFeature.STOP
)
_attr_current_cover_position: int | None = None
def __init__(
self,
@@ -51,44 +53,69 @@ class PowerViewCover(CoverEntity):
self._attr_name = CoverDeviceClass.SHADE
self._coord = coordinator
self._attr_device_info = self._coord.device_info
self._target_position: int | None = None
self._attr_unique_id = (
f"{DOMAIN}_{format_mac(self._coord.address)}_{CoverDeviceClass.SHADE}"
)
self._attr_current_cover_position: int | None = 0
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."""
return bool(self._coord.data.get("is_opening"))
@property
def is_closing(self) -> bool | None: # type: ignore[reportIncompatibleVariableOverride]
"""Return if the cover is closing or not."""
return bool(self._coord.data.get("is_closing"))
@property
def is_closed(self) -> bool: # type: ignore[reportIncompatibleVariableOverride]
"""Return if the cover is closed."""
return self._attr_current_cover_position == 0
return self.current_cover_position == CLOSED_POSITION
@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") or self._coord.data.get("battery_charging"):
return CoverEntityFeature(0)
return super().supported_features
@property
def current_cover_position(self) -> int | None: # type: ignore[reportIncompatibleVariableOverride]
"""Return current position of cover.
None is unknown, 0 is closed, 100 is fully open.
"""
if ATTR_CURRENT_POSITION in self._coord.data:
pos = self._coord.data.get(ATTR_CURRENT_POSITION)
return int(pos) if pos is not None else None
return None
async def async_set_cover_position(self, **kwargs: Any) -> None:
"""Move the cover to a specific position."""
if pos := kwargs[ATTR_POSITION]:
LOGGER.debug("set cover to position %i", pos)
try:
await self._coord.api.set_position(pos)
self._attr_current_cover_position = pos
except TimeoutError:
pass
self._target_position = kwargs.get(ATTR_POSITION, None)
if self._target_position is not None:
LOGGER.debug("set cover to position %i", self._target_position)
await self._coord.api.set_position(self._target_position)
async def async_open_cover(self, **kwargs: Any) -> None:
"""Open the cover."""
LOGGER.debug("open cover")
try:
await self._coord.api.activate_scene(2)
self._attr_current_cover_position = 100
except TimeoutError:
pass
await self._coord.api.set_position(OPEN_POSITION)
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close the cover tilt."""
LOGGER.debug("close cover")
try:
await self._coord.api.activate_scene(3)
self._attr_current_cover_position = 0
except TimeoutError:
pass
await self._coord.api.set_position(CLOSED_POSITION)
async def async_stop_cover(self, **kwargs: Any) -> None:
"""Stop the cover."""
LOGGER.debug("stop cover")
await self._coord.api.stop()

4
custom_components/hunterdouglas_powerview_ble/manifest.json
View File

@@ -5,7 +5,7 @@
{
"service_uuid": "0000fdc1-0000-1000-8000-00805f9b34fb",
"manufacturer_id": 2073,
"manufacturer_data_start": [0,0,42,0]
"manufacturer_data_start": [0,0]
}
],
"codeowners": ["@patman15"],
@@ -17,5 +17,5 @@
"issue_tracker": "https://github.com/patman15/hdpv_ble/issues",
"loggers": ["hunterdouglas_powerview_ble"],
"requirements": [],
"version": "0.1"
"version": 0.11
}

77
custom_components/hunterdouglas_powerview_ble/sensor.py Normal file
View File

@@ -0,0 +1,77 @@
"""Platform for sensor integration."""
from homeassistant.components.bluetooth.passive_update_coordinator import (
PassiveBluetoothCoordinatorEntity,
)
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import (
ATTR_BATTERY_LEVEL,
PERCENTAGE,
SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
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 ATTR_RSSI, DOMAIN
from .coordinator import PVCoordinator
SENSOR_TYPES: list[SensorEntityDescription] = [
SensorEntityDescription(
key=ATTR_BATTERY_LEVEL,
translation_key=ATTR_BATTERY_LEVEL,
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.BATTERY,
),
SensorEntityDescription(
key=ATTR_RSSI,
translation_key=ATTR_RSSI,
native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
entity_registry_enabled_default=False,
entity_category=EntityCategory.DIAGNOSTIC,
),
]
async def async_setup_entry(
_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))]
)
class PVSensor(PassiveBluetoothCoordinatorEntity[PVCoordinator], SensorEntity): # type: ignore[reportIncompatibleMethodOverride]
"""The generic BMS sensor implementation."""
_attr_has_entity_name = True
def __init__(
self, pv_dev: PVCoordinator, descr: SensorEntityDescription, unique_id: str
) -> None:
"""Intitialize the BMS sensor."""
self._attr_unique_id = f"{DOMAIN}-{unique_id}-{descr.key}"
self._attr_device_info = pv_dev.device_info
self.entity_description = descr
super().__init__(pv_dev)
@property
def native_value(self) -> int | float | None: # type: ignore[reportIncompatibleVariableOverride]
"""Return the sensor value."""
return self.coordinator.data.get(self.entity_description.key)