mirror of https://github.com/home-assistant/core
Add Crownstone integration (#50677)
This commit is contained in:
Ricardo Steijn
GitHub
parent
bac55b78fe
commit
2a51bb5bba
|
|
@ -171,6 +171,13 @@ omit =
|
|||
homeassistant/components/coolmaster/const.py
|
||||
homeassistant/components/cppm_tracker/device_tracker.py
|
||||
homeassistant/components/cpuspeed/sensor.py
|
||||
homeassistant/components/crownstone/__init__.py
|
||||
homeassistant/components/crownstone/const.py
|
||||
homeassistant/components/crownstone/listeners.py
|
||||
homeassistant/components/crownstone/helpers.py
|
||||
homeassistant/components/crownstone/devices.py
|
||||
homeassistant/components/crownstone/entry_manager.py
|
||||
homeassistant/components/crownstone/light.py
|
||||
homeassistant/components/cups/sensor.py
|
||||
homeassistant/components/currencylayer/sensor.py
|
||||
homeassistant/components/daikin/*
|
||||
|
|
|
|||
|
|
@ -27,6 +27,7 @@ homeassistant.components.calendar.*
|
|||
homeassistant.components.camera.*
|
||||
homeassistant.components.canary.*
|
||||
homeassistant.components.cover.*
|
||||
homeassistant.components.crownstone.*
|
||||
homeassistant.components.device_automation.*
|
||||
homeassistant.components.device_tracker.*
|
||||
homeassistant.components.devolo_home_control.*
|
||||
|
|
|
|||
|
|
@ -104,6 +104,7 @@ homeassistant/components/coronavirus/* @home-assistant/core
|
|||
homeassistant/components/counter/* @fabaff
|
||||
homeassistant/components/cover/* @home-assistant/core
|
||||
homeassistant/components/cpuspeed/* @fabaff
|
||||
homeassistant/components/crownstone/* @Crownstone @RicArch97
|
||||
homeassistant/components/cups/* @fabaff
|
||||
homeassistant/components/daikin/* @fredrike
|
||||
homeassistant/components/darksky/* @fabaff
|
||||
|
|
|
|||
|
|
@ -0,0 +1,23 @@
|
|||
"""Integration for Crownstone."""
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from .const import DOMAIN
|
||||
from .entry_manager import CrownstoneEntryManager
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Initiate setup for a Crownstone config entry."""
|
||||
manager = CrownstoneEntryManager(hass, entry)
|
||||
|
||||
return await manager.async_setup()
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
unload_ok: bool = await hass.data[DOMAIN][entry.entry_id].async_unload()
|
||||
if len(hass.data[DOMAIN]) == 0:
|
||||
hass.data.pop(DOMAIN)
|
||||
return unload_ok
|
||||
|
|
@ -0,0 +1,299 @@
|
|||
"""Flow handler for Crownstone."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from crownstone_cloud import CrownstoneCloud
|
||||
from crownstone_cloud.exceptions import (
|
||||
CrownstoneAuthenticationError,
|
||||
CrownstoneUnknownError,
|
||||
)
|
||||
import serial.tools.list_ports
|
||||
from serial.tools.list_ports_common import ListPortInfo
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components import usb
|
||||
from homeassistant.config_entries import ConfigEntry, ConfigFlow, OptionsFlow
|
||||
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
from homeassistant.helpers import aiohttp_client
|
||||
|
||||
from .const import (
|
||||
CONF_USB_MANUAL_PATH,
|
||||
CONF_USB_PATH,
|
||||
CONF_USB_SPHERE,
|
||||
CONF_USB_SPHERE_OPTION,
|
||||
CONF_USE_USB_OPTION,
|
||||
DOMAIN,
|
||||
DONT_USE_USB,
|
||||
MANUAL_PATH,
|
||||
REFRESH_LIST,
|
||||
)
|
||||
from .entry_manager import CrownstoneEntryManager
|
||||
from .helpers import list_ports_as_str
|
||||
|
||||
|
||||
class CrownstoneConfigFlowHandler(ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for Crownstone."""
|
||||
|
||||
VERSION = 1
|
||||
cloud: CrownstoneCloud
|
||||
|
||||
@staticmethod
|
||||
@callback
|
||||
def async_get_options_flow(
|
||||
config_entry: ConfigEntry,
|
||||
) -> CrownstoneOptionsFlowHandler:
|
||||
"""Return the Crownstone options."""
|
||||
return CrownstoneOptionsFlowHandler(config_entry)
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Initialize the flow."""
|
||||
self.login_info: dict[str, Any] = {}
|
||||
self.usb_path: str | None = None
|
||||
self.usb_sphere_id: str | None = None
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle the initial step."""
|
||||
errors: dict[str, str] = {}
|
||||
if user_input is None:
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=vol.Schema(
|
||||
{vol.Required(CONF_EMAIL): str, vol.Required(CONF_PASSWORD): str}
|
||||
),
|
||||
)
|
||||
|
||||
self.cloud = CrownstoneCloud(
|
||||
email=user_input[CONF_EMAIL],
|
||||
password=user_input[CONF_PASSWORD],
|
||||
clientsession=aiohttp_client.async_get_clientsession(self.hass),
|
||||
)
|
||||
# Login & sync all user data
|
||||
try:
|
||||
await self.cloud.async_initialize()
|
||||
except CrownstoneAuthenticationError as auth_error:
|
||||
if auth_error.type == "LOGIN_FAILED":
|
||||
errors["base"] = "invalid_auth"
|
||||
elif auth_error.type == "LOGIN_FAILED_EMAIL_NOT_VERIFIED":
|
||||
errors["base"] = "account_not_verified"
|
||||
except CrownstoneUnknownError:
|
||||
errors["base"] = "unknown_error"
|
||||
|
||||
# show form again, with the errors
|
||||
if errors:
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=vol.Schema(
|
||||
{vol.Required(CONF_EMAIL): str, vol.Required(CONF_PASSWORD): str}
|
||||
),
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
await self.async_set_unique_id(self.cloud.cloud_data.user_id)
|
||||
self._abort_if_unique_id_configured()
|
||||
|
||||
self.login_info = user_input
|
||||
return await self.async_step_usb_config()
|
||||
|
||||
async def async_step_usb_config(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Set up a Crownstone USB dongle."""
|
||||
list_of_ports = await self.hass.async_add_executor_job(
|
||||
serial.tools.list_ports.comports
|
||||
)
|
||||
ports_as_string = list_ports_as_str(list_of_ports)
|
||||
|
||||
if user_input is not None:
|
||||
selection = user_input[CONF_USB_PATH]
|
||||
|
||||
if selection == DONT_USE_USB:
|
||||
return self.async_create_new_entry()
|
||||
if selection == MANUAL_PATH:
|
||||
return await self.async_step_usb_manual_config()
|
||||
if selection != REFRESH_LIST:
|
||||
selected_port: ListPortInfo = list_of_ports[
|
||||
(ports_as_string.index(selection) - 1)
|
||||
]
|
||||
self.usb_path = await self.hass.async_add_executor_job(
|
||||
usb.get_serial_by_id, selected_port.device
|
||||
)
|
||||
return await self.async_step_usb_sphere_config()
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="usb_config",
|
||||
data_schema=vol.Schema(
|
||||
{vol.Required(CONF_USB_PATH): vol.In(ports_as_string)}
|
||||
),
|
||||
)
|
||||
|
||||
async def async_step_usb_manual_config(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Manually enter Crownstone USB dongle path."""
|
||||
if user_input is None:
|
||||
return self.async_show_form(
|
||||
step_id="usb_manual_config",
|
||||
data_schema=vol.Schema({vol.Required(CONF_USB_MANUAL_PATH): str}),
|
||||
)
|
||||
|
||||
self.usb_path = user_input[CONF_USB_MANUAL_PATH]
|
||||
return await self.async_step_usb_sphere_config()
|
||||
|
||||
async def async_step_usb_sphere_config(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Select a Crownstone sphere that the USB operates in."""
|
||||
spheres = {sphere.name: sphere.cloud_id for sphere in self.cloud.cloud_data}
|
||||
# no need to select if there's only 1 option
|
||||
sphere_id: str | None = None
|
||||
if len(spheres) == 1:
|
||||
sphere_id = next(iter(spheres.values()))
|
||||
|
||||
if user_input is None and sphere_id is None:
|
||||
return self.async_show_form(
|
||||
step_id="usb_sphere_config",
|
||||
data_schema=vol.Schema({CONF_USB_SPHERE: vol.In(spheres.keys())}),
|
||||
)
|
||||
|
||||
if sphere_id:
|
||||
self.usb_sphere_id = sphere_id
|
||||
elif user_input:
|
||||
self.usb_sphere_id = spheres[user_input[CONF_USB_SPHERE]]
|
||||
|
||||
return self.async_create_new_entry()
|
||||
|
||||
def async_create_new_entry(self) -> FlowResult:
|
||||
"""Create a new entry."""
|
||||
return self.async_create_entry(
|
||||
title=f"Account: {self.login_info[CONF_EMAIL]}",
|
||||
data={
|
||||
CONF_EMAIL: self.login_info[CONF_EMAIL],
|
||||
CONF_PASSWORD: self.login_info[CONF_PASSWORD],
|
||||
},
|
||||
options={CONF_USB_PATH: self.usb_path, CONF_USB_SPHERE: self.usb_sphere_id},
|
||||
)
|
||||
|
||||
|
||||
class CrownstoneOptionsFlowHandler(OptionsFlow):
|
||||
"""Handle Crownstone options."""
|
||||
|
||||
def __init__(self, config_entry: ConfigEntry) -> None:
|
||||
"""Initialize Crownstone options."""
|
||||
self.entry = config_entry
|
||||
self.updated_options = config_entry.options.copy()
|
||||
self.spheres: dict[str, str] = {}
|
||||
|
||||
async def async_step_init(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Manage Crownstone options."""
|
||||
manager: CrownstoneEntryManager = self.hass.data[DOMAIN][self.entry.entry_id]
|
||||
|
||||
spheres = {sphere.name: sphere.cloud_id for sphere in manager.cloud.cloud_data}
|
||||
usb_path = self.entry.options.get(CONF_USB_PATH)
|
||||
usb_sphere = self.entry.options.get(CONF_USB_SPHERE)
|
||||
|
||||
options_schema = vol.Schema(
|
||||
{vol.Optional(CONF_USE_USB_OPTION, default=usb_path is not None): bool}
|
||||
)
|
||||
if usb_path is not None and len(spheres) > 1:
|
||||
options_schema = options_schema.extend(
|
||||
{
|
||||
vol.Optional(
|
||||
CONF_USB_SPHERE_OPTION,
|
||||
default=manager.cloud.cloud_data.spheres[usb_sphere].name,
|
||||
): vol.In(spheres.keys())
|
||||
}
|
||||
)
|
||||
|
||||
if user_input is not None:
|
||||
if user_input[CONF_USE_USB_OPTION] and usb_path is None:
|
||||
self.spheres = spheres
|
||||
return await self.async_step_usb_config_option()
|
||||
if not user_input[CONF_USE_USB_OPTION] and usb_path is not None:
|
||||
self.updated_options[CONF_USB_PATH] = None
|
||||
self.updated_options[CONF_USB_SPHERE] = None
|
||||
elif (
|
||||
CONF_USB_SPHERE_OPTION in user_input
|
||||
and spheres[user_input[CONF_USB_SPHERE_OPTION]] != usb_sphere
|
||||
):
|
||||
sphere_id = spheres[user_input[CONF_USB_SPHERE_OPTION]]
|
||||
user_input[CONF_USB_SPHERE_OPTION] = sphere_id
|
||||
self.updated_options[CONF_USB_SPHERE] = sphere_id
|
||||
|
||||
return self.async_create_entry(title="", data=self.updated_options)
|
||||
|
||||
return self.async_show_form(step_id="init", data_schema=options_schema)
|
||||
|
||||
async def async_step_usb_config_option(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Set up a Crownstone USB dongle."""
|
||||
list_of_ports = await self.hass.async_add_executor_job(
|
||||
serial.tools.list_ports.comports
|
||||
)
|
||||
ports_as_string = list_ports_as_str(list_of_ports, False)
|
||||
|
||||
if user_input is not None:
|
||||
selection = user_input[CONF_USB_PATH]
|
||||
|
||||
if selection == MANUAL_PATH:
|
||||
return await self.async_step_usb_manual_config_option()
|
||||
if selection != REFRESH_LIST:
|
||||
selected_port: ListPortInfo = list_of_ports[
|
||||
ports_as_string.index(selection)
|
||||
]
|
||||
usb_path = await self.hass.async_add_executor_job(
|
||||
usb.get_serial_by_id, selected_port.device
|
||||
)
|
||||
self.updated_options[CONF_USB_PATH] = usb_path
|
||||
return await self.async_step_usb_sphere_config_option()
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="usb_config_option",
|
||||
data_schema=vol.Schema(
|
||||
{vol.Required(CONF_USB_PATH): vol.In(ports_as_string)}
|
||||
),
|
||||
)
|
||||
|
||||
async def async_step_usb_manual_config_option(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Manually enter Crownstone USB dongle path."""
|
||||
if user_input is None:
|
||||
return self.async_show_form(
|
||||
step_id="usb_manual_config_option",
|
||||
data_schema=vol.Schema({vol.Required(CONF_USB_MANUAL_PATH): str}),
|
||||
)
|
||||
|
||||
self.updated_options[CONF_USB_PATH] = user_input[CONF_USB_MANUAL_PATH]
|
||||
return await self.async_step_usb_sphere_config_option()
|
||||
|
||||
async def async_step_usb_sphere_config_option(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Select a Crownstone sphere that the USB operates in."""
|
||||
# no need to select if there's only 1 option
|
||||
sphere_id: str | None = None
|
||||
if len(self.spheres) == 1:
|
||||
sphere_id = next(iter(self.spheres.values()))
|
||||
|
||||
if user_input is None and sphere_id is None:
|
||||
return self.async_show_form(
|
||||
step_id="usb_sphere_config_option",
|
||||
data_schema=vol.Schema({CONF_USB_SPHERE: vol.In(self.spheres.keys())}),
|
||||
)
|
||||
|
||||
if sphere_id:
|
||||
self.updated_options[CONF_USB_SPHERE] = sphere_id
|
||||
elif user_input:
|
||||
self.updated_options[CONF_USB_SPHERE] = self.spheres[
|
||||
user_input[CONF_USB_SPHERE]
|
||||
]
|
||||
|
||||
return self.async_create_entry(title="", data=self.updated_options)
|
||||
|
|
@ -0,0 +1,45 @@
|
|||
"""Constants for the crownstone integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Final
|
||||
|
||||
# Platforms
|
||||
DOMAIN: Final = "crownstone"
|
||||
PLATFORMS: Final[list[str]] = ["light"]
|
||||
|
||||
# Listeners
|
||||
SSE_LISTENERS: Final = "sse_listeners"
|
||||
UART_LISTENERS: Final = "uart_listeners"
|
||||
|
||||
# Unique ID suffixes
|
||||
CROWNSTONE_SUFFIX: Final = "crownstone"
|
||||
|
||||
# Signals (within integration)
|
||||
SIG_CROWNSTONE_STATE_UPDATE: Final = "crownstone.crownstone_state_update"
|
||||
SIG_CROWNSTONE_UPDATE: Final = "crownstone.crownstone_update"
|
||||
SIG_UART_STATE_CHANGE: Final = "crownstone.uart_state_change"
|
||||
|
||||
# Abilities state
|
||||
ABILITY_STATE: Final[dict[bool, str]] = {True: "Enabled", False: "Disabled"}
|
||||
|
||||
# Config flow
|
||||
CONF_USB_PATH: Final = "usb_path"
|
||||
CONF_USB_MANUAL_PATH: Final = "usb_manual_path"
|
||||
CONF_USB_SPHERE: Final = "usb_sphere"
|
||||
# Options flow
|
||||
CONF_USE_USB_OPTION: Final = "use_usb_option"
|
||||
CONF_USB_SPHERE_OPTION: Final = "usb_sphere_option"
|
||||
# USB config list entries
|
||||
DONT_USE_USB: Final = "Don't use USB"
|
||||
REFRESH_LIST: Final = "Refresh list"
|
||||
MANUAL_PATH: Final = "Enter manually"
|
||||
|
||||
# Crownstone entity
|
||||
CROWNSTONE_INCLUDE_TYPES: Final[dict[str, str]] = {
|
||||
"PLUG": "Plug",
|
||||
"BUILTIN": "Built-in",
|
||||
"BUILTIN_ONE": "Built-in One",
|
||||
}
|
||||
|
||||
# Crownstone USB Dongle
|
||||
CROWNSTONE_USB: Final = "CROWNSTONE_USB"
|
||||
|
|
@ -0,0 +1,43 @@
|
|||
"""Base classes for Crownstone devices."""
|
||||
from __future__ import annotations
|
||||
|
||||
from crownstone_cloud.cloud_models.crownstones import Crownstone
|
||||
|
||||
from homeassistant.const import (
|
||||
ATTR_IDENTIFIERS,
|
||||
ATTR_MANUFACTURER,
|
||||
ATTR_MODEL,
|
||||
ATTR_NAME,
|
||||
ATTR_SW_VERSION,
|
||||
)
|
||||
from homeassistant.helpers.entity import DeviceInfo
|
||||
|
||||
from .const import CROWNSTONE_INCLUDE_TYPES, DOMAIN
|
||||
|
||||
|
||||
class CrownstoneDevice:
|
||||
"""Representation of a Crownstone device."""
|
||||
|
||||
def __init__(self, device: Crownstone) -> None:
|
||||
"""Initialize the device."""
|
||||
self.device = device
|
||||
|
||||
@property
|
||||
def cloud_id(self) -> str:
|
||||
"""
|
||||
Return the unique identifier for this device.
|
||||
|
||||
Used as device ID and to generate unique entity ID's.
|
||||
"""
|
||||
return str(self.device.cloud_id)
|
||||
|
||||
@property
|
||||
def device_info(self) -> DeviceInfo:
|
||||
"""Return device info."""
|
||||
return {
|
||||
ATTR_IDENTIFIERS: {(DOMAIN, self.cloud_id)},
|
||||
ATTR_NAME: self.device.name,
|
||||
ATTR_MANUFACTURER: "Crownstone",
|
||||
ATTR_MODEL: CROWNSTONE_INCLUDE_TYPES[self.device.type],
|
||||
ATTR_SW_VERSION: self.device.sw_version,
|
||||
}
|
||||
|
|
@ -0,0 +1,190 @@
|
|||
"""Manager to set up IO with Crownstone devices for a config entry."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from crownstone_cloud import CrownstoneCloud
|
||||
from crownstone_cloud.exceptions import (
|
||||
CrownstoneAuthenticationError,
|
||||
CrownstoneUnknownError,
|
||||
)
|
||||
from crownstone_sse import CrownstoneSSEAsync
|
||||
from crownstone_uart import CrownstoneUart, UartEventBus
|
||||
from crownstone_uart.Exceptions import UartException
|
||||
|
||||
from homeassistant.components import persistent_notification
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD, EVENT_HOMEASSISTANT_STOP
|
||||
from homeassistant.core import Event, HomeAssistant, callback
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
from homeassistant.helpers import aiohttp_client
|
||||
|
||||
from .const import (
|
||||
CONF_USB_PATH,
|
||||
CONF_USB_SPHERE,
|
||||
DOMAIN,
|
||||
PLATFORMS,
|
||||
SSE_LISTENERS,
|
||||
UART_LISTENERS,
|
||||
)
|
||||
from .helpers import get_port
|
||||
from .listeners import setup_sse_listeners, setup_uart_listeners
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class CrownstoneEntryManager:
|
||||
"""Manage a Crownstone config entry."""
|
||||
|
||||
uart: CrownstoneUart | None = None
|
||||
cloud: CrownstoneCloud
|
||||
sse: CrownstoneSSEAsync
|
||||
|
||||
def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry) -> None:
|
||||
"""Initialize the hub."""
|
||||
self.hass = hass
|
||||
self.config_entry = config_entry
|
||||
self.listeners: dict[str, Any] = {}
|
||||
self.usb_sphere_id: str | None = None
|
||||
|
||||
async def async_setup(self) -> bool:
|
||||
"""
|
||||
Set up a Crownstone config entry.
|
||||
|
||||
Returns True if the setup was successful.
|
||||
"""
|
||||
email = self.config_entry.data[CONF_EMAIL]
|
||||
password = self.config_entry.data[CONF_PASSWORD]
|
||||
|
||||
self.cloud = CrownstoneCloud(
|
||||
email=email,
|
||||
password=password,
|
||||
clientsession=aiohttp_client.async_get_clientsession(self.hass),
|
||||
)
|
||||
# Login & sync all user data
|
||||
try:
|
||||
await self.cloud.async_initialize()
|
||||
except CrownstoneAuthenticationError as auth_err:
|
||||
_LOGGER.error(
|
||||
"Auth error during login with type: %s and message: %s",
|
||||
auth_err.type,
|
||||
auth_err.message,
|
||||
)
|
||||
return False
|
||||
except CrownstoneUnknownError as unknown_err:
|
||||
_LOGGER.error("Unknown error during login")
|
||||
raise ConfigEntryNotReady from unknown_err
|
||||
|
||||
# A new clientsession is created because the default one does not cleanup on unload
|
||||
self.sse = CrownstoneSSEAsync(
|
||||
email=email,
|
||||
password=password,
|
||||
access_token=self.cloud.access_token,
|
||||
websession=aiohttp_client.async_create_clientsession(self.hass),
|
||||
)
|
||||
# Listen for events in the background, without task tracking
|
||||
asyncio.create_task(self.async_process_events(self.sse))
|
||||
setup_sse_listeners(self)
|
||||
|
||||
# Set up a Crownstone USB only if path exists
|
||||
if self.config_entry.options[CONF_USB_PATH] is not None:
|
||||
await self.async_setup_usb()
|
||||
|
||||
# Save the sphere where the USB is located
|
||||
# Makes HA aware of the Crownstone environment HA is placed in, a user can have multiple
|
||||
self.usb_sphere_id = self.config_entry.options[CONF_USB_SPHERE]
|
||||
|
||||
self.hass.data.setdefault(DOMAIN, {})[self.config_entry.entry_id] = self
|
||||
self.hass.config_entries.async_setup_platforms(self.config_entry, PLATFORMS)
|
||||
|
||||
# HA specific listeners
|
||||
self.config_entry.async_on_unload(
|
||||
self.config_entry.add_update_listener(_async_update_listener)
|
||||
)
|
||||
self.config_entry.async_on_unload(
|
||||
self.hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, self.on_shutdown)
|
||||
)
|
||||
|
||||
return True
|
||||
|
||||
async def async_process_events(self, sse_client: CrownstoneSSEAsync) -> None:
|
||||
"""Asynchronous iteration of Crownstone SSE events."""
|
||||
async with sse_client as client:
|
||||
async for event in client:
|
||||
if event is not None:
|
||||
# Make SSE updates, like ability change, available to the user
|
||||
self.hass.bus.async_fire(f"{DOMAIN}_{event.type}", event.data)
|
||||
|
||||
async def async_setup_usb(self) -> None:
|
||||
"""Attempt setup of a Crownstone usb dongle."""
|
||||
# Trace by-id symlink back to the serial port
|
||||
serial_port = await self.hass.async_add_executor_job(
|
||||
get_port, self.config_entry.options[CONF_USB_PATH]
|
||||
)
|
||||
if serial_port is None:
|
||||
return
|
||||
|
||||
self.uart = CrownstoneUart()
|
||||
# UartException is raised when serial controller fails to open
|
||||
try:
|
||||
await self.uart.initialize_usb(serial_port)
|
||||
except UartException:
|
||||
self.uart = None
|
||||
# Set entry options for usb to null
|
||||
updated_options = self.config_entry.options.copy()
|
||||
updated_options[CONF_USB_PATH] = None
|
||||
updated_options[CONF_USB_SPHERE] = None
|
||||
# Ensure that the user can configure an USB again from options
|
||||
self.hass.config_entries.async_update_entry(
|
||||
self.config_entry, options=updated_options
|
||||
)
|
||||
# Show notification to ensure the user knows the cloud is now used
|
||||
persistent_notification.async_create(
|
||||
self.hass,
|
||||
f"Setup of Crownstone USB dongle was unsuccessful on port {serial_port}.\n \
|
||||
Crownstone Cloud will be used to switch Crownstones.\n \
|
||||
Please check if your port is correct and set up the USB again from integration options.",
|
||||
"Crownstone",
|
||||
"crownstone_usb_dongle_setup",
|
||||
)
|
||||
return
|
||||
|
||||
setup_uart_listeners(self)
|
||||
|
||||
async def async_unload(self) -> bool:
|
||||
"""Unload the current config entry."""
|
||||
# Authentication failed
|
||||
if self.cloud.cloud_data is None:
|
||||
return True
|
||||
|
||||
self.sse.close_client()
|
||||
for sse_unsub in self.listeners[SSE_LISTENERS]:
|
||||
sse_unsub()
|
||||
|
||||
if self.uart:
|
||||
self.uart.stop()
|
||||
for subscription_id in self.listeners[UART_LISTENERS]:
|
||||
UartEventBus.unsubscribe(subscription_id)
|
||||
|
||||
unload_ok = await self.hass.config_entries.async_unload_platforms(
|
||||
self.config_entry, PLATFORMS
|
||||
)
|
||||
|
||||
if unload_ok:
|
||||
self.hass.data[DOMAIN].pop(self.config_entry.entry_id)
|
||||
|
||||
return unload_ok
|
||||
|
||||
@callback
|
||||
def on_shutdown(self, _: Event) -> None:
|
||||
"""Close all IO connections."""
|
||||
self.sse.close_client()
|
||||
if self.uart:
|
||||
self.uart.stop()
|
||||
|
||||
|
||||
async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
|
||||
"""Handle options update."""
|
||||
await hass.config_entries.async_reload(entry.entry_id)
|
||||
|
|
@ -0,0 +1,59 @@
|
|||
"""Helper functions for the Crownstone integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
from serial.tools.list_ports_common import ListPortInfo
|
||||
|
||||
from homeassistant.components import usb
|
||||
|
||||
from .const import DONT_USE_USB, MANUAL_PATH, REFRESH_LIST
|
||||
|
||||
|
||||
def list_ports_as_str(
|
||||
serial_ports: list[ListPortInfo], no_usb_option: bool = True
|
||||
) -> list[str]:
|
||||
"""
|
||||
Represent currently available serial ports as string.
|
||||
|
||||
Adds option to not use usb on top of the list,
|
||||
option to use manual path or refresh list at the end.
|
||||
"""
|
||||
ports_as_string: list[str] = []
|
||||
|
||||
if no_usb_option:
|
||||
ports_as_string.append(DONT_USE_USB)
|
||||
|
||||
for port in serial_ports:
|
||||
ports_as_string.append(
|
||||
usb.human_readable_device_name(
|
||||
port.device,
|
||||
port.serial_number,
|
||||
port.manufacturer,
|
||||
port.description,
|
||||
f"{hex(port.vid)[2:]:0>4}".upper(),
|
||||
f"{hex(port.pid)[2:]:0>4}".upper(),
|
||||
)
|
||||
)
|
||||
ports_as_string.append(MANUAL_PATH)
|
||||
ports_as_string.append(REFRESH_LIST)
|
||||
|
||||
return ports_as_string
|
||||
|
||||
|
||||
def get_port(dev_path: str) -> str | None:
|
||||
"""Get the port that the by-id link points to."""
|
||||
# not a by-id link, but just given path
|
||||
by_id = "/dev/serial/by-id"
|
||||
if by_id not in dev_path:
|
||||
return dev_path
|
||||
|
||||
try:
|
||||
return f"/dev/{os.path.basename(os.readlink(dev_path))}"
|
||||
except FileNotFoundError:
|
||||
return None
|
||||
|
||||
|
||||
def map_from_to(val: int, in_min: int, in_max: int, out_min: int, out_max: int) -> int:
|
||||
"""Map a value from a range to another."""
|
||||
return int((val - in_min) * (out_max - out_min) / (in_max - in_min) + out_min)
|
||||
|
|
@ -0,0 +1,204 @@
|
|||
"""Support for Crownstone devices."""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from functools import partial
|
||||
import logging
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from crownstone_cloud.cloud_models.crownstones import Crownstone
|
||||
from crownstone_cloud.const import (
|
||||
DIMMING_ABILITY,
|
||||
SWITCHCRAFT_ABILITY,
|
||||
TAP_TO_TOGGLE_ABILITY,
|
||||
)
|
||||
from crownstone_cloud.exceptions import CrownstoneAbilityError
|
||||
from crownstone_uart import CrownstoneUart
|
||||
|
||||
from homeassistant.components.light import (
|
||||
ATTR_BRIGHTNESS,
|
||||
SUPPORT_BRIGHTNESS,
|
||||
LightEntity,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.dispatcher import async_dispatcher_connect
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
|
||||
from .const import (
|
||||
ABILITY_STATE,
|
||||
CROWNSTONE_INCLUDE_TYPES,
|
||||
CROWNSTONE_SUFFIX,
|
||||
DOMAIN,
|
||||
SIG_CROWNSTONE_STATE_UPDATE,
|
||||
SIG_UART_STATE_CHANGE,
|
||||
)
|
||||
from .devices import CrownstoneDevice
|
||||
from .helpers import map_from_to
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .entry_manager import CrownstoneEntryManager
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
config_entry: ConfigEntry,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up crownstones from a config entry."""
|
||||
manager: CrownstoneEntryManager = hass.data[DOMAIN][config_entry.entry_id]
|
||||
|
||||
entities: list[CrownstoneEntity] = []
|
||||
|
||||
# Add Crownstone entities that support switching/dimming
|
||||
for sphere in manager.cloud.cloud_data:
|
||||
for crownstone in sphere.crownstones:
|
||||
if crownstone.type in CROWNSTONE_INCLUDE_TYPES:
|
||||
# Crownstone can communicate with Crownstone USB
|
||||
if manager.uart and sphere.cloud_id == manager.usb_sphere_id:
|
||||
entities.append(CrownstoneEntity(crownstone, manager.uart))
|
||||
# Crownstone can't communicate with Crownstone USB
|
||||
else:
|
||||
entities.append(CrownstoneEntity(crownstone))
|
||||
|
||||
async_add_entities(entities)
|
||||
|
||||
|
||||
def crownstone_state_to_hass(value: int) -> int:
|
||||
"""Crownstone 0..100 to hass 0..255."""
|
||||
return map_from_to(value, 0, 100, 0, 255)
|
||||
|
||||
|
||||
def hass_to_crownstone_state(value: int) -> int:
|
||||
"""Hass 0..255 to Crownstone 0..100."""
|
||||
return map_from_to(value, 0, 255, 0, 100)
|
||||
|
||||
|
||||
class CrownstoneEntity(CrownstoneDevice, LightEntity):
|
||||
"""
|
||||
Representation of a crownstone.
|
||||
|
||||
Light platform is used to support dimming.
|
||||
"""
|
||||
|
||||
_attr_should_poll = False
|
||||
_attr_icon = "mdi:power-socket-de"
|
||||
|
||||
def __init__(self, crownstone_data: Crownstone, usb: CrownstoneUart = None) -> None:
|
||||
"""Initialize the crownstone."""
|
||||
super().__init__(crownstone_data)
|
||||
self.usb = usb
|
||||
# Entity class attributes
|
||||
self._attr_name = str(self.device.name)
|
||||
self._attr_unique_id = f"{self.cloud_id}-{CROWNSTONE_SUFFIX}"
|
||||
|
||||
@property
|
||||
def usb_available(self) -> bool:
|
||||
"""Return if this entity can use a usb dongle."""
|
||||
return self.usb is not None and self.usb.is_ready()
|
||||
|
||||
@property
|
||||
def brightness(self) -> int | None:
|
||||
"""Return the brightness if dimming enabled."""
|
||||
return crownstone_state_to_hass(self.device.state)
|
||||
|
||||
@property
|
||||
def is_on(self) -> bool:
|
||||
"""Return if the device is on."""
|
||||
return crownstone_state_to_hass(self.device.state) > 0
|
||||
|
||||
@property
|
||||
def supported_features(self) -> int:
|
||||
"""Return the supported features of this Crownstone."""
|
||||
if self.device.abilities.get(DIMMING_ABILITY).is_enabled:
|
||||
return SUPPORT_BRIGHTNESS
|
||||
return 0
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self) -> Mapping[str, Any] | None:
|
||||
"""State attributes for Crownstone devices."""
|
||||
attributes: dict[str, Any] = {}
|
||||
# switch method
|
||||
if self.usb_available:
|
||||
attributes["switch_method"] = "Crownstone USB Dongle"
|
||||
else:
|
||||
attributes["switch_method"] = "Crownstone Cloud"
|
||||
|
||||
# crownstone abilities
|
||||
attributes["dimming"] = ABILITY_STATE.get(
|
||||
self.device.abilities.get(DIMMING_ABILITY).is_enabled
|
||||
)
|
||||
attributes["tap_to_toggle"] = ABILITY_STATE.get(
|
||||
self.device.abilities.get(TAP_TO_TOGGLE_ABILITY).is_enabled
|
||||
)
|
||||
attributes["switchcraft"] = ABILITY_STATE.get(
|
||||
self.device.abilities.get(SWITCHCRAFT_ABILITY).is_enabled
|
||||
)
|
||||
|
||||
return attributes
|
||||
|
||||
async def async_added_to_hass(self) -> None:
|
||||
"""Set up a listener when this entity is added to HA."""
|
||||
# new state received
|
||||
self.async_on_remove(
|
||||
async_dispatcher_connect(
|
||||
self.hass, SIG_CROWNSTONE_STATE_UPDATE, self.async_write_ha_state
|
||||
)
|
||||
)
|
||||
# updates state attributes when usb connects/disconnects
|
||||
self.async_on_remove(
|
||||
async_dispatcher_connect(
|
||||
self.hass, SIG_UART_STATE_CHANGE, self.async_write_ha_state
|
||||
)
|
||||
)
|
||||
|
||||
async def async_turn_on(self, **kwargs: Any) -> None:
|
||||
"""Turn on this light via dongle or cloud."""
|
||||
if ATTR_BRIGHTNESS in kwargs:
|
||||
if self.usb_available:
|
||||
await self.hass.async_add_executor_job(
|
||||
partial(
|
||||
self.usb.dim_crownstone,
|
||||
self.device.unique_id,
|
||||
hass_to_crownstone_state(kwargs[ATTR_BRIGHTNESS]),
|
||||
)
|
||||
)
|
||||
else:
|
||||
try:
|
||||
await self.device.async_set_brightness(
|
||||
hass_to_crownstone_state(kwargs[ATTR_BRIGHTNESS])
|
||||
)
|
||||
except CrownstoneAbilityError as ability_error:
|
||||
_LOGGER.error(ability_error)
|
||||
return
|
||||
|
||||
# assume brightness is set on device
|
||||
self.device.state = hass_to_crownstone_state(kwargs[ATTR_BRIGHTNESS])
|
||||
self.async_write_ha_state()
|
||||
|
||||
elif self.usb_available:
|
||||
await self.hass.async_add_executor_job(
|
||||
partial(self.usb.switch_crownstone, self.device.unique_id, on=True)
|
||||
)
|
||||
self.device.state = 100
|
||||
self.async_write_ha_state()
|
||||
|
||||
else:
|
||||
await self.device.async_turn_on()
|
||||
self.device.state = 100
|
||||
self.async_write_ha_state()
|
||||
|
||||
async def async_turn_off(self, **kwargs: Any) -> None:
|
||||
"""Turn off this device via dongle or cloud."""
|
||||
if self.usb_available:
|
||||
await self.hass.async_add_executor_job(
|
||||
partial(self.usb.switch_crownstone, self.device.unique_id, on=False)
|
||||
)
|
||||
|
||||
else:
|
||||
await self.device.async_turn_off()
|
||||
|
||||
self.device.state = 0
|
||||
self.async_write_ha_state()
|
||||
|
|
@ -0,0 +1,147 @@
|
|||
"""
|
||||
Listeners for updating data in the Crownstone integration.
|
||||
|
||||
For data updates, Cloud Push is used in form of an SSE server that sends out events.
|
||||
For fast device switching Local Push is used in form of a USB dongle that hooks into a BLE mesh.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from functools import partial
|
||||
from typing import TYPE_CHECKING, cast
|
||||
|
||||
from crownstone_core.packets.serviceDataParsers.containers.AdvExternalCrownstoneState import (
|
||||
AdvExternalCrownstoneState,
|
||||
)
|
||||
from crownstone_core.packets.serviceDataParsers.containers.elements.AdvTypes import (
|
||||
AdvType,
|
||||
)
|
||||
from crownstone_core.protocol.SwitchState import SwitchState
|
||||
from crownstone_sse.const import (
|
||||
EVENT_ABILITY_CHANGE,
|
||||
EVENT_ABILITY_CHANGE_DIMMING,
|
||||
EVENT_SWITCH_STATE_UPDATE,
|
||||
)
|
||||
from crownstone_sse.events import AbilityChangeEvent, SwitchStateUpdateEvent
|
||||
from crownstone_uart import UartEventBus, UartTopics
|
||||
from crownstone_uart.topics.SystemTopics import SystemTopics
|
||||
|
||||
from homeassistant.core import Event, callback
|
||||
from homeassistant.helpers.dispatcher import async_dispatcher_send, dispatcher_send
|
||||
|
||||
from .const import (
|
||||
DOMAIN,
|
||||
SIG_CROWNSTONE_STATE_UPDATE,
|
||||
SIG_UART_STATE_CHANGE,
|
||||
SSE_LISTENERS,
|
||||
UART_LISTENERS,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .entry_manager import CrownstoneEntryManager
|
||||
|
||||
|
||||
@callback
|
||||
def async_update_crwn_state_sse(
|
||||
manager: CrownstoneEntryManager, ha_event: Event
|
||||
) -> None:
|
||||
"""Update the state of a Crownstone when switched externally."""
|
||||
switch_event = SwitchStateUpdateEvent(ha_event.data)
|
||||
try:
|
||||
updated_crownstone = manager.cloud.get_crownstone_by_id(switch_event.cloud_id)
|
||||
except KeyError:
|
||||
return
|
||||
|
||||
# only update on change.
|
||||
if updated_crownstone.state != switch_event.switch_state:
|
||||
updated_crownstone.state = switch_event.switch_state
|
||||
async_dispatcher_send(manager.hass, SIG_CROWNSTONE_STATE_UPDATE)
|
||||
|
||||
|
||||
@callback
|
||||
def async_update_crwn_ability(manager: CrownstoneEntryManager, ha_event: Event) -> None:
|
||||
"""Update the ability information of a Crownstone."""
|
||||
ability_event = AbilityChangeEvent(ha_event.data)
|
||||
try:
|
||||
updated_crownstone = manager.cloud.get_crownstone_by_id(ability_event.cloud_id)
|
||||
except KeyError:
|
||||
return
|
||||
|
||||
ability_type = ability_event.ability_type
|
||||
ability_enabled = ability_event.ability_enabled
|
||||
# only update on a change in state
|
||||
if updated_crownstone.abilities[ability_type].is_enabled == ability_enabled:
|
||||
return
|
||||
|
||||
# write the change to the crownstone entity.
|
||||
updated_crownstone.abilities[ability_type].is_enabled = ability_enabled
|
||||
|
||||
if ability_event.sub_type == EVENT_ABILITY_CHANGE_DIMMING:
|
||||
# reload the config entry because dimming is part of supported features
|
||||
manager.hass.async_create_task(
|
||||
manager.hass.config_entries.async_reload(manager.config_entry.entry_id)
|
||||
)
|
||||
else:
|
||||
async_dispatcher_send(manager.hass, SIG_CROWNSTONE_STATE_UPDATE)
|
||||
|
||||
|
||||
def update_uart_state(manager: CrownstoneEntryManager, _: bool | None) -> None:
|
||||
"""Update the uart ready state for entities that use USB."""
|
||||
# update availability of power usage entities.
|
||||
dispatcher_send(manager.hass, SIG_UART_STATE_CHANGE)
|
||||
|
||||
|
||||
def update_crwn_state_uart(
|
||||
manager: CrownstoneEntryManager, data: AdvExternalCrownstoneState
|
||||
) -> None:
|
||||
"""Update the state of a Crownstone when switched externally."""
|
||||
if data.type != AdvType.EXTERNAL_STATE:
|
||||
return
|
||||
try:
|
||||
updated_crownstone = manager.cloud.get_crownstone_by_uid(
|
||||
data.crownstoneId, manager.usb_sphere_id
|
||||
)
|
||||
except KeyError:
|
||||
return
|
||||
|
||||
if data.switchState is None:
|
||||
return
|
||||
# update on change
|
||||
updated_state = cast(SwitchState, data.switchState)
|
||||
if updated_crownstone.state != updated_state.intensity:
|
||||
updated_crownstone.state = updated_state.intensity
|
||||
|
||||
dispatcher_send(manager.hass, SIG_CROWNSTONE_STATE_UPDATE)
|
||||
|
||||
|
||||
def setup_sse_listeners(manager: CrownstoneEntryManager) -> None:
|
||||
"""Set up SSE listeners."""
|
||||
# save unsub function for when entry removed
|
||||
manager.listeners[SSE_LISTENERS] = [
|
||||
manager.hass.bus.async_listen(
|
||||
f"{DOMAIN}_{EVENT_SWITCH_STATE_UPDATE}",
|
||||
partial(async_update_crwn_state_sse, manager),
|
||||
),
|
||||
manager.hass.bus.async_listen(
|
||||
f"{DOMAIN}_{EVENT_ABILITY_CHANGE}",
|
||||
partial(async_update_crwn_ability, manager),
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
def setup_uart_listeners(manager: CrownstoneEntryManager) -> None:
|
||||
"""Set up UART listeners."""
|
||||
# save subscription id to unsub
|
||||
manager.listeners[UART_LISTENERS] = [
|
||||
UartEventBus.subscribe(
|
||||
SystemTopics.connectionEstablished,
|
||||
partial(update_uart_state, manager),
|
||||
),
|
||||
UartEventBus.subscribe(
|
||||
SystemTopics.connectionClosed,
|
||||
partial(update_uart_state, manager),
|
||||
),
|
||||
UartEventBus.subscribe(
|
||||
UartTopics.newDataAvailable,
|
||||
partial(update_crwn_state_uart, manager),
|
||||
),
|
||||
]
|
||||
|
|
@ -0,0 +1,15 @@
|
|||
{
|
||||
"domain": "crownstone",
|
||||
"name": "Crownstone",
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/integrations/crownstone",
|
||||
"requirements": [
|
||||
"crownstone-cloud==1.4.7",
|
||||
"crownstone-sse==2.0.2",
|
||||
"crownstone-uart==2.1.0",
|
||||
"pyserial==3.5"
|
||||
],
|
||||
"codeowners": ["@Crownstone", "@RicArch97"],
|
||||
"after_dependencies": ["usb"],
|
||||
"iot_class": "cloud_push"
|
||||
}
|
||||
|
|
@ -0,0 +1,75 @@
|
|||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
|
||||
"usb_setup_complete": "Crownstone USB setup complete.",
|
||||
"usb_setup_unsuccessful": "Crownstone USB setup was unsuccessful."
|
||||
},
|
||||
"error": {
|
||||
"account_not_verified": "Account not verified. Please activate your account through the activation email from Crownstone.",
|
||||
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
|
||||
"unknown": "[%key:common::config_flow::error::unknown%]"
|
||||
},
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"email": "[%key:common::config_flow::data::email%]",
|
||||
"password": "[%key:common::config_flow::data::password%]"
|
||||
},
|
||||
"title": "Crownstone account"
|
||||
},
|
||||
"usb_config": {
|
||||
"data": {
|
||||
"usb_path": "[%key:common::config_flow::data::usb_path%]"
|
||||
},
|
||||
"title": "Crownstone USB dongle configuration",
|
||||
"description": "Select the serial port of the Crownstone USB dongle, or select 'Don't use USB' if you don't want to setup a USB dongle.\n\nLook for a device with VID 10C4 and PID EA60."
|
||||
},
|
||||
"usb_manual_config": {
|
||||
"data": {
|
||||
"usb_manual_path": "[%key:common::config_flow::data::usb_path%]"
|
||||
},
|
||||
"title": "Crownstone USB dongle manual path",
|
||||
"description": "Manually enter the path of a Crownstone USB dongle."
|
||||
},
|
||||
"usb_sphere_config": {
|
||||
"data": {
|
||||
"usb_sphere": "Crownstone Sphere"
|
||||
},
|
||||
"title": "Crownstone USB Sphere",
|
||||
"description": "Select a Crownstone Sphere where the USB is located."
|
||||
}
|
||||
}
|
||||
},
|
||||
"options": {
|
||||
"step": {
|
||||
"init": {
|
||||
"data": {
|
||||
"use_usb_option": "Use a Crownstone USB dongle for local data transmission",
|
||||
"usb_sphere_option": "Crownstone Sphere where the USB is located"
|
||||
}
|
||||
},
|
||||
"usb_config_option": {
|
||||
"data": {
|
||||
"usb_path": "[%key:common::config_flow::data::usb_path%]"
|
||||
},
|
||||
"title": "Crownstone USB dongle configuration",
|
||||
"description": "Select the serial port of the Crownstone USB dongle.\n\nLook for a device with VID 10C4 and PID EA60."
|
||||
},
|
||||
"usb_manual_config_option": {
|
||||
"data": {
|
||||
"usb_manual_path": "[%key:common::config_flow::data::usb_path%]"
|
||||
},
|
||||
"title": "Crownstone USB dongle manual path",
|
||||
"description": "Manually enter the path of a Crownstone USB dongle."
|
||||
},
|
||||
"usb_sphere_config_option": {
|
||||
"data": {
|
||||
"usb_sphere": "Crownstone Sphere"
|
||||
},
|
||||
"title": "Crownstone USB Sphere",
|
||||
"description": "Select a Crownstone Sphere where the USB is located."
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,75 @@
|
|||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"already_configured": "Account is already configured",
|
||||
"usb_setup_complete": "Crownstone USB setup complete.",
|
||||
"usb_setup_unsuccessful": "Crownstone USB setup was unsuccessful."
|
||||
},
|
||||
"error": {
|
||||
"account_not_verified": "Account not verified. Please activate your account through the activation email from Crownstone.",
|
||||
"invalid_auth": "Invalid authentication",
|
||||
"unknown": "Unexpected error"
|
||||
},
|
||||
"step": {
|
||||
"usb_config": {
|
||||
"data": {
|
||||
"usb_path": "USB Device Path"
|
||||
},
|
||||
"description": "Select the serial port of the Crownstone USB dongle, or select 'Don't use USB' if you don't want to setup a USB dongle.\n\nLook for a device with VID 10C4 and PID EA60.",
|
||||
"title": "Crownstone USB dongle configuration"
|
||||
},
|
||||
"usb_manual_config": {
|
||||
"data": {
|
||||
"usb_manual_path": "USB Device Path"
|
||||
},
|
||||
"description": "Manually enter the path of a Crownstone USB dongle.",
|
||||
"title": "Crownstone USB dongle manual path"
|
||||
},
|
||||
"usb_sphere_config": {
|
||||
"data": {
|
||||
"usb_sphere": "Crownstone Sphere"
|
||||
},
|
||||
"description": "Select a Crownstone Sphere where the USB is located.",
|
||||
"title": "Crownstone USB Sphere"
|
||||
},
|
||||
"user": {
|
||||
"data": {
|
||||
"email": "Email",
|
||||
"password": "Password"
|
||||
},
|
||||
"title": "Crownstone account"
|
||||
}
|
||||
}
|
||||
},
|
||||
"options": {
|
||||
"step": {
|
||||
"init": {
|
||||
"data": {
|
||||
"usb_sphere_option": "Crownstone Sphere where the USB is located",
|
||||
"use_usb_option": "Use a Crownstone USB dongle for local data transmission"
|
||||
}
|
||||
},
|
||||
"usb_config_option": {
|
||||
"data": {
|
||||
"usb_path": "USB Device Path"
|
||||
},
|
||||
"description": "Select the serial port of the Crownstone USB dongle.\n\nLook for a device with VID 10C4 and PID EA60.",
|
||||
"title": "Crownstone USB dongle configuration"
|
||||
},
|
||||
"usb_manual_config_option": {
|
||||
"data": {
|
||||
"usb_manual_path": "USB Device Path"
|
||||
},
|
||||
"description": "Manually enter the path of a Crownstone USB dongle.",
|
||||
"title": "Crownstone USB dongle manual path"
|
||||
},
|
||||
"usb_sphere_config_option": {
|
||||
"data": {
|
||||
"usb_sphere": "Crownstone Sphere"
|
||||
},
|
||||
"description": "Select a Crownstone Sphere where the USB is located.",
|
||||
"title": "Crownstone USB Sphere"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -52,6 +52,7 @@ FLOWS = [
|
|||
"control4",
|
||||
"coolmaster",
|
||||
"coronavirus",
|
||||
"crownstone",
|
||||
"daikin",
|
||||
"deconz",
|
||||
"denonavr",
|
||||
|
|
|
|||
11
mypy.ini
11
mypy.ini
|
|
@ -308,6 +308,17 @@ no_implicit_optional = true
|
|||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.crownstone.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
disallow_subclassing_any = true
|
||||
disallow_untyped_calls = true
|
||||
disallow_untyped_decorators = true
|
||||
disallow_untyped_defs = true
|
||||
no_implicit_optional = true
|
||||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.device_automation.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
|
|
|
|||
|
|
@ -486,6 +486,15 @@ coronavirus==1.1.1
|
|||
# homeassistant.components.utility_meter
|
||||
croniter==1.0.6
|
||||
|
||||
# homeassistant.components.crownstone
|
||||
crownstone-cloud==1.4.7
|
||||
|
||||
# homeassistant.components.crownstone
|
||||
crownstone-sse==2.0.2
|
||||
|
||||
# homeassistant.components.crownstone
|
||||
crownstone-uart==2.1.0
|
||||
|
||||
# homeassistant.components.datadog
|
||||
datadog==0.15.0
|
||||
|
||||
|
|
@ -1761,6 +1770,7 @@ pysensibo==1.0.3
|
|||
pyserial-asyncio==0.5
|
||||
|
||||
# homeassistant.components.acer_projector
|
||||
# homeassistant.components.crownstone
|
||||
# homeassistant.components.usb
|
||||
# homeassistant.components.zha
|
||||
pyserial==3.5
|
||||
|
|
|
|||
|
|
@ -285,6 +285,15 @@ coronavirus==1.1.1
|
|||
# homeassistant.components.utility_meter
|
||||
croniter==1.0.6
|
||||
|
||||
# homeassistant.components.crownstone
|
||||
crownstone-cloud==1.4.7
|
||||
|
||||
# homeassistant.components.crownstone
|
||||
crownstone-sse==2.0.2
|
||||
|
||||
# homeassistant.components.crownstone
|
||||
crownstone-uart==2.1.0
|
||||
|
||||
# homeassistant.components.datadog
|
||||
datadog==0.15.0
|
||||
|
||||
|
|
@ -1026,6 +1035,7 @@ pyruckus==0.12
|
|||
pyserial-asyncio==0.5
|
||||
|
||||
# homeassistant.components.acer_projector
|
||||
# homeassistant.components.crownstone
|
||||
# homeassistant.components.usb
|
||||
# homeassistant.components.zha
|
||||
pyserial==3.5
|
||||
|
|
|
|||
|
|
@ -0,0 +1 @@
|
|||
"""Tests for the Crownstone integration."""
|
||||
|
|
@ -0,0 +1,531 @@
|
|||
"""Tests for the Crownstone integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from crownstone_cloud.cloud_models.spheres import Spheres
|
||||
from crownstone_cloud.exceptions import (
|
||||
CrownstoneAuthenticationError,
|
||||
CrownstoneUnknownError,
|
||||
)
|
||||
import pytest
|
||||
from serial.tools.list_ports_common import ListPortInfo
|
||||
|
||||
from homeassistant import data_entry_flow
|
||||
from homeassistant.components import usb
|
||||
from homeassistant.components.crownstone.const import (
|
||||
CONF_USB_MANUAL_PATH,
|
||||
CONF_USB_PATH,
|
||||
CONF_USB_SPHERE,
|
||||
CONF_USB_SPHERE_OPTION,
|
||||
CONF_USE_USB_OPTION,
|
||||
DOMAIN,
|
||||
DONT_USE_USB,
|
||||
MANUAL_PATH,
|
||||
)
|
||||
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
@pytest.fixture(name="crownstone_setup", autouse=True)
|
||||
def crownstone_setup():
|
||||
"""Mock Crownstone entry setup."""
|
||||
with patch(
|
||||
"homeassistant.components.crownstone.async_setup_entry", return_value=True
|
||||
):
|
||||
yield
|
||||
|
||||
|
||||
def get_mocked_crownstone_cloud(spheres: dict[str, MagicMock] | None = None):
|
||||
"""Return a mocked Crownstone Cloud instance."""
|
||||
mock_cloud = MagicMock()
|
||||
mock_cloud.async_initialize = AsyncMock()
|
||||
mock_cloud.cloud_data = Spheres(MagicMock(), "account_id")
|
||||
mock_cloud.cloud_data.spheres = spheres
|
||||
|
||||
return mock_cloud
|
||||
|
||||
|
||||
def create_mocked_spheres(amount: int) -> dict[str, MagicMock]:
|
||||
"""Return a dict with mocked sphere instances."""
|
||||
spheres: dict[str, MagicMock] = {}
|
||||
for i in range(amount):
|
||||
spheres[f"sphere_id_{i}"] = MagicMock()
|
||||
spheres[f"sphere_id_{i}"].name = f"sphere_name_{i}"
|
||||
spheres[f"sphere_id_{i}"].cloud_id = f"sphere_id_{i}"
|
||||
|
||||
return spheres
|
||||
|
||||
|
||||
def get_mocked_com_port():
|
||||
"""Mock of a serial port."""
|
||||
port = ListPortInfo("/dev/ttyUSB1234")
|
||||
port.device = "/dev/ttyUSB1234"
|
||||
port.serial_number = "1234567"
|
||||
port.manufacturer = "crownstone"
|
||||
port.description = "crownstone dongle - crownstone dongle"
|
||||
port.vid = 1234
|
||||
port.pid = 5678
|
||||
|
||||
return port
|
||||
|
||||
|
||||
def create_mocked_entry_data_conf(email: str, password: str):
|
||||
"""Set a result for the entry data for comparison."""
|
||||
mock_data: dict[str, str | None] = {}
|
||||
mock_data[CONF_EMAIL] = email
|
||||
mock_data[CONF_PASSWORD] = password
|
||||
|
||||
return mock_data
|
||||
|
||||
|
||||
def create_mocked_entry_options_conf(usb_path: str | None, usb_sphere: str | None):
|
||||
"""Set a result for the entry options for comparison."""
|
||||
mock_options: dict[str, str | None] = {}
|
||||
mock_options[CONF_USB_PATH] = usb_path
|
||||
mock_options[CONF_USB_SPHERE] = usb_sphere
|
||||
|
||||
return mock_options
|
||||
|
||||
|
||||
async def start_config_flow(hass: HomeAssistant, mocked_cloud: MagicMock):
|
||||
"""Patch Crownstone Cloud and start the flow."""
|
||||
mocked_login_input = {
|
||||
CONF_EMAIL: "example@homeassistant.com",
|
||||
CONF_PASSWORD: "homeassistantisawesome",
|
||||
}
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.crownstone.config_flow.CrownstoneCloud",
|
||||
return_value=mocked_cloud,
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": "user"}, data=mocked_login_input
|
||||
)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
async def start_options_flow(
|
||||
hass: HomeAssistant, entry_id: str, mocked_cloud: MagicMock
|
||||
):
|
||||
"""Patch CrownstoneEntryManager and start the flow."""
|
||||
mocked_manager = MagicMock()
|
||||
mocked_manager.cloud = mocked_cloud
|
||||
hass.data.setdefault(DOMAIN, {})
|
||||
hass.data[DOMAIN][entry_id] = mocked_manager
|
||||
|
||||
return await hass.config_entries.options.async_init(entry_id)
|
||||
|
||||
|
||||
async def test_no_user_input(hass: HomeAssistant):
|
||||
"""Test the flow done in the correct way."""
|
||||
# test if a form is returned if no input is provided
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": "user"}
|
||||
)
|
||||
# show the login form
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
|
||||
async def test_abort_if_configured(hass: HomeAssistant):
|
||||
"""Test flow with correct login input and abort if sphere already configured."""
|
||||
# create mock entry conf
|
||||
configured_entry_data = create_mocked_entry_data_conf(
|
||||
email="example@homeassistant.com",
|
||||
password="homeassistantisawesome",
|
||||
)
|
||||
configured_entry_options = create_mocked_entry_options_conf(
|
||||
usb_path="/dev/serial/by-id/crownstone-usb",
|
||||
usb_sphere="sphere_id",
|
||||
)
|
||||
|
||||
# create mocked entry
|
||||
MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data=configured_entry_data,
|
||||
options=configured_entry_options,
|
||||
unique_id="account_id",
|
||||
).add_to_hass(hass)
|
||||
|
||||
result = await start_config_flow(hass, get_mocked_crownstone_cloud())
|
||||
|
||||
# test if we abort if we try to configure the same entry
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
async def test_authentication_errors(hass: HomeAssistant):
|
||||
"""Test flow with wrong auth errors."""
|
||||
cloud = get_mocked_crownstone_cloud()
|
||||
# side effect: auth error login failed
|
||||
cloud.async_initialize.side_effect = CrownstoneAuthenticationError(
|
||||
exception_type="LOGIN_FAILED"
|
||||
)
|
||||
|
||||
result = await start_config_flow(hass, cloud)
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "invalid_auth"}
|
||||
|
||||
# side effect: auth error account not verified
|
||||
cloud.async_initialize.side_effect = CrownstoneAuthenticationError(
|
||||
exception_type="LOGIN_FAILED_EMAIL_NOT_VERIFIED"
|
||||
)
|
||||
|
||||
result = await start_config_flow(hass, cloud)
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "account_not_verified"}
|
||||
|
||||
|
||||
async def test_unknown_error(hass: HomeAssistant):
|
||||
"""Test flow with unknown error."""
|
||||
cloud = get_mocked_crownstone_cloud()
|
||||
# side effect: unknown error
|
||||
cloud.async_initialize.side_effect = CrownstoneUnknownError
|
||||
|
||||
result = await start_config_flow(hass, cloud)
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "unknown_error"}
|
||||
|
||||
|
||||
async def test_successful_login_no_usb(hass: HomeAssistant):
|
||||
"""Test a successful login without configuring a USB."""
|
||||
entry_data_without_usb = create_mocked_entry_data_conf(
|
||||
email="example@homeassistant.com",
|
||||
password="homeassistantisawesome",
|
||||
)
|
||||
entry_options_without_usb = create_mocked_entry_options_conf(
|
||||
usb_path=None,
|
||||
usb_sphere=None,
|
||||
)
|
||||
|
||||
result = await start_config_flow(hass, get_mocked_crownstone_cloud())
|
||||
# should show usb form
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "usb_config"
|
||||
|
||||
# don't setup USB dongle, create entry
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input={CONF_USB_PATH: DONT_USE_USB}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["data"] == entry_data_without_usb
|
||||
assert result["options"] == entry_options_without_usb
|
||||
|
||||
|
||||
@patch(
|
||||
"serial.tools.list_ports.comports", MagicMock(return_value=[get_mocked_com_port()])
|
||||
)
|
||||
@patch(
|
||||
"homeassistant.components.usb.get_serial_by_id",
|
||||
return_value="/dev/serial/by-id/crownstone-usb",
|
||||
)
|
||||
async def test_successful_login_with_usb(serial_mock: MagicMock, hass: HomeAssistant):
|
||||
"""Test flow with correct login and usb configuration."""
|
||||
entry_data_with_usb = create_mocked_entry_data_conf(
|
||||
email="example@homeassistant.com",
|
||||
password="homeassistantisawesome",
|
||||
)
|
||||
entry_options_with_usb = create_mocked_entry_options_conf(
|
||||
usb_path="/dev/serial/by-id/crownstone-usb",
|
||||
usb_sphere="sphere_id_1",
|
||||
)
|
||||
|
||||
result = await start_config_flow(
|
||||
hass, get_mocked_crownstone_cloud(create_mocked_spheres(2))
|
||||
)
|
||||
# should show usb form
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "usb_config"
|
||||
|
||||
# create a mocked port
|
||||
port = get_mocked_com_port()
|
||||
port_select = usb.human_readable_device_name(
|
||||
port.device,
|
||||
port.serial_number,
|
||||
port.manufacturer,
|
||||
port.description,
|
||||
f"{hex(port.vid)[2:]:0>4}".upper(),
|
||||
f"{hex(port.pid)[2:]:0>4}".upper(),
|
||||
)
|
||||
|
||||
# select a port from the list
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input={CONF_USB_PATH: port_select}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "usb_sphere_config"
|
||||
assert serial_mock.call_count == 1
|
||||
|
||||
# select a sphere
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input={CONF_USB_SPHERE: "sphere_name_1"}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["data"] == entry_data_with_usb
|
||||
assert result["options"] == entry_options_with_usb
|
||||
|
||||
|
||||
@patch(
|
||||
"serial.tools.list_ports.comports", MagicMock(return_value=[get_mocked_com_port()])
|
||||
)
|
||||
async def test_successful_login_with_manual_usb_path(hass: HomeAssistant):
|
||||
"""Test flow with correct login and usb configuration."""
|
||||
entry_data_with_manual_usb = create_mocked_entry_data_conf(
|
||||
email="example@homeassistant.com",
|
||||
password="homeassistantisawesome",
|
||||
)
|
||||
entry_options_with_manual_usb = create_mocked_entry_options_conf(
|
||||
usb_path="/dev/crownstone-usb",
|
||||
usb_sphere="sphere_id_0",
|
||||
)
|
||||
|
||||
result = await start_config_flow(
|
||||
hass, get_mocked_crownstone_cloud(create_mocked_spheres(1))
|
||||
)
|
||||
# should show usb form
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "usb_config"
|
||||
|
||||
# select manual from the list
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input={CONF_USB_PATH: MANUAL_PATH}
|
||||
)
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "usb_manual_config"
|
||||
|
||||
# enter USB path
|
||||
path = "/dev/crownstone-usb"
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input={CONF_USB_MANUAL_PATH: path}
|
||||
)
|
||||
|
||||
# since we only have 1 sphere here, test that it's automatically selected and
|
||||
# creating entry without asking for user input
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["data"] == entry_data_with_manual_usb
|
||||
assert result["options"] == entry_options_with_manual_usb
|
||||
|
||||
|
||||
@patch(
|
||||
"serial.tools.list_ports.comports", MagicMock(return_value=[get_mocked_com_port()])
|
||||
)
|
||||
@patch(
|
||||
"homeassistant.components.usb.get_serial_by_id",
|
||||
return_value="/dev/serial/by-id/crownstone-usb",
|
||||
)
|
||||
async def test_options_flow_setup_usb(serial_mock: MagicMock, hass: HomeAssistant):
|
||||
"""Test options flow init."""
|
||||
configured_entry_data = create_mocked_entry_data_conf(
|
||||
email="example@homeassistant.com",
|
||||
password="homeassistantisawesome",
|
||||
)
|
||||
configured_entry_options = create_mocked_entry_options_conf(
|
||||
usb_path=None,
|
||||
usb_sphere=None,
|
||||
)
|
||||
|
||||
# create mocked entry
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data=configured_entry_data,
|
||||
options=configured_entry_options,
|
||||
unique_id="account_id",
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
result = await start_options_flow(
|
||||
hass, entry.entry_id, get_mocked_crownstone_cloud(create_mocked_spheres(2))
|
||||
)
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "init"
|
||||
|
||||
schema = result["data_schema"].schema
|
||||
for schema_key in schema:
|
||||
if schema_key == CONF_USE_USB_OPTION:
|
||||
assert not schema_key.default()
|
||||
|
||||
# USB is not set up, so this should not be in the options
|
||||
assert CONF_USB_SPHERE_OPTION not in schema
|
||||
|
||||
result = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"], user_input={CONF_USE_USB_OPTION: True}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "usb_config_option"
|
||||
|
||||
# create a mocked port
|
||||
port = get_mocked_com_port()
|
||||
port_select = usb.human_readable_device_name(
|
||||
port.device,
|
||||
port.serial_number,
|
||||
port.manufacturer,
|
||||
port.description,
|
||||
f"{hex(port.vid)[2:]:0>4}".upper(),
|
||||
f"{hex(port.pid)[2:]:0>4}".upper(),
|
||||
)
|
||||
|
||||
# select a port from the list
|
||||
result = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"], user_input={CONF_USB_PATH: port_select}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "usb_sphere_config_option"
|
||||
assert serial_mock.call_count == 1
|
||||
|
||||
# select a sphere
|
||||
result = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"], user_input={CONF_USB_SPHERE: "sphere_name_1"}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["data"] == create_mocked_entry_options_conf(
|
||||
usb_path="/dev/serial/by-id/crownstone-usb", usb_sphere="sphere_id_1"
|
||||
)
|
||||
|
||||
|
||||
async def test_options_flow_remove_usb(hass: HomeAssistant):
|
||||
"""Test selecting to set up an USB dongle."""
|
||||
configured_entry_data = create_mocked_entry_data_conf(
|
||||
email="example@homeassistant.com",
|
||||
password="homeassistantisawesome",
|
||||
)
|
||||
configured_entry_options = create_mocked_entry_options_conf(
|
||||
usb_path="/dev/serial/by-id/crownstone-usb",
|
||||
usb_sphere="sphere_id_0",
|
||||
)
|
||||
|
||||
# create mocked entry
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data=configured_entry_data,
|
||||
options=configured_entry_options,
|
||||
unique_id="account_id",
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
result = await start_options_flow(
|
||||
hass, entry.entry_id, get_mocked_crownstone_cloud(create_mocked_spheres(2))
|
||||
)
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "init"
|
||||
|
||||
schema = result["data_schema"].schema
|
||||
for schema_key in schema:
|
||||
if schema_key == CONF_USE_USB_OPTION:
|
||||
assert schema_key.default()
|
||||
if schema_key == CONF_USB_SPHERE_OPTION:
|
||||
assert schema_key.default() == "sphere_name_0"
|
||||
|
||||
result = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_USE_USB_OPTION: False,
|
||||
CONF_USB_SPHERE_OPTION: "sphere_name_0",
|
||||
},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["data"] == create_mocked_entry_options_conf(
|
||||
usb_path=None, usb_sphere=None
|
||||
)
|
||||
|
||||
|
||||
@patch(
|
||||
"serial.tools.list_ports.comports", MagicMock(return_value=[get_mocked_com_port()])
|
||||
)
|
||||
async def test_options_flow_manual_usb_path(hass: HomeAssistant):
|
||||
"""Test flow with correct login and usb configuration."""
|
||||
configured_entry_data = create_mocked_entry_data_conf(
|
||||
email="example@homeassistant.com",
|
||||
password="homeassistantisawesome",
|
||||
)
|
||||
configured_entry_options = create_mocked_entry_options_conf(
|
||||
usb_path=None,
|
||||
usb_sphere=None,
|
||||
)
|
||||
|
||||
# create mocked entry
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data=configured_entry_data,
|
||||
options=configured_entry_options,
|
||||
unique_id="account_id",
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
result = await start_options_flow(
|
||||
hass, entry.entry_id, get_mocked_crownstone_cloud(create_mocked_spheres(1))
|
||||
)
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "init"
|
||||
|
||||
result = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"], user_input={CONF_USE_USB_OPTION: True}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "usb_config_option"
|
||||
|
||||
# select manual from the list
|
||||
result = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"], user_input={CONF_USB_PATH: MANUAL_PATH}
|
||||
)
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "usb_manual_config_option"
|
||||
|
||||
# enter USB path
|
||||
path = "/dev/crownstone-usb"
|
||||
result = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"], user_input={CONF_USB_MANUAL_PATH: path}
|
||||
)
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["data"] == create_mocked_entry_options_conf(
|
||||
usb_path=path, usb_sphere="sphere_id_0"
|
||||
)
|
||||
|
||||
|
||||
async def test_options_flow_change_usb_sphere(hass: HomeAssistant):
|
||||
"""Test changing the usb sphere in the options."""
|
||||
configured_entry_data = create_mocked_entry_data_conf(
|
||||
email="example@homeassistant.com",
|
||||
password="homeassistantisawesome",
|
||||
)
|
||||
configured_entry_options = create_mocked_entry_options_conf(
|
||||
usb_path="/dev/serial/by-id/crownstone-usb",
|
||||
usb_sphere="sphere_id_0",
|
||||
)
|
||||
|
||||
# create mocked entry
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data=configured_entry_data,
|
||||
options=configured_entry_options,
|
||||
unique_id="account_id",
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
result = await start_options_flow(
|
||||
hass, entry.entry_id, get_mocked_crownstone_cloud(create_mocked_spheres(3))
|
||||
)
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "init"
|
||||
|
||||
result = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_USE_USB_OPTION: True, CONF_USB_SPHERE_OPTION: "sphere_name_2"},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["data"] == create_mocked_entry_options_conf(
|
||||
usb_path="/dev/serial/by-id/crownstone-usb", usb_sphere="sphere_id_2"
|
||||
)
|
||||
Loading…
Reference in New Issue