evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Code cleanup yale_smart_alarm (#67701)

Browse source
This commit is contained in:
G Johansson 2022-03-06 06:27:06 +01:00 • committed by GitHub
parent f5d25eaf8f
commit 1358aed016
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 68 additions and 136 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
homeassistant/components/yale_smart_alarm/__init__.py
View file

@ -5,31 +5,25 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from .const import COORDINATOR, DOMAIN, LOGGER, PLATFORMS
from .const import COORDINATOR, DOMAIN, PLATFORMS
from .coordinator import YaleDataUpdateCoordinator
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Yale from a config entry."""
hass.data.setdefault(DOMAIN, {})
title = entry.title
coordinator = YaleDataUpdateCoordinator(hass, entry=entry)
coordinator = YaleDataUpdateCoordinator(hass, entry)
if not await hass.async_add_executor_job(coordinator.get_updates):
raise ConfigEntryAuthFailed
await coordinator.async_config_entry_first_refresh()
hass.data[DOMAIN][entry.entry_id] = {
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = {
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
@ -41,12 +35,7 @@ async def update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
title = entry.title
if unload_ok:
if await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
hass.data[DOMAIN].pop(entry.entry_id)
LOGGER.debug("Unloaded entry for %s", title)
return unload_ok
return True
return False

85
homeassistant/components/yale_smart_alarm/alarm_control_panel.py
View file

@ -9,7 +9,6 @@ from yalesmartalarmclient.const import (
YALE_STATE_ARM_PARTIAL,
YALE_STATE_DISARM,
)
from yalesmartalarmclient.exceptions import AuthenticationError, UnknownError
from homeassistant.components.alarm_control_panel import (
PLATFORM_SCHEMA as PARENT_PLATFORM_SCHEMA,
@ -39,6 +38,7 @@ from .const import (
MANUFACTURER,
MODEL,
STATE_MAP,
YALE_ALL_ERRORS,
)
from .coordinator import YaleDataUpdateCoordinator
@ -103,81 +103,46 @@ 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"
try:
alarm_state = await self.hass.async_add_executor_job(
self.coordinator.yale.disarm
)
except (
AuthenticationError,
ConnectionError,
TimeoutError,
UnknownError,
) as error:
raise HomeAssistantError(
f"Could not verify disarmed for {self._attr_name}: {error}"
) from error
LOGGER.debug("Alarm disarmed: %s", alarm_state)
if alarm_state:
self.coordinator.data["alarm"] = YALE_STATE_DISARM
self.async_write_ha_state()
return
raise HomeAssistantError("Could not disarm, check system ready for disarming.")
return await self.async_set_alarm(YALE_STATE_DISARM, code)
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"
try:
alarm_state = await self.hass.async_add_executor_job(
self.coordinator.yale.arm_partial
)
except (
AuthenticationError,
ConnectionError,
TimeoutError,
UnknownError,
) as error:
raise HomeAssistantError(
f"Could not verify armed home for {self._attr_name}: {error}"
) from error
LOGGER.debug("Alarm armed home: %s", alarm_state)
if alarm_state:
self.coordinator.data["alarm"] = YALE_STATE_ARM_PARTIAL
self.async_write_ha_state()
return
raise HomeAssistantError("Could not arm home, check system ready for arming.")
return await self.async_set_alarm(YALE_STATE_ARM_PARTIAL, code)
async def async_alarm_arm_away(self, code=None) -> None:
"""Send arm away command."""
return await self.async_set_alarm(YALE_STATE_ARM_FULL, code)
async def async_set_alarm(self, command: str, code: str | None = None) -> None:
"""Set alarm."""
if TYPE_CHECKING:
assert self.coordinator.yale, "Connection to API is missing"
try:
alarm_state = await self.hass.async_add_executor_job(
self.coordinator.yale.arm_full
)
except (
AuthenticationError,
ConnectionError,
TimeoutError,
UnknownError,
) as error:
if command == YALE_STATE_ARM_FULL:
alarm_state = await self.hass.async_add_executor_job(
self.coordinator.yale.arm_full
)
if command == YALE_STATE_ARM_PARTIAL:
alarm_state = await self.hass.async_add_executor_job(
self.coordinator.yale.arm_partial
)
if command == YALE_STATE_DISARM:
alarm_state = await self.hass.async_add_executor_job(
self.coordinator.yale.disarm
)
except YALE_ALL_ERRORS as error:
raise HomeAssistantError(
f"Could not verify armed away for {self._attr_name}: {error}"
f"Could not set alarm for {self._attr_name}: {error}"
) from error
LOGGER.debug("Alarm armed away: %s", alarm_state)
if alarm_state:
self.coordinator.data["alarm"] = YALE_STATE_ARM_FULL
self.coordinator.data["alarm"] = command
self.async_write_ha_state()
return
raise HomeAssistantError("Could not arm away, check system ready for arming.")
raise HomeAssistantError(
"Could not change alarm check system ready for arming."
)
@property
def available(self) -> bool:

7
homeassistant/components/yale_smart_alarm/config_flow.py
View file

@ -5,7 +5,7 @@ from typing import Any
import voluptuous as vol
from yalesmartalarmclient.client import YaleSmartAlarmClient
from yalesmartalarmclient.exceptions import AuthenticationError, UnknownError
from yalesmartalarmclient.exceptions import AuthenticationError
from homeassistant.config_entries import ConfigEntry, ConfigFlow, OptionsFlow
from homeassistant.const import CONF_CODE, CONF_NAME, CONF_PASSWORD, CONF_USERNAME
@ -21,6 +21,7 @@ from .const import (
DEFAULT_NAME,
DOMAIN,
LOGGER,
YALE_BASE_ERRORS,
)
DATA_SCHEMA = vol.Schema(
@ -81,7 +82,7 @@ class YaleConfigFlow(ConfigFlow, domain=DOMAIN):
except AuthenticationError as error:
LOGGER.error("Authentication failed. Check credentials %s", error)
errors = {"base": "invalid_auth"}
except (ConnectionError, TimeoutError, UnknownError) as error:
except YALE_BASE_ERRORS as error:
LOGGER.error("Connection to API failed %s", error)
errors = {"base": "cannot_connect"}
@ -124,7 +125,7 @@ class YaleConfigFlow(ConfigFlow, domain=DOMAIN):
except AuthenticationError as error:
LOGGER.error("Authentication failed. Check credentials %s", error)
errors = {"base": "invalid_auth"}
except (ConnectionError, TimeoutError, UnknownError) as error:
except YALE_BASE_ERRORS as error:
LOGGER.error("Connection to API failed %s", error)
errors = {"base": "cannot_connect"}

8
homeassistant/components/yale_smart_alarm/const.py
View file

@ -6,6 +6,7 @@ from yalesmartalarmclient.client import (
YALE_STATE_ARM_PARTIAL,
YALE_STATE_DISARM,
)
from yalesmartalarmclient.exceptions import AuthenticationError, UnknownError
from homeassistant.const import (
STATE_ALARM_ARMED_AWAY,
@ -40,3 +41,10 @@ STATE_MAP = {
YALE_STATE_ARM_PARTIAL: STATE_ALARM_ARMED_HOME,
YALE_STATE_ARM_FULL: STATE_ALARM_ARMED_AWAY,
}
YALE_BASE_ERRORS = (
ConnectionError,
TimeoutError,
UnknownError,
)
YALE_ALL_ERRORS = (*YALE_BASE_ERRORS, AuthenticationError)

13
homeassistant/components/yale_smart_alarm/coordinator.py
View file

@ -2,9 +2,10 @@
from __future__ import annotations
from datetime import timedelta
from typing import Any
from yalesmartalarmclient.client import YaleSmartAlarmClient
from yalesmartalarmclient.exceptions import AuthenticationError, UnknownError
from yalesmartalarmclient.exceptions import AuthenticationError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
@ -12,7 +13,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DEFAULT_SCAN_INTERVAL, DOMAIN, LOGGER
from .const import DEFAULT_SCAN_INTERVAL, DOMAIN, LOGGER, YALE_BASE_ERRORS
class YaleDataUpdateCoordinator(DataUpdateCoordinator):
@ -29,7 +30,7 @@ class YaleDataUpdateCoordinator(DataUpdateCoordinator):
update_interval=timedelta(seconds=DEFAULT_SCAN_INTERVAL),
)
async def _async_update_data(self) -> dict:
async def _async_update_data(self) -> dict[str, Any]:
"""Fetch data from Yale."""
updates = await self.hass.async_add_executor_job(self.get_updates)
@ -120,7 +121,7 @@ class YaleDataUpdateCoordinator(DataUpdateCoordinator):
"lock_map": _lock_map,
}
def get_updates(self) -> dict:
def get_updates(self) -> dict[str, Any]:
"""Fetch data from Yale."""
if self.yale is None:
@ -130,7 +131,7 @@ class YaleDataUpdateCoordinator(DataUpdateCoordinator):
)
except AuthenticationError as error:
raise ConfigEntryAuthFailed from error
except (ConnectionError, TimeoutError, UnknownError) as error:
except YALE_BASE_ERRORS as error:
raise UpdateFailed from error
try:
@ -141,7 +142,7 @@ class YaleDataUpdateCoordinator(DataUpdateCoordinator):
except AuthenticationError as error:
raise ConfigEntryAuthFailed from error
except (ConnectionError, TimeoutError, UnknownError) as error:
except YALE_BASE_ERRORS as error:
raise UpdateFailed from error
return {

70
homeassistant/components/yale_smart_alarm/lock.py
View file

@ -3,8 +3,6 @@ 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
@ -17,7 +15,7 @@ from .const import (
COORDINATOR,
DEFAULT_LOCK_CODE_DIGITS,
DOMAIN,
LOGGER,
YALE_ALL_ERRORS,
)
from .coordinator import YaleDataUpdateCoordinator
from .entity import YaleEntity
@ -51,44 +49,15 @@ class YaleDoorlock(YaleEntity, LockEntity):
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:
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")
return await self.async_set_lock("unlocked", code)
async def async_lock(self, **kwargs) -> None:
"""Send lock command."""
return await self.async_set_lock("locked", None)
async def async_set_lock(self, command: str, code: str | None) -> None:
"""Set lock."""
if TYPE_CHECKING:
assert self.coordinator.yale, "Connection to API is missing"
@ -96,26 +65,25 @@ class YaleDoorlock(YaleEntity, LockEntity):
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:
if command == "locked":
lock_state = await self.hass.async_add_executor_job(
self.coordinator.yale.lock_api.close_lock,
get_lock,
)
if command == "unlocked":
lock_state = await self.hass.async_add_executor_job(
self.coordinator.yale.lock_api.open_lock, get_lock, code
)
except YALE_ALL_ERRORS as error:
raise HomeAssistantError(
f"Could not verify unlocking for {self._attr_name}: {error}"
f"Could not set lock for {self._attr_name}: {error}"
) from error
LOGGER.debug("Door unlock: %s", lock_state)
if lock_state:
self.coordinator.data["lock_map"][self._attr_unique_id] = "unlocked"
self.coordinator.data["lock_map"][self._attr_unique_id] = command
self.async_write_ha_state()
return
raise HomeAssistantError("Could not unlock, check system ready for unlocking")
raise HomeAssistantError("Could set lock, check system ready for lock.")
@property
def is_locked(self) -> bool | None: