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

Migrate Tuya integration to new sharing SDK (#109155)

Browse source 
* Scan QR code to log in And Migrate Tuya integration to new sharing SDK (#104767)

* Remove non-opt-in/out reporting

* Improve setup, fix unload

* Cleanup token listner, remove logging of sensitive data

* Collection of fixes after extensive testing

* Tests happy user config flow path

* Test unhappy paths

* Add reauth

* Fix translation key

* Prettier manifest

* Ruff format

* Cleanup of const

* Process review comments

* Adjust update token handling

---------

Co-authored-by: melo <411787243@qq.com>
This commit is contained in:
Franck Nijhof 2024-01-31 03:22:22 +01:00 • committed by GitHub
parent 712ba2fdca
commit 82e1ed43f8
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
28 changed files with 829 additions and 708 deletions
Download patch file Download diff file Expand all files Collapse all files

181
homeassistant/components/tuya/__init__.py
View file

@ -1,103 +1,75 @@
"""Support for Tuya Smart devices."""
from __future__ import annotations
from typing import NamedTuple
import logging
from typing import Any, NamedTuple
import requests
from tuya_iot import (
AuthType,
TuyaDevice,
TuyaDeviceListener,
TuyaDeviceManager,
TuyaHomeManager,
TuyaOpenAPI,
TuyaOpenMQ,
from tuya_sharing import (
CustomerDevice,
Manager,
SharingDeviceListener,
SharingTokenListener,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_COUNTRY_CODE, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.dispatcher import dispatcher_send
from .const import (
CONF_ACCESS_ID,
CONF_ACCESS_SECRET,
CONF_APP_TYPE,
CONF_AUTH_TYPE,
CONF_ENDPOINT,
CONF_TERMINAL_ID,
CONF_TOKEN_INFO,
CONF_USER_CODE,
DOMAIN,
LOGGER,
PLATFORMS,
TUYA_CLIENT_ID,
TUYA_DISCOVERY_NEW,
TUYA_HA_SIGNAL_UPDATE_ENTITY,
)
# Suppress logs from the library, it logs unneeded on error
logging.getLogger("tuya_sharing").setLevel(logging.CRITICAL)
class HomeAssistantTuyaData(NamedTuple):
"""Tuya data stored in the Home Assistant data object."""
device_listener: TuyaDeviceListener
device_manager: TuyaDeviceManager
home_manager: TuyaHomeManager
manager: Manager
listener: SharingDeviceListener
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Async setup hass config entry."""
hass.data.setdefault(DOMAIN, {})
if CONF_APP_TYPE in entry.data:
raise ConfigEntryAuthFailed("Authentication failed. Please re-authenticate.")
auth_type = AuthType(entry.data[CONF_AUTH_TYPE])
api = TuyaOpenAPI(
endpoint=entry.data[CONF_ENDPOINT],
access_id=entry.data[CONF_ACCESS_ID],
access_secret=entry.data[CONF_ACCESS_SECRET],
auth_type=auth_type,
token_listener = TokenListener(hass, entry)
manager = Manager(
TUYA_CLIENT_ID,
entry.data[CONF_USER_CODE],
entry.data[CONF_TERMINAL_ID],
entry.data[CONF_ENDPOINT],
entry.data[CONF_TOKEN_INFO],
token_listener,
)
api.set_dev_channel("hass")
try:
if auth_type == AuthType.CUSTOM:
response = await hass.async_add_executor_job(
api.connect, entry.data[CONF_USERNAME], entry.data[CONF_PASSWORD]
)
else:
response = await hass.async_add_executor_job(
api.connect,
entry.data[CONF_USERNAME],
entry.data[CONF_PASSWORD],
entry.data[CONF_COUNTRY_CODE],
entry.data[CONF_APP_TYPE],
)
except requests.exceptions.RequestException as err:
raise ConfigEntryNotReady(err) from err
if response.get("success", False) is False:
raise ConfigEntryNotReady(response)
tuya_mq = TuyaOpenMQ(api)
tuya_mq.start()
device_ids: set[str] = set()
device_manager = TuyaDeviceManager(api, tuya_mq)
home_manager = TuyaHomeManager(api, tuya_mq, device_manager)
listener = DeviceListener(hass, device_manager, device_ids)
device_manager.add_device_listener(listener)
hass.data[DOMAIN][entry.entry_id] = HomeAssistantTuyaData(
device_listener=listener,
device_manager=device_manager,
home_manager=home_manager,
listener = DeviceListener(hass, manager)
manager.add_device_listener(listener)
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = HomeAssistantTuyaData(
manager=manager, listener=listener
)
# Get devices & clean up device entities
await hass.async_add_executor_job(home_manager.update_device_cache)
await cleanup_device_registry(hass, device_manager)
await hass.async_add_executor_job(manager.update_device_cache)
await cleanup_device_registry(hass, manager)
# Register known device IDs
device_registry = dr.async_get(hass)
for device in device_manager.device_map.values():
for device in manager.device_map.values():
device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
identifiers={(DOMAIN, device.id)},
@ -105,15 +77,15 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
name=device.name,
model=f"{device.product_name} (unsupported)",
)
device_ids.add(device.id)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
# If the device does not register any entities, the device does not need to subscribe
# So the subscription is here
await hass.async_add_executor_job(manager.refresh_mq)
return True
async def cleanup_device_registry(
hass: HomeAssistant, device_manager: TuyaDeviceManager
) -> None:
async def cleanup_device_registry(hass: HomeAssistant, device_manager: Manager) -> None:
"""Remove deleted device registry entry if there are no remaining entities."""
device_registry = dr.async_get(hass)
for dev_id, device_entry in list(device_registry.devices.items()):
@ -125,59 +97,44 @@ async def cleanup_device_registry(
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unloading the Tuya platforms."""
unload = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload:
hass_data: HomeAssistantTuyaData = hass.data[DOMAIN][entry.entry_id]
hass_data.device_manager.mq.stop()
hass_data.device_manager.remove_device_listener(hass_data.device_listener)
hass.data[DOMAIN].pop(entry.entry_id)
if not hass.data[DOMAIN]:
hass.data.pop(DOMAIN)
return unload
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
tuya: HomeAssistantTuyaData = hass.data[DOMAIN][entry.entry_id]
if tuya.manager.mq is not None:
tuya.manager.mq.stop()
tuya.manager.remove_device_listener(tuya.listener)
await hass.async_add_executor_job(tuya.manager.unload)
del hass.data[DOMAIN][entry.entry_id]
return unload_ok
class DeviceListener(TuyaDeviceListener):
class DeviceListener(SharingDeviceListener):
"""Device Update Listener."""
def __init__(
self,
hass: HomeAssistant,
device_manager: TuyaDeviceManager,
device_ids: set[str],
manager: Manager,
) -> None:
"""Init DeviceListener."""
self.hass = hass
self.device_manager = device_manager
self.device_ids = device_ids
self.manager = manager
def update_device(self, device: TuyaDevice) -> None:
def update_device(self, device: CustomerDevice) -> None:
"""Update device status."""
if device.id in self.device_ids:
LOGGER.debug(
"Received update for device %s: %s",
device.id,
self.device_manager.device_map[device.id].status,
self.manager.device_map[device.id].status,
)
dispatcher_send(self.hass, f"{TUYA_HA_SIGNAL_UPDATE_ENTITY}_{device.id}")
def add_device(self, device: TuyaDevice) -> None:
def add_device(self, device: CustomerDevice) -> None:
"""Add device added listener."""
# Ensure the device isn't present stale
self.hass.add_job(self.async_remove_device, device.id)
self.device_ids.add(device.id)
dispatcher_send(self.hass, TUYA_DISCOVERY_NEW, [device.id])
device_manager = self.device_manager
device_manager.mq.stop()
tuya_mq = TuyaOpenMQ(device_manager.api)
tuya_mq.start()
device_manager.mq = tuya_mq
tuya_mq.add_message_listener(device_manager.on_message)
def remove_device(self, device_id: str) -> None:
"""Add device removed listener."""
self.hass.add_job(self.async_remove_device, device_id)
@ -192,4 +149,36 @@ class DeviceListener(TuyaDeviceListener):
)
if device_entry is not None:
device_registry.async_remove_device(device_entry.id)
self.device_ids.discard(device_id)
class TokenListener(SharingTokenListener):
"""Token listener for upstream token updates."""
def __init__(
self,
hass: HomeAssistant,
entry: ConfigEntry,
) -> None:
"""Init TokenListener."""
self.hass = hass
self.entry = entry
def update_token(self, token_info: dict[str, Any]) -> None:
"""Update token info in config entry."""
data = {
**self.entry.data,
CONF_TOKEN_INFO: {
"t": token_info["t"],
"uid": token_info["uid"],
"expire_time": token_info["expire_time"],
"access_token": token_info["access_token"],
"refresh_token": token_info["refresh_token"],
},
}
@callback
def async_update_entry() -> None:
"""Update config entry."""
self.hass.config_entries.async_update_entry(self.entry, data=data)
self.hass.add_job(async_update_entry)

14
homeassistant/components/tuya/alarm_control_panel.py
View file

@ -3,7 +3,7 @@ from __future__ import annotations
from enum import StrEnum
from tuya_iot import TuyaDevice, TuyaDeviceManager
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components.alarm_control_panel import (
AlarmControlPanelEntity,
@ -68,18 +68,16 @@ async def async_setup_entry(
"""Discover and add a discovered Tuya siren."""
entities: list[TuyaAlarmEntity] = []
for device_id in device_ids:
device = hass_data.device_manager.device_map[device_id]
device = hass_data.manager.device_map[device_id]
if descriptions := ALARM.get(device.category):
for description in descriptions:
if description.key in device.status:
entities.append(
TuyaAlarmEntity(
device, hass_data.device_manager, description
)
TuyaAlarmEntity(device, hass_data.manager, description)
)
async_add_entities(entities)
async_discover_device([*hass_data.device_manager.device_map])
async_discover_device([*hass_data.manager.device_map])
entry.async_on_unload(
async_dispatcher_connect(hass, TUYA_DISCOVERY_NEW, async_discover_device)
@ -94,8 +92,8 @@ class TuyaAlarmEntity(TuyaEntity, AlarmControlPanelEntity):
def __init__(
self,
device: TuyaDevice,
device_manager: TuyaDeviceManager,
device: CustomerDevice,
device_manager: Manager,
description: AlarmControlPanelEntityDescription,
) -> None:
"""Init Tuya Alarm."""

6
homeassistant/components/tuya/base.py
View file

@ -7,7 +7,7 @@ import json
import struct
from typing import Any, Literal, Self, overload
from tuya_iot import TuyaDevice, TuyaDeviceManager
from tuya_sharing import CustomerDevice, Manager
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.dispatcher import async_dispatcher_connect
@ -135,9 +135,11 @@ class TuyaEntity(Entity):
_attr_has_entity_name = True
_attr_should_poll = False
def __init__(self, device: TuyaDevice, device_manager: TuyaDeviceManager) -> None:
def __init__(self, device: CustomerDevice, device_manager: Manager) -> None:
"""Init TuyaHaEntity."""
self._attr_unique_id = f"tuya.{device.id}"
# TuyaEntity initialize mq can subscribe
device.set_up = True
self.device = device
self.device_manager = device_manager

12
homeassistant/components/tuya/binary_sensor.py
View file

@ -3,7 +3,7 @@ from __future__ import annotations
from dataclasses import dataclass
from tuya_iot import TuyaDevice, TuyaDeviceManager
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
@ -354,20 +354,20 @@ async def async_setup_entry(
"""Discover and add a discovered Tuya binary sensor."""
entities: list[TuyaBinarySensorEntity] = []
for device_id in device_ids:
device = hass_data.device_manager.device_map[device_id]
device = hass_data.manager.device_map[device_id]
if descriptions := BINARY_SENSORS.get(device.category):
for description in descriptions:
dpcode = description.dpcode or description.key
if dpcode in device.status:
entities.append(
TuyaBinarySensorEntity(
device, hass_data.device_manager, description
device, hass_data.manager, description
)
)
async_add_entities(entities)
async_discover_device([*hass_data.device_manager.device_map])
async_discover_device([*hass_data.manager.device_map])
entry.async_on_unload(
async_dispatcher_connect(hass, TUYA_DISCOVERY_NEW, async_discover_device)
@ -381,8 +381,8 @@ class TuyaBinarySensorEntity(TuyaEntity, BinarySensorEntity):
def __init__(
self,
device: TuyaDevice,
device_manager: TuyaDeviceManager,
device: CustomerDevice,
device_manager: Manager,
description: TuyaBinarySensorEntityDescription,
) -> None:
"""Init Tuya binary sensor."""

14
homeassistant/components/tuya/button.py
View file

@ -1,7 +1,7 @@
"""Support for Tuya buttons."""
from __future__ import annotations
from tuya_iot import TuyaDevice, TuyaDeviceManager
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components.button import ButtonEntity, ButtonEntityDescription
from homeassistant.config_entries import ConfigEntry
@ -74,19 +74,17 @@ async def async_setup_entry(
"""Discover and add a discovered Tuya buttons."""
entities: list[TuyaButtonEntity] = []
for device_id in device_ids:
device = hass_data.device_manager.device_map[device_id]
device = hass_data.manager.device_map[device_id]
if descriptions := BUTTONS.get(device.category):
for description in descriptions:
if description.key in device.status:
entities.append(
TuyaButtonEntity(
device, hass_data.device_manager, description
)
TuyaButtonEntity(device, hass_data.manager, description)
)
async_add_entities(entities)
async_discover_device([*hass_data.device_manager.device_map])
async_discover_device([*hass_data.manager.device_map])
entry.async_on_unload(
async_dispatcher_connect(hass, TUYA_DISCOVERY_NEW, async_discover_device)
@ -98,8 +96,8 @@ class TuyaButtonEntity(TuyaEntity, ButtonEntity):
def __init__(
self,
device: TuyaDevice,
device_manager: TuyaDeviceManager,
device: CustomerDevice,
device_manager: Manager,
description: ButtonEntityDescription,
) -> None:
"""Init Tuya button."""

12
homeassistant/components/tuya/camera.py
View file

@ -1,7 +1,7 @@
"""Support for Tuya cameras."""
from __future__ import annotations
from tuya_iot import TuyaDevice, TuyaDeviceManager
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components import ffmpeg
from homeassistant.components.camera import Camera as CameraEntity, CameraEntityFeature
@ -34,13 +34,13 @@ async def async_setup_entry(
"""Discover and add a discovered Tuya camera."""
entities: list[TuyaCameraEntity] = []
for device_id in device_ids:
device = hass_data.device_manager.device_map[device_id]
device = hass_data.manager.device_map[device_id]
if device.category in CAMERAS:
entities.append(TuyaCameraEntity(device, hass_data.device_manager))
entities.append(TuyaCameraEntity(device, hass_data.manager))
async_add_entities(entities)
async_discover_device([*hass_data.device_manager.device_map])
async_discover_device([*hass_data.manager.device_map])
entry.async_on_unload(
async_dispatcher_connect(hass, TUYA_DISCOVERY_NEW, async_discover_device)
@ -56,8 +56,8 @@ class TuyaCameraEntity(TuyaEntity, CameraEntity):
def __init__(
self,
device: TuyaDevice,
device_manager: TuyaDeviceManager,
device: CustomerDevice,
device_manager: Manager,
) -> None:
"""Init Tuya Camera."""
super().__init__(device, device_manager)

16
homeassistant/components/tuya/climate.py
View file

@ -4,7 +4,7 @@ from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from tuya_iot import TuyaDevice, TuyaDeviceManager
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components.climate import (
SWING_BOTH,
@ -98,18 +98,19 @@ async def async_setup_entry(
"""Discover and add a discovered Tuya climate."""
entities: list[TuyaClimateEntity] = []
for device_id in device_ids:
device = hass_data.device_manager.device_map[device_id]
device = hass_data.manager.device_map[device_id]
if device and device.category in CLIMATE_DESCRIPTIONS:
entities.append(
TuyaClimateEntity(
device,
hass_data.device_manager,
hass_data.manager,
CLIMATE_DESCRIPTIONS[device.category],
hass.config.units.temperature_unit,
)
)
async_add_entities(entities)
async_discover_device([*hass_data.device_manager.device_map])
async_discover_device([*hass_data.manager.device_map])
entry.async_on_unload(
async_dispatcher_connect(hass, TUYA_DISCOVERY_NEW, async_discover_device)
@ -129,9 +130,10 @@ class TuyaClimateEntity(TuyaEntity, ClimateEntity):
def __init__(
self,
device: TuyaDevice,
device_manager: TuyaDeviceManager,
device: CustomerDevice,
device_manager: Manager,
description: TuyaClimateEntityDescription,
system_temperature_unit: UnitOfTemperature,
) -> None:
"""Determine which values to use."""
self._attr_target_temperature_step = 1.0
@ -157,7 +159,7 @@ class TuyaClimateEntity(TuyaEntity, ClimateEntity):
prefered_temperature_unit = UnitOfTemperature.FAHRENHEIT
# Default to System Temperature Unit
self._attr_temperature_unit = self.hass.config.units.temperature_unit
self._attr_temperature_unit = system_temperature_unit
# Figure out current temperature, use preferred unit or what is available
celsius_type = self.find_dpcode(

267
homeassistant/components/tuya/config_flow.py
View file

@ -1,115 +1,65 @@
"""Config flow for Tuya."""
from __future__ import annotations
from collections.abc import Mapping
from io import BytesIO
from typing import Any
from tuya_iot import AuthType, TuyaOpenAPI
import segno
from tuya_sharing import LoginControl
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_COUNTRY_CODE, CONF_PASSWORD, CONF_USERNAME
from homeassistant.config_entries import ConfigEntry, ConfigFlow
from homeassistant.data_entry_flow import FlowResult
from .const import (
CONF_ACCESS_ID,
CONF_ACCESS_SECRET,
CONF_APP_TYPE,
CONF_AUTH_TYPE,
CONF_ENDPOINT,
CONF_TERMINAL_ID,
CONF_TOKEN_INFO,
CONF_USER_CODE,
DOMAIN,
LOGGER,
SMARTLIFE_APP,
TUYA_COUNTRIES,
TUYA_CLIENT_ID,
TUYA_RESPONSE_CODE,
TUYA_RESPONSE_MSG,
TUYA_RESPONSE_PLATFORM_URL,
TUYA_RESPONSE_QR_CODE,
TUYA_RESPONSE_RESULT,
TUYA_RESPONSE_SUCCESS,
TUYA_SMART_APP,
TUYA_SCHEMA,
)
class TuyaConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Tuya Config Flow."""
class TuyaConfigFlow(ConfigFlow, domain=DOMAIN):
"""Tuya config flow."""
@staticmethod
def _try_login(user_input: dict[str, Any]) -> tuple[dict[Any, Any], dict[str, Any]]:
"""Try login."""
response = {}
__user_code: str
__qr_code: str
__qr_image: str
__reauth_entry: ConfigEntry | None = None
country = [
country
for country in TUYA_COUNTRIES
if country.name == user_input[CONF_COUNTRY_CODE]
][0]
def __init__(self) -> None:
"""Initialize the config flow."""
self.__login_control = LoginControl()
data = {
CONF_ENDPOINT: country.endpoint,
CONF_AUTH_TYPE: AuthType.CUSTOM,
CONF_ACCESS_ID: user_input[CONF_ACCESS_ID],
CONF_ACCESS_SECRET: user_input[CONF_ACCESS_SECRET],
CONF_USERNAME: user_input[CONF_USERNAME],
CONF_PASSWORD: user_input[CONF_PASSWORD],
CONF_COUNTRY_CODE: country.country_code,
}
for app_type in ("", TUYA_SMART_APP, SMARTLIFE_APP):
data[CONF_APP_TYPE] = app_type
if data[CONF_APP_TYPE] == "":
data[CONF_AUTH_TYPE] = AuthType.CUSTOM
else:
data[CONF_AUTH_TYPE] = AuthType.SMART_HOME
api = TuyaOpenAPI(
endpoint=data[CONF_ENDPOINT],
access_id=data[CONF_ACCESS_ID],
access_secret=data[CONF_ACCESS_SECRET],
auth_type=data[CONF_AUTH_TYPE],
)
api.set_dev_channel("hass")
response = api.connect(
username=data[CONF_USERNAME],
password=data[CONF_PASSWORD],
country_code=data[CONF_COUNTRY_CODE],
schema=data[CONF_APP_TYPE],
)
LOGGER.debug("Response %s", response)
if response.get(TUYA_RESPONSE_SUCCESS, False):
break
return response, data
async def async_step_user(self, user_input=None):
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Step user."""
errors = {}
placeholders = {}
if user_input is not None:
response, data = await self.hass.async_add_executor_job(
self._try_login, user_input
success, response = await self.__async_get_qr_code(
user_input[CONF_USER_CODE]
)
if success:
return await self.async_step_scan()
if response.get(TUYA_RESPONSE_SUCCESS, False):
if endpoint := response.get(TUYA_RESPONSE_RESULT, {}).get(
TUYA_RESPONSE_PLATFORM_URL
):
data[CONF_ENDPOINT] = endpoint
data[CONF_AUTH_TYPE] = data[CONF_AUTH_TYPE].value
return self.async_create_entry(
title=user_input[CONF_USERNAME],
data=data,
)
errors["base"] = "login_error"
placeholders = {
TUYA_RESPONSE_CODE: response.get(TUYA_RESPONSE_CODE),
TUYA_RESPONSE_MSG: response.get(TUYA_RESPONSE_MSG),
TUYA_RESPONSE_MSG: response.get(TUYA_RESPONSE_MSG, "Unknown error"),
TUYA_RESPONSE_CODE: response.get(TUYA_RESPONSE_CODE, "0"),
}
if user_input is None:
else:
user_input = {}
return self.async_show_form(
@ -117,27 +67,146 @@ class TuyaConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
data_schema=vol.Schema(
{
vol.Required(
CONF_COUNTRY_CODE,
default=user_input.get(CONF_COUNTRY_CODE, "United States"),
): vol.In(
# We don't pass a dict {code:name} because country codes can be duplicate.
[country.name for country in TUYA_COUNTRIES]
),
vol.Required(
CONF_ACCESS_ID, default=user_input.get(CONF_ACCESS_ID, "")
): str,
vol.Required(
CONF_ACCESS_SECRET,
default=user_input.get(CONF_ACCESS_SECRET, ""),
): str,
vol.Required(
CONF_USERNAME, default=user_input.get(CONF_USERNAME, "")
): str,
vol.Required(
CONF_PASSWORD, default=user_input.get(CONF_PASSWORD, "")
CONF_USER_CODE, default=user_input.get(CONF_USER_CODE, "")
): str,
}
),
errors=errors,
description_placeholders=placeholders,
)
async def async_step_scan(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Step scan."""
if user_input is None:
return self.async_show_form(
step_id="scan",
description_placeholders={
TUYA_RESPONSE_QR_CODE: self.__qr_image,
},
)
ret, info = await self.hass.async_add_executor_job(
self.__login_control.login_result,
self.__qr_code,
TUYA_CLIENT_ID,
self.__user_code,
)
if not ret:
return self.async_show_form(
step_id="scan",
errors={"base": "login_error"},
description_placeholders={
TUYA_RESPONSE_QR_CODE: self.__qr_image,
TUYA_RESPONSE_MSG: info.get(TUYA_RESPONSE_MSG, "Unknown error"),
TUYA_RESPONSE_CODE: info.get(TUYA_RESPONSE_CODE, 0),
},
)
entry_data = {
CONF_USER_CODE: self.__user_code,
CONF_TOKEN_INFO: {
"t": info["t"],
"uid": info["uid"],
"expire_time": info["expire_time"],
"access_token": info["access_token"],
"refresh_token": info["refresh_token"],
},
CONF_TERMINAL_ID: info[CONF_TERMINAL_ID],
CONF_ENDPOINT: info[CONF_ENDPOINT],
}
if self.__reauth_entry:
return self.async_update_reload_and_abort(
self.__reauth_entry,
data=entry_data,
)
return self.async_create_entry(
title=info.get("username"),
data=entry_data,
)
async def async_step_reauth(self, _: Mapping[str, Any]) -> FlowResult:
"""Handle initiation of re-authentication with Tuya."""
self.__reauth_entry = self.hass.config_entries.async_get_entry(
self.context["entry_id"]
)
if self.__reauth_entry and CONF_USER_CODE in self.__reauth_entry.data:
success, _ = await self.__async_get_qr_code(
self.__reauth_entry.data[CONF_USER_CODE]
)
if success:
return await self.async_step_scan()
return await self.async_step_reauth_user_code()
async def async_step_reauth_user_code(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle re-authentication with a Tuya."""
errors = {}
placeholders = {}
if user_input is not None:
success, response = await self.__async_get_qr_code(
user_input[CONF_USER_CODE]
)
if success:
return await self.async_step_scan()
errors["base"] = "login_error"
placeholders = {
TUYA_RESPONSE_MSG: response.get(TUYA_RESPONSE_MSG, "Unknown error"),
TUYA_RESPONSE_CODE: response.get(TUYA_RESPONSE_CODE, "0"),
}
else:
user_input = {}
return self.async_show_form(
step_id="reauth_user_code",
data_schema=vol.Schema(
{
vol.Required(
CONF_USER_CODE, default=user_input.get(CONF_USER_CODE, "")
): str,
}
),
errors=errors,
description_placeholders=placeholders,
)
async def __async_get_qr_code(self, user_code: str) -> tuple[bool, dict[str, Any]]:
"""Get the QR code."""
response = await self.hass.async_add_executor_job(
self.__login_control.qr_code,
TUYA_CLIENT_ID,
TUYA_SCHEMA,
user_code,
)
if success := response.get(TUYA_RESPONSE_SUCCESS, False):
self.__user_code = user_code
self.__qr_code = response[TUYA_RESPONSE_RESULT][TUYA_RESPONSE_QR_CODE]
self.__qr_image = _generate_qr_code(self.__qr_code)
return success, response
def _generate_qr_code(data: str) -> str:
"""Create an SVG QR code that can be scanned with the Smart Life app."""
qr_code = segno.make(f"tuyaSmart--qrLogin?token={data}", error="h")
with BytesIO() as buffer:
qr_code.save(
buffer,
kind="svg",
border=5,
scale=5,
xmldecl=False,
svgns=False,
svgclass=None,
lineclass=None,
svgversion=2,
dark="#1abcf2",
)
return str(buffer.getvalue().decode("ascii"))

278
homeassistant/components/tuya/const.py
View file

@ -6,8 +6,6 @@ from dataclasses import dataclass, field
from enum import StrEnum
import logging
from tuya_iot import TuyaCloudOpenAPIEndpoint
from homeassistant.components.sensor import SensorDeviceClass
from homeassistant.const import (
CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
@ -31,24 +29,24 @@ from homeassistant.const import (
DOMAIN = "tuya"
LOGGER = logging.getLogger(__package__)
CONF_AUTH_TYPE = "auth_type"
CONF_PROJECT_TYPE = "tuya_project_type"
CONF_ENDPOINT = "endpoint"
CONF_ACCESS_ID = "access_id"
CONF_ACCESS_SECRET = "access_secret"
CONF_APP_TYPE = "tuya_app_type"
CONF_ENDPOINT = "endpoint"
CONF_TERMINAL_ID = "terminal_id"
CONF_TOKEN_INFO = "token_info"
CONF_USER_CODE = "user_code"
CONF_USERNAME = "username"
TUYA_CLIENT_ID = "HA_3y9q4ak7g4ephrvke"
TUYA_SCHEMA = "haauthorize"
TUYA_DISCOVERY_NEW = "tuya_discovery_new"
TUYA_HA_SIGNAL_UPDATE_ENTITY = "tuya_entry_update"
TUYA_RESPONSE_CODE = "code"
TUYA_RESPONSE_RESULT = "result"
TUYA_RESPONSE_MSG = "msg"
TUYA_RESPONSE_QR_CODE = "qrcode"
TUYA_RESPONSE_RESULT = "result"
TUYA_RESPONSE_SUCCESS = "success"
TUYA_RESPONSE_PLATFORM_URL = "platform_url"
TUYA_SMART_APP = "tuyaSmart"
SMARTLIFE_APP = "smartlife"
PLATFORMS = [
Platform.ALARM_CONTROL_PANEL,
@ -570,259 +568,3 @@ for uom in UNITS:
DEVICE_CLASS_UNITS.setdefault(device_class, {})[uom.unit] = uom
for unit_alias in uom.aliases:
DEVICE_CLASS_UNITS[device_class][unit_alias] = uom
@dataclass
class Country:
"""Describe a supported country."""
name: str
country_code: str
endpoint: str = TuyaCloudOpenAPIEndpoint.AMERICA
# https://developer.tuya.com/en/docs/iot/oem-app-data-center-distributed?id=Kafi0ku9l07qb
TUYA_COUNTRIES = [
Country("Afghanistan", "93", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Albania", "355", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Algeria", "213", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("American Samoa", "1-684", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Andorra", "376", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Angola", "244", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Anguilla", "1-264", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Antarctica", "672", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Antigua and Barbuda", "1-268", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Argentina", "54", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Armenia", "374", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Aruba", "297", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Australia", "61", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Austria", "43", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Azerbaijan", "994", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Bahamas", "1-242", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Bahrain", "973", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Bangladesh", "880", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Barbados", "1-246", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Belarus", "375", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Belgium", "32", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Belize", "501", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Benin", "229", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Bermuda", "1-441", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Bhutan", "975", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Bolivia", "591", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Bosnia and Herzegovina", "387", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Botswana", "267", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Brazil", "55", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("British Indian Ocean Territory", "246", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("British Virgin Islands", "1-284", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Brunei", "673", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Bulgaria", "359", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Burkina Faso", "226", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Burundi", "257", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Cambodia", "855", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Cameroon", "237", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Canada", "1", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Capo Verde", "238", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Cayman Islands", "1-345", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Central African Republic", "236", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Chad", "235", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Chile", "56", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("China", "86", TuyaCloudOpenAPIEndpoint.CHINA),
Country("Christmas Island", "61"),
Country("Cocos Islands", "61"),
Country("Colombia", "57", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Comoros", "269", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Cook Islands", "682", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Costa Rica", "506", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Croatia", "385", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Cuba", "53"),
Country("Curacao", "599", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Cyprus", "357", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Czech Republic", "420", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Democratic Republic of the Congo", "243", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Denmark", "45", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Djibouti", "253", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Dominica", "1-767", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Dominican Republic", "1-809", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("East Timor", "670", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Ecuador", "593", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Egypt", "20", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("El Salvador", "503", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Equatorial Guinea", "240", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Eritrea", "291", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Estonia", "372", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Ethiopia", "251", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Falkland Islands", "500", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Faroe Islands", "298", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Fiji", "679", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Finland", "358", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("France", "33", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("French Polynesia", "689", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Gabon", "241", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Gambia", "220", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Georgia", "995", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Germany", "49", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Ghana", "233", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Gibraltar", "350", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Greece", "30", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Greenland", "299", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Grenada", "1-473", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Guam", "1-671", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Guatemala", "502", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Guernsey", "44-1481"),
Country("Guinea", "224"),
Country("Guinea-Bissau", "245", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Guyana", "592", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Haiti", "509", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Honduras", "504", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Hong Kong", "852", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Hungary", "36", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Iceland", "354", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("India", "91", TuyaCloudOpenAPIEndpoint.INDIA),
Country("Indonesia", "62", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Iran", "98"),
Country("Iraq", "964", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Ireland", "353", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Isle of Man", "44-1624"),
Country("Israel", "972", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Italy", "39", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Ivory Coast", "225", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Jamaica", "1-876", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Japan", "81", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Jersey", "44-1534"),
Country("Jordan", "962", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Kazakhstan", "7", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Kenya", "254", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Kiribati", "686", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Kosovo", "383"),
Country("Kuwait", "965", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Kyrgyzstan", "996", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Laos", "856", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Latvia", "371", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Lebanon", "961", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Lesotho", "266", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Liberia", "231", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Libya", "218", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Liechtenstein", "423", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Lithuania", "370", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Luxembourg", "352", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Macao", "853", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Macedonia", "389", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Madagascar", "261", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Malawi", "265", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Malaysia", "60", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Maldives", "960", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Mali", "223", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Malta", "356", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Marshall Islands", "692", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Mauritania", "222", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Mauritius", "230", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Mayotte", "262", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Mexico", "52", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Micronesia", "691", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Moldova", "373", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Monaco", "377", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Mongolia", "976", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Montenegro", "382", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Montserrat", "1-664", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Morocco", "212", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Mozambique", "258", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Myanmar", "95", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Namibia", "264", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Nauru", "674", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Nepal", "977", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Netherlands", "31", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Netherlands Antilles", "599"),
Country("New Caledonia", "687", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("New Zealand", "64", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Nicaragua", "505", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Niger", "227", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Nigeria", "234", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Niue", "683", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("North Korea", "850"),
Country("Northern Mariana Islands", "1-670", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Norway", "47", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Oman", "968", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Pakistan", "92", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Palau", "680", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Palestine", "970", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Panama", "507", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Papua New Guinea", "675", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Paraguay", "595", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Peru", "51", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Philippines", "63", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Pitcairn", "64"),
Country("Poland", "48", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Portugal", "351", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Puerto Rico", "1-787, 1-939", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Qatar", "974", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Republic of the Congo", "242", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Reunion", "262", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Romania", "40", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Russia", "7", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Rwanda", "250", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Saint Barthelemy", "590", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Saint Helena", "290"),
Country("Saint Kitts and Nevis", "1-869", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Saint Lucia", "1-758", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Saint Martin", "590", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Saint Pierre and Miquelon", "508", TuyaCloudOpenAPIEndpoint.EUROPE),
Country(
"Saint Vincent and the Grenadines", "1-784", TuyaCloudOpenAPIEndpoint.EUROPE
),
Country("Samoa", "685", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("San Marino", "378", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Sao Tome and Principe", "239", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Saudi Arabia", "966", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Senegal", "221", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Serbia", "381", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Seychelles", "248", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Sierra Leone", "232", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Singapore", "65", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Sint Maarten", "1-721", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Slovakia", "421", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Slovenia", "386", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Solomon Islands", "677", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Somalia", "252", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("South Africa", "27", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("South Korea", "82", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("South Sudan", "211"),
Country("Spain", "34", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Sri Lanka", "94", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Sudan", "249"),
Country("Suriname", "597", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Svalbard and Jan Mayen", "4779", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Swaziland", "268", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Sweden", "46", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Switzerland", "41", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Syria", "963"),
Country("Taiwan", "886", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Tajikistan", "992", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Tanzania", "255", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Thailand", "66", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Togo", "228", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Tokelau", "690", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Tonga", "676", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Trinidad and Tobago", "1-868", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Tunisia", "216", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Turkey", "90", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Turkmenistan", "993", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Turks and Caicos Islands", "1-649", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Tuvalu", "688", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("U.S. Virgin Islands", "1-340", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Uganda", "256", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Ukraine", "380", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("United Arab Emirates", "971", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("United Kingdom", "44", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("United States", "1", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Uruguay", "598", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Uzbekistan", "998", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Vanuatu", "678", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Vatican", "379", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Venezuela", "58", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Vietnam", "84", TuyaCloudOpenAPIEndpoint.AMERICA),
Country("Wallis and Futuna", "681", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Western Sahara", "212", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Yemen", "967", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Zambia", "260", TuyaCloudOpenAPIEndpoint.EUROPE),
Country("Zimbabwe", "263", TuyaCloudOpenAPIEndpoint.EUROPE),
]

14
homeassistant/components/tuya/cover.py
View file

@ -4,7 +4,7 @@ from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from tuya_iot import TuyaDevice, TuyaDeviceManager
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components.cover import (
ATTR_POSITION,
@ -152,7 +152,7 @@ async def async_setup_entry(
"""Discover and add a discovered tuya cover."""
entities: list[TuyaCoverEntity] = []
for device_id in device_ids:
device = hass_data.device_manager.device_map[device_id]
device = hass_data.manager.device_map[device_id]
if descriptions := COVERS.get(device.category):
for description in descriptions:
if (
@ -160,14 +160,12 @@ async def async_setup_entry(
or description.key in device.status_range
):
entities.append(
TuyaCoverEntity(
device, hass_data.device_manager, description
)
TuyaCoverEntity(device, hass_data.manager, description)
)
async_add_entities(entities)
async_discover_device([*hass_data.device_manager.device_map])
async_discover_device([*hass_data.manager.device_map])
entry.async_on_unload(
async_dispatcher_connect(hass, TUYA_DISCOVERY_NEW, async_discover_device)
@ -184,8 +182,8 @@ class TuyaCoverEntity(TuyaEntity, CoverEntity):
def __init__(
self,
device: TuyaDevice,
device_manager: TuyaDeviceManager,
device: CustomerDevice,
device_manager: Manager,
description: TuyaCoverEntityDescription,
) -> None:
"""Init Tuya Cover."""

27
homeassistant/components/tuya/diagnostics.py
View file

@ -5,18 +5,17 @@ from contextlib import suppress
import json
from typing import Any, cast
from tuya_iot import TuyaDevice
from tuya_sharing import CustomerDevice
from homeassistant.components.diagnostics import REDACTED
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_COUNTRY_CODE
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.device_registry import DeviceEntry
from homeassistant.util import dt as dt_util
from . import HomeAssistantTuyaData
from .const import CONF_APP_TYPE, CONF_AUTH_TYPE, CONF_ENDPOINT, DOMAIN, DPCode
from .const import DOMAIN, DPCode
async def async_get_config_entry_diagnostics(
@ -43,14 +42,12 @@ def _async_get_diagnostics(
hass_data: HomeAssistantTuyaData = hass.data[DOMAIN][entry.entry_id]
mqtt_connected = None
if hass_data.home_manager.mq.client:
mqtt_connected = hass_data.home_manager.mq.client.is_connected()
if hass_data.manager.mq.client:
mqtt_connected = hass_data.manager.mq.client.is_connected()
data = {
"endpoint": entry.data[CONF_ENDPOINT],
"auth_type": entry.data[CONF_AUTH_TYPE],
"country_code": entry.data[CONF_COUNTRY_CODE],
"app_type": entry.data[CONF_APP_TYPE],
"endpoint": hass_data.manager.customer_api.endpoint,
"terminal_id": hass_data.manager.terminal_id,
"mqtt_connected": mqtt_connected,
"disabled_by": entry.disabled_by,
"disabled_polling": entry.pref_disable_polling,
@ -59,13 +56,13 @@ def _async_get_diagnostics(
if device:
tuya_device_id = next(iter(device.identifiers))[1]
data |= _async_device_as_dict(
hass, hass_data.device_manager.device_map[tuya_device_id]
hass, hass_data.manager.device_map[tuya_device_id]
)
else:
data.update(
devices=[
_async_device_as_dict(hass, device)
for device in hass_data.device_manager.device_map.values()
for device in hass_data.manager.device_map.values()
]
)
@ -73,13 +70,15 @@ def _async_get_diagnostics(
@callback
def _async_device_as_dict(hass: HomeAssistant, device: TuyaDevice) -> dict[str, Any]:
def _async_device_as_dict(
hass: HomeAssistant, device: CustomerDevice
) -> dict[str, Any]:
"""Represent a Tuya device as a dictionary."""
# Base device information, without sensitive information.
data = {
"id": device.id,
"name": device.name,
"model": device.model if hasattr(device, "model") else None,
"category": device.category,
"product_id": device.product_id,
"product_name": device.product_name,
@ -93,6 +92,8 @@ def _async_device_as_dict(hass: HomeAssistant, device: TuyaDevice) -> dict[str,
"status_range": {},
"status": {},
"home_assistant": {},
"set_up": device.set_up,
"support_local": device.support_local,
}
# Gather Tuya states

12
homeassistant/components/tuya/fan.py
View file

@ -3,7 +3,7 @@ from __future__ import annotations
from typing import Any
from tuya_iot import TuyaDevice, TuyaDeviceManager
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components.fan import (
DIRECTION_FORWARD,
@ -44,12 +44,12 @@ async def async_setup_entry(
"""Discover and add a discovered tuya fan."""
entities: list[TuyaFanEntity] = []
for device_id in device_ids:
device = hass_data.device_manager.device_map[device_id]
device = hass_data.manager.device_map[device_id]
if device and device.category in TUYA_SUPPORT_TYPE:
entities.append(TuyaFanEntity(device, hass_data.device_manager))
entities.append(TuyaFanEntity(device, hass_data.manager))
async_add_entities(entities)
async_discover_device([*hass_data.device_manager.device_map])
async_discover_device([*hass_data.manager.device_map])
entry.async_on_unload(
async_dispatcher_connect(hass, TUYA_DISCOVERY_NEW, async_discover_device)
@ -69,8 +69,8 @@ class TuyaFanEntity(TuyaEntity, FanEntity):
def __init__(
self,
device: TuyaDevice,
device_manager: TuyaDeviceManager,
device: CustomerDevice,
device_manager: Manager,
) -> None:
"""Init Tuya Fan Device."""
super().__init__(device, device_manager)

12
homeassistant/components/tuya/humidifier.py
View file

@ -3,7 +3,7 @@ from __future__ import annotations
from dataclasses import dataclass
from tuya_iot import TuyaDevice, TuyaDeviceManager
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components.humidifier import (
HumidifierDeviceClass,
@ -65,14 +65,14 @@ async def async_setup_entry(
"""Discover and add a discovered Tuya (de)humidifier."""
entities: list[TuyaHumidifierEntity] = []
for device_id in device_ids:
device = hass_data.device_manager.device_map[device_id]
device = hass_data.manager.device_map[device_id]
if description := HUMIDIFIERS.get(device.category):
entities.append(
TuyaHumidifierEntity(device, hass_data.device_manager, description)
TuyaHumidifierEntity(device, hass_data.manager, description)
)
async_add_entities(entities)
async_discover_device([*hass_data.device_manager.device_map])
async_discover_device([*hass_data.manager.device_map])
entry.async_on_unload(
async_dispatcher_connect(hass, TUYA_DISCOVERY_NEW, async_discover_device)
@ -90,8 +90,8 @@ class TuyaHumidifierEntity(TuyaEntity, HumidifierEntity):
def __init__(
self,
device: TuyaDevice,
device_manager: TuyaDeviceManager,
device: CustomerDevice,
device_manager: Manager,
description: TuyaHumidifierEntityDescription,
) -> None:
"""Init Tuya (de)humidifier."""

14
homeassistant/components/tuya/light.py
View file

@ -5,7 +5,7 @@ from dataclasses import dataclass, field
import json
from typing import Any, cast
from tuya_iot import TuyaDevice, TuyaDeviceManager
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
@ -413,19 +413,17 @@ async def async_setup_entry(
"""Discover and add a discovered tuya light."""
entities: list[TuyaLightEntity] = []
for device_id in device_ids:
device = hass_data.device_manager.device_map[device_id]
device = hass_data.manager.device_map[device_id]
if descriptions := LIGHTS.get(device.category):
for description in descriptions:
if description.key in device.status:
entities.append(
TuyaLightEntity(
device, hass_data.device_manager, description
)
TuyaLightEntity(device, hass_data.manager, description)
)
async_add_entities(entities)
async_discover_device([*hass_data.device_manager.device_map])
async_discover_device([*hass_data.manager.device_map])
entry.async_on_unload(
async_dispatcher_connect(hass, TUYA_DISCOVERY_NEW, async_discover_device)
@ -447,8 +445,8 @@ class TuyaLightEntity(TuyaEntity, LightEntity):
def __init__(
self,
device: TuyaDevice,
device_manager: TuyaDeviceManager,
device: CustomerDevice,
device_manager: Manager,
description: TuyaLightEntityDescription,
) -> None:
"""Init TuyaHaLight."""

2
homeassistant/components/tuya/manifest.json
View file

@ -43,5 +43,5 @@
"integration_type": "hub",
"iot_class": "cloud_push",
"loggers": ["tuya_iot"],
"requirements": ["tuya-iot-py-sdk==0.6.6"]
"requirements": ["tuya-device-sharing-sdk==0.1.9", "segno==1.5.3"]
}

14
homeassistant/components/tuya/number.py
View file

@ -1,7 +1,7 @@
"""Support for Tuya number."""
from __future__ import annotations
from tuya_iot import TuyaDevice, TuyaDeviceManager
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components.number import (
NumberDeviceClass,
@ -323,19 +323,17 @@ async def async_setup_entry(
"""Discover and add a discovered Tuya number."""
entities: list[TuyaNumberEntity] = []
for device_id in device_ids:
device = hass_data.device_manager.device_map[device_id]
device = hass_data.manager.device_map[device_id]
if descriptions := NUMBERS.get(device.category):
for description in descriptions:
if description.key in device.status:
entities.append(
TuyaNumberEntity(
device, hass_data.device_manager, description
)
TuyaNumberEntity(device, hass_data.manager, description)
)
async_add_entities(entities)
async_discover_device([*hass_data.device_manager.device_map])
async_discover_device([*hass_data.manager.device_map])
entry.async_on_unload(
async_dispatcher_connect(hass, TUYA_DISCOVERY_NEW, async_discover_device)
@ -349,8 +347,8 @@ class TuyaNumberEntity(TuyaEntity, NumberEntity):
def __init__(
self,
device: TuyaDevice,
device_manager: TuyaDeviceManager,
device: CustomerDevice,
device_manager: Manager,
description: NumberEntityDescription,
) -> None:
"""Init Tuya sensor."""

10
homeassistant/components/tuya/scene.py
View file

@ -3,7 +3,7 @@ from __future__ import annotations
from typing import Any
from tuya_iot import TuyaHomeManager, TuyaScene
from tuya_sharing import Manager, SharingScene
from homeassistant.components.scene import Scene
from homeassistant.config_entries import ConfigEntry
@ -20,10 +20,8 @@ async def async_setup_entry(
) -> None:
"""Set up Tuya scenes."""
hass_data: HomeAssistantTuyaData = hass.data[DOMAIN][entry.entry_id]
scenes = await hass.async_add_executor_job(hass_data.home_manager.query_scenes)
async_add_entities(
TuyaSceneEntity(hass_data.home_manager, scene) for scene in scenes
)
scenes = await hass.async_add_executor_job(hass_data.manager.query_scenes)
async_add_entities(TuyaSceneEntity(hass_data.manager, scene) for scene in scenes)
class TuyaSceneEntity(Scene):
@ -33,7 +31,7 @@ class TuyaSceneEntity(Scene):
_attr_has_entity_name = True
_attr_name = None
def __init__(self, home_manager: TuyaHomeManager, scene: TuyaScene) -> None:
def __init__(self, home_manager: Manager, scene: SharingScene) -> None:
"""Init Tuya Scene."""
super().__init__()
self._attr_unique_id = f"tys{scene.scene_id}"

14
homeassistant/components/tuya/select.py
View file

@ -1,7 +1,7 @@
"""Support for Tuya select."""
from __future__ import annotations
from tuya_iot import TuyaDevice, TuyaDeviceManager
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components.select import SelectEntity, SelectEntityDescription
from homeassistant.config_entries import ConfigEntry
@ -356,19 +356,17 @@ async def async_setup_entry(
"""Discover and add a discovered Tuya select."""
entities: list[TuyaSelectEntity] = []
for device_id in device_ids:
device = hass_data.device_manager.device_map[device_id]
device = hass_data.manager.device_map[device_id]
if descriptions := SELECTS.get(device.category):
for description in descriptions:
if description.key in device.status:
entities.append(
TuyaSelectEntity(
device, hass_data.device_manager, description
)
TuyaSelectEntity(device, hass_data.manager, description)
)
async_add_entities(entities)
async_discover_device([*hass_data.device_manager.device_map])
async_discover_device([*hass_data.manager.device_map])
entry.async_on_unload(
async_dispatcher_connect(hass, TUYA_DISCOVERY_NEW, async_discover_device)
@ -380,8 +378,8 @@ class TuyaSelectEntity(TuyaEntity, SelectEntity):
def __init__(
self,
device: TuyaDevice,
device_manager: TuyaDeviceManager,
device: CustomerDevice,
device_manager: Manager,
description: SelectEntityDescription,
) -> None:
"""Init Tuya sensor."""

18
homeassistant/components/tuya/sensor.py
View file

@ -3,8 +3,8 @@ from __future__ import annotations
from dataclasses import dataclass
from tuya_iot import TuyaDevice, TuyaDeviceManager
from tuya_iot.device import TuyaDeviceStatusRange
from tuya_sharing import CustomerDevice, Manager
from tuya_sharing.device import DeviceStatusRange
from homeassistant.components.sensor import (
SensorDeviceClass,
@ -1112,19 +1112,17 @@ async def async_setup_entry(
"""Discover and add a discovered Tuya sensor."""
entities: list[TuyaSensorEntity] = []
for device_id in device_ids:
device = hass_data.device_manager.device_map[device_id]
device = hass_data.manager.device_map[device_id]
if descriptions := SENSORS.get(device.category):
for description in descriptions:
if description.key in device.status:
entities.append(
TuyaSensorEntity(
device, hass_data.device_manager, description
)
TuyaSensorEntity(device, hass_data.manager, description)
)
async_add_entities(entities)
async_discover_device([*hass_data.device_manager.device_map])
async_discover_device([*hass_data.manager.device_map])
entry.async_on_unload(
async_dispatcher_connect(hass, TUYA_DISCOVERY_NEW, async_discover_device)
@ -1136,15 +1134,15 @@ class TuyaSensorEntity(TuyaEntity, SensorEntity):
entity_description: TuyaSensorEntityDescription
_status_range: TuyaDeviceStatusRange | None = None
_status_range: DeviceStatusRange | None = None
_type: DPType | None = None
_type_data: IntegerTypeData | EnumTypeData | None = None
_uom: UnitOfMeasurement | None = None
def __init__(
self,
device: TuyaDevice,
device_manager: TuyaDeviceManager,
device: CustomerDevice,
device_manager: Manager,
description: TuyaSensorEntityDescription,
) -> None:
"""Init Tuya sensor."""

14
homeassistant/components/tuya/siren.py
View file

@ -3,7 +3,7 @@ from __future__ import annotations
from typing import Any
from tuya_iot import TuyaDevice, TuyaDeviceManager
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components.siren import (
SirenEntity,
@ -57,19 +57,17 @@ async def async_setup_entry(
"""Discover and add a discovered Tuya siren."""
entities: list[TuyaSirenEntity] = []
for device_id in device_ids:
device = hass_data.device_manager.device_map[device_id]
device = hass_data.manager.device_map[device_id]
if descriptions := SIRENS.get(device.category):
for description in descriptions:
if description.key in device.status:
entities.append(
TuyaSirenEntity(
device, hass_data.device_manager, description
)
TuyaSirenEntity(device, hass_data.manager, description)
)
async_add_entities(entities)
async_discover_device([*hass_data.device_manager.device_map])
async_discover_device([*hass_data.manager.device_map])
entry.async_on_unload(
async_dispatcher_connect(hass, TUYA_DISCOVERY_NEW, async_discover_device)
@ -84,8 +82,8 @@ class TuyaSirenEntity(TuyaEntity, SirenEntity):
def __init__(
self,
device: TuyaDevice,
device_manager: TuyaDeviceManager,
device: CustomerDevice,
device_manager: Manager,
description: SirenEntityDescription,
) -> None:
"""Init Tuya Siren."""

22
homeassistant/components/tuya/strings.json
View file

@ -1,20 +1,26 @@
{
"config": {
"step": {
"user": {
"description": "Enter your Tuya credentials",
"reauth_user_code": {
"description": "The Tuya integration now uses an improved login method. To reauthenticate with your Smart Life or Tuya Smart account, you need to enter your user code.\n\nYou can find this code in the Smart Life app or Tuya Smart app in **Settings** > **Account and Security** screen, and enter the code shown on the **User Code** field. The user code is case sensitive, please be sure to enter it exactly as shown in the app.",
"data": {
"country_code": "Country",
"access_id": "Tuya IoT Access ID",
"access_secret": "Tuya IoT Access Secret",
"username": "Account",
"password": "[%key:common::config_flow::data::password%]"
"user_code": "User code"
}
},
"user": {
"description": "Enter your Smart Life or Tuya Smart user code.\n\nYou can find this code in the Smart Life app or Tuya Smart app in **Settings** > **Account and Security** screen, and enter the code shown on the **User Code** field. The user code is case sensitive, please be sure to enter it exactly as shown in the app.",
"data": {
"user_code": "User code"
}
},
"scan": {
"description": "Use Smart Life app or Tuya Smart app to scan the following QR-code to complete the login:\n\n {qrcode} \n\nContinue to the next step once you have completed this step in the app."
}
},
"error": {
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"login_error": "Login error ({code}): {msg}"
"login_error": "Login error ({code}): {msg}",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
}
},
"entity": {

14
homeassistant/components/tuya/switch.py
View file

@ -3,7 +3,7 @@ from __future__ import annotations
from typing import Any
from tuya_iot import TuyaDevice, TuyaDeviceManager
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components.switch import (
SwitchDeviceClass,
@ -730,19 +730,17 @@ async def async_setup_entry(
"""Discover and add a discovered tuya sensor."""
entities: list[TuyaSwitchEntity] = []
for device_id in device_ids:
device = hass_data.device_manager.device_map[device_id]
device = hass_data.manager.device_map[device_id]
if descriptions := SWITCHES.get(device.category):
for description in descriptions:
if description.key in device.status:
entities.append(
TuyaSwitchEntity(
device, hass_data.device_manager, description
)
TuyaSwitchEntity(device, hass_data.manager, description)
)
async_add_entities(entities)
async_discover_device([*hass_data.device_manager.device_map])
async_discover_device([*hass_data.manager.device_map])
entry.async_on_unload(
async_dispatcher_connect(hass, TUYA_DISCOVERY_NEW, async_discover_device)
@ -754,8 +752,8 @@ class TuyaSwitchEntity(TuyaEntity, SwitchEntity):
def __init__(
self,
device: TuyaDevice,
device_manager: TuyaDeviceManager,
device: CustomerDevice,
device_manager: Manager,
description: SwitchEntityDescription,
) -> None:
"""Init TuyaHaSwitch."""

10
homeassistant/components/tuya/vacuum.py
View file

@ -3,7 +3,7 @@ from __future__ import annotations
from typing import Any
from tuya_iot import TuyaDevice, TuyaDeviceManager
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components.vacuum import (
STATE_CLEANING,
@ -61,12 +61,12 @@ async def async_setup_entry(
"""Discover and add a discovered Tuya vacuum."""
entities: list[TuyaVacuumEntity] = []
for device_id in device_ids:
device = hass_data.device_manager.device_map[device_id]
device = hass_data.manager.device_map[device_id]
if device.category == "sd":
entities.append(TuyaVacuumEntity(device, hass_data.device_manager))
entities.append(TuyaVacuumEntity(device, hass_data.manager))
async_add_entities(entities)
async_discover_device([*hass_data.device_manager.device_map])
async_discover_device([*hass_data.manager.device_map])
entry.async_on_unload(
async_dispatcher_connect(hass, TUYA_DISCOVERY_NEW, async_discover_device)
@ -80,7 +80,7 @@ class TuyaVacuumEntity(TuyaEntity, StateVacuumEntity):
_battery_level: IntegerTypeData | None = None
_attr_name = None
def __init__(self, device: TuyaDevice, device_manager: TuyaDeviceManager) -> None:
def __init__(self, device: CustomerDevice, device_manager: Manager) -> None:
"""Init Tuya vacuum."""
super().__init__(device, device_manager)

5
requirements_all.txt
View file

@ -2487,6 +2487,9 @@ scsgate==0.1.0
# homeassistant.components.backup
securetar==2023.3.0
# homeassistant.components.tuya
segno==1.5.3
# homeassistant.components.sendgrid
sendgrid==6.8.2
@ -2723,7 +2726,7 @@ transmission-rpc==7.0.3
ttls==1.5.1
# homeassistant.components.tuya
tuya-iot-py-sdk==0.6.6
tuya-device-sharing-sdk==0.1.9
# homeassistant.components.twentemilieu
twentemilieu==2.0.1

5
requirements_test_all.txt
View file

@ -1891,6 +1891,9 @@ screenlogicpy==0.10.0
# homeassistant.components.backup
securetar==2023.3.0
# homeassistant.components.tuya
segno==1.5.3
# homeassistant.components.emulated_kasa
# homeassistant.components.sense
sense-energy==0.12.2
@ -2067,7 +2070,7 @@ transmission-rpc==7.0.3
ttls==1.5.1
# homeassistant.components.tuya
tuya-iot-py-sdk==0.6.6
tuya-device-sharing-sdk==0.1.9
# homeassistant.components.twentemilieu
twentemilieu==2.0.1

69
tests/components/tuya/conftest.py Normal file
View file

@ -0,0 +1,69 @@
"""Fixtures for the Tuya integration tests."""
from __future__ import annotations
from collections.abc import Generator
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from homeassistant.components.tuya.const import CONF_APP_TYPE, CONF_USER_CODE, DOMAIN
from tests.common import MockConfigEntry
@pytest.fixture
def mock_old_config_entry() -> MockConfigEntry:
"""Mock an old config entry that can be migrated."""
return MockConfigEntry(
title="Old Tuya configuration entry",
domain=DOMAIN,
data={CONF_APP_TYPE: "tuyaSmart"},
unique_id="12345",
)
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Mock an config entry."""
return MockConfigEntry(
title="12345",
domain=DOMAIN,
data={CONF_USER_CODE: "12345"},
unique_id="12345",
)
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock, None, None]:
"""Mock setting up a config entry."""
with patch("homeassistant.components.tuya.async_setup_entry", return_value=True):
yield
@pytest.fixture
def mock_tuya_login_control() -> Generator[MagicMock, None, None]:
"""Return a mocked Tuya login control."""
with patch(
"homeassistant.components.tuya.config_flow.LoginControl", autospec=True
) as login_control_mock:
login_control = login_control_mock.return_value
login_control.qr_code.return_value = {
"success": True,
"result": {
"qrcode": "mocked_qr_code",
},
}
login_control.login_result.return_value = (
True,
{
"t": "mocked_t",
"uid": "mocked_uid",
"username": "mocked_username",
"expire_time": "mocked_expire_time",
"access_token": "mocked_access_token",
"refresh_token": "mocked_refresh_token",
"terminal_id": "mocked_terminal_id",
"endpoint": "mocked_endpoint",
},
)
yield login_control

112
tests/components/tuya/snapshots/test_config_flow.ambr Normal file
View file

@ -0,0 +1,112 @@
# serializer version: 1
# name: test_reauth_flow
ConfigEntrySnapshot({
'data': dict({
'endpoint': 'mocked_endpoint',
'terminal_id': 'mocked_terminal_id',
'token_info': dict({
'access_token': 'mocked_access_token',
'expire_time': 'mocked_expire_time',
'refresh_token': 'mocked_refresh_token',
't': 'mocked_t',
'uid': 'mocked_uid',
}),
'user_code': '12345',
}),
'disabled_by': None,
'domain': 'tuya',
'entry_id': <ANY>,
'minor_version': 1,
'options': dict({
}),
'pref_disable_new_entities': False,
'pref_disable_polling': False,
'source': 'user',
'title': '12345',
'unique_id': '12345',
'version': 1,
})
# ---
# name: test_reauth_flow_migration
ConfigEntrySnapshot({
'data': dict({
'endpoint': 'mocked_endpoint',
'terminal_id': 'mocked_terminal_id',
'token_info': dict({
'access_token': 'mocked_access_token',
'expire_time': 'mocked_expire_time',
'refresh_token': 'mocked_refresh_token',
't': 'mocked_t',
'uid': 'mocked_uid',
}),
'user_code': '12345',
}),
'disabled_by': None,
'domain': 'tuya',
'entry_id': <ANY>,
'minor_version': 1,
'options': dict({
}),
'pref_disable_new_entities': False,
'pref_disable_polling': False,
'source': 'user',
'title': 'Old Tuya configuration entry',
'unique_id': '12345',
'version': 1,
})
# ---
# name: test_user_flow
FlowResultSnapshot({
'context': dict({
'source': 'user',
}),
'data': dict({
'endpoint': 'mocked_endpoint',
'terminal_id': 'mocked_terminal_id',
'token_info': dict({
'access_token': 'mocked_access_token',
'expire_time': 'mocked_expire_time',
'refresh_token': 'mocked_refresh_token',
't': 'mocked_t',
'uid': 'mocked_uid',
}),
'user_code': '12345',
}),
'description': None,
'description_placeholders': None,
'flow_id': <ANY>,
'handler': 'tuya',
'minor_version': 1,
'options': dict({
}),
'result': ConfigEntrySnapshot({
'data': dict({
'endpoint': 'mocked_endpoint',
'terminal_id': 'mocked_terminal_id',
'token_info': dict({
'access_token': 'mocked_access_token',
'expire_time': 'mocked_expire_time',
'refresh_token': 'mocked_refresh_token',
't': 'mocked_t',
'uid': 'mocked_uid',
}),
'user_code': '12345',
}),
'disabled_by': None,
'domain': 'tuya',
'entry_id': <ANY>,
'minor_version': 1,
'options': dict({
}),
'pref_disable_new_entities': False,
'pref_disable_polling': False,
'source': 'user',
'title': 'mocked_username',
'unique_id': None,
'version': 1,
}),
'title': 'mocked_username',
'type': <FlowResultType.CREATE_ENTRY: 'create_entry'>,
'version': 1,
})
# ---

339
tests/components/tuya/test_config_flow.py
View file

@ -1,127 +1,270 @@
"""Tests for the Tuya config flow."""
from __future__ import annotations
from typing import Any
from unittest.mock import MagicMock, patch
from unittest.mock import MagicMock
import pytest
from tuya_iot import TuyaCloudOpenAPIEndpoint
from syrupy.assertion import SnapshotAssertion
from homeassistant import config_entries, data_entry_flow
from homeassistant.components.tuya.const import (
CONF_ACCESS_ID,
CONF_ACCESS_SECRET,
CONF_APP_TYPE,
CONF_AUTH_TYPE,
CONF_ENDPOINT,
DOMAIN,
SMARTLIFE_APP,
TUYA_COUNTRIES,
TUYA_SMART_APP,
)
from homeassistant.const import CONF_COUNTRY_CODE, CONF_PASSWORD, CONF_USERNAME
from homeassistant.components.tuya.const import CONF_APP_TYPE, CONF_USER_CODE, DOMAIN
from homeassistant.config_entries import SOURCE_REAUTH, SOURCE_USER
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
MOCK_SMART_HOME_PROJECT_TYPE = 0
MOCK_INDUSTRY_PROJECT_TYPE = 1
from tests.common import ANY, MockConfigEntry
MOCK_COUNTRY = "India"
MOCK_ACCESS_ID = "myAccessId"
MOCK_ACCESS_SECRET = "myAccessSecret"
MOCK_USERNAME = "myUsername"
MOCK_PASSWORD = "myPassword"
MOCK_ENDPOINT = TuyaCloudOpenAPIEndpoint.INDIA
TUYA_INPUT_DATA = {
CONF_COUNTRY_CODE: MOCK_COUNTRY,
CONF_ACCESS_ID: MOCK_ACCESS_ID,
CONF_ACCESS_SECRET: MOCK_ACCESS_SECRET,
CONF_USERNAME: MOCK_USERNAME,
CONF_PASSWORD: MOCK_PASSWORD,
}
RESPONSE_SUCCESS = {
"success": True,
"code": 1024,
"result": {"platform_url": MOCK_ENDPOINT},
}
RESPONSE_ERROR = {"success": False, "code": 123, "msg": "Error"}
pytestmark = pytest.mark.usefixtures("mock_setup_entry")
@pytest.fixture(name="tuya")
def tuya_fixture() -> MagicMock:
"""Patch libraries."""
with patch("homeassistant.components.tuya.config_flow.TuyaOpenAPI") as tuya:
yield tuya
@pytest.fixture(name="tuya_setup", autouse=True)
def tuya_setup_fixture() -> None:
"""Mock tuya entry setup."""
with patch("homeassistant.components.tuya.async_setup_entry", return_value=True):
yield
@pytest.mark.parametrize(
("app_type", "side_effects", "project_type"),
[
("", [RESPONSE_SUCCESS], 1),
(TUYA_SMART_APP, [RESPONSE_ERROR, RESPONSE_SUCCESS], 0),
(SMARTLIFE_APP, [RESPONSE_ERROR, RESPONSE_ERROR, RESPONSE_SUCCESS], 0),
],
)
@pytest.mark.usefixtures("mock_tuya_login_control")
async def test_user_flow(
hass: HomeAssistant,
tuya: MagicMock,
app_type: str,
side_effects: list[dict[str, Any]],
project_type: int,
):
"""Test user flow."""
snapshot: SnapshotAssertion,
) -> None:
"""Test the full happy path user flow from start to finish."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
DOMAIN,
context={"source": SOURCE_USER},
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "user"
assert result.get("type") == FlowResultType.FORM
assert result.get("step_id") == "user"
tuya().connect = MagicMock(side_effect=side_effects)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=TUYA_INPUT_DATA
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_USER_CODE: "12345"},
)
await hass.async_block_till_done()
country = [country for country in TUYA_COUNTRIES if country.name == MOCK_COUNTRY][0]
assert result2.get("type") == FlowResultType.FORM
assert result2.get("step_id") == "scan"
assert result2.get("description_placeholders") == {"qrcode": ANY}
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["title"] == MOCK_USERNAME
assert result["data"][CONF_ACCESS_ID] == MOCK_ACCESS_ID
assert result["data"][CONF_ACCESS_SECRET] == MOCK_ACCESS_SECRET
assert result["data"][CONF_USERNAME] == MOCK_USERNAME
assert result["data"][CONF_PASSWORD] == MOCK_PASSWORD
assert result["data"][CONF_ENDPOINT] == country.endpoint
assert result["data"][CONF_APP_TYPE] == app_type
assert result["data"][CONF_AUTH_TYPE] == project_type
assert result["data"][CONF_COUNTRY_CODE] == country.country_code
assert not result["result"].unique_id
result3 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={},
)
assert result3.get("type") == FlowResultType.CREATE_ENTRY
assert result3 == snapshot
async def test_error_on_invalid_credentials(hass: HomeAssistant, tuya) -> None:
"""Test when we have invalid credentials."""
async def test_user_flow_failed_qr_code(
hass: HomeAssistant,
mock_tuya_login_control: MagicMock,
) -> None:
"""Test an error occurring while retrieving the QR code."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
assert result.get("type") == FlowResultType.FORM
assert result.get("step_id") == "user"
# Something went wrong getting the QR code (like an invalid user code)
mock_tuya_login_control.qr_code.return_value["success"] = False
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_USER_CODE: "12345"},
)
assert result2.get("type") == FlowResultType.FORM
assert result2.get("errors") == {"base": "login_error"}
# This time it worked out
mock_tuya_login_control.qr_code.return_value["success"] = True
result3 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_USER_CODE: "12345"},
)
assert result3.get("step_id") == "scan"
result3 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={},
)
assert result3.get("type") == FlowResultType.CREATE_ENTRY
async def test_user_flow_failed_scan(
hass: HomeAssistant,
mock_tuya_login_control: MagicMock,
) -> None:
"""Test an error occurring while verifying login."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
assert result.get("type") == FlowResultType.FORM
assert result.get("step_id") == "user"
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_USER_CODE: "12345"},
)
assert result2.get("type") == FlowResultType.FORM
assert result2.get("step_id") == "scan"
# Access has been denied, or the code hasn't been scanned yet
good_values = mock_tuya_login_control.login_result.return_value
mock_tuya_login_control.login_result.return_value = (
False,
{"msg": "oops", "code": 42},
)
result3 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={},
)
assert result3.get("type") == FlowResultType.FORM
assert result3.get("errors") == {"base": "login_error"}
# This time it worked out
mock_tuya_login_control.login_result.return_value = good_values
result4 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={},
)
assert result4.get("type") == FlowResultType.CREATE_ENTRY
@pytest.mark.usefixtures("mock_tuya_login_control")
async def test_reauth_flow(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
) -> None:
"""Test the reauthentication configuration flow."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
DOMAIN,
context={
"source": SOURCE_REAUTH,
"unique_id": mock_config_entry.unique_id,
"entry_id": mock_config_entry.entry_id,
},
data=mock_config_entry.data,
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "user"
assert result.get("type") == FlowResultType.FORM
assert result.get("step_id") == "scan"
assert result.get("description_placeholders") == {"qrcode": ANY}
tuya().connect = MagicMock(return_value=RESPONSE_ERROR)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=TUYA_INPUT_DATA
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={},
)
await hass.async_block_till_done()
assert result["errors"]["base"] == "login_error"
assert result["description_placeholders"]["code"] == RESPONSE_ERROR["code"]
assert result["description_placeholders"]["msg"] == RESPONSE_ERROR["msg"]
assert result2.get("type") == FlowResultType.ABORT
assert result2.get("reason") == "reauth_successful"
assert mock_config_entry == snapshot
@pytest.mark.usefixtures("mock_tuya_login_control")
async def test_reauth_flow_migration(
hass: HomeAssistant,
mock_old_config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
) -> None:
"""Test the reauthentication configuration flow.
This flow tests the migration from an old config entry.
"""
mock_old_config_entry.add_to_hass(hass)
# Ensure old data is there, new data is missing
assert CONF_APP_TYPE in mock_old_config_entry.data
assert CONF_USER_CODE not in mock_old_config_entry.data
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={
"source": SOURCE_REAUTH,
"unique_id": mock_old_config_entry.unique_id,
"entry_id": mock_old_config_entry.entry_id,
},
data=mock_old_config_entry.data,
)
assert result.get("type") == FlowResultType.FORM
assert result.get("step_id") == "reauth_user_code"
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_USER_CODE: "12345"},
)
assert result2.get("type") == FlowResultType.FORM
assert result2.get("step_id") == "scan"
assert result2.get("description_placeholders") == {"qrcode": ANY}
result3 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={},
)
assert result3.get("type") == FlowResultType.ABORT
assert result3.get("reason") == "reauth_successful"
# Ensure the old data is gone, new data is present
assert CONF_APP_TYPE not in mock_old_config_entry.data
assert CONF_USER_CODE in mock_old_config_entry.data
assert mock_old_config_entry == snapshot
async def test_reauth_flow_failed_qr_code(
hass: HomeAssistant,
mock_tuya_login_control: MagicMock,
mock_old_config_entry: MockConfigEntry,
) -> None:
"""Test an error occurring while retrieving the QR code."""
mock_old_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={
"source": SOURCE_REAUTH,
"unique_id": mock_old_config_entry.unique_id,
"entry_id": mock_old_config_entry.entry_id,
},
data=mock_old_config_entry.data,
)
# Something went wrong getting the QR code (like an invalid user code)
mock_tuya_login_control.qr_code.return_value["success"] = False
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_USER_CODE: "12345"},
)
assert result2.get("type") == FlowResultType.FORM
assert result2.get("errors") == {"base": "login_error"}
# This time it worked out
mock_tuya_login_control.qr_code.return_value["success"] = True
result3 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_USER_CODE: "12345"},
)
assert result3.get("step_id") == "scan"
result3 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={},
)
assert result3.get("type") == FlowResultType.ABORT
assert result3.get("reason") == "reauth_successful"