Implement binary_sensor in yale_smart_alarm (#63937)

This commit is contained in:
G Johansson 2022-01-13 10:34:43 +01:00 committed by GitHub
parent b196064540
commit c021e58ee2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 98 additions and 44 deletions

View File

@ -1332,8 +1332,10 @@ omit =
homeassistant/components/xs1/*
homeassistant/components/yale_smart_alarm/__init__.py
homeassistant/components/yale_smart_alarm/alarm_control_panel.py
homeassistant/components/yale_smart_alarm/binary_sensor.py
homeassistant/components/yale_smart_alarm/const.py
homeassistant/components/yale_smart_alarm/coordinator.py
homeassistant/components/yale_smart_alarm/entity.py
homeassistant/components/yale_smart_alarm/lock.py
homeassistant/components/yamaha_musiccast/__init__.py
homeassistant/components/yamaha_musiccast/media_player.py

View File

@ -84,13 +84,14 @@ async def async_setup_entry(
class YaleAlarmDevice(CoordinatorEntity, AlarmControlPanelEntity):
"""Represent a Yale Smart Alarm."""
coordinator: YaleDataUpdateCoordinator
_attr_code_arm_required = False
_attr_supported_features = SUPPORT_ALARM_ARM_HOME | SUPPORT_ALARM_ARM_AWAY
def __init__(self, coordinator: YaleDataUpdateCoordinator) -> None:
"""Initialize the Yale Alarm Device."""
super().__init__(coordinator)
self._coordinator = coordinator
self._attr_name = coordinator.entry.data[CONF_NAME]
self._attr_unique_id = coordinator.entry.entry_id
self._attr_device_info = DeviceInfo(
@ -103,11 +104,11 @@ class YaleAlarmDevice(CoordinatorEntity, AlarmControlPanelEntity):
async def async_alarm_disarm(self, code=None) -> None:
"""Send disarm command."""
if TYPE_CHECKING:
assert self._coordinator.yale, "Connection to API is missing"
assert self.coordinator.yale, "Connection to API is missing"
try:
alarm_state = await self.hass.async_add_executor_job(
self._coordinator.yale.disarm
self.coordinator.yale.disarm
)
except (
AuthenticationError,
@ -129,11 +130,11 @@ class YaleAlarmDevice(CoordinatorEntity, AlarmControlPanelEntity):
async def async_alarm_arm_home(self, code=None) -> None:
"""Send arm home command."""
if TYPE_CHECKING:
assert self._coordinator.yale, "Connection to API is missing"
assert self.coordinator.yale, "Connection to API is missing"
try:
alarm_state = await self.hass.async_add_executor_job(
self._coordinator.yale.arm_partial
self.coordinator.yale.arm_partial
)
except (
AuthenticationError,
@ -155,11 +156,11 @@ class YaleAlarmDevice(CoordinatorEntity, AlarmControlPanelEntity):
async def async_alarm_arm_away(self, code=None) -> None:
"""Send arm away command."""
if TYPE_CHECKING:
assert self._coordinator.yale, "Connection to API is missing"
assert self.coordinator.yale, "Connection to API is missing"
try:
alarm_state = await self.hass.async_add_executor_job(
self._coordinator.yale.arm_full
self.coordinator.yale.arm_full
)
except (
AuthenticationError,

View File

@ -0,0 +1,39 @@
"""Binary sensors for Yale Alarm."""
from __future__ import annotations
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import COORDINATOR, DOMAIN
from .coordinator import YaleDataUpdateCoordinator
from .entity import YaleEntity
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the Yale binary sensor entry."""
coordinator: YaleDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][
COORDINATOR
]
async_add_entities(
YaleBinarySensor(coordinator, data) for data in coordinator.data["door_windows"]
)
class YaleBinarySensor(YaleEntity, BinarySensorEntity):
"""Representation of a Yale binary sensor."""
_attr_device_class = BinarySensorDeviceClass.DOOR
@property
def is_on(self) -> bool:
"""Return true if the binary sensor is on."""
return self.coordinator.data["sensor_map"][self._attr_unique_id] == "open"

View File

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

View File

@ -105,12 +105,19 @@ class YaleDataUpdateCoordinator(DataUpdateCoordinator):
door_windows.append(device)
continue
_sensor_map = {
contact["address"]: contact["_state"] for contact in door_windows
}
_lock_map = {lock["address"]: lock["_state"] for lock in locks}
return {
"alarm": updates["arm_status"],
"locks": locks,
"door_windows": door_windows,
"status": updates["status"],
"online": updates["online"],
"sensor_map": _sensor_map,
"lock_map": _lock_map,
}
def get_updates(self) -> dict:

View File

@ -0,0 +1,27 @@
"""Base class for yale_smart_alarm entity."""
from homeassistant.const import CONF_USERNAME
from homeassistant.helpers.entity import DeviceInfo, Entity
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN, MANUFACTURER, MODEL
from .coordinator import YaleDataUpdateCoordinator
class YaleEntity(CoordinatorEntity, Entity):
"""Base implementation for Yale device."""
coordinator: YaleDataUpdateCoordinator
def __init__(self, coordinator: YaleDataUpdateCoordinator, data: dict) -> None:
"""Initialize an Yale device."""
super().__init__(coordinator)
self._attr_name: str = data["name"]
self._attr_unique_id: str = data["address"]
self._attr_device_info: DeviceInfo = DeviceInfo(
name=self._attr_name,
manufacturer=MANUFACTURER,
model=MODEL,
identifiers={(DOMAIN, data["address"])},
via_device=(DOMAIN, self.coordinator.entry.data[CONF_USERNAME]),
)

View File

@ -7,12 +7,10 @@ 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.const import ATTR_CODE, CONF_CODE
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,
@ -20,10 +18,9 @@ from .const import (
DEFAULT_LOCK_CODE_DIGITS,
DOMAIN,
LOGGER,
MANUFACTURER,
MODEL,
)
from .coordinator import YaleDataUpdateCoordinator
from .entity import YaleEntity
async def async_setup_entry(
@ -42,32 +39,22 @@ async def async_setup_entry(
)
class YaleDoorlock(CoordinatorEntity, LockEntity):
class YaleDoorlock(YaleEntity, 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]),
)
super().__init__(coordinator, data)
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"
assert self.coordinator.yale, "Connection to API is missing"
code = kwargs.get(ATTR_CODE, self._coordinator.entry.options.get(CONF_CODE))
code = kwargs.get(ATTR_CODE, self.coordinator.entry.options.get(CONF_CODE))
if not code:
raise HomeAssistantError(
@ -76,10 +63,10 @@ class YaleDoorlock(CoordinatorEntity, LockEntity):
try:
get_lock = await self.hass.async_add_executor_job(
self._coordinator.yale.lock_api.get, self._attr_name
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,
self.coordinator.yale.lock_api.open_lock,
get_lock,
code,
)
@ -95,10 +82,7 @@ class YaleDoorlock(CoordinatorEntity, LockEntity):
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.coordinator.data["lock_map"][self._attr_unique_id] = "unlocked"
self.async_write_ha_state()
return
raise HomeAssistantError("Could not unlock, check system ready for unlocking")
@ -106,14 +90,14 @@ class YaleDoorlock(CoordinatorEntity, LockEntity):
async def async_lock(self, **kwargs) -> None:
"""Send lock command."""
if TYPE_CHECKING:
assert self._coordinator.yale, "Connection to API is missing"
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
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,
self.coordinator.yale.lock_api.close_lock,
get_lock,
)
except (
@ -128,9 +112,7 @@ class YaleDoorlock(CoordinatorEntity, LockEntity):
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.coordinator.data["lock_map"][self._attr_unique_id] = "unlocked"
self.async_write_ha_state()
return
raise HomeAssistantError("Could not unlock, check system ready for unlocking")
@ -138,8 +120,4 @@ class YaleDoorlock(CoordinatorEntity, LockEntity):
@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
return self.coordinator.data["lock_map"][self._attr_unique_id] == "locked"