Add YoLink lock V2 support (#124202)

* Add Lock V2 Support

* Change as suggestions
pull/124707/head
Matrix 2024-08-28 20:18:55 +08:00 committed by GitHub
parent c4e5d67551
commit f8ac952cd7
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 30 additions and 9 deletions

View File

@ -1,11 +1,11 @@
"""YoLink Lock."""
"""YoLink Lock V1/V2."""
from __future__ import annotations
from typing import Any
from yolink.client_request import ClientRequest
from yolink.const import ATTR_DEVICE_LOCK
from yolink.const import ATTR_DEVICE_LOCK, ATTR_DEVICE_LOCK_V2
from homeassistant.components.lock import LockEntity
from homeassistant.config_entries import ConfigEntry
@ -27,7 +27,8 @@ async def async_setup_entry(
entities = [
YoLinkLockEntity(config_entry, device_coordinator)
for device_coordinator in device_coordinators.values()
if device_coordinator.device.device_type == ATTR_DEVICE_LOCK
if device_coordinator.device.device_type
in [ATTR_DEVICE_LOCK, ATTR_DEVICE_LOCK_V2]
]
async_add_entities(entities)
@ -50,21 +51,41 @@ class YoLinkLockEntity(YoLinkEntity, LockEntity):
def update_entity_state(self, state: dict[str, Any]) -> None:
"""Update HA Entity State."""
state_value = state.get("state")
self._attr_is_locked = (
state_value == "locked" if state_value is not None else None
)
if self.coordinator.device.device_type == ATTR_DEVICE_LOCK_V2:
self._attr_is_locked = (
state_value["lock"] == "locked" if state_value is not None else None
)
else:
self._attr_is_locked = (
state_value == "locked" if state_value is not None else None
)
self.async_write_ha_state()
async def call_lock_state_change(self, state: str) -> None:
"""Call setState api to change lock state."""
await self.call_device(ClientRequest("setState", {"state": state}))
if self.coordinator.device.device_type == ATTR_DEVICE_LOCK_V2:
await self.call_device(
ClientRequest("setState", {"state": {"lock": state}})
)
else:
await self.call_device(ClientRequest("setState", {"state": state}))
self._attr_is_locked = state == "lock"
self.async_write_ha_state()
async def async_lock(self, **kwargs: Any) -> None:
"""Lock device."""
await self.call_lock_state_change("lock")
state_param = (
"locked"
if self.coordinator.device.device_type == ATTR_DEVICE_LOCK_V2
else "lock"
)
await self.call_lock_state_change(state_param)
async def async_unlock(self, **kwargs: Any) -> None:
"""Unlock device."""
await self.call_lock_state_change("unlock")
state_param = (
"unlocked"
if self.coordinator.device.device_type == ATTR_DEVICE_LOCK_V2
else "unlock"
)
await self.call_lock_state_change(state_param)