initialize device info
This commit is contained in:
@@ -1,4 +1,10 @@
|
|||||||
"""The Hunter Douglas PowerView (BLE) integration."""
|
"""The Hunter Douglas PowerView (BLE) integration.
|
||||||
|
|
||||||
|
@author: patman15
|
||||||
|
@license: Apache-2.0 license
|
||||||
|
"""
|
||||||
|
|
||||||
|
from bleak.exc import BleakError
|
||||||
|
|
||||||
from homeassistant.components.bluetooth import async_ble_device_from_address
|
from homeassistant.components.bluetooth import async_ble_device_from_address
|
||||||
from homeassistant.config_entries import ConfigEntry
|
from homeassistant.config_entries import ConfigEntry
|
||||||
@@ -31,6 +37,10 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntryType) -> bool
|
|||||||
)
|
)
|
||||||
|
|
||||||
coordinator = PVCoordinator(hass, ble_device, entry.data.copy())
|
coordinator = PVCoordinator(hass, ble_device, entry.data.copy())
|
||||||
|
try:
|
||||||
|
await coordinator.query_dev_info()
|
||||||
|
except BleakError as err:
|
||||||
|
raise ConfigEntryNotReady("Unable to query device info.") from err
|
||||||
|
|
||||||
# Insert the coordinator in the global registry
|
# Insert the coordinator in the global registry
|
||||||
hass.data.setdefault(DOMAIN, {})
|
hass.data.setdefault(DOMAIN, {})
|
||||||
|
|||||||
@@ -62,7 +62,7 @@ class ShadeCmd(Enum):
|
|||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class DeviceInfo:
|
class PVDeviceInfo:
|
||||||
"""Dataclass holding available PowerView device information."""
|
"""Dataclass holding available PowerView device information."""
|
||||||
|
|
||||||
manufacturer: str = ""
|
manufacturer: str = ""
|
||||||
@@ -85,10 +85,10 @@ class PowerViewBLE:
|
|||||||
self._client: BleakClient | None = None
|
self._client: BleakClient | None = None
|
||||||
self._data_event = asyncio.Event()
|
self._data_event = asyncio.Event()
|
||||||
self._data: bytearray
|
self._data: bytearray
|
||||||
self._info: DeviceInfo = DeviceInfo()
|
|
||||||
# self._connect_lock = (
|
# self._connect_lock = (
|
||||||
# asyncio.Lock()
|
# asyncio.Lock()
|
||||||
# ) # TODO: try get rid of (device_info vs. normal cmds)
|
# ) # TODO: try get rid of (device_info vs. normal cmds)
|
||||||
|
self._info: PVDeviceInfo = PVDeviceInfo()
|
||||||
self._cmd_lock: Final = asyncio.Lock()
|
self._cmd_lock: Final = asyncio.Lock()
|
||||||
self._cmd_next = None
|
self._cmd_next = None
|
||||||
self._cipher: Final = (
|
self._cipher: Final = (
|
||||||
@@ -102,7 +102,6 @@ class PowerViewBLE:
|
|||||||
self._data_event.clear()
|
self._data_event.clear()
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def info(self) -> DeviceInfo:
|
|
||||||
"""Return device information, e.g. SW version."""
|
"""Return device information, e.g. SW version."""
|
||||||
return self._info
|
return self._info
|
||||||
|
|
||||||
@@ -159,8 +158,12 @@ class PowerViewBLE:
|
|||||||
LOGGER.debug("not a V2 record!")
|
LOGGER.debug("not a V2 record!")
|
||||||
return []
|
return []
|
||||||
pos = int.from_bytes(data[3:5], byteorder="little")
|
pos = int.from_bytes(data[3:5], byteorder="little")
|
||||||
|
pos2 = ((int(data[5]) << 4) + (int(data[4]) >> 4))
|
||||||
return [
|
return [
|
||||||
(ATTR_CURRENT_POSITION, ((pos >> 2) / 10)),
|
(ATTR_CURRENT_POSITION, ((pos >> 2) / 10)),
|
||||||
|
("position2", pos2 >> 2),
|
||||||
|
("position3", int(data[6])),
|
||||||
|
("tilt", int(data[7])),
|
||||||
("home_id", int.from_bytes(data[0:2], byteorder="little")),
|
("home_id", int.from_bytes(data[0:2], byteorder="little")),
|
||||||
("type_id", int.from_bytes(data[2:3])),
|
("type_id", int.from_bytes(data[2:3])),
|
||||||
("is_opening", bool(pos & 0x3 == 0x2)),
|
("is_opening", bool(pos & 0x3 == 0x2)),
|
||||||
|
|||||||
@@ -1,6 +1,5 @@
|
|||||||
"""Home Assistant coordinator for Hunter Douglas PowerView (BLE) integration."""
|
"""Home Assistant coordinator for Hunter Douglas PowerView (BLE) integration."""
|
||||||
|
|
||||||
import asyncio
|
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
from bleak.backends.device import BLEDevice
|
from bleak.backends.device import BLEDevice
|
||||||
@@ -30,7 +29,6 @@ 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._dev_info_task: asyncio.Task | None = None
|
|
||||||
|
|
||||||
LOGGER.debug(
|
LOGGER.debug(
|
||||||
"Initializing coordinator for %s (%s)",
|
"Initializing coordinator for %s (%s)",
|
||||||
@@ -44,17 +42,15 @@ class PVCoordinator(PassiveBluetoothDataUpdateCoordinator):
|
|||||||
bluetooth.BluetoothScanningMode.ACTIVE,
|
bluetooth.BluetoothScanningMode.ACTIVE,
|
||||||
)
|
)
|
||||||
|
|
||||||
async def _get_device_info(self) -> None:
|
async def query_dev_info(self) -> None:
|
||||||
self.dev_details = await self.api.query_dev_info()
|
"""Receive detailed information from device."""
|
||||||
|
LOGGER.debug("%s: querying device info", self.name)
|
||||||
|
self.dev_details.update(await self.api.query_dev_info())
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def device_info(self) -> DeviceInfo:
|
def device_info(self) -> DeviceInfo:
|
||||||
"""Return detailed device information for GUI."""
|
"""Return detailed device information for GUI."""
|
||||||
LOGGER.debug("device_info, %s", self.dev_details)
|
LOGGER.debug("%s: device_info, %s", self.name, self.dev_details)
|
||||||
if self._dev_info_task is None or not self._dev_info_task.done:
|
|
||||||
self._dev_info_task = self.hass.async_create_task(
|
|
||||||
self._get_device_info(), "query_device_details", False
|
|
||||||
)
|
|
||||||
return DeviceInfo(
|
return DeviceInfo(
|
||||||
identifiers={
|
identifiers={
|
||||||
(DOMAIN, self.name),
|
(DOMAIN, self.name),
|
||||||
|
|||||||
Reference in New Issue
Block a user