Add config flow to Verisure (#47880)

Co-authored-by: Paulus Schoutsen <balloob@gmail.com>
This commit is contained in:
Franck Nijhof 2021-03-15 20:30:44 +01:00 committed by GitHub
parent 9f4c2f6260
commit 059e9e8307
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
19 changed files with 996 additions and 200 deletions

View File

@ -1064,7 +1064,14 @@ omit =
homeassistant/components/velbus/switch.py
homeassistant/components/velux/*
homeassistant/components/venstar/climate.py
homeassistant/components/verisure/*
homeassistant/components/verisure/__init__.py
homeassistant/components/verisure/alarm_control_panel.py
homeassistant/components/verisure/binary_sensor.py
homeassistant/components/verisure/camera.py
homeassistant/components/verisure/coordinator.py
homeassistant/components/verisure/lock.py
homeassistant/components/verisure/sensor.py
homeassistant/components/verisure/switch.py
homeassistant/components/versasense/*
homeassistant/components/vesync/__init__.py
homeassistant/components/vesync/common.py

View File

@ -1,6 +1,10 @@
"""Support for Verisure devices."""
from __future__ import annotations
import asyncio
import os
from typing import Any
from verisure import Error as VerisureError
import voluptuous as vol
@ -12,34 +16,28 @@ from homeassistant.components.camera import DOMAIN as CAMERA_DOMAIN
from homeassistant.components.lock import DOMAIN as LOCK_DOMAIN
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
CONF_EMAIL,
CONF_PASSWORD,
CONF_SCAN_INTERVAL,
CONF_USERNAME,
EVENT_HOMEASSISTANT_STOP,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import discovery
from homeassistant.exceptions import ConfigEntryNotReady
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType
from homeassistant.helpers.storage import STORAGE_DIR
from .const import (
ATTR_DEVICE_SERIAL,
CONF_ALARM,
CONF_CODE_DIGITS,
CONF_DEFAULT_LOCK_CODE,
CONF_DOOR_WINDOW,
CONF_GIID,
CONF_HYDROMETERS,
CONF_LOCKS,
CONF_MOUSE,
CONF_SMARTCAM,
CONF_SMARTPLUGS,
CONF_THERMOMETERS,
DEFAULT_SCAN_INTERVAL,
CONF_LOCK_CODE_DIGITS,
CONF_LOCK_DEFAULT_CODE,
DEFAULT_LOCK_CODE_DIGITS,
DOMAIN,
LOGGER,
MIN_SCAN_INTERVAL,
SERVICE_CAPTURE_SMARTCAM,
SERVICE_DISABLE_AUTOLOCK,
SERVICE_ENABLE_AUTOLOCK,
@ -56,54 +54,101 @@ PLATFORMS = [
]
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.Schema(
{
vol.Required(CONF_PASSWORD): cv.string,
vol.Required(CONF_USERNAME): cv.string,
vol.Optional(CONF_ALARM, default=True): cv.boolean,
vol.Optional(CONF_CODE_DIGITS, default=4): cv.positive_int,
vol.Optional(CONF_DOOR_WINDOW, default=True): cv.boolean,
vol.Optional(CONF_GIID): cv.string,
vol.Optional(CONF_HYDROMETERS, default=True): cv.boolean,
vol.Optional(CONF_LOCKS, default=True): cv.boolean,
vol.Optional(CONF_DEFAULT_LOCK_CODE): cv.string,
vol.Optional(CONF_MOUSE, default=True): cv.boolean,
vol.Optional(CONF_SMARTPLUGS, default=True): cv.boolean,
vol.Optional(CONF_THERMOMETERS, default=True): cv.boolean,
vol.Optional(CONF_SMARTCAM, default=True): cv.boolean,
vol.Optional(CONF_SCAN_INTERVAL, default=DEFAULT_SCAN_INTERVAL): (
vol.All(cv.time_period, vol.Clamp(min=MIN_SCAN_INTERVAL))
),
}
)
},
vol.All(
cv.deprecated(DOMAIN),
{
DOMAIN: vol.Schema(
{
vol.Required(CONF_PASSWORD): cv.string,
vol.Required(CONF_USERNAME): cv.string,
vol.Optional(CONF_CODE_DIGITS): cv.positive_int,
vol.Optional(CONF_GIID): cv.string,
vol.Optional(CONF_DEFAULT_LOCK_CODE): cv.string,
},
extra=vol.ALLOW_EXTRA,
)
},
),
extra=vol.ALLOW_EXTRA,
)
DEVICE_SERIAL_SCHEMA = vol.Schema({vol.Required(ATTR_DEVICE_SERIAL): cv.string})
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
async def async_setup(hass: HomeAssistant, config: dict[str, Any]) -> bool:
"""Set up the Verisure integration."""
coordinator = VerisureDataUpdateCoordinator(hass, config=config[DOMAIN])
if DOMAIN in config:
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={
CONF_EMAIL: config[DOMAIN][CONF_USERNAME],
CONF_PASSWORD: config[DOMAIN][CONF_PASSWORD],
CONF_GIID: config[DOMAIN].get(CONF_GIID),
CONF_LOCK_CODE_DIGITS: config[DOMAIN].get(CONF_CODE_DIGITS),
CONF_LOCK_DEFAULT_CODE: config[DOMAIN].get(CONF_LOCK_DEFAULT_CODE),
},
)
)
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Verisure from a config entry."""
# Migrate old YAML settings (hidden in the config entry),
# to config entry options. Can be removed after YAML support is gone.
if CONF_LOCK_CODE_DIGITS in entry.data or CONF_DEFAULT_LOCK_CODE in entry.data:
options = entry.options.copy()
if (
CONF_LOCK_CODE_DIGITS in entry.data
and CONF_LOCK_CODE_DIGITS not in entry.options
and entry.data[CONF_LOCK_CODE_DIGITS] != DEFAULT_LOCK_CODE_DIGITS
):
options.update(
{
CONF_LOCK_CODE_DIGITS: entry.data[CONF_LOCK_CODE_DIGITS],
}
)
if (
CONF_DEFAULT_LOCK_CODE in entry.data
and CONF_DEFAULT_LOCK_CODE not in entry.options
):
options.update(
{
CONF_DEFAULT_LOCK_CODE: entry.data[CONF_DEFAULT_LOCK_CODE],
}
)
data = entry.data.copy()
data.pop(CONF_LOCK_CODE_DIGITS, None)
data.pop(CONF_DEFAULT_LOCK_CODE, None)
hass.config_entries.async_update_entry(entry, data=data, options=options)
# Continue as normal...
coordinator = VerisureDataUpdateCoordinator(hass, entry=entry)
if not await coordinator.async_login():
LOGGER.error("Login failed")
LOGGER.error("Could not login to Verisure, aborting setting up integration")
return False
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, coordinator.async_logout)
await coordinator.async_refresh()
if not coordinator.last_update_success:
LOGGER.error("Update failed")
return False
raise ConfigEntryNotReady
hass.data[DOMAIN] = coordinator
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][entry.entry_id] = coordinator
# Set up all platforms for this device/entry.
for platform in PLATFORMS:
hass.async_create_task(
discovery.async_load_platform(hass, platform, DOMAIN, {}, config)
hass.config_entries.async_forward_entry_setup(entry, platform)
)
async def capture_smartcam(service):
@ -145,3 +190,31 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
DOMAIN, SERVICE_ENABLE_AUTOLOCK, enable_autolock, schema=DEVICE_SERIAL_SCHEMA
)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload Verisure config entry."""
unload_ok = all(
await asyncio.gather(
*(
hass.config_entries.async_forward_entry_unload(entry, platform)
for platform in PLATFORMS
)
)
)
if not unload_ok:
return False
cookie_file = hass.config.path(STORAGE_DIR, f"verisure_{entry.entry_id}")
try:
await hass.async_add_executor_job(os.unlink, cookie_file)
except FileNotFoundError:
pass
del hass.data[DOMAIN][entry.entry_id]
if not hass.data[DOMAIN]:
del hass.data[DOMAIN]
return True

View File

@ -2,7 +2,7 @@
from __future__ import annotations
import asyncio
from typing import Any, Callable
from typing import Callable, Iterable
from homeassistant.components.alarm_control_panel import (
FORMAT_NUMBER,
@ -12,6 +12,7 @@ from homeassistant.components.alarm_control_panel.const import (
SUPPORT_ALARM_ARM_AWAY,
SUPPORT_ALARM_ARM_HOME,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
STATE_ALARM_ARMED_AWAY,
STATE_ALARM_ARMED_HOME,
@ -22,22 +23,17 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import CONF_ALARM, CONF_GIID, DOMAIN, LOGGER
from .const import CONF_GIID, DOMAIN, LOGGER
from .coordinator import VerisureDataUpdateCoordinator
def setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: dict[str, Any],
add_entities: Callable[[list[Entity], bool], None],
discovery_info: dict[str, Any] | None = None,
entry: ConfigEntry,
async_add_entities: Callable[[Iterable[Entity]], None],
) -> None:
"""Set up the Verisure platform."""
coordinator = hass.data[DOMAIN]
alarms = []
if int(coordinator.config.get(CONF_ALARM, 1)):
alarms.append(VerisureAlarm(coordinator))
add_entities(alarms)
"""Set up Verisure alarm control panel from a config entry."""
async_add_entities([VerisureAlarm(coordinator=hass.data[DOMAIN][entry.entry_id])])
class VerisureAlarm(CoordinatorEntity, AlarmControlPanelEntity):
@ -53,17 +49,12 @@ class VerisureAlarm(CoordinatorEntity, AlarmControlPanelEntity):
@property
def name(self) -> str:
"""Return the name of the device."""
giid = self.coordinator.config.get(CONF_GIID)
if giid is not None:
aliass = {
i["giid"]: i["alias"] for i in self.coordinator.verisure.installations
}
if giid in aliass:
return "{} alarm".format(aliass[giid])
return "Verisure Alarm"
LOGGER.error("Verisure installation giid not found: %s", giid)
return "{} alarm".format(self.coordinator.verisure.installations[0]["alias"])
@property
def unique_id(self) -> str:
"""Return the unique ID for this alarm control panel."""
return self.coordinator.entry.data[CONF_GIID]
@property
def state(self) -> str | None:

View File

@ -1,40 +1,38 @@
"""Support for Verisure binary sensors."""
from __future__ import annotations
from typing import Any, Callable
from typing import Callable, Iterable
from homeassistant.components.binary_sensor import (
DEVICE_CLASS_CONNECTIVITY,
DEVICE_CLASS_OPENING,
BinarySensorEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import CONF_DOOR_WINDOW, DOMAIN
from . import DOMAIN
from .coordinator import VerisureDataUpdateCoordinator
def setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: dict[str, Any],
add_entities: Callable[[list[CoordinatorEntity]], None],
discovery_info: dict[str, Any] | None = None,
entry: ConfigEntry,
async_add_entities: Callable[[Iterable[Entity]], None],
) -> None:
"""Set up the Verisure binary sensors."""
coordinator = hass.data[DOMAIN]
"""Set up Verisure sensors based on a config entry."""
coordinator: VerisureDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
sensors: list[CoordinatorEntity] = [VerisureEthernetStatus(coordinator)]
sensors: list[Entity] = [VerisureEthernetStatus(coordinator)]
if int(coordinator.config.get(CONF_DOOR_WINDOW, 1)):
sensors.extend(
[
VerisureDoorWindowSensor(coordinator, serial_number)
for serial_number in coordinator.data["door_window"]
]
)
sensors.extend(
VerisureDoorWindowSensor(coordinator, serial_number)
for serial_number in coordinator.data["door_window"]
)
add_entities(sensors)
async_add_entities(sensors)
class VerisureDoorWindowSensor(CoordinatorEntity, BinarySensorEntity):

View File

@ -3,34 +3,31 @@ from __future__ import annotations
import errno
import os
from typing import Any, Callable
from typing import Callable, Iterable
from homeassistant.components.camera import Camera
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EVENT_HOMEASSISTANT_STOP
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import CONF_SMARTCAM, DOMAIN, LOGGER
from .const import DOMAIN, LOGGER
from .coordinator import VerisureDataUpdateCoordinator
def setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: dict[str, Any],
add_entities: Callable[[list[VerisureSmartcam]], None],
discovery_info: dict[str, Any] | None = None,
entry: ConfigEntry,
async_add_entities: Callable[[Iterable[Entity]], None],
) -> None:
"""Set up the Verisure Camera."""
coordinator: VerisureDataUpdateCoordinator = hass.data[DOMAIN]
if not int(coordinator.config.get(CONF_SMARTCAM, 1)):
return
"""Set up Verisure sensors based on a config entry."""
coordinator: VerisureDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
assert hass.config.config_dir
add_entities(
[
VerisureSmartcam(hass, coordinator, serial_number, hass.config.config_dir)
for serial_number in coordinator.data["cameras"]
]
async_add_entities(
VerisureSmartcam(hass, coordinator, serial_number, hass.config.config_dir)
for serial_number in coordinator.data["cameras"]
)

View File

@ -0,0 +1,186 @@
"""Config flow for Verisure integration."""
from __future__ import annotations
from typing import Any
from verisure import (
Error as VerisureError,
LoginError as VerisureLoginError,
ResponseError as VerisureResponseError,
Session as Verisure,
)
import voluptuous as vol
from homeassistant.config_entries import (
CONN_CLASS_CLOUD_POLL,
ConfigEntry,
ConfigFlow,
OptionsFlow,
)
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.core import callback
from .const import ( # pylint:disable=unused-import
CONF_GIID,
CONF_LOCK_CODE_DIGITS,
CONF_LOCK_DEFAULT_CODE,
DEFAULT_LOCK_CODE_DIGITS,
DOMAIN,
LOGGER,
)
class VerisureConfigFlowHandler(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Verisure."""
VERSION = 1
CONNECTION_CLASS = CONN_CLASS_CLOUD_POLL
installations: dict[str, str]
email: str
password: str
# These can be removed after YAML import has been removed.
giid: str | None = None
settings: dict[str, int | str]
def __init__(self):
"""Initialize."""
self.settings = {}
@staticmethod
@callback
def async_get_options_flow(config_entry: ConfigEntry) -> VerisureOptionsFlowHandler:
"""Get the options flow for this handler."""
return VerisureOptionsFlowHandler(config_entry)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> dict[str, Any]:
"""Handle the initial step."""
errors: dict[str, str] = {}
if user_input is not None:
verisure = Verisure(
username=user_input[CONF_EMAIL], password=user_input[CONF_PASSWORD]
)
try:
await self.hass.async_add_executor_job(verisure.login)
except VerisureLoginError as ex:
LOGGER.debug("Could not log in to Verisure, %s", ex)
errors["base"] = "invalid_auth"
except (VerisureError, VerisureResponseError) as ex:
LOGGER.debug("Unexpected response from Verisure, %s", ex)
errors["base"] = "unknown"
else:
self.email = user_input[CONF_EMAIL]
self.password = user_input[CONF_PASSWORD]
self.installations = {
inst["giid"]: f"{inst['alias']} ({inst['street']})"
for inst in verisure.installations
}
return await self.async_step_installation()
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_EMAIL): str,
vol.Required(CONF_PASSWORD): str,
}
),
errors=errors,
)
async def async_step_installation(
self, user_input: dict[str, Any] | None = None
) -> dict[str, Any]:
"""Select Verisure installation to add."""
if len(self.installations) == 1:
user_input = {CONF_GIID: list(self.installations)[0]}
elif self.giid and self.giid in self.installations:
user_input = {CONF_GIID: self.giid}
if user_input is None:
return self.async_show_form(
step_id="installation",
data_schema=vol.Schema(
{vol.Required(CONF_GIID): vol.In(self.installations)}
),
)
await self.async_set_unique_id(user_input[CONF_GIID])
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=self.installations[user_input[CONF_GIID]],
data={
CONF_EMAIL: self.email,
CONF_PASSWORD: self.password,
CONF_GIID: user_input[CONF_GIID],
**self.settings,
},
)
async def async_step_import(self, user_input: dict[str, Any]) -> dict[str, Any]:
"""Import Verisure YAML configuration."""
if user_input[CONF_GIID]:
self.giid = user_input[CONF_GIID]
await self.async_set_unique_id(self.giid)
self._abort_if_unique_id_configured()
else:
# The old YAML configuration could handle 1 single Verisure instance.
# Therefore, if we don't know the GIID, we can use the discovery
# without a unique ID logic, to prevent re-import/discovery.
await self._async_handle_discovery_without_unique_id()
# Settings, later to be converted to config entry options
if user_input[CONF_LOCK_CODE_DIGITS]:
self.settings[CONF_LOCK_CODE_DIGITS] = user_input[CONF_LOCK_CODE_DIGITS]
if user_input[CONF_LOCK_DEFAULT_CODE]:
self.settings[CONF_LOCK_DEFAULT_CODE] = user_input[CONF_LOCK_DEFAULT_CODE]
return await self.async_step_user(user_input)
class VerisureOptionsFlowHandler(OptionsFlow):
"""Handle Verisure options."""
def __init__(self, entry: ConfigEntry) -> None:
"""Initialize Verisure options flow."""
self.entry = entry
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> dict[str, Any]:
"""Manage Verisure options."""
errors = {}
if user_input is not None:
if len(user_input[CONF_LOCK_DEFAULT_CODE]) not in [
0,
user_input[CONF_LOCK_CODE_DIGITS],
]:
errors["base"] = "code_format_mismatch"
else:
return self.async_create_entry(title="", data=user_input)
return self.async_show_form(
step_id="init",
data_schema=vol.Schema(
{
vol.Optional(
CONF_LOCK_CODE_DIGITS,
default=self.entry.options.get(
CONF_LOCK_CODE_DIGITS, DEFAULT_LOCK_CODE_DIGITS
),
): int,
vol.Optional(
CONF_LOCK_DEFAULT_CODE,
default=self.entry.options.get(CONF_LOCK_DEFAULT_CODE),
): str,
}
),
errors=errors,
)

View File

@ -8,21 +8,17 @@ LOGGER = logging.getLogger(__package__)
ATTR_DEVICE_SERIAL = "device_serial"
CONF_ALARM = "alarm"
CONF_CODE_DIGITS = "code_digits"
CONF_DOOR_WINDOW = "door_window"
CONF_GIID = "giid"
CONF_HYDROMETERS = "hygrometers"
CONF_LOCKS = "locks"
CONF_DEFAULT_LOCK_CODE = "default_lock_code"
CONF_MOUSE = "mouse"
CONF_SMARTPLUGS = "smartplugs"
CONF_THERMOMETERS = "thermometers"
CONF_SMARTCAM = "smartcam"
CONF_LOCK_CODE_DIGITS = "lock_code_digits"
CONF_LOCK_DEFAULT_CODE = "lock_default_code"
DEFAULT_SCAN_INTERVAL = timedelta(minutes=1)
MIN_SCAN_INTERVAL = timedelta(minutes=1)
DEFAULT_LOCK_CODE_DIGITS = 4
SERVICE_CAPTURE_SMARTCAM = "capture_smartcam"
SERVICE_DISABLE_AUTOLOCK = "disable_autolock"
SERVICE_ENABLE_AUTOLOCK = "enable_autolock"
# Legacy; to remove after YAML removal
CONF_CODE_DIGITS = "code_digits"
CONF_DEFAULT_LOCK_CODE = "default_lock_code"

View File

@ -9,9 +9,10 @@ from verisure import (
Session as Verisure,
)
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, HTTP_SERVICE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.helpers.typing import ConfigType
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD, HTTP_SERVICE_UNAVAILABLE
from homeassistant.core import Event, HomeAssistant
from homeassistant.helpers.storage import STORAGE_DIR
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from homeassistant.util import Throttle
@ -21,14 +22,15 @@ from .const import CONF_GIID, DEFAULT_SCAN_INTERVAL, DOMAIN, LOGGER
class VerisureDataUpdateCoordinator(DataUpdateCoordinator):
"""A Verisure Data Update Coordinator."""
def __init__(self, hass: HomeAssistant, config: ConfigType) -> None:
def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Initialize the Verisure hub."""
self.imageseries = {}
self.config = config
self.giid = config.get(CONF_GIID)
self.entry = entry
self.verisure = Verisure(
username=config[CONF_USERNAME], password=config[CONF_PASSWORD]
username=entry.data[CONF_EMAIL],
password=entry.data[CONF_PASSWORD],
cookieFileName=hass.config.path(STORAGE_DIR, f"verisure_{entry.entry_id}"),
)
super().__init__(
@ -42,11 +44,14 @@ class VerisureDataUpdateCoordinator(DataUpdateCoordinator):
except VerisureError as ex:
LOGGER.error("Could not log in to verisure, %s", ex)
return False
if self.giid:
return await self.async_set_giid()
await self.hass.async_add_executor_job(
self.verisure.set_giid, self.entry.data[CONF_GIID]
)
return True
async def async_logout(self) -> bool:
async def async_logout(self, _event: Event) -> bool:
"""Logout from Verisure."""
try:
await self.hass.async_add_executor_job(self.verisure.logout)
@ -55,15 +60,6 @@ class VerisureDataUpdateCoordinator(DataUpdateCoordinator):
return False
return True
async def async_set_giid(self) -> bool:
"""Set installation GIID."""
try:
await self.hass.async_add_executor_job(self.verisure.set_giid, self.giid)
except VerisureError as ex:
LOGGER.error("Could not set installation GIID, %s", ex)
return False
return True
async def _async_update_data(self) -> dict:
"""Fetch data from Verisure."""
try:

View File

@ -2,35 +2,36 @@
from __future__ import annotations
import asyncio
from typing import Any, Callable
from typing import Callable, Iterable
from homeassistant.components.lock import LockEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_CODE, STATE_LOCKED, STATE_UNLOCKED
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import CONF_CODE_DIGITS, CONF_DEFAULT_LOCK_CODE, CONF_LOCKS, DOMAIN, LOGGER
from .const import (
CONF_LOCK_CODE_DIGITS,
CONF_LOCK_DEFAULT_CODE,
DEFAULT_LOCK_CODE_DIGITS,
DOMAIN,
LOGGER,
)
from .coordinator import VerisureDataUpdateCoordinator
def setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: dict[str, Any],
add_entities: Callable[[list[VerisureDoorlock]], None],
discovery_info: dict[str, Any] | None = None,
entry: ConfigEntry,
async_add_entities: Callable[[Iterable[Entity]], None],
) -> None:
"""Set up the Verisure lock platform."""
coordinator = hass.data[DOMAIN]
locks = []
if int(coordinator.config.get(CONF_LOCKS, 1)):
locks.extend(
[
VerisureDoorlock(coordinator, serial_number)
for serial_number in coordinator.data["locks"]
]
)
add_entities(locks)
"""Set up Verisure alarm control panel from a config entry."""
coordinator: VerisureDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
async_add_entities(
VerisureDoorlock(coordinator, serial_number)
for serial_number in coordinator.data["locks"]
)
class VerisureDoorlock(CoordinatorEntity, LockEntity):
@ -45,8 +46,9 @@ class VerisureDoorlock(CoordinatorEntity, LockEntity):
super().__init__(coordinator)
self.serial_number = serial_number
self._state = None
self._digits = coordinator.config.get(CONF_CODE_DIGITS)
self._default_lock_code = coordinator.config.get(CONF_DEFAULT_LOCK_CODE)
self._digits = coordinator.entry.options.get(
CONF_LOCK_CODE_DIGITS, DEFAULT_LOCK_CODE_DIGITS
)
@property
def name(self) -> str:
@ -80,7 +82,9 @@ class VerisureDoorlock(CoordinatorEntity, LockEntity):
async def async_unlock(self, **kwargs) -> None:
"""Send unlock command."""
code = kwargs.get(ATTR_CODE, self._default_lock_code)
code = kwargs.get(
ATTR_CODE, self.coordinator.entry.options.get(CONF_LOCK_DEFAULT_CODE)
)
if code is None:
LOGGER.error("Code required but none provided")
return
@ -89,7 +93,9 @@ class VerisureDoorlock(CoordinatorEntity, LockEntity):
async def async_lock(self, **kwargs) -> None:
"""Send lock command."""
code = kwargs.get(ATTR_CODE, self._default_lock_code)
code = kwargs.get(
ATTR_CODE, self.coordinator.entry.options.get(CONF_LOCK_DEFAULT_CODE)
)
if code is None:
LOGGER.error("Code required but none provided")
return

View File

@ -3,5 +3,7 @@
"name": "Verisure",
"documentation": "https://www.home-assistant.io/integrations/verisure",
"requirements": ["vsure==1.7.3"],
"codeowners": ["@frenck"]
"codeowners": ["@frenck"],
"config_flow": true,
"dhcp": [{ "macaddress": "0023C1*" }]
}

View File

@ -1,54 +1,44 @@
"""Support for Verisure sensors."""
from __future__ import annotations
from typing import Any, Callable
from typing import Callable, Iterable
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import PERCENTAGE, TEMP_CELSIUS
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import CONF_HYDROMETERS, CONF_MOUSE, CONF_THERMOMETERS, DOMAIN
from .const import DOMAIN
from .coordinator import VerisureDataUpdateCoordinator
def setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: dict[str, Any],
add_entities: Callable[[list[CoordinatorEntity], bool], None],
discovery_info: dict[str, Any] | None = None,
entry: ConfigEntry,
async_add_entities: Callable[[Iterable[Entity]], None],
) -> None:
"""Set up the Verisure platform."""
coordinator = hass.data[DOMAIN]
"""Set up Verisure sensors based on a config entry."""
coordinator: VerisureDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
sensors: list[CoordinatorEntity] = []
if int(coordinator.config.get(CONF_THERMOMETERS, 1)):
sensors.extend(
[
VerisureThermometer(coordinator, serial_number)
for serial_number, values in coordinator.data["climate"].items()
if "temperature" in values
]
)
sensors: list[Entity] = [
VerisureThermometer(coordinator, serial_number)
for serial_number, values in coordinator.data["climate"].items()
if "temperature" in values
]
if int(coordinator.config.get(CONF_HYDROMETERS, 1)):
sensors.extend(
[
VerisureHygrometer(coordinator, serial_number)
for serial_number, values in coordinator.data["climate"].items()
if "humidity" in values
]
)
sensors.extend(
VerisureHygrometer(coordinator, serial_number)
for serial_number, values in coordinator.data["climate"].items()
if "humidity" in values
)
if int(coordinator.config.get(CONF_MOUSE, 1)):
sensors.extend(
[
VerisureMouseDetection(coordinator, serial_number)
for serial_number in coordinator.data["mice"]
]
)
sensors.extend(
VerisureMouseDetection(coordinator, serial_number)
for serial_number in coordinator.data["mice"]
)
add_entities(sensors)
async_add_entities(sensors)
class VerisureThermometer(CoordinatorEntity, Entity):

View File

@ -0,0 +1,39 @@
{
"config": {
"step": {
"user": {
"data": {
"description": "Sign-in with your Verisure My Pages account.",
"email": "[%key:common::config_flow::data::email%]",
"password": "[%key:common::config_flow::data::password%]"
}
},
"installation": {
"description": "Home Assistant found multiple Verisure installations in your My Pages account. Please, select the installation to add to Home Assistant.",
"data": {
"giid": "Installation"
}
}
},
"error": {
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
}
},
"options": {
"step": {
"init": {
"data": {
"lock_code_digits": "Number of digits in PIN code for locks",
"lock_default_code": "Default PIN code for locks, used if none is given"
}
}
},
"error": {
"code_format_mismatch": "The default PIN code does not match the required number of digits"
}
}
}

View File

@ -2,33 +2,28 @@
from __future__ import annotations
from time import monotonic
from typing import Any, Callable
from typing import Callable, Iterable
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import CONF_SMARTPLUGS, DOMAIN
from .const import DOMAIN
from .coordinator import VerisureDataUpdateCoordinator
def setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: dict[str, Any],
add_entities: Callable[[list[CoordinatorEntity]], None],
discovery_info: dict[str, Any] | None = None,
entry: ConfigEntry,
async_add_entities: Callable[[Iterable[Entity]], None],
) -> None:
"""Set up the Verisure switch platform."""
coordinator = hass.data[DOMAIN]
if not int(coordinator.config.get(CONF_SMARTPLUGS, 1)):
return
add_entities(
[
VerisureSmartplug(coordinator, serial_number)
for serial_number in coordinator.data["smart_plugs"]
]
"""Set up Verisure alarm control panel from a config entry."""
coordinator: VerisureDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
async_add_entities(
VerisureSmartplug(coordinator, serial_number)
for serial_number in coordinator.data["smart_plugs"]
)
@ -51,6 +46,11 @@ class VerisureSmartplug(CoordinatorEntity, SwitchEntity):
"""Return the name or location of the smartplug."""
return self.coordinator.data["smart_plugs"][self.serial_number]["area"]
@property
def unique_id(self) -> str:
"""Return the unique ID for this alarm control panel."""
return self.serial_number
@property
def is_on(self) -> bool:
"""Return true if on."""

View File

@ -0,0 +1,39 @@
{
"config": {
"abort": {
"already_configured": "Account is already configured"
},
"error": {
"invalid_auth": "Invalid authentication",
"unknown": "Unexpected error"
},
"step": {
"installation": {
"data": {
"giid": "Installation"
},
"description": "Home Assistant found multiple Verisure installations in your My Pages account. Please, select the installation to add to Home Assistant."
},
"user": {
"data": {
"description": "Sign-in with your Verisure My Pages account.",
"email": "Email",
"password": "Password"
}
}
}
},
"options": {
"error": {
"code_format_mismatch": "The default PIN code does not match the required number of digits"
},
"step": {
"init": {
"data": {
"lock_code_digits": "Number of digits in PIN code for locks",
"lock_default_code": "Default PIN code for locks, used if none is given"
}
}
}
}
}

View File

@ -247,6 +247,7 @@ FLOWS = [
"upnp",
"velbus",
"vera",
"verisure",
"vesync",
"vilfo",
"vizio",

View File

@ -153,5 +153,9 @@ DHCP = [
"domain": "toon",
"hostname": "eneco-*",
"macaddress": "74C63B*"
},
{
"domain": "verisure",
"macaddress": "0023C1*"
}
]

View File

@ -1171,6 +1171,9 @@ uvcclient==0.11.0
# homeassistant.components.vilfo
vilfo-api-client==0.3.2
# homeassistant.components.verisure
vsure==1.7.3
# homeassistant.components.vultr
vultr==0.1.2

View File

@ -0,0 +1 @@
"""Tests for the Verisure integration."""

View File

@ -0,0 +1,467 @@
"""Test the Verisure config flow."""
from __future__ import annotations
from unittest.mock import PropertyMock, patch
import pytest
from verisure import Error as VerisureError, LoginError as VerisureLoginError
from homeassistant import config_entries
from homeassistant.components.dhcp import MAC_ADDRESS
from homeassistant.components.verisure.const import (
CONF_GIID,
CONF_LOCK_CODE_DIGITS,
CONF_LOCK_DEFAULT_CODE,
DEFAULT_LOCK_CODE_DIGITS,
DOMAIN,
)
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import (
RESULT_TYPE_ABORT,
RESULT_TYPE_CREATE_ENTRY,
RESULT_TYPE_FORM,
)
from tests.common import MockConfigEntry
TEST_INSTALLATIONS = [
{"giid": "12345", "alias": "ascending", "street": "12345th street"},
{"giid": "54321", "alias": "descending", "street": "54321th street"},
]
TEST_INSTALLATION = [TEST_INSTALLATIONS[0]]
async def test_full_user_flow_single_installation(hass: HomeAssistant) -> None:
"""Test a full user initiated configuration flow with a single installation."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["step_id"] == "user"
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] == {}
with patch(
"homeassistant.components.verisure.config_flow.Verisure",
) as mock_verisure, patch(
"homeassistant.components.verisure.async_setup", return_value=True
) as mock_setup, patch(
"homeassistant.components.verisure.async_setup_entry",
return_value=True,
) as mock_setup_entry:
type(mock_verisure.return_value).installations = PropertyMock(
return_value=TEST_INSTALLATION
)
mock_verisure.login.return_value = True
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"email": "verisure_my_pages@example.com",
"password": "SuperS3cr3t!",
},
)
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_CREATE_ENTRY
assert result2["title"] == "ascending (12345th street)"
assert result2["data"] == {
CONF_GIID: "12345",
CONF_EMAIL: "verisure_my_pages@example.com",
CONF_PASSWORD: "SuperS3cr3t!",
}
assert len(mock_verisure.mock_calls) == 2
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
async def test_full_user_flow_multiple_installations(hass: HomeAssistant) -> None:
"""Test a full user initiated configuration flow with multiple installations."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["step_id"] == "user"
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] == {}
with patch(
"homeassistant.components.verisure.config_flow.Verisure",
) as mock_verisure:
type(mock_verisure.return_value).installations = PropertyMock(
return_value=TEST_INSTALLATIONS
)
mock_verisure.login.return_value = True
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"email": "verisure_my_pages@example.com",
"password": "SuperS3cr3t!",
},
)
await hass.async_block_till_done()
assert result2["step_id"] == "installation"
assert result2["type"] == RESULT_TYPE_FORM
assert result2["errors"] is None
with patch(
"homeassistant.components.verisure.async_setup", return_value=True
) as mock_setup, patch(
"homeassistant.components.verisure.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"], {"giid": "54321"}
)
await hass.async_block_till_done()
assert result3["type"] == RESULT_TYPE_CREATE_ENTRY
assert result3["title"] == "descending (54321th street)"
assert result3["data"] == {
CONF_GIID: "54321",
CONF_EMAIL: "verisure_my_pages@example.com",
CONF_PASSWORD: "SuperS3cr3t!",
}
assert len(mock_verisure.mock_calls) == 2
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
async def test_invalid_login(hass: HomeAssistant) -> None:
"""Test a flow with an invalid Verisure My Pages login."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.verisure.config_flow.Verisure.login",
side_effect=VerisureLoginError,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"email": "verisure_my_pages@example.com",
"password": "SuperS3cr3t!",
},
)
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_FORM
assert result2["step_id"] == "user"
assert result2["errors"] == {"base": "invalid_auth"}
async def test_unknown_error(hass: HomeAssistant) -> None:
"""Test a flow with an invalid Verisure My Pages login."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.verisure.config_flow.Verisure.login",
side_effect=VerisureError,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"email": "verisure_my_pages@example.com",
"password": "SuperS3cr3t!",
},
)
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_FORM
assert result2["step_id"] == "user"
assert result2["errors"] == {"base": "unknown"}
async def test_dhcp(hass: HomeAssistant) -> None:
"""Test that DHCP discovery works."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
data={MAC_ADDRESS: "01:23:45:67:89:ab"},
context={"source": config_entries.SOURCE_DHCP},
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user"
@pytest.mark.parametrize(
"input,output",
[
(
{
CONF_LOCK_CODE_DIGITS: 5,
CONF_LOCK_DEFAULT_CODE: "12345",
},
{
CONF_LOCK_CODE_DIGITS: 5,
CONF_LOCK_DEFAULT_CODE: "12345",
},
),
(
{
CONF_LOCK_DEFAULT_CODE: "",
},
{
CONF_LOCK_DEFAULT_CODE: "",
CONF_LOCK_CODE_DIGITS: DEFAULT_LOCK_CODE_DIGITS,
},
),
],
)
async def test_options_flow(
hass: HomeAssistant, input: dict[str, int | str], output: dict[str, int | str]
) -> None:
"""Test options config flow."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="12345",
data={},
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.verisure.async_setup", return_value=True
), patch(
"homeassistant.components.verisure.async_setup_entry",
return_value=True,
):
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input=input,
)
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["data"] == output
async def test_options_flow_code_format_mismatch(hass: HomeAssistant) -> None:
"""Test options config flow with a code format mismatch."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="12345",
data={},
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.verisure.async_setup", return_value=True
), patch(
"homeassistant.components.verisure.async_setup_entry",
return_value=True,
):
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "init"
assert result["errors"] == {}
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_LOCK_CODE_DIGITS: 5,
CONF_LOCK_DEFAULT_CODE: "123",
},
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "init"
assert result["errors"] == {"base": "code_format_mismatch"}
#
# Below this line are tests that can be removed once the YAML configuration
# has been removed from this integration.
#
@pytest.mark.parametrize(
"giid,installations",
[
("12345", TEST_INSTALLATION),
("12345", TEST_INSTALLATIONS),
(None, TEST_INSTALLATION),
],
)
async def test_imports(
hass: HomeAssistant, giid: str | None, installations: dict[str, str]
) -> None:
"""Test a YAML import with/without known giid on single/multiple installations."""
with patch(
"homeassistant.components.verisure.config_flow.Verisure",
) as mock_verisure, patch(
"homeassistant.components.verisure.async_setup", return_value=True
) as mock_setup, patch(
"homeassistant.components.verisure.async_setup_entry",
return_value=True,
) as mock_setup_entry:
type(mock_verisure.return_value).installations = PropertyMock(
return_value=installations
)
mock_verisure.login.return_value = True
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_EMAIL: "verisure_my_pages@example.com",
CONF_GIID: giid,
CONF_LOCK_CODE_DIGITS: 10,
CONF_LOCK_DEFAULT_CODE: "123456",
CONF_PASSWORD: "SuperS3cr3t!",
},
)
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "ascending (12345th street)"
assert result["data"] == {
CONF_EMAIL: "verisure_my_pages@example.com",
CONF_GIID: "12345",
CONF_LOCK_CODE_DIGITS: 10,
CONF_LOCK_DEFAULT_CODE: "123456",
CONF_PASSWORD: "SuperS3cr3t!",
}
assert len(mock_verisure.mock_calls) == 2
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
async def test_imports_invalid_login(hass: HomeAssistant) -> None:
"""Test a YAML import that results in a invalid login."""
with patch(
"homeassistant.components.verisure.config_flow.Verisure.login",
side_effect=VerisureLoginError,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_EMAIL: "verisure_my_pages@example.com",
CONF_GIID: None,
CONF_LOCK_CODE_DIGITS: None,
CONF_LOCK_DEFAULT_CODE: None,
CONF_PASSWORD: "SuperS3cr3t!",
},
)
assert result["step_id"] == "user"
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] == {"base": "invalid_auth"}
with patch(
"homeassistant.components.verisure.config_flow.Verisure",
) as mock_verisure, patch(
"homeassistant.components.verisure.async_setup", return_value=True
) as mock_setup, patch(
"homeassistant.components.verisure.async_setup_entry",
return_value=True,
) as mock_setup_entry:
type(mock_verisure.return_value).installations = PropertyMock(
return_value=TEST_INSTALLATION
)
mock_verisure.login.return_value = True
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"email": "verisure_my_pages@example.com",
"password": "SuperS3cr3t!",
},
)
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_CREATE_ENTRY
assert result2["title"] == "ascending (12345th street)"
assert result2["data"] == {
CONF_GIID: "12345",
CONF_EMAIL: "verisure_my_pages@example.com",
CONF_PASSWORD: "SuperS3cr3t!",
}
assert len(mock_verisure.mock_calls) == 2
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
async def test_imports_needs_user_installation_choice(hass: HomeAssistant) -> None:
"""Test a YAML import that needs to use to decide on the installation."""
with patch(
"homeassistant.components.verisure.config_flow.Verisure",
) as mock_verisure:
type(mock_verisure.return_value).installations = PropertyMock(
return_value=TEST_INSTALLATIONS
)
mock_verisure.login.return_value = True
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_EMAIL: "verisure_my_pages@example.com",
CONF_GIID: None,
CONF_LOCK_CODE_DIGITS: None,
CONF_LOCK_DEFAULT_CODE: None,
CONF_PASSWORD: "SuperS3cr3t!",
},
)
assert result["step_id"] == "installation"
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] is None
with patch(
"homeassistant.components.verisure.async_setup", return_value=True
) as mock_setup, patch(
"homeassistant.components.verisure.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], {"giid": "12345"}
)
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_CREATE_ENTRY
assert result2["title"] == "ascending (12345th street)"
assert result2["data"] == {
CONF_GIID: "12345",
CONF_EMAIL: "verisure_my_pages@example.com",
CONF_PASSWORD: "SuperS3cr3t!",
}
assert len(mock_verisure.mock_calls) == 2
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.parametrize("giid", ["12345", None])
async def test_import_already_exists(hass: HomeAssistant, giid: str | None) -> None:
"""Test that import flow aborts if exists."""
MockConfigEntry(domain=DOMAIN, data={}, unique_id="12345").add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_EMAIL: "verisure_my_pages@example.com",
CONF_PASSWORD: "SuperS3cr3t!",
CONF_GIID: giid,
},
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"