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

fixed opening/closing detection

Browse Source
This commit is contained in:
patman15
2024-08-30 22:56:37 +02:00
parent b1959ec043
commit e0185f19e7
2 changed files with 52 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
custom_components/hunterdouglas_powerview_ble/api.py
View File

@@ -85,9 +85,6 @@ 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._connect_lock = (
# asyncio.Lock()
# ) # TODO: try get rid of (device_info vs. normal cmds)
self._info: PVDeviceInfo = PVDeviceInfo() self._info: PVDeviceInfo = PVDeviceInfo()
self._cmd_lock: Final = asyncio.Lock() self._cmd_lock: Final = asyncio.Lock()
self._cmd_next = None self._cmd_next = None
@@ -102,6 +99,7 @@ class PowerViewBLE:
self._data_event.clear() self._data_event.clear()
@property @property
def info(self) -> PVDeviceInfo:
"""Return device information, e.g. SW version.""" """Return device information, e.g. SW version."""
return self._info return self._info
@@ -112,7 +110,7 @@ class PowerViewBLE:
# general cmd: uint16_t cmd, uint8_t seqID, uint8_t data_len # general cmd: uint16_t cmd, uint8_t seqID, uint8_t data_len
async def _cmd( async def _cmd(
self, cmd: tuple[ShadeCmd, bytearray], disconnect: bool = False self, cmd: tuple[ShadeCmd, bytearray], disconnect: bool = True
) -> None: ) -> None:
self._cmd_next = cmd self._cmd_next = cmd
if self._cmd_lock.locked(): if self._cmd_lock.locked():
@@ -175,7 +173,9 @@ class PowerViewBLE:
] ]
# position cmd: uint16_t pos1, uint16_t pos2, uint16_t pos3, uint16_t tilt, uint8_t velocity # position cmd: uint16_t pos1, uint16_t pos2, uint16_t pos3, uint16_t tilt, uint8_t velocity
async def _set_position(self, value: int, disconnect: bool = False) -> None: async def set_position(self, value: int, disconnect: bool = True) -> None:
"""Set position of device."""
LOGGER.debug("%s setting position to %i", self.name, value)
await self._cmd( await self._cmd(
( (
ShadeCmd.SET_POSITION, ShadeCmd.SET_POSITION,
@@ -187,15 +187,10 @@ class PowerViewBLE:
disconnect, disconnect,
) )
async def set_position(self, value: int) -> None:
"""Set position of device."""
LOGGER.debug("%s setting position to %i", self.name, value)
await self._set_position(value, disconnect=True)
async def open(self) -> None: async def open(self) -> 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) await self.set_position(OPEN_POSITION, disconnect=False)
async def stop(self) -> None: async def stop(self) -> None:
"""Stop device movement.""" """Stop device movement."""
@@ -205,7 +200,7 @@ class PowerViewBLE:
async def close(self) -> None: async def close(self) -> 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) await self.set_position(CLOSED_POSITION, disconnect=False)
# uint8_t scene#, uint8_t unknown # uint8_t scene#, uint8_t unknown
# open: scene 2 # open: scene 2
@@ -257,24 +252,22 @@ class PowerViewBLE:
"sw_rev": "2a28", "sw_rev": "2a28",
} }
try: async with self._cmd_lock:
await self._connect() try:
assert self._client is not None await self._connect()
assert self._client is not None
for key, uuid in uuids.items(): for key, uuid in uuids.items():
LOGGER.debug("querying %s(%s)", key, uuid) LOGGER.debug("querying %s(%s)", key, uuid)
data[key] = ( data[key] = (
(await self._client.read_gatt_char(normalize_uuid_str(uuid))) (await self._client.read_gatt_char(normalize_uuid_str(uuid)))
.copy() .copy()
.decode("UTF-8") .decode("UTF-8")
) )
except BleakError as ex: finally:
LOGGER.debug("%s error: %s", self.name, ex) await self._disconnect()
return {}
finally:
await self._disconnect()
LOGGER.debug("%s device data: %s", self.name, data) LOGGER.debug("%s device data: %s", self.name, data)
return data return data.copy()
def _on_disconnect(self, client: BleakClient) -> None: def _on_disconnect(self, client: BleakClient) -> None:
"""Disconnect callback function.""" """Disconnect callback function."""

37
custom_components/hunterdouglas_powerview_ble/cover.py
View File

@@ -16,7 +16,6 @@ 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.exceptions import HomeAssistantError
from homeassistant.helpers.device_registry import DeviceInfo, format_mac from homeassistant.helpers.device_registry import DeviceInfo, format_mac
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
@@ -56,6 +55,9 @@ class PowerViewCover(PassiveBluetoothCoordinatorEntity[PVCoordinator], CoverEnti
self._attr_name = CoverDeviceClass.SHADE self._attr_name = CoverDeviceClass.SHADE
self._coord = coordinator self._coord = coordinator
self._attr_device_info = self._coord.device_info self._attr_device_info = self._coord.device_info
self._target_position: int | None = round(
self._coord.data.get(ATTR_CURRENT_POSITION, OPEN_POSITION)
)
self._attr_unique_id = ( self._attr_unique_id = (
f"{DOMAIN}_{format_mac(self._coord.address)}_{CoverDeviceClass.SHADE}" f"{DOMAIN}_{format_mac(self._coord.address)}_{CoverDeviceClass.SHADE}"
) )
@@ -69,12 +71,22 @@ class PowerViewCover(PassiveBluetoothCoordinatorEntity[PVCoordinator], CoverEnti
@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."""
return bool(self._coord.data.get("is_opening")) return bool(self._coord.data.get("is_opening")) or (
isinstance(self._target_position, int)
and isinstance(self.current_cover_position, int)
and self._target_position > self.current_cover_position
and self._coord.api.is_connected
)
@property @property
def is_closing(self) -> bool | None: # type: ignore[reportIncompatibleVariableOverride] def is_closing(self) -> bool | None: # type: ignore[reportIncompatibleVariableOverride]
"""Return if the cover is closing or not.""" """Return if the cover is closing or not."""
return bool(self._coord.data.get("is_closing")) return bool(self._coord.data.get("is_closing")) or (
isinstance(self._target_position, int)
and isinstance(self.current_cover_position, int)
and self._target_position < self.current_cover_position
and self._coord.api.is_connected
)
@property @property
def is_closed(self) -> bool: # type: ignore[reportIncompatibleVariableOverride] def is_closed(self) -> bool: # type: ignore[reportIncompatibleVariableOverride]
@@ -97,34 +109,42 @@ class PowerViewCover(PassiveBluetoothCoordinatorEntity[PVCoordinator], CoverEnti
None is unknown, 0 is closed, 100 is fully open. None is unknown, 0 is closed, 100 is fully open.
""" """
pos = self._coord.data.get(ATTR_CURRENT_POSITION) pos: Final = self._coord.data.get(ATTR_CURRENT_POSITION)
return round(pos) if pos is not None else None return round(pos) if pos is not None else None
async def async_set_cover_position(self, **kwargs: Any) -> None: async def async_set_cover_position(self, **kwargs: Any) -> None:
"""Move the cover to a specific position.""" """Move the cover to a specific position."""
target_position: Final[int | None] = kwargs.get(ATTR_POSITION) target_position: Final = kwargs.get(ATTR_POSITION)
if target_position is not None: if target_position is not None:
LOGGER.debug("set cover to position %i", target_position) LOGGER.debug("set cover to position %f", target_position)
if self.current_cover_position == round(target_position) and not ( if self.current_cover_position == round(target_position) and not (
self.is_closing or self.is_opening self.is_closing or self.is_opening
): ):
return return
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))
self.async_write_ha_state()
except BleakError as err: except BleakError as err:
LOGGER.error( LOGGER.error(
f"Failed to move cover '{self.name}' to {target_position}%: {err}" f"Failed to move cover '{self.name}' to {target_position}%: {err}"
) )
def _reset_target_position(self) -> None:
self._target_position = None
async def async_open_cover(self, **kwargs: Any) -> None: async def async_open_cover(self, **kwargs: Any) -> None:
"""Open the cover.""" """Open the cover."""
LOGGER.debug("open cover") LOGGER.debug("open cover")
if self.current_cover_position == OPEN_POSITION: if self.current_cover_position == OPEN_POSITION:
return return
try: try:
self._target_position = OPEN_POSITION
await self._coord.api.open() await self._coord.api.open()
self.async_write_ha_state()
except BleakError as err: except BleakError as err:
LOGGER.error(f"Failed to open cover '{self.name}': {err}") LOGGER.error(f"Failed to open cover '{self.name}': {err}")
self._reset_target_position()
async def async_close_cover(self, **kwargs: Any) -> None: async def async_close_cover(self, **kwargs: Any) -> None:
"""Close the cover tilt.""" """Close the cover tilt."""
@@ -132,14 +152,19 @@ class PowerViewCover(PassiveBluetoothCoordinatorEntity[PVCoordinator], CoverEnti
if self.current_cover_position == CLOSED_POSITION: if self.current_cover_position == CLOSED_POSITION:
return return
try: try:
self._target_position = CLOSED_POSITION
await self._coord.api.close() await self._coord.api.close()
self.async_write_ha_state()
except BleakError as err: except BleakError as err:
LOGGER.error(f"Failed to close cover '{self.name}': {err}") LOGGER.error(f"Failed to close cover '{self.name}': {err}")
self._reset_target_position()
async def async_stop_cover(self, **kwargs: Any) -> None: async def async_stop_cover(self, **kwargs: Any) -> None:
"""Stop the cover.""" """Stop the cover."""
LOGGER.debug("stop cover") LOGGER.debug("stop cover")
try: try:
await self._coord.api.stop() await self._coord.api.stop()
self._reset_target_position()
self.async_write_ha_state()
except BleakError as err: except BleakError as err:
LOGGER.error(f"Failed to stop cover '{self.name}': {err}") LOGGER.error(f"Failed to stop cover '{self.name}': {err}")