From 2a51bb5bba0105a7b4eaf3d6f81c02e65dcb4aa2 Mon Sep 17 00:00:00 2001 From: Ricardo Steijn <61013287+RicArch97@users.noreply.github.com> Date: Tue, 14 Sep 2021 21:46:52 +0200 Subject: [PATCH] Add Crownstone integration (#50677) --- .coveragerc | 7 + .strict-typing | 1 + CODEOWNERS | 1 + .../components/crownstone/__init__.py | 23 + .../components/crownstone/config_flow.py | 299 ++++++++++ homeassistant/components/crownstone/const.py | 45 ++ .../components/crownstone/devices.py | 43 ++ .../components/crownstone/entry_manager.py | 190 +++++++ .../components/crownstone/helpers.py | 59 ++ homeassistant/components/crownstone/light.py | 204 +++++++ .../components/crownstone/listeners.py | 147 +++++ .../components/crownstone/manifest.json | 15 + .../components/crownstone/strings.json | 75 +++ .../crownstone/translations/en.json | 75 +++ homeassistant/generated/config_flows.py | 1 + mypy.ini | 11 + requirements_all.txt | 10 + requirements_test_all.txt | 10 + tests/components/crownstone/__init__.py | 1 + .../components/crownstone/test_config_flow.py | 531 ++++++++++++++++++ 20 files changed, 1748 insertions(+) create mode 100644 homeassistant/components/crownstone/__init__.py create mode 100644 homeassistant/components/crownstone/config_flow.py create mode 100644 homeassistant/components/crownstone/const.py create mode 100644 homeassistant/components/crownstone/devices.py create mode 100644 homeassistant/components/crownstone/entry_manager.py create mode 100644 homeassistant/components/crownstone/helpers.py create mode 100644 homeassistant/components/crownstone/light.py create mode 100644 homeassistant/components/crownstone/listeners.py create mode 100644 homeassistant/components/crownstone/manifest.json create mode 100644 homeassistant/components/crownstone/strings.json create mode 100644 homeassistant/components/crownstone/translations/en.json create mode 100644 tests/components/crownstone/__init__.py create mode 100644 tests/components/crownstone/test_config_flow.py diff --git a/.coveragerc b/.coveragerc index 6ac8779ad75..119f0345f3f 100644 --- a/.coveragerc +++ b/.coveragerc @@ -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/* diff --git a/.strict-typing b/.strict-typing index 68c8f62daf6..b664fc3b886 100644 --- a/.strict-typing +++ b/.strict-typing @@ -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.* diff --git a/CODEOWNERS b/CODEOWNERS index c05b11a5b02..9982fc8de7b 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -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 diff --git a/homeassistant/components/crownstone/__init__.py b/homeassistant/components/crownstone/__init__.py new file mode 100644 index 00000000000..bd4aae79665 --- /dev/null +++ b/homeassistant/components/crownstone/__init__.py @@ -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 diff --git a/homeassistant/components/crownstone/config_flow.py b/homeassistant/components/crownstone/config_flow.py new file mode 100644 index 00000000000..72edeef7910 --- /dev/null +++ b/homeassistant/components/crownstone/config_flow.py @@ -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) diff --git a/homeassistant/components/crownstone/const.py b/homeassistant/components/crownstone/const.py new file mode 100644 index 00000000000..2238701dcaf --- /dev/null +++ b/homeassistant/components/crownstone/const.py @@ -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" diff --git a/homeassistant/components/crownstone/devices.py b/homeassistant/components/crownstone/devices.py new file mode 100644 index 00000000000..49965bc8fcd --- /dev/null +++ b/homeassistant/components/crownstone/devices.py @@ -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, + } diff --git a/homeassistant/components/crownstone/entry_manager.py b/homeassistant/components/crownstone/entry_manager.py new file mode 100644 index 00000000000..b01316a771a --- /dev/null +++ b/homeassistant/components/crownstone/entry_manager.py @@ -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) diff --git a/homeassistant/components/crownstone/helpers.py b/homeassistant/components/crownstone/helpers.py new file mode 100644 index 00000000000..ad12c28d464 --- /dev/null +++ b/homeassistant/components/crownstone/helpers.py @@ -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) diff --git a/homeassistant/components/crownstone/light.py b/homeassistant/components/crownstone/light.py new file mode 100644 index 00000000000..b2d4d8411b7 --- /dev/null +++ b/homeassistant/components/crownstone/light.py @@ -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() diff --git a/homeassistant/components/crownstone/listeners.py b/homeassistant/components/crownstone/listeners.py new file mode 100644 index 00000000000..ae316bc0029 --- /dev/null +++ b/homeassistant/components/crownstone/listeners.py @@ -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), + ), + ] diff --git a/homeassistant/components/crownstone/manifest.json b/homeassistant/components/crownstone/manifest.json new file mode 100644 index 00000000000..a7caa6a8d7f --- /dev/null +++ b/homeassistant/components/crownstone/manifest.json @@ -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" +} diff --git a/homeassistant/components/crownstone/strings.json b/homeassistant/components/crownstone/strings.json new file mode 100644 index 00000000000..7437d458ea7 --- /dev/null +++ b/homeassistant/components/crownstone/strings.json @@ -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." + } + } + } +} \ No newline at end of file diff --git a/homeassistant/components/crownstone/translations/en.json b/homeassistant/components/crownstone/translations/en.json new file mode 100644 index 00000000000..e8b552ba53c --- /dev/null +++ b/homeassistant/components/crownstone/translations/en.json @@ -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" + } + } + } +} \ No newline at end of file diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py index 1ad789c33d7..6265360700a 100644 --- a/homeassistant/generated/config_flows.py +++ b/homeassistant/generated/config_flows.py @@ -52,6 +52,7 @@ FLOWS = [ "control4", "coolmaster", "coronavirus", + "crownstone", "daikin", "deconz", "denonavr", diff --git a/mypy.ini b/mypy.ini index 7c195414135..6a27e352595 100644 --- a/mypy.ini +++ b/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 diff --git a/requirements_all.txt b/requirements_all.txt index ed0d0547fa6..5b4bc9e0f73 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -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 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index 97e3c63a95e..e2925d55244 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -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 diff --git a/tests/components/crownstone/__init__.py b/tests/components/crownstone/__init__.py new file mode 100644 index 00000000000..de960619d1d --- /dev/null +++ b/tests/components/crownstone/__init__.py @@ -0,0 +1 @@ +"""Tests for the Crownstone integration.""" diff --git a/tests/components/crownstone/test_config_flow.py b/tests/components/crownstone/test_config_flow.py new file mode 100644 index 00000000000..227657a65a2 --- /dev/null +++ b/tests/components/crownstone/test_config_flow.py @@ -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" + )