home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2025-06-25 01:21:51 +02:00
Code Issues Packages Projects Releases Wiki Activity

Add config flow to skybell (#70887)

Browse Source
This commit is contained in:
Robert Hillis
2022-06-04 22:37:08 -04:00
committed by GitHub
parent cbea919c3d
commit a502a8798f
20 changed files with 664 additions and 349 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
.coveragerc
View File

@ -1064,7 +1064,14 @@ omit =
homeassistant/components/sisyphus/*
homeassistant/components/sky_hub/*
homeassistant/components/skybeacon/sensor.py
homeassistant/components/skybell/*
homeassistant/components/skybell/__init__.py
homeassistant/components/skybell/binary_sensor.py
homeassistant/components/skybell/camera.py
homeassistant/components/skybell/coordinator.py
homeassistant/components/skybell/entity.py
homeassistant/components/skybell/light.py
homeassistant/components/skybell/sensor.py
homeassistant/components/skybell/switch.py
homeassistant/components/slack/__init__.py
homeassistant/components/slack/notify.py
homeassistant/components/sia/__init__.py

2
CODEOWNERS
View File

@ -933,6 +933,8 @@ build.json @home-assistant/supervisor
/tests/components/siren/ @home-assistant/core @raman325
/homeassistant/components/sisyphus/ @jkeljo
/homeassistant/components/sky_hub/ @rogerselwyn
/homeassistant/components/skybell/ @tkdrob
/tests/components/skybell/ @tkdrob
/homeassistant/components/slack/ @bachya @tkdrob
/tests/components/slack/ @bachya @tkdrob
/homeassistant/components/sleepiq/ @mfugate1 @kbickar

174
homeassistant/components/skybell/__init__.py
View File

@ -1,101 +1,117 @@
"""Support for the Skybell HD Doorbell."""
import logging
from __future__ import annotations
from requests.exceptions import ConnectTimeout, HTTPError
from skybellpy import Skybell
import asyncio
import os
from aioskybell import Skybell
from aioskybell.exceptions import SkybellAuthenticationException, SkybellException
import voluptuous as vol
from homeassistant.components import persistent_notification
from homeassistant.const import (
ATTR_ATTRIBUTION,
CONF_PASSWORD,
CONF_USERNAME,
__version__,
)
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD, CONF_USERNAME, Platform
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.typing import ConfigType
_LOGGER = logging.getLogger(__name__)
ATTRIBUTION = "Data provided by Skybell.com"
NOTIFICATION_ID = "skybell_notification"
NOTIFICATION_TITLE = "Skybell Sensor Setup"
DOMAIN = "skybell"
DEFAULT_CACHEDB = "./skybell_cache.pickle"
DEFAULT_ENTITY_NAMESPACE = "skybell"
AGENT_IDENTIFIER = f"HomeAssistant/{__version__}"
from .const import DEFAULT_CACHEDB, DOMAIN
from .coordinator import SkybellDataUpdateCoordinator
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.Schema(
{
vol.Required(CONF_USERNAME): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
}
)
},
vol.All(
# Deprecated in Home Assistant 2022.6
cv.deprecated(DOMAIN),
{
DOMAIN: vol.Schema(
{
vol.Required(CONF_USERNAME): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
}
)
},
),
extra=vol.ALLOW_EXTRA,
)
PLATFORMS = [
Platform.BINARY_SENSOR,
Platform.CAMERA,
Platform.LIGHT,
Platform.SENSOR,
Platform.SWITCH,
]
def setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Skybell component."""
conf = config[DOMAIN]
username = conf.get(CONF_USERNAME)
password = conf.get(CONF_PASSWORD)
try:
cache = hass.config.path(DEFAULT_CACHEDB)
skybell = Skybell(
username=username,
password=password,
get_devices=True,
cache_path=cache,
agent_identifier=AGENT_IDENTIFIER,
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the SkyBell component."""
hass.data.setdefault(DOMAIN, {})
entry_config = {}
if DOMAIN not in config:
return True
for parameter, value in config[DOMAIN].items():
if parameter == CONF_USERNAME:
entry_config[CONF_EMAIL] = value
else:
entry_config[parameter] = value
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=entry_config,
)
)
hass.data[DOMAIN] = skybell
except (ConnectTimeout, HTTPError) as ex:
_LOGGER.error("Unable to connect to Skybell service: %s", str(ex))
persistent_notification.create(
hass,
"Error: {}<br />"
"You will need to restart hass after fixing."
"".format(ex),
title=NOTIFICATION_TITLE,
notification_id=NOTIFICATION_ID,
)
return False
# Clean up unused cache file since we are using an account specific name
# Remove with import
def clean_cache():
"""Clean old cache filename."""
if os.path.exists(hass.config.path(DEFAULT_CACHEDB)):
os.remove(hass.config.path(DEFAULT_CACHEDB))
await hass.async_add_executor_job(clean_cache)
return True
class SkybellDevice(Entity):
"""A HA implementation for Skybell devices."""
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Skybell from a config entry."""
email = entry.data[CONF_EMAIL]
password = entry.data[CONF_PASSWORD]
def __init__(self, device):
"""Initialize a sensor for Skybell device."""
self._device = device
api = Skybell(
username=email,
password=password,
get_devices=True,
cache_path=hass.config.path(f"./skybell_{entry.unique_id}.pickle"),
session=async_get_clientsession(hass),
)
try:
devices = await api.async_initialize()
except SkybellAuthenticationException:
return False
except SkybellException as ex:
raise ConfigEntryNotReady(f"Unable to connect to Skybell service: {ex}") from ex
def update(self):
"""Update automation state."""
self._device.refresh()
device_coordinators: list[SkybellDataUpdateCoordinator] = [
SkybellDataUpdateCoordinator(hass, device) for device in devices
]
await asyncio.gather(
*[
coordinator.async_config_entry_first_refresh()
for coordinator in device_coordinators
]
)
hass.data[DOMAIN][entry.entry_id] = device_coordinators
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
@property
def extra_state_attributes(self):
"""Return the state attributes."""
return {
ATTR_ATTRIBUTION: ATTRIBUTION,
"device_id": self._device.device_id,
"status": self._device.status,
"location": self._device.location,
"wifi_ssid": self._device.wifi_ssid,
"wifi_status": self._device.wifi_status,
"last_check_in": self._device.last_check_in,
"motion_threshold": self._device.motion_threshold,
"video_profile": self._device.video_profile,
}
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok

89
homeassistant/components/skybell/binary_sensor.py
View File

@ -1,9 +1,7 @@
"""Binary sensor support for the Skybell HD Doorbell."""
from __future__ import annotations
from datetime import timedelta
from typing import Any
from aioskybell.helpers import const as CONST
import voluptuous as vol
from homeassistant.components.binary_sensor import (
@ -12,36 +10,33 @@ from homeassistant.components.binary_sensor import (
BinarySensorEntity,
BinarySensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ENTITY_NAMESPACE, CONF_MONITORED_CONDITIONS
from homeassistant.core import HomeAssistant
from homeassistant.core import HomeAssistant, callback
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import DEFAULT_ENTITY_NAMESPACE, DOMAIN as SKYBELL_DOMAIN, SkybellDevice
from . import DOMAIN
from .coordinator import SkybellDataUpdateCoordinator
from .entity import SkybellEntity
SCAN_INTERVAL = timedelta(seconds=10)
BINARY_SENSOR_TYPES: dict[str, BinarySensorEntityDescription] = {
"button": BinarySensorEntityDescription(
key="device:sensor:button",
BINARY_SENSOR_TYPES: tuple[BinarySensorEntityDescription, ...] = (
BinarySensorEntityDescription(
key="button",
name="Button",
device_class=BinarySensorDeviceClass.OCCUPANCY,
),
"motion": BinarySensorEntityDescription(
key="device:sensor:motion",
BinarySensorEntityDescription(
key="motion",
name="Motion",
device_class=BinarySensorDeviceClass.MOTION,
),
}
)
# Deprecated in Home Assistant 2022.6
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Optional(
CONF_ENTITY_NAMESPACE, default=DEFAULT_ENTITY_NAMESPACE
): cv.string,
vol.Optional(CONF_ENTITY_NAMESPACE, default=DOMAIN): cv.string,
vol.Required(CONF_MONITORED_CONDITIONS, default=[]): vol.All(
cv.ensure_list, [vol.In(BINARY_SENSOR_TYPES)]
),
@ -49,53 +44,41 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
)
def setup_platform(
hass: HomeAssistant,
config: ConfigType,
add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the platform for a Skybell device."""
skybell = hass.data[SKYBELL_DOMAIN]
binary_sensors = [
SkybellBinarySensor(device, BINARY_SENSOR_TYPES[sensor_type])
for device in skybell.get_devices()
for sensor_type in config[CONF_MONITORED_CONDITIONS]
]
add_entities(binary_sensors, True)
"""Set up Skybell switch."""
async_add_entities(
SkybellBinarySensor(coordinator, sensor)
for sensor in BINARY_SENSOR_TYPES
for coordinator in hass.data[DOMAIN][entry.entry_id]
)
class SkybellBinarySensor(SkybellDevice, BinarySensorEntity):
class SkybellBinarySensor(SkybellEntity, BinarySensorEntity):
"""A binary sensor implementation for Skybell devices."""
def __init__(
self,
device,
coordinator: SkybellDataUpdateCoordinator,
description: BinarySensorEntityDescription,
):
) -> None:
"""Initialize a binary sensor for a Skybell device."""
super().__init__(device)
self.entity_description = description
self._attr_name = f"{self._device.name} {description.name}"
self._event: dict[Any, Any] = {}
super().__init__(coordinator, description)
self._event: dict[str, str] = {}
@property
def extra_state_attributes(self):
def extra_state_attributes(self) -> dict[str, str | int | tuple[str, str]]:
"""Return the state attributes."""
attrs = super().extra_state_attributes
attrs["event_date"] = self._event.get("createdAt")
if event := self._event.get(CONST.CREATED_AT):
attrs["event_date"] = event
return attrs
def update(self):
"""Get the latest data and updates the state."""
super().update()
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
event = self._device.latest(self.entity_description.key)
self._attr_is_on = bool(event and event.get("id") != self._event.get("id"))
self._event = event or {}
self._attr_is_on = bool(event.get(CONST.ID) != self._event.get(CONST.ID))
self._event = event
super()._handle_coordinator_update()

112
homeassistant/components/skybell/camera.py
View File

@ -1,31 +1,31 @@
"""Camera support for the Skybell HD Doorbell."""
from __future__ import annotations
from datetime import timedelta
import logging
import requests
import voluptuous as vol
from homeassistant.components.camera import PLATFORM_SCHEMA, Camera
from homeassistant.components.camera import (
PLATFORM_SCHEMA,
Camera,
CameraEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_MONITORED_CONDITIONS
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import EntityDescription
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import DOMAIN as SKYBELL_DOMAIN, SkybellDevice
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(seconds=90)
IMAGE_AVATAR = "avatar"
IMAGE_ACTIVITY = "activity"
CONF_ACTIVITY_NAME = "activity_name"
CONF_AVATAR_NAME = "avatar_name"
from .const import (
CONF_ACTIVITY_NAME,
CONF_AVATAR_NAME,
DOMAIN,
IMAGE_ACTIVITY,
IMAGE_AVATAR,
)
from .coordinator import SkybellDataUpdateCoordinator
from .entity import SkybellEntity
# Deprecated in Home Assistant 2022.6
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Optional(CONF_MONITORED_CONDITIONS, default=[IMAGE_AVATAR]): vol.All(
@ -36,71 +36,37 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
}
)
CAMERA_TYPES: tuple[CameraEntityDescription, ...] = (
CameraEntityDescription(key="activity", name="Last Activity"),
CameraEntityDescription(key="avatar", name="Camera"),
)
def setup_platform(
hass: HomeAssistant,
config: ConfigType,
add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the platform for a Skybell device."""
cond = config[CONF_MONITORED_CONDITIONS]
names = {}
names[IMAGE_ACTIVITY] = config.get(CONF_ACTIVITY_NAME)
names[IMAGE_AVATAR] = config.get(CONF_AVATAR_NAME)
skybell = hass.data[SKYBELL_DOMAIN]
sensors = []
for device in skybell.get_devices():
for camera_type in cond:
sensors.append(SkybellCamera(device, camera_type, names.get(camera_type)))
add_entities(sensors, True)
"""Set up Skybell switch."""
async_add_entities(
SkybellCamera(coordinator, description)
for description in CAMERA_TYPES
for coordinator in hass.data[DOMAIN][entry.entry_id]
)
class SkybellCamera(SkybellDevice, Camera):
class SkybellCamera(SkybellEntity, Camera):
"""A camera implementation for Skybell devices."""
def __init__(self, device, camera_type, name=None):
def __init__(
self,
coordinator: SkybellDataUpdateCoordinator,
description: EntityDescription,
) -> None:
"""Initialize a camera for a Skybell device."""
self._type = camera_type
SkybellDevice.__init__(self, device)
super().__init__(coordinator, description)
Camera.__init__(self)
if name is not None:
self._name = f"{self._device.name} {name}"
else:
self._name = self._device.name
self._url = None
self._response = None
@property
def name(self):
"""Return the name of the sensor."""
return self._name
@property
def image_url(self):
"""Get the camera image url based on type."""
if self._type == IMAGE_ACTIVITY:
return self._device.activity_image
return self._device.image
def camera_image(
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Get the latest camera image."""
super().update()
if self._url != self.image_url:
self._url = self.image_url
try:
self._response = requests.get(self._url, stream=True, timeout=10)
except requests.HTTPError as err:
_LOGGER.warning("Failed to get camera image: %s", err)
self._response = None
if not self._response:
return None
return self._response.content
return self._device.images[self.entity_description.key]

76
homeassistant/components/skybell/config_flow.py Normal file
View File

@ -0,0 +1,76 @@
"""Config flow for Skybell integration."""
from __future__ import annotations
from typing import Any
from aioskybell import Skybell, exceptions
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.typing import ConfigType
from .const import DOMAIN
class SkybellFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Skybell."""
async def async_step_import(self, user_input: ConfigType) -> FlowResult:
"""Import a config entry from configuration.yaml."""
if self._async_current_entries():
return self.async_abort(reason="already_configured")
return await self.async_step_user(user_input)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initiated by the user."""
errors = {}
if user_input is not None:
email = user_input[CONF_EMAIL].lower()
password = user_input[CONF_PASSWORD]
self._async_abort_entries_match({CONF_EMAIL: email})
user_id, error = await self._async_validate_input(email, password)
if error is None:
await self.async_set_unique_id(user_id)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=email,
data={CONF_EMAIL: email, CONF_PASSWORD: password},
)
errors["base"] = error
user_input = user_input or {}
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_EMAIL, default=user_input.get(CONF_EMAIL)): str,
vol.Required(CONF_PASSWORD): str,
}
),
errors=errors,
)
async def _async_validate_input(self, email: str, password: str) -> tuple:
"""Validate login credentials."""
skybell = Skybell(
username=email,
password=password,
disable_cache=True,
session=async_get_clientsession(self.hass),
)
try:
await skybell.async_initialize()
except exceptions.SkybellAuthenticationException:
return None, "invalid_auth"
except exceptions.SkybellException:
return None, "cannot_connect"
except Exception: # pylint: disable=broad-except
return None, "unknown"
return skybell.user_id, None

14
homeassistant/components/skybell/const.py Normal file
View File

@ -0,0 +1,14 @@
"""Constants for the Skybell HD Doorbell."""
import logging
from typing import Final
CONF_ACTIVITY_NAME = "activity_name"
CONF_AVATAR_NAME = "avatar_name"
DEFAULT_CACHEDB = "./skybell_cache.pickle"
DEFAULT_NAME = "SkyBell"
DOMAIN: Final = "skybell"
IMAGE_AVATAR = "avatar"
IMAGE_ACTIVITY = "activity"
LOGGER = logging.getLogger(__package__)

34
homeassistant/components/skybell/coordinator.py Normal file
View File

@ -0,0 +1,34 @@
"""Data update coordinator for the Skybell integration."""
from datetime import timedelta
from aioskybell import SkybellDevice, SkybellException
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import LOGGER
class SkybellDataUpdateCoordinator(DataUpdateCoordinator):
"""Data update coordinator for the Skybell integration."""
config_entry: ConfigEntry
def __init__(self, hass: HomeAssistant, device: SkybellDevice) -> None:
"""Initialize the coordinator."""
super().__init__(
hass=hass,
logger=LOGGER,
name=device.name,
update_interval=timedelta(seconds=30),
)
self.device = device
async def _async_update_data(self) -> None:
"""Fetch data from API endpoint."""
try:
await self.device.async_update()
except SkybellException as err:
raise UpdateFailed(f"Failed to communicate with device: {err}") from err

65
homeassistant/components/skybell/entity.py Normal file
View File

@ -0,0 +1,65 @@
"""Entity representing a Skybell HD Doorbell."""
from __future__ import annotations
from aioskybell import SkybellDevice
from homeassistant.const import ATTR_CONNECTIONS
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.entity import DeviceInfo, EntityDescription
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DEFAULT_NAME, DOMAIN
from .coordinator import SkybellDataUpdateCoordinator
class SkybellEntity(CoordinatorEntity[SkybellDataUpdateCoordinator]):
"""An HA implementation for Skybell entity."""
_attr_attribution = "Data provided by Skybell.com"
def __init__(
self, coordinator: SkybellDataUpdateCoordinator, description: EntityDescription
) -> None:
"""Initialize a SkyBell entity."""
super().__init__(coordinator)
self.entity_description = description
if description.name != coordinator.device.name:
self._attr_name = f"{self._device.name} {description.name}"
self._attr_unique_id = f"{self._device.device_id}_{description.key}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, self._device.device_id)},
manufacturer=DEFAULT_NAME,
model=self._device.type,
name=self._device.name,
sw_version=self._device.firmware_ver,
)
if self._device.mac:
self._attr_device_info[ATTR_CONNECTIONS] = {
(dr.CONNECTION_NETWORK_MAC, self._device.mac)
}
@property
def _device(self) -> SkybellDevice:
"""Return the device."""
return self.coordinator.device
@property
def extra_state_attributes(self) -> dict[str, str | int | tuple[str, str]]:
"""Return the state attributes."""
attr: dict[str, str | int | tuple[str, str]] = {
"device_id": self._device.device_id,
"status": self._device.status,
"location": self._device.location,
"motion_threshold": self._device.motion_threshold,
"video_profile": self._device.video_profile,
}
if self._device.owner:
attr["wifi_ssid"] = self._device.wifi_ssid
attr["wifi_status"] = self._device.wifi_status
attr["last_check_in"] = self._device.last_check_in
return attr
async def async_added_to_hass(self) -> None:
"""When entity is added to hass."""
await super().async_added_to_hass()
self._handle_coordinator_update()

89
homeassistant/components/skybell/light.py
View File

@ -1,82 +1,63 @@
"""Light/LED support for the Skybell HD Doorbell."""
from __future__ import annotations
from typing import Any
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
ATTR_HS_COLOR,
ATTR_RGB_COLOR,
ColorMode,
LightEntity,
LightEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
import homeassistant.util.color as color_util
from . import DOMAIN as SKYBELL_DOMAIN, SkybellDevice
from .const import DOMAIN
from .entity import SkybellEntity
def setup_platform(
hass: HomeAssistant,
config: ConfigType,
add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the platform for a Skybell device."""
skybell = hass.data[SKYBELL_DOMAIN]
sensors = []
for device in skybell.get_devices():
sensors.append(SkybellLight(device))
add_entities(sensors, True)
"""Set up Skybell switch."""
async_add_entities(
SkybellLight(
coordinator,
LightEntityDescription(
key=coordinator.device.name,
name=coordinator.device.name,
),
)
for coordinator in hass.data[DOMAIN][entry.entry_id]
)
def _to_skybell_level(level):
"""Convert the given Home Assistant light level (0-255) to Skybell (0-100)."""
return int((level * 100) / 255)
class SkybellLight(SkybellEntity, LightEntity):
"""A light implementation for Skybell devices."""
_attr_supported_color_modes = {ColorMode.BRIGHTNESS, ColorMode.RGB}
def _to_hass_level(level):
"""Convert the given Skybell (0-100) light level to Home Assistant (0-255)."""
return int((level * 255) / 100)
class SkybellLight(SkybellDevice, LightEntity):
"""A binary sensor implementation for Skybell devices."""
_attr_color_mode = ColorMode.HS
_attr_supported_color_modes = {ColorMode.HS}
def __init__(self, device):
"""Initialize a light for a Skybell device."""
super().__init__(device)
self._attr_name = device.name
def turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the light."""
if ATTR_HS_COLOR in kwargs:
rgb = color_util.color_hs_to_RGB(*kwargs[ATTR_HS_COLOR])
self._device.led_rgb = rgb
elif ATTR_BRIGHTNESS in kwargs:
self._device.led_intensity = _to_skybell_level(kwargs[ATTR_BRIGHTNESS])
else:
self._device.led_intensity = _to_skybell_level(255)
if ATTR_RGB_COLOR in kwargs:
rgb = kwargs[ATTR_RGB_COLOR]
await self._device.async_set_setting(ATTR_RGB_COLOR, rgb)
if ATTR_BRIGHTNESS in kwargs:
level = int((kwargs.get(ATTR_BRIGHTNESS, 0) * 100) / 255)
await self._device.async_set_setting(ATTR_BRIGHTNESS, level)
def turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the light."""
self._device.led_intensity = 0
await self._device.async_set_setting(ATTR_BRIGHTNESS, 0)
@property
def is_on(self):
def is_on(self) -> bool:
"""Return true if device is on."""
return self._device.led_intensity > 0
@property
def brightness(self):
def brightness(self) -> int:
"""Return the brightness of the light."""
return _to_hass_level(self._device.led_intensity)
@property
def hs_color(self):
"""Return the color of the light."""
return color_util.color_RGB_to_hs(*self._device.led_rgb)
return int((self._device.led_intensity * 255) / 100)

7
homeassistant/components/skybell/manifest.json
View File

@ -1,9 +1,10 @@
{
"domain": "skybell",
"name": "SkyBell",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/skybell",
"requirements": ["skybellpy==0.6.3"],
"codeowners": [],
"requirements": ["aioskybell==22.3.0"],
"codeowners": ["@tkdrob"],
"iot_class": "cloud_polling",
"loggers": ["skybellpy"]
"loggers": ["aioskybell"]
}

59
homeassistant/components/skybell/sensor.py
View File

@ -1,8 +1,6 @@
"""Sensor support for Skybell Doorbells."""
from __future__ import annotations
from datetime import timedelta
import voluptuous as vol
from homeassistant.components.sensor import (
@ -10,15 +8,13 @@ from homeassistant.components.sensor import (
SensorEntity,
SensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ENTITY_NAMESPACE, CONF_MONITORED_CONDITIONS
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import DEFAULT_ENTITY_NAMESPACE, DOMAIN as SKYBELL_DOMAIN, SkybellDevice
SCAN_INTERVAL = timedelta(seconds=30)
from .entity import DOMAIN, SkybellEntity
SENSOR_TYPES: tuple[SensorEntityDescription, ...] = (
SensorEntityDescription(
@ -27,14 +23,13 @@ SENSOR_TYPES: tuple[SensorEntityDescription, ...] = (
icon="mdi:bell-ring",
),
)
MONITORED_CONDITIONS: list[str] = [desc.key for desc in SENSOR_TYPES]
MONITORED_CONDITIONS = SENSOR_TYPES
# Deprecated in Home Assistant 2022.6
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Optional(
CONF_ENTITY_NAMESPACE, default=DEFAULT_ENTITY_NAMESPACE
): cv.string,
vol.Optional(CONF_ENTITY_NAMESPACE, default=DOMAIN): cv.string,
vol.Required(CONF_MONITORED_CONDITIONS, default=[]): vol.All(
cv.ensure_list, [vol.In(MONITORED_CONDITIONS)]
),
@ -42,41 +37,21 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
)
def setup_platform(
hass: HomeAssistant,
config: ConfigType,
add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the platform for a Skybell device."""
skybell = hass.data[SKYBELL_DOMAIN]
sensors = [
SkybellSensor(device, description)
for device in skybell.get_devices()
"""Set up Skybell sensor."""
async_add_entities(
SkybellSensor(coordinator, description)
for coordinator in hass.data[DOMAIN][entry.entry_id]
for description in SENSOR_TYPES
if description.key in config[CONF_MONITORED_CONDITIONS]
]
add_entities(sensors, True)
)
class SkybellSensor(SkybellDevice, SensorEntity):
class SkybellSensor(SkybellEntity, SensorEntity):
"""A sensor implementation for Skybell devices."""
def __init__(
self,
device,
description: SensorEntityDescription,
):
"""Initialize a sensor for a Skybell device."""
super().__init__(device)
self.entity_description = description
self._attr_name = f"{self._device.name} {description.name}"
def update(self):
"""Get the latest data and updates the state."""
super().update()
if self.entity_description.key == "chime_level":
self._attr_native_value = self._device.outdoor_chime_level
@property
def native_value(self) -> int:
"""Return the state of the sensor."""
return self._device.outdoor_chime_level

21
homeassistant/components/skybell/strings.json Normal file
View File

@ -0,0 +1,21 @@
{
"config": {
"step": {
"user": {
"data": {
"email": "[%key:common::config_flow::data::email%]",
"password": "[%key:common::config_flow::data::password%]"
}
}
},
"error": {
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
}
}
}

64
homeassistant/components/skybell/switch.py
View File

@ -1,6 +1,8 @@
"""Switch support for the Skybell HD Doorbell."""
from __future__ import annotations
from typing import Any, cast
import voluptuous as vol
from homeassistant.components.switch import (
@ -8,13 +10,14 @@ from homeassistant.components.switch import (
SwitchEntity,
SwitchEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ENTITY_NAMESPACE, CONF_MONITORED_CONDITIONS
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import DEFAULT_ENTITY_NAMESPACE, DOMAIN as SKYBELL_DOMAIN, SkybellDevice
from .const import DOMAIN
from .entity import SkybellEntity
SWITCH_TYPES: tuple[SwitchEntityDescription, ...] = (
SwitchEntityDescription(
@ -26,62 +29,41 @@ SWITCH_TYPES: tuple[SwitchEntityDescription, ...] = (
name="Motion Sensor",
),
)
MONITORED_CONDITIONS: list[str] = [desc.key for desc in SWITCH_TYPES]
# Deprecated in Home Assistant 2022.6
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Optional(
CONF_ENTITY_NAMESPACE, default=DEFAULT_ENTITY_NAMESPACE
): cv.string,
vol.Optional(CONF_ENTITY_NAMESPACE, default=DOMAIN): cv.string,
vol.Required(CONF_MONITORED_CONDITIONS, default=[]): vol.All(
cv.ensure_list, [vol.In(MONITORED_CONDITIONS)]
cv.ensure_list, [vol.In(SWITCH_TYPES)]
),
}
)
def setup_platform(
hass: HomeAssistant,
config: ConfigType,
add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the platform for a Skybell device."""
skybell = hass.data[SKYBELL_DOMAIN]
switches = [
SkybellSwitch(device, description)
for device in skybell.get_devices()
"""Set up the SkyBell switch."""
async_add_entities(
SkybellSwitch(coordinator, description)
for coordinator in hass.data[DOMAIN][entry.entry_id]
for description in SWITCH_TYPES
if description.key in config[CONF_MONITORED_CONDITIONS]
]
add_entities(switches, True)
)
class SkybellSwitch(SkybellDevice, SwitchEntity):
class SkybellSwitch(SkybellEntity, SwitchEntity):
"""A switch implementation for Skybell devices."""
def __init__(
self,
device,
description: SwitchEntityDescription,
):
"""Initialize a light for a Skybell device."""
super().__init__(device)
self.entity_description = description
self._attr_name = f"{self._device.name} {description.name}"
def turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the switch."""
setattr(self._device, self.entity_description.key, True)
await self._device.async_set_setting(self.entity_description.key, True)
def turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the switch."""
setattr(self._device, self.entity_description.key, False)
await self._device.async_set_setting(self.entity_description.key, False)
@property
def is_on(self):
"""Return true if device is on."""
return getattr(self._device, self.entity_description.key)
def is_on(self) -> bool:
"""Return true if entity is on."""
return cast(bool, getattr(self._device, self.entity_description.key))

21
homeassistant/components/skybell/translations/en.json Normal file
View File

@ -0,0 +1,21 @@
{
"config": {
"step": {
"user": {
"data": {
"email": "Email",
"password": "Password"
}
}
},
"error": {
"invalid_auth": "Invalid authentication",
"cannot_connect": "Failed to connect",
"unknown": "Unexpected error"
},
"abort": {
"already_configured": "Account is already configured",
"reauth_successful": "Re-authentication was successful"
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -311,6 +311,7 @@ FLOWS = {
"shopping_list",
"sia",
"simplisafe",
"skybell",
"slack",
"sleepiq",
"slimproto",

6
requirements_all.txt
View File

@ -240,6 +240,9 @@ aiosenz==1.0.0
# homeassistant.components.shelly
aioshelly==2.0.0
# homeassistant.components.skybell
aioskybell==22.3.0
# homeassistant.components.slimproto
aioslimproto==2.0.1
@ -2173,9 +2176,6 @@ simplisafe-python==2022.05.2
# homeassistant.components.sisyphus
sisyphus-control==3.1.2
# homeassistant.components.skybell
skybellpy==0.6.3
# homeassistant.components.slack
slackclient==2.5.0

3
requirements_test_all.txt
View File

@ -209,6 +209,9 @@ aiosenz==1.0.0
# homeassistant.components.shelly
aioshelly==2.0.0
# homeassistant.components.skybell
aioskybell==22.3.0
# homeassistant.components.slimproto
aioslimproto==2.0.1

30
tests/components/skybell/__init__.py Normal file
View File

@ -0,0 +1,30 @@
"""Tests for the SkyBell integration."""
from unittest.mock import AsyncMock, patch
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
USERNAME = "user"
PASSWORD = "password"
USER_ID = "123456789012345678901234"
CONF_CONFIG_FLOW = {
CONF_EMAIL: USERNAME,
CONF_PASSWORD: PASSWORD,
}
def _patch_skybell_devices() -> None:
mocked_skybell = AsyncMock()
mocked_skybell.user_id = USER_ID
return patch(
"homeassistant.components.skybell.config_flow.Skybell.async_get_devices",
return_value=[mocked_skybell],
)
def _patch_skybell() -> None:
return patch(
"homeassistant.components.skybell.config_flow.Skybell.async_send_request",
return_value={"id": USER_ID},
)

137
tests/components/skybell/test_config_flow.py Normal file
View File

@ -0,0 +1,137 @@
"""Test SkyBell config flow."""
from unittest.mock import patch
from aioskybell import exceptions
from homeassistant.components.skybell.const import DOMAIN
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import (
RESULT_TYPE_ABORT,
RESULT_TYPE_CREATE_ENTRY,
RESULT_TYPE_FORM,
)
from . import CONF_CONFIG_FLOW, _patch_skybell, _patch_skybell_devices
from tests.common import MockConfigEntry
def _patch_setup_entry() -> None:
return patch(
"homeassistant.components.skybell.async_setup_entry",
return_value=True,
)
def _patch_setup() -> None:
return patch(
"homeassistant.components.skybell.async_setup",
return_value=True,
)
async def test_flow_user(hass: HomeAssistant) -> None:
"""Test that the user step works."""
with _patch_skybell(), _patch_skybell_devices(), _patch_setup_entry(), _patch_setup():
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONF_CONFIG_FLOW,
)
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "user"
assert result["data"] == CONF_CONFIG_FLOW
async def test_flow_user_already_configured(hass: HomeAssistant) -> None:
"""Test user initialized flow with duplicate server."""
entry = MockConfigEntry(
domain=DOMAIN,
data=CONF_CONFIG_FLOW,
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONF_CONFIG_FLOW
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_flow_user_cannot_connect(hass: HomeAssistant) -> None:
"""Test user initialized flow with unreachable server."""
with _patch_skybell() as skybell_mock:
skybell_mock.side_effect = exceptions.SkybellException(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONF_CONFIG_FLOW
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "cannot_connect"}
async def test_invalid_credentials(hass: HomeAssistant) -> None:
"""Test that invalid credentials throws an error."""
with patch("homeassistant.components.skybell.Skybell.async_login") as skybell_mock:
skybell_mock.side_effect = exceptions.SkybellAuthenticationException(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONF_CONFIG_FLOW
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "invalid_auth"}
async def test_flow_user_unknown_error(hass: HomeAssistant) -> None:
"""Test user initialized flow with unreachable server."""
with _patch_skybell_devices() as skybell_mock:
skybell_mock.side_effect = Exception
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONF_CONFIG_FLOW
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "unknown"}
async def test_flow_import(hass: HomeAssistant) -> None:
"""Test import step."""
with _patch_skybell(), _patch_skybell_devices(), _patch_setup_entry(), _patch_setup():
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONF_CONFIG_FLOW,
)
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "user"
assert result["data"] == CONF_CONFIG_FLOW
async def test_flow_import_already_configured(hass: HomeAssistant) -> None:
"""Test import step already configured."""
entry = MockConfigEntry(
domain=DOMAIN, unique_id="123456789012345678901234", data=CONF_CONFIG_FLOW
)
entry.add_to_hass(hass)
with _patch_skybell():
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"