Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Implement lock to yale_smart_alarm (#63643)

pull/63330/head^2
G Johansson 2022-01-11 20:20:15 +01:00 committed by GitHub
parent d3da791168
commit 17bf51a855
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 155 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -1332,6 +1332,7 @@ omit =
homeassistant/components/yale_smart_alarm/alarm_control_panel.py
homeassistant/components/yale_smart_alarm/const.py
homeassistant/components/yale_smart_alarm/coordinator.py
homeassistant/components/yale_smart_alarm/lock.py
homeassistant/components/yamaha_musiccast/__init__.py
homeassistant/components/yamaha_musiccast/media_player.py
homeassistant/components/yamaha_musiccast/number.py

10
homeassistant/components/yale_smart_alarm/__init__.py
View File

@ -5,7 +5,7 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from .const import DOMAIN, LOGGER, PLATFORMS
from .const import COORDINATOR, DOMAIN, LOGGER, PLATFORMS
from .coordinator import YaleDataUpdateCoordinator
@ -22,16 +22,22 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
await coordinator.async_config_entry_first_refresh()
hass.data[DOMAIN][entry.entry_id] = {
"coordinator": coordinator,
COORDINATOR: coordinator,
}
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(update_listener))
LOGGER.debug("Loaded entry for %s", title)
return True
async def update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Handle options update."""
await hass.config_entries.async_reload(entry.entry_id)
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""

2
homeassistant/components/yale_smart_alarm/const.py
View File

@ -33,7 +33,7 @@ LOGGER = logging.getLogger(__package__)
ATTR_ONLINE = "online"
ATTR_STATUS = "status"
PLATFORMS = [Platform.ALARM_CONTROL_PANEL]
PLATFORMS = [Platform.ALARM_CONTROL_PANEL, Platform.LOCK]
STATE_MAP = {
YALE_STATE_DISARM: STATE_ALARM_DISARMED,

12
homeassistant/components/yale_smart_alarm/coordinator.py
View File

@ -42,7 +42,6 @@ class YaleDataUpdateCoordinator(DataUpdateCoordinator):
if device["type"] == "device_type.door_lock":
lock_status_str = device["minigw_lock_status"]
lock_status = int(str(lock_status_str or 0), 16)
jammed = (lock_status & 48) == 48
closed = (lock_status & 16) == 16
locked = (lock_status & 1) == 1
if not lock_status and "device_status.lock" in state:
@ -55,17 +54,6 @@ class YaleDataUpdateCoordinator(DataUpdateCoordinator):
device["_state2"] = "unknown"
locks.append(device)
continue
if (
lock_status
and (
"device_status.lock" in state or "device_status.unlock" in state
)
and jammed
):
device["_state"] = "jammed"
device["_state2"] = "closed"
locks.append(device)
continue
if (
lock_status
and (

145
homeassistant/components/yale_smart_alarm/lock.py Normal file
View File

@ -0,0 +1,145 @@
"""Lock for Yale Alarm."""
from __future__ import annotations
from typing import TYPE_CHECKING
from yalesmartalarmclient.exceptions import AuthenticationError, UnknownError
from homeassistant.components.lock import LockEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_CODE, CONF_CODE, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import (
CONF_LOCK_CODE_DIGITS,
COORDINATOR,
DEFAULT_LOCK_CODE_DIGITS,
DOMAIN,
LOGGER,
MANUFACTURER,
MODEL,
)
from .coordinator import YaleDataUpdateCoordinator
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the Yale lock entry."""
coordinator: YaleDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][
COORDINATOR
]
code_format = entry.options.get(CONF_LOCK_CODE_DIGITS, DEFAULT_LOCK_CODE_DIGITS)
async_add_entities(
YaleDoorlock(coordinator, data, code_format)
for data in coordinator.data["locks"]
)
class YaleDoorlock(CoordinatorEntity, LockEntity):
"""Representation of a Yale doorlock."""
def __init__(
self, coordinator: YaleDataUpdateCoordinator, data: dict, code_format: int
) -> None:
"""Initialize the Yale Lock Device."""
super().__init__(coordinator)
self._coordinator = coordinator
self._attr_name = data["name"]
self._attr_unique_id = data["address"]
self._attr_device_info = DeviceInfo(
name=self._attr_name,
manufacturer=MANUFACTURER,
model=MODEL,
identifiers={(DOMAIN, data["address"])},
via_device=(DOMAIN, self._coordinator.entry.data[CONF_USERNAME]),
)
self._attr_code_format = f"^\\d{code_format}$"
async def async_unlock(self, **kwargs) -> None:
"""Send unlock command."""
if TYPE_CHECKING:
assert self._coordinator.yale, "Connection to API is missing"
code = kwargs.get(ATTR_CODE, self._coordinator.entry.options.get(CONF_CODE))
if not code:
raise HomeAssistantError(
f"No code provided, {self._attr_name} not unlocked"
)
try:
get_lock = await self.hass.async_add_executor_job(
self._coordinator.yale.lock_api.get, self._attr_name
)
lock_state = await self.hass.async_add_executor_job(
self._coordinator.yale.lock_api.open_lock,
get_lock,
code,
)
except (
AuthenticationError,
ConnectionError,
TimeoutError,
UnknownError,
) as error:
raise HomeAssistantError(
f"Could not verify unlocking for {self._attr_name}: {error}"
) from error
LOGGER.debug("Door unlock: %s", lock_state)
if lock_state:
for lock in self.coordinator.data["locks"]:
if lock["address"] == self._attr_unique_id:
lock["_state"] = "unlocked"
LOGGER.debug("lock data %s", self.coordinator.data["locks"])
self.async_write_ha_state()
return
raise HomeAssistantError("Could not unlock, check system ready for unlocking")
async def async_lock(self, **kwargs) -> None:
"""Send lock command."""
if TYPE_CHECKING:
assert self._coordinator.yale, "Connection to API is missing"
try:
get_lock = await self.hass.async_add_executor_job(
self._coordinator.yale.lock_api.get, self._attr_name
)
lock_state = await self.hass.async_add_executor_job(
self._coordinator.yale.lock_api.close_lock,
get_lock,
)
except (
AuthenticationError,
ConnectionError,
TimeoutError,
UnknownError,
) as error:
raise HomeAssistantError(
f"Could not verify unlocking for {self._attr_name}: {error}"
) from error
LOGGER.debug("Door unlock: %s", lock_state)
if lock_state:
for lock in self.coordinator.data["locks"]:
if lock["address"] == self._attr_unique_id:
lock["_state"] = "locked"
self.async_write_ha_state()
return
raise HomeAssistantError("Could not unlock, check system ready for unlocking")
@property
def is_locked(self) -> bool | None:
"""Return true if the lock is locked."""
for lock in self.coordinator.data["locks"]:
return bool(
lock["address"] == self._attr_unique_id and lock["_state"] == "locked"
)
return None