Add LG ThinQ integration (#129299)

Co-authored-by: jangwon.lee <jangwon.lee@lge.com>
pull/127785/head^2
LG-ThinQ-Integration 2024-10-29 01:22:24 +09:00 committed by GitHub
parent 8eb68b54d9
commit 420538e6e7
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
27 changed files with 4379 additions and 0 deletions

View File

@ -821,6 +821,8 @@ build.json @home-assistant/supervisor
/tests/components/lektrico/ @lektrico
/homeassistant/components/lg_netcast/ @Drafteed @splinter98
/tests/components/lg_netcast/ @Drafteed @splinter98
/homeassistant/components/lg_thinq/ @LG-ThinQ-Integration
/tests/components/lg_thinq/ @LG-ThinQ-Integration
/homeassistant/components/lidarr/ @tkdrob
/tests/components/lidarr/ @tkdrob
/homeassistant/components/lifx/ @Djelibeybi

View File

@ -0,0 +1,166 @@
"""Support for LG ThinQ Connect device."""
from __future__ import annotations
import asyncio
from dataclasses import dataclass, field
import logging
from thinqconnect import ThinQApi, ThinQAPIException
from thinqconnect.integration import async_get_ha_bridge_list
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_ACCESS_TOKEN,
CONF_COUNTRY,
EVENT_HOMEASSISTANT_STOP,
Platform,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.event import async_track_time_interval
from .const import CONF_CONNECT_CLIENT_ID, MQTT_SUBSCRIPTION_INTERVAL
from .coordinator import DeviceDataUpdateCoordinator, async_setup_device_coordinator
from .mqtt import ThinQMQTT
@dataclass(kw_only=True)
class ThinqData:
"""A class that holds runtime data."""
coordinators: dict[str, DeviceDataUpdateCoordinator] = field(default_factory=dict)
mqtt_client: ThinQMQTT | None = None
type ThinqConfigEntry = ConfigEntry[ThinqData]
PLATFORMS = [
Platform.BINARY_SENSOR,
Platform.CLIMATE,
Platform.EVENT,
Platform.FAN,
Platform.NUMBER,
Platform.SELECT,
Platform.SENSOR,
Platform.SWITCH,
Platform.VACUUM,
]
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ThinqConfigEntry) -> bool:
"""Set up an entry."""
entry.runtime_data = ThinqData()
access_token = entry.data[CONF_ACCESS_TOKEN]
client_id = entry.data[CONF_CONNECT_CLIENT_ID]
country_code = entry.data[CONF_COUNTRY]
thinq_api = ThinQApi(
session=async_get_clientsession(hass),
access_token=access_token,
country_code=country_code,
client_id=client_id,
)
# Setup coordinators and register devices.
await async_setup_coordinators(hass, entry, thinq_api)
# Set up all platforms for this device/entry.
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
# Set up MQTT connection.
await async_setup_mqtt(hass, entry, thinq_api, client_id)
# Clean up devices they are no longer in use.
async_cleanup_device_registry(hass, entry)
return True
async def async_setup_coordinators(
hass: HomeAssistant,
entry: ThinqConfigEntry,
thinq_api: ThinQApi,
) -> None:
"""Set up coordinators and register devices."""
# Get a list of ha bridge.
try:
bridge_list = await async_get_ha_bridge_list(thinq_api)
except ThinQAPIException as exc:
raise ConfigEntryNotReady(exc.message) from exc
if not bridge_list:
return
# Setup coordinator per device.
task_list = [
hass.async_create_task(async_setup_device_coordinator(hass, bridge))
for bridge in bridge_list
]
task_result = await asyncio.gather(*task_list)
for coordinator in task_result:
entry.runtime_data.coordinators[coordinator.unique_id] = coordinator
@callback
def async_cleanup_device_registry(hass: HomeAssistant, entry: ThinqConfigEntry) -> None:
"""Clean up device registry."""
new_device_unique_ids = [
coordinator.unique_id
for coordinator in entry.runtime_data.coordinators.values()
]
device_registry = dr.async_get(hass)
existing_entries = dr.async_entries_for_config_entry(
device_registry, entry.entry_id
)
# Remove devices that are no longer exist.
for old_entry in existing_entries:
old_unique_id = next(iter(old_entry.identifiers))[1]
if old_unique_id not in new_device_unique_ids:
device_registry.async_remove_device(old_entry.id)
_LOGGER.debug("Remove device_registry: device_id=%s", old_entry.id)
async def async_setup_mqtt(
hass: HomeAssistant, entry: ThinqConfigEntry, thinq_api: ThinQApi, client_id: str
) -> None:
"""Set up MQTT connection."""
mqtt_client = ThinQMQTT(hass, thinq_api, client_id, entry.runtime_data.coordinators)
entry.runtime_data.mqtt_client = mqtt_client
# Try to connect.
result = await mqtt_client.async_connect()
if not result:
_LOGGER.error("Failed to set up mqtt connection")
return
# Ready to subscribe.
await mqtt_client.async_start_subscribes()
entry.async_on_unload(
async_track_time_interval(
hass,
mqtt_client.async_refresh_subscribe,
MQTT_SUBSCRIPTION_INTERVAL,
cancel_on_shutdown=True,
)
)
entry.async_on_unload(
hass.bus.async_listen_once(
EVENT_HOMEASSISTANT_STOP, mqtt_client.async_disconnect
)
)
async def async_unload_entry(hass: HomeAssistant, entry: ThinqConfigEntry) -> bool:
"""Unload the entry."""
if entry.runtime_data.mqtt_client:
await entry.runtime_data.mqtt_client.async_disconnect()
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

View File

@ -0,0 +1,181 @@
"""Support for binary sensor entities."""
from __future__ import annotations
from dataclasses import dataclass
import logging
from thinqconnect import DeviceType
from thinqconnect.devices.const import Property as ThinQProperty
from thinqconnect.integration import ActiveMode
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
BinarySensorEntityDescription,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import ThinqConfigEntry
from .entity import ThinQEntity
@dataclass(frozen=True, kw_only=True)
class ThinQBinarySensorEntityDescription(BinarySensorEntityDescription):
"""Describes ThinQ sensor entity."""
on_key: str | None = None
BINARY_SENSOR_DESC: dict[ThinQProperty, ThinQBinarySensorEntityDescription] = {
ThinQProperty.RINSE_REFILL: ThinQBinarySensorEntityDescription(
key=ThinQProperty.RINSE_REFILL,
translation_key=ThinQProperty.RINSE_REFILL,
),
ThinQProperty.ECO_FRIENDLY_MODE: ThinQBinarySensorEntityDescription(
key=ThinQProperty.ECO_FRIENDLY_MODE,
translation_key=ThinQProperty.ECO_FRIENDLY_MODE,
),
ThinQProperty.POWER_SAVE_ENABLED: ThinQBinarySensorEntityDescription(
key=ThinQProperty.POWER_SAVE_ENABLED,
translation_key=ThinQProperty.POWER_SAVE_ENABLED,
),
ThinQProperty.REMOTE_CONTROL_ENABLED: ThinQBinarySensorEntityDescription(
key=ThinQProperty.REMOTE_CONTROL_ENABLED,
translation_key=ThinQProperty.REMOTE_CONTROL_ENABLED,
),
ThinQProperty.SABBATH_MODE: ThinQBinarySensorEntityDescription(
key=ThinQProperty.SABBATH_MODE,
translation_key=ThinQProperty.SABBATH_MODE,
),
ThinQProperty.DOOR_STATE: ThinQBinarySensorEntityDescription(
key=ThinQProperty.DOOR_STATE,
device_class=BinarySensorDeviceClass.DOOR,
on_key="open",
),
ThinQProperty.MACHINE_CLEAN_REMINDER: ThinQBinarySensorEntityDescription(
key=ThinQProperty.MACHINE_CLEAN_REMINDER,
translation_key=ThinQProperty.MACHINE_CLEAN_REMINDER,
on_key="mcreminder_on",
),
ThinQProperty.SIGNAL_LEVEL: ThinQBinarySensorEntityDescription(
key=ThinQProperty.SIGNAL_LEVEL,
translation_key=ThinQProperty.SIGNAL_LEVEL,
on_key="signallevel_on",
),
ThinQProperty.CLEAN_LIGHT_REMINDER: ThinQBinarySensorEntityDescription(
key=ThinQProperty.CLEAN_LIGHT_REMINDER,
translation_key=ThinQProperty.CLEAN_LIGHT_REMINDER,
on_key="cleanlreminder_on",
),
ThinQProperty.HOOD_OPERATION_MODE: ThinQBinarySensorEntityDescription(
key=ThinQProperty.HOOD_OPERATION_MODE,
translation_key="operation_mode",
on_key="power_on",
),
ThinQProperty.WATER_HEATER_OPERATION_MODE: ThinQBinarySensorEntityDescription(
key=ThinQProperty.WATER_HEATER_OPERATION_MODE,
translation_key="operation_mode",
on_key="power_on",
),
ThinQProperty.ONE_TOUCH_FILTER: ThinQBinarySensorEntityDescription(
key=ThinQProperty.ONE_TOUCH_FILTER,
translation_key=ThinQProperty.ONE_TOUCH_FILTER,
on_key="on",
),
}
DEVICE_TYPE_BINARY_SENSOR_MAP: dict[
DeviceType, tuple[ThinQBinarySensorEntityDescription, ...]
] = {
DeviceType.COOKTOP: (BINARY_SENSOR_DESC[ThinQProperty.REMOTE_CONTROL_ENABLED],),
DeviceType.DISH_WASHER: (
BINARY_SENSOR_DESC[ThinQProperty.DOOR_STATE],
BINARY_SENSOR_DESC[ThinQProperty.RINSE_REFILL],
BINARY_SENSOR_DESC[ThinQProperty.REMOTE_CONTROL_ENABLED],
BINARY_SENSOR_DESC[ThinQProperty.MACHINE_CLEAN_REMINDER],
BINARY_SENSOR_DESC[ThinQProperty.SIGNAL_LEVEL],
BINARY_SENSOR_DESC[ThinQProperty.CLEAN_LIGHT_REMINDER],
),
DeviceType.DRYER: (BINARY_SENSOR_DESC[ThinQProperty.REMOTE_CONTROL_ENABLED],),
DeviceType.HOOD: (BINARY_SENSOR_DESC[ThinQProperty.HOOD_OPERATION_MODE],),
DeviceType.OVEN: (BINARY_SENSOR_DESC[ThinQProperty.REMOTE_CONTROL_ENABLED],),
DeviceType.REFRIGERATOR: (
BINARY_SENSOR_DESC[ThinQProperty.DOOR_STATE],
BINARY_SENSOR_DESC[ThinQProperty.ECO_FRIENDLY_MODE],
BINARY_SENSOR_DESC[ThinQProperty.POWER_SAVE_ENABLED],
BINARY_SENSOR_DESC[ThinQProperty.SABBATH_MODE],
),
DeviceType.KIMCHI_REFRIGERATOR: (
BINARY_SENSOR_DESC[ThinQProperty.ONE_TOUCH_FILTER],
),
DeviceType.STYLER: (BINARY_SENSOR_DESC[ThinQProperty.REMOTE_CONTROL_ENABLED],),
DeviceType.WASHCOMBO_MAIN: (
BINARY_SENSOR_DESC[ThinQProperty.REMOTE_CONTROL_ENABLED],
),
DeviceType.WASHCOMBO_MINI: (
BINARY_SENSOR_DESC[ThinQProperty.REMOTE_CONTROL_ENABLED],
),
DeviceType.WASHER: (BINARY_SENSOR_DESC[ThinQProperty.REMOTE_CONTROL_ENABLED],),
DeviceType.WASHTOWER_DRYER: (
BINARY_SENSOR_DESC[ThinQProperty.REMOTE_CONTROL_ENABLED],
),
DeviceType.WASHTOWER: (BINARY_SENSOR_DESC[ThinQProperty.REMOTE_CONTROL_ENABLED],),
DeviceType.WASHTOWER_WASHER: (
BINARY_SENSOR_DESC[ThinQProperty.REMOTE_CONTROL_ENABLED],
),
DeviceType.WATER_HEATER: (
BINARY_SENSOR_DESC[ThinQProperty.WATER_HEATER_OPERATION_MODE],
),
DeviceType.WINE_CELLAR: (BINARY_SENSOR_DESC[ThinQProperty.SABBATH_MODE],),
}
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
entry: ThinqConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up an entry for binary sensor platform."""
entities: list[ThinQBinarySensorEntity] = []
for coordinator in entry.runtime_data.coordinators.values():
if (
descriptions := DEVICE_TYPE_BINARY_SENSOR_MAP.get(
coordinator.api.device.device_type
)
) is not None:
for description in descriptions:
entities.extend(
ThinQBinarySensorEntity(coordinator, description, property_id)
for property_id in coordinator.api.get_active_idx(
description.key, ActiveMode.READ_ONLY
)
)
if entities:
async_add_entities(entities)
class ThinQBinarySensorEntity(ThinQEntity, BinarySensorEntity):
"""Represent a thinq binary sensor platform."""
entity_description: ThinQBinarySensorEntityDescription
def _update_status(self) -> None:
"""Update status itself."""
super()._update_status()
if (key := self.entity_description.on_key) is not None:
self._attr_is_on = self.data.value == key
else:
self._attr_is_on = self.data.is_on
_LOGGER.debug(
"[%s:%s] update status: %s -> %s",
self.coordinator.device_name,
self.property_id,
self.data.value,
self.is_on,
)

View File

@ -0,0 +1,334 @@
"""Support for climate entities."""
from __future__ import annotations
from dataclasses import dataclass
import logging
from typing import Any
from thinqconnect import DeviceType
from thinqconnect.integration import ExtendedProperty
from homeassistant.components.climate import (
ATTR_TARGET_TEMP_HIGH,
ATTR_TARGET_TEMP_LOW,
FAN_OFF,
ClimateEntity,
ClimateEntityDescription,
ClimateEntityFeature,
HVACMode,
)
from homeassistant.const import ATTR_TEMPERATURE, UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.temperature import display_temp
from . import ThinqConfigEntry
from .coordinator import DeviceDataUpdateCoordinator
from .entity import ThinQEntity
@dataclass(frozen=True, kw_only=True)
class ThinQClimateEntityDescription(ClimateEntityDescription):
"""Describes ThinQ climate entity."""
min_temp: float | None = None
max_temp: float | None = None
step: float | None = None
DEVIE_TYPE_CLIMATE_MAP: dict[DeviceType, tuple[ThinQClimateEntityDescription, ...]] = {
DeviceType.AIR_CONDITIONER: (
ThinQClimateEntityDescription(
key=ExtendedProperty.CLIMATE_AIR_CONDITIONER,
name=None,
translation_key=ExtendedProperty.CLIMATE_AIR_CONDITIONER,
),
),
DeviceType.SYSTEM_BOILER: (
ThinQClimateEntityDescription(
key=ExtendedProperty.CLIMATE_SYSTEM_BOILER,
name=None,
min_temp=16,
max_temp=30,
step=1,
),
),
}
STR_TO_HVAC: dict[str, HVACMode] = {
"air_dry": HVACMode.DRY,
"auto": HVACMode.AUTO,
"cool": HVACMode.COOL,
"fan": HVACMode.FAN_ONLY,
"heat": HVACMode.HEAT,
}
HVAC_TO_STR: dict[HVACMode, str] = {
HVACMode.AUTO: "auto",
HVACMode.COOL: "cool",
HVACMode.DRY: "air_dry",
HVACMode.FAN_ONLY: "fan",
HVACMode.HEAT: "heat",
}
THINQ_PRESET_MODE: list[str] = ["air_clean", "aroma", "energy_saving"]
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
entry: ThinqConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up an entry for climate platform."""
entities: list[ThinQClimateEntity] = []
for coordinator in entry.runtime_data.coordinators.values():
if (
descriptions := DEVIE_TYPE_CLIMATE_MAP.get(
coordinator.api.device.device_type
)
) is not None:
for description in descriptions:
entities.extend(
ThinQClimateEntity(coordinator, description, property_id)
for property_id in coordinator.api.get_active_idx(description.key)
)
if entities:
async_add_entities(entities)
class ThinQClimateEntity(ThinQEntity, ClimateEntity):
"""Represent a thinq climate platform."""
entity_description: ThinQClimateEntityDescription
def __init__(
self,
coordinator: DeviceDataUpdateCoordinator,
entity_description: ThinQClimateEntityDescription,
property_id: str,
) -> None:
"""Initialize a climate entity."""
super().__init__(coordinator, entity_description, property_id)
self._attr_supported_features = (
ClimateEntityFeature.TARGET_TEMPERATURE
| ClimateEntityFeature.TURN_ON
| ClimateEntityFeature.TURN_OFF
)
self._attr_hvac_modes = [HVACMode.OFF]
self._attr_hvac_mode = HVACMode.OFF
self._attr_preset_modes = []
self._attr_temperature_unit = UnitOfTemperature.CELSIUS
self._requested_hvac_mode: str | None = None
# Set up HVAC modes.
for mode in self.data.hvac_modes:
if mode in STR_TO_HVAC:
self._attr_hvac_modes.append(STR_TO_HVAC[mode])
elif mode in THINQ_PRESET_MODE:
self._attr_preset_modes.append(mode)
self._attr_supported_features |= ClimateEntityFeature.PRESET_MODE
# Set up fan modes.
self._attr_fan_modes = self.data.fan_modes
if self.fan_modes:
self._attr_supported_features |= ClimateEntityFeature.FAN_MODE
# Supports target temperature range.
if self.data.support_temperature_range:
self._attr_supported_features |= (
ClimateEntityFeature.TARGET_TEMPERATURE_RANGE
)
def _update_status(self) -> None:
"""Update status itself."""
super()._update_status()
# Update fan, hvac and preset mode.
if self.data.is_on:
if self.supported_features & ClimateEntityFeature.FAN_MODE:
self._attr_fan_mode = self.data.fan_mode
hvac_mode = self._requested_hvac_mode or self.data.hvac_mode
if hvac_mode in STR_TO_HVAC:
self._attr_hvac_mode = STR_TO_HVAC.get(hvac_mode)
self._attr_preset_mode = None
elif hvac_mode in THINQ_PRESET_MODE:
self._attr_preset_mode = hvac_mode
else:
if self.supported_features & ClimateEntityFeature.FAN_MODE:
self._attr_fan_mode = FAN_OFF
self._attr_hvac_mode = HVACMode.OFF
self._attr_preset_mode = None
self.reset_requested_hvac_mode()
self._attr_current_humidity = self.data.humidity
self._attr_current_temperature = self.data.current_temp
if (max_temp := self.entity_description.max_temp) is not None or (
max_temp := self.data.max
) is not None:
self._attr_max_temp = max_temp
if (min_temp := self.entity_description.min_temp) is not None or (
min_temp := self.data.min
) is not None:
self._attr_min_temp = min_temp
if (step := self.entity_description.step) is not None or (
step := self.data.step
) is not None:
self._attr_target_temperature_step = step
# Update target temperatures.
if (
self.supported_features & ClimateEntityFeature.TARGET_TEMPERATURE_RANGE
and self.hvac_mode == HVACMode.AUTO
):
self._attr_target_temperature = None
self._attr_target_temperature_high = self.data.target_temp_high
self._attr_target_temperature_low = self.data.target_temp_low
else:
self._attr_target_temperature = self.data.target_temp
self._attr_target_temperature_high = None
self._attr_target_temperature_low = None
_LOGGER.debug(
"[%s:%s] update status: %s/%s -> %s/%s, hvac:%s, unit:%s, step:%s",
self.coordinator.device_name,
self.property_id,
self.data.current_temp,
self.data.target_temp,
self.current_temperature,
self.target_temperature,
self.hvac_mode,
self.temperature_unit,
self.target_temperature_step,
)
def reset_requested_hvac_mode(self) -> None:
"""Cancel request to set hvac mode."""
self._requested_hvac_mode = None
async def async_turn_on(self) -> None:
"""Turn the entity on."""
_LOGGER.debug(
"[%s:%s] async_turn_on", self.coordinator.device_name, self.property_id
)
await self.async_call_api(self.coordinator.api.async_turn_on(self.property_id))
async def async_turn_off(self) -> None:
"""Turn the entity off."""
_LOGGER.debug(
"[%s:%s] async_turn_off", self.coordinator.device_name, self.property_id
)
await self.async_call_api(self.coordinator.api.async_turn_off(self.property_id))
async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Set new target hvac mode."""
if hvac_mode == HVACMode.OFF:
await self.async_turn_off()
return
# If device is off, turn on first.
if not self.data.is_on:
await self.async_turn_on()
# When we request hvac mode while turning on the device, the previously set
# hvac mode is displayed first and then switches to the requested hvac mode.
# To prevent this, set the requested hvac mode here so that it will be set
# immediately on the next update.
self._requested_hvac_mode = HVAC_TO_STR.get(hvac_mode)
_LOGGER.debug(
"[%s:%s] async_set_hvac_mode: %s",
self.coordinator.device_name,
self.property_id,
hvac_mode,
)
await self.async_call_api(
self.coordinator.api.async_set_hvac_mode(
self.property_id, self._requested_hvac_mode
),
self.reset_requested_hvac_mode,
)
async def async_set_preset_mode(self, preset_mode: str) -> None:
"""Set new preset mode."""
_LOGGER.debug(
"[%s:%s] async_set_preset_mode: %s",
self.coordinator.device_name,
self.property_id,
preset_mode,
)
await self.async_call_api(
self.coordinator.api.async_set_hvac_mode(self.property_id, preset_mode)
)
async def async_set_fan_mode(self, fan_mode: str) -> None:
"""Set new target fan mode."""
_LOGGER.debug(
"[%s:%s] async_set_fan_mode: %s",
self.coordinator.device_name,
self.property_id,
fan_mode,
)
await self.async_call_api(
self.coordinator.api.async_set_fan_mode(self.property_id, fan_mode)
)
def _round_by_step(self, temperature: float) -> float:
"""Round the value by step."""
if (
target_temp := display_temp(
self.coordinator.hass,
temperature,
self.coordinator.hass.config.units.temperature_unit,
self.target_temperature_step or 1,
)
) is not None:
return target_temp
return temperature
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperature."""
_LOGGER.debug(
"[%s:%s] async_set_temperature: %s",
self.coordinator.device_name,
self.property_id,
kwargs,
)
if (temperature := kwargs.get(ATTR_TEMPERATURE)) is not None:
if (
target_temp := self._round_by_step(temperature)
) != self.target_temperature:
await self.async_call_api(
self.coordinator.api.async_set_target_temperature(
self.property_id, target_temp
)
)
if (temperature_low := kwargs.get(ATTR_TARGET_TEMP_LOW)) is not None:
if (
target_temp_low := self._round_by_step(temperature_low)
) != self.target_temperature_low:
await self.async_call_api(
self.coordinator.api.async_set_target_temperature_low(
self.property_id, target_temp_low
)
)
if (temperature_high := kwargs.get(ATTR_TARGET_TEMP_HIGH)) is not None:
if (
target_temp_high := self._round_by_step(temperature_high)
) != self.target_temperature_high:
await self.async_call_api(
self.coordinator.api.async_set_target_temperature_high(
self.property_id, target_temp_high
)
)

View File

@ -0,0 +1,103 @@
"""Config flow for LG ThinQ."""
from __future__ import annotations
import logging
from typing import Any
import uuid
from thinqconnect import ThinQApi, ThinQAPIException
from thinqconnect.country import Country
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_COUNTRY
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.selector import CountrySelector, CountrySelectorConfig
from .const import (
CLIENT_PREFIX,
CONF_CONNECT_CLIENT_ID,
DEFAULT_COUNTRY,
DOMAIN,
THINQ_DEFAULT_NAME,
THINQ_PAT_URL,
)
SUPPORTED_COUNTRIES = [country.value for country in Country]
_LOGGER = logging.getLogger(__name__)
class ThinQFlowHandler(ConfigFlow, domain=DOMAIN):
"""Handle a config flow."""
VERSION = 1
def _get_default_country_code(self) -> str:
"""Get the default country code based on config."""
country = self.hass.config.country
if country is not None and country in SUPPORTED_COUNTRIES:
return country
return DEFAULT_COUNTRY
async def _validate_and_create_entry(
self, access_token: str, country_code: str
) -> ConfigFlowResult:
"""Create an entry for the flow."""
connect_client_id = f"{CLIENT_PREFIX}-{uuid.uuid4()!s}"
# To verify PAT, create an api to retrieve the device list.
await ThinQApi(
session=async_get_clientsession(self.hass),
access_token=access_token,
country_code=country_code,
client_id=connect_client_id,
).async_get_device_list()
# If verification is success, create entry.
return self.async_create_entry(
title=THINQ_DEFAULT_NAME,
data={
CONF_ACCESS_TOKEN: access_token,
CONF_CONNECT_CLIENT_ID: connect_client_id,
CONF_COUNTRY: country_code,
},
)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle a flow initiated by the user."""
errors: dict[str, str] = {}
if user_input is not None:
access_token = user_input[CONF_ACCESS_TOKEN]
country_code = user_input[CONF_COUNTRY]
# Check if PAT is already configured.
await self.async_set_unique_id(access_token)
self._abort_if_unique_id_configured()
try:
return await self._validate_and_create_entry(access_token, country_code)
except ThinQAPIException:
errors["base"] = "token_unauthorized"
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_ACCESS_TOKEN): cv.string,
vol.Required(
CONF_COUNTRY, default=self._get_default_country_code()
): CountrySelector(
CountrySelectorConfig(countries=SUPPORTED_COUNTRIES)
),
}
),
description_placeholders={"pat_url": THINQ_PAT_URL},
errors=errors,
)

View File

@ -0,0 +1,20 @@
"""Constants for LG ThinQ."""
from datetime import timedelta
from typing import Final
# Config flow
DOMAIN = "lg_thinq"
COMPANY = "LGE"
DEFAULT_COUNTRY: Final = "US"
THINQ_DEFAULT_NAME: Final = "LG ThinQ"
THINQ_PAT_URL: Final = "https://connect-pat.lgthinq.com"
CLIENT_PREFIX: Final = "home-assistant"
CONF_CONNECT_CLIENT_ID: Final = "connect_client_id"
# MQTT
MQTT_SUBSCRIPTION_INTERVAL: Final = timedelta(days=1)
# MQTT: Message types
DEVICE_PUSH_MESSAGE: Final = "DEVICE_PUSH"
DEVICE_STATUS_MESSAGE: Final = "DEVICE_STATUS"

View File

@ -0,0 +1,81 @@
"""DataUpdateCoordinator for the LG ThinQ device."""
from __future__ import annotations
import logging
from typing import Any
from thinqconnect import ThinQAPIException
from thinqconnect.integration import HABridge
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
class DeviceDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
"""LG Device's Data Update Coordinator."""
def __init__(self, hass: HomeAssistant, ha_bridge: HABridge) -> None:
"""Initialize data coordinator."""
super().__init__(
hass,
_LOGGER,
name=f"{DOMAIN}_{ha_bridge.device.device_id}",
)
self.data = {}
self.api = ha_bridge
self.device_id = ha_bridge.device.device_id
self.sub_id = ha_bridge.sub_id
alias = ha_bridge.device.alias
# The device name is usually set to 'alias'.
# But, if the sub_id exists, it will be set to 'alias {sub_id}'.
# e.g. alias='MyWashTower', sub_id='dryer' then 'MyWashTower dryer'.
self.device_name = f"{alias} {self.sub_id}" if self.sub_id else alias
# The unique id is usually set to 'device_id'.
# But, if the sub_id exists, it will be set to 'device_id_{sub_id}'.
# e.g. device_id='TQSXXXX', sub_id='dryer' then 'TQSXXXX_dryer'.
self.unique_id = (
f"{self.device_id}_{self.sub_id}" if self.sub_id else self.device_id
)
async def _async_update_data(self) -> dict[str, Any]:
"""Request to the server to update the status from full response data."""
try:
return await self.api.fetch_data()
except ThinQAPIException as e:
raise UpdateFailed(e) from e
def refresh_status(self) -> None:
"""Refresh current status."""
self.async_set_updated_data(self.data)
def handle_update_status(self, status: dict[str, Any]) -> None:
"""Handle the status received from the mqtt connection."""
data = self.api.update_status(status)
if data is not None:
self.async_set_updated_data(data)
def handle_notification_message(self, message: str | None) -> None:
"""Handle the status received from the mqtt connection."""
data = self.api.update_notification(message)
if data is not None:
self.async_set_updated_data(data)
async def async_setup_device_coordinator(
hass: HomeAssistant, ha_bridge: HABridge
) -> DeviceDataUpdateCoordinator:
"""Create DeviceDataUpdateCoordinator and device_api per device."""
coordinator = DeviceDataUpdateCoordinator(hass, ha_bridge)
await coordinator.async_refresh()
_LOGGER.debug("Setup device's coordinator: %s", coordinator.device_name)
return coordinator

View File

@ -0,0 +1,114 @@
"""Base class for ThinQ entities."""
from __future__ import annotations
from collections.abc import Callable, Coroutine
import logging
from typing import Any
from thinqconnect import ThinQAPIException
from thinqconnect.devices.const import Location
from thinqconnect.integration import PropertyState
from homeassistant.const import UnitOfTemperature
from homeassistant.core import callback
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.entity import EntityDescription
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import COMPANY, DOMAIN
from .coordinator import DeviceDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
EMPTY_STATE = PropertyState()
UNIT_CONVERSION_MAP: dict[str, str] = {
"F": UnitOfTemperature.FAHRENHEIT,
"C": UnitOfTemperature.CELSIUS,
}
class ThinQEntity(CoordinatorEntity[DeviceDataUpdateCoordinator]):
"""The base implementation of all lg thinq entities."""
_attr_has_entity_name = True
def __init__(
self,
coordinator: DeviceDataUpdateCoordinator,
entity_description: EntityDescription,
property_id: str,
) -> None:
"""Initialize an entity."""
super().__init__(coordinator)
self.entity_description = entity_description
self.property_id = property_id
self.location = self.coordinator.api.get_location_for_idx(self.property_id)
self._attr_device_info = dr.DeviceInfo(
identifiers={(DOMAIN, coordinator.unique_id)},
manufacturer=COMPANY,
model=coordinator.api.device.model_name,
name=coordinator.device_name,
)
self._attr_unique_id = f"{coordinator.unique_id}_{self.property_id}"
if self.location is not None and self.location not in (
Location.MAIN,
Location.OVEN,
coordinator.sub_id,
):
self._attr_translation_placeholders = {"location": self.location}
self._attr_translation_key = (
f"{entity_description.translation_key}_for_location"
)
@property
def data(self) -> PropertyState:
"""Return the state data of entity."""
return self.coordinator.data.get(self.property_id, EMPTY_STATE)
def _get_unit_of_measurement(self, unit: str | None) -> str | None:
"""Convert thinq unit string to HA unit string."""
if unit is None:
return None
return UNIT_CONVERSION_MAP.get(unit)
def _update_status(self) -> None:
"""Update status itself.
All inherited classes can update their own status in here.
"""
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
self._update_status()
self.async_write_ha_state()
async def async_added_to_hass(self) -> None:
"""Call when entity is added to hass."""
await super().async_added_to_hass()
self._handle_coordinator_update()
async def async_call_api(
self,
target: Coroutine[Any, Any, Any],
on_fail_method: Callable[[], None] | None = None,
) -> None:
"""Call the given api and handle exception."""
try:
await target
except ThinQAPIException as exc:
if on_fail_method:
on_fail_method()
raise ServiceValidationError(
exc.message, translation_domain=DOMAIN, translation_key=exc.code
) from exc
except ValueError as exc:
if on_fail_method:
on_fail_method()
raise ServiceValidationError(exc) from exc

View File

@ -0,0 +1,115 @@
"""Support for event entity."""
from __future__ import annotations
import logging
from thinqconnect import DeviceType
from thinqconnect.integration import ActiveMode, ThinQPropertyEx
from homeassistant.components.event import EventEntity, EventEntityDescription
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import ThinqConfigEntry
from .coordinator import DeviceDataUpdateCoordinator
from .entity import ThinQEntity
NOTIFICATION_EVENT_DESC = EventEntityDescription(
key=ThinQPropertyEx.NOTIFICATION,
translation_key=ThinQPropertyEx.NOTIFICATION,
)
ERROR_EVENT_DESC = EventEntityDescription(
key=ThinQPropertyEx.ERROR,
translation_key=ThinQPropertyEx.ERROR,
)
ALL_EVENTS: tuple[EventEntityDescription, ...] = (
ERROR_EVENT_DESC,
NOTIFICATION_EVENT_DESC,
)
DEVICE_TYPE_EVENT_MAP: dict[DeviceType, tuple[EventEntityDescription, ...]] = {
DeviceType.AIR_CONDITIONER: (NOTIFICATION_EVENT_DESC,),
DeviceType.AIR_PURIFIER_FAN: (NOTIFICATION_EVENT_DESC,),
DeviceType.AIR_PURIFIER: (NOTIFICATION_EVENT_DESC,),
DeviceType.DEHUMIDIFIER: (NOTIFICATION_EVENT_DESC,),
DeviceType.DISH_WASHER: ALL_EVENTS,
DeviceType.DRYER: ALL_EVENTS,
DeviceType.HUMIDIFIER: (NOTIFICATION_EVENT_DESC,),
DeviceType.KIMCHI_REFRIGERATOR: (NOTIFICATION_EVENT_DESC,),
DeviceType.MICROWAVE_OVEN: (NOTIFICATION_EVENT_DESC,),
DeviceType.OVEN: (NOTIFICATION_EVENT_DESC,),
DeviceType.REFRIGERATOR: (NOTIFICATION_EVENT_DESC,),
DeviceType.ROBOT_CLEANER: ALL_EVENTS,
DeviceType.STICK_CLEANER: (NOTIFICATION_EVENT_DESC,),
DeviceType.STYLER: ALL_EVENTS,
DeviceType.WASHCOMBO_MAIN: ALL_EVENTS,
DeviceType.WASHCOMBO_MINI: ALL_EVENTS,
DeviceType.WASHER: ALL_EVENTS,
DeviceType.WASHTOWER_DRYER: ALL_EVENTS,
DeviceType.WASHTOWER: ALL_EVENTS,
DeviceType.WASHTOWER_WASHER: ALL_EVENTS,
DeviceType.WINE_CELLAR: (NOTIFICATION_EVENT_DESC,),
}
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
entry: ThinqConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up an entry for event platform."""
entities: list[ThinQEventEntity] = []
for coordinator in entry.runtime_data.coordinators.values():
if (
descriptions := DEVICE_TYPE_EVENT_MAP.get(
coordinator.api.device.device_type
)
) is not None:
for description in descriptions:
entities.extend(
ThinQEventEntity(coordinator, description, property_id)
for property_id in coordinator.api.get_active_idx(
description.key, ActiveMode.READ_ONLY
)
)
if entities:
async_add_entities(entities)
class ThinQEventEntity(ThinQEntity, EventEntity):
"""Represent an thinq event platform."""
def __init__(
self,
coordinator: DeviceDataUpdateCoordinator,
entity_description: EventEntityDescription,
property_id: str,
) -> None:
"""Initialize an event platform."""
super().__init__(coordinator, entity_description, property_id)
# For event types.
self._attr_event_types = self.data.options
def _update_status(self) -> None:
"""Update status itself."""
super()._update_status()
_LOGGER.debug(
"[%s:%s] update status: %s, event_types=%s",
self.coordinator.device_name,
self.property_id,
self.data.value,
self.event_types,
)
# Handle an event.
if (value := self.data.value) is not None and value in self.event_types:
self._async_handle_update(value)
def _async_handle_update(self, value: str) -> None:
"""Handle the event."""
self._trigger_event(value)
self.async_write_ha_state()

View File

@ -0,0 +1,150 @@
"""Support for fan entities."""
from __future__ import annotations
import logging
from typing import Any
from thinqconnect import DeviceType
from thinqconnect.integration import ExtendedProperty
from homeassistant.components.fan import (
FanEntity,
FanEntityDescription,
FanEntityFeature,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util.percentage import (
ordered_list_item_to_percentage,
percentage_to_ordered_list_item,
)
from . import ThinqConfigEntry
from .coordinator import DeviceDataUpdateCoordinator
from .entity import ThinQEntity
DEVICE_TYPE_FAN_MAP: dict[DeviceType, tuple[FanEntityDescription, ...]] = {
DeviceType.CEILING_FAN: (
FanEntityDescription(
key=ExtendedProperty.FAN,
name=None,
),
),
}
FOUR_STEP_SPEEDS = ["low", "mid", "high", "turbo"]
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
entry: ThinqConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up an entry for fan platform."""
entities: list[ThinQFanEntity] = []
for coordinator in entry.runtime_data.coordinators.values():
if (
descriptions := DEVICE_TYPE_FAN_MAP.get(coordinator.api.device.device_type)
) is not None:
for description in descriptions:
entities.extend(
ThinQFanEntity(coordinator, description, property_id)
for property_id in coordinator.api.get_active_idx(description.key)
)
if entities:
async_add_entities(entities)
class ThinQFanEntity(ThinQEntity, FanEntity):
"""Represent a thinq fan platform."""
def __init__(
self,
coordinator: DeviceDataUpdateCoordinator,
entity_description: FanEntityDescription,
property_id: str,
) -> None:
"""Initialize fan platform."""
super().__init__(coordinator, entity_description, property_id)
self._ordered_named_fan_speeds = []
self._attr_supported_features |= FanEntityFeature.SET_SPEED
if (fan_modes := self.data.fan_modes) is not None:
self._attr_speed_count = len(fan_modes)
if self.speed_count == 4:
self._ordered_named_fan_speeds = FOUR_STEP_SPEEDS
def _update_status(self) -> None:
"""Update status itself."""
super()._update_status()
# Update power on state.
self._attr_is_on = self.data.is_on
# Update fan speed.
if (
self.data.is_on
and (mode := self.data.fan_mode) in self._ordered_named_fan_speeds
):
self._attr_percentage = ordered_list_item_to_percentage(
self._ordered_named_fan_speeds, mode
)
else:
self._attr_percentage = 0
_LOGGER.debug(
"[%s:%s] update status: %s -> %s (percntage=%s)",
self.coordinator.device_name,
self.property_id,
self.data.is_on,
self.is_on,
self.percentage,
)
async def async_set_percentage(self, percentage: int) -> None:
"""Set the speed percentage of the fan."""
if percentage == 0:
await self.async_turn_off()
return
try:
value = percentage_to_ordered_list_item(
self._ordered_named_fan_speeds, percentage
)
except ValueError:
_LOGGER.exception("Failed to async_set_percentage")
return
_LOGGER.debug(
"[%s:%s] async_set_percentage. percntage=%s, value=%s",
self.coordinator.device_name,
self.property_id,
percentage,
value,
)
await self.async_call_api(
self.coordinator.api.async_set_fan_mode(self.property_id, value)
)
async def async_turn_on(
self,
percentage: int | None = None,
preset_mode: str | None = None,
**kwargs: Any,
) -> None:
"""Turn on the fan."""
_LOGGER.debug(
"[%s:%s] async_turn_on", self.coordinator.device_name, self.property_id
)
await self.async_call_api(self.coordinator.api.async_turn_on(self.property_id))
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the fan off."""
_LOGGER.debug(
"[%s:%s] async_turn_off", self.coordinator.device_name, self.property_id
)
await self.async_call_api(self.coordinator.api.async_turn_off(self.property_id))

View File

@ -0,0 +1,407 @@
{
"entity": {
"switch": {
"auto_mode": {
"default": "mdi:cogs"
},
"express_mode": {
"default": "mdi:snowflake-variant"
},
"hot_water_mode": {
"default": "mdi:list-status"
},
"humidity_warm_mode": {
"default": "mdi:heat-wave"
},
"hygiene_dry_mode": {
"default": "mdi:format-list-bulleted"
},
"mood_lamp_state": {
"default": "mdi:lamp"
},
"operation_power": {
"default": "mdi:power"
},
"optimal_humidity": {
"default": "mdi:water-percent"
},
"power_save_enabled": {
"default": "mdi:hydro-power"
},
"rapid_freeze": {
"default": "mdi:snowflake"
},
"sleep_mode": {
"default": "mdi:format-list-bulleted"
},
"uv_nano": {
"default": "mdi:air-filter"
},
"warm_mode": {
"default": "mdi:heat-wave"
}
},
"binary_sensor": {
"eco_friendly_mode": {
"default": "mdi:sprout"
},
"power_save_enabled": {
"default": "mdi:meter-electric"
},
"remote_control_enabled": {
"default": "mdi:remote"
},
"remote_control_enabled_for_location": {
"default": "mdi:remote"
},
"rinse_refill": {
"default": "mdi:tune-vertical-variant"
},
"sabbath_mode": {
"default": "mdi:food-off-outline"
},
"machine_clean_reminder": {
"default": "mdi:tune-vertical-variant"
},
"signal_level": {
"default": "mdi:tune-vertical-variant"
},
"clean_light_reminder": {
"default": "mdi:tune-vertical-variant"
},
"operation_mode": {
"default": "mdi:power"
},
"one_touch_filter": {
"default": "mdi:air-filter"
}
},
"climate": {
"climate_air_conditioner": {
"state_attributes": {
"fan_mode": {
"state": {
"slow": "mdi:fan-chevron-down",
"low": "mdi:fan-speed-1",
"mid": "mdi:fan-speed-2",
"high": "mdi:fan-speed-3",
"power": "mdi:fan-chevron-up",
"auto": "mdi:fan-auto"
}
}
}
}
},
"event": {
"error": {
"default": "mdi:alert-circle-outline"
},
"notification": {
"default": "mdi:message-badge-outline"
}
},
"number": {
"target_temperature": {
"default": "mdi:thermometer"
},
"target_temperature_for_location": {
"default": "mdi:thermometer"
},
"light_status": {
"default": "mdi:television-ambient-light"
},
"fan_speed": {
"default": "mdi:wind-power-outline"
},
"lamp_brightness": {
"default": "mdi:alarm-light-outline"
},
"wind_temperature": {
"default": "mdi:thermometer"
},
"relative_hour_to_start": {
"default": "mdi:timer-edit-outline"
},
"relative_hour_to_start_for_location": {
"default": "mdi:timer-edit-outline"
},
"relative_hour_to_start_wm": {
"default": "mdi:timer-edit-outline"
},
"relative_hour_to_start_wm_for_location": {
"default": "mdi:timer-edit-outline"
},
"relative_hour_to_stop": {
"default": "mdi:timer-edit-outline"
},
"relative_hour_to_stop_for_location": {
"default": "mdi:timer-edit-outline"
},
"relative_hour_to_stop_wm": {
"default": "mdi:timer-edit-outline"
},
"relative_hour_to_stop_wm_for_location": {
"default": "mdi:timer-edit-outline"
},
"sleep_timer_relative_hour_to_stop": {
"default": "mdi:bed-clock"
},
"sleep_timer_relative_hour_to_stop_for_location": {
"default": "mdi:bed-clock"
}
},
"select": {
"wind_strength": {
"default": "mdi:wind-power-outline"
},
"monitoring_enabled": {
"default": "mdi:monitor-eye"
},
"current_job_mode": {
"default": "mdi:format-list-bulleted"
},
"operation_mode": {
"default": "mdi:gesture-tap-button"
},
"operation_mode_for_location": {
"default": "mdi:gesture-tap-button"
},
"air_clean_operation_mode": {
"default": "mdi:air-filter"
},
"cook_mode": {
"default": "mdi:chef-hat"
},
"cook_mode_for_location": {
"default": "mdi:chef-hat"
},
"light_brightness": {
"default": "mdi:list-status"
},
"wind_angle": {
"default": "mdi:rotate-360"
},
"display_light": {
"default": "mdi:brightness-6"
},
"fresh_air_filter": {
"default": "mdi:air-filter"
},
"hygiene_dry_mode": {
"default": "mdi:format-list-bulleted"
}
},
"sensor": {
"odor_level": {
"default": "mdi:scent"
},
"current_temperature": {
"default": "mdi:thermometer"
},
"temperature": {
"default": "mdi:thermometer"
},
"total_pollution_level": {
"default": "mdi:air-filter"
},
"monitoring_enabled": {
"default": "mdi:monitor-eye"
},
"growth_mode": {
"default": "mdi:sprout-outline"
},
"growth_mode_for_location": {
"default": "mdi:sprout-outline"
},
"wind_volume": {
"default": "mdi:wind-power-outline"
},
"wind_volume_for_location": {
"default": "mdi:wind-power-outline"
},
"brightness": {
"default": "mdi:tune-vertical-variant"
},
"brightness_for_location": {
"default": "mdi:tune-vertical-variant"
},
"duration": {
"default": "mdi:tune-vertical-variant"
},
"duration_for_location": {
"default": "mdi:tune-vertical-variant"
},
"day_target_temperature": {
"default": "mdi:thermometer"
},
"day_target_temperature_for_location": {
"default": "mdi:thermometer"
},
"night_target_temperature": {
"default": "mdi:thermometer"
},
"night_target_temperature_for_location": {
"default": "mdi:thermometer"
},
"temperature_state": {
"default": "mdi:thermometer"
},
"temperature_state_for_location": {
"default": "mdi:thermometer"
},
"current_state": {
"default": "mdi:list-status"
},
"current_state_for_location": {
"default": "mdi:list-status"
},
"fresh_air_filter": {
"default": "mdi:air-filter"
},
"filter_lifetime": {
"default": "mdi:air-filter"
},
"used_time": {
"default": "mdi:air-filter"
},
"current_job_mode": {
"default": "mdi:dots-circle"
},
"current_job_mode_stick_cleaner": {
"default": "mdi:dots-circle"
},
"personalization_mode": {
"default": "mdi:dots-circle"
},
"current_dish_washing_course": {
"default": "mdi:format-list-checks"
},
"rinse_level": {
"default": "mdi:tune-vertical-variant"
},
"softening_level": {
"default": "mdi:tune-vertical-variant"
},
"cock_state": {
"default": "mdi:air-filter"
},
"sterilizing_state": {
"default": "mdi:water-alert-outline"
},
"water_type": {
"default": "mdi:water"
},
"target_temperature": {
"default": "mdi:thermometer"
},
"target_temperature_for_location": {
"default": "mdi:thermometer"
},
"elapsed_day_state": {
"default": "mdi:calendar-range-outline"
},
"elapsed_day_total": {
"default": "mdi:calendar-range-outline"
},
"recipe_name": {
"default": "mdi:information-box-outline"
},
"wort_info": {
"default": "mdi:information-box-outline"
},
"yeast_info": {
"default": "mdi:information-box-outline"
},
"hop_oil_info": {
"default": "mdi:information-box-outline"
},
"flavor_info": {
"default": "mdi:information-box-outline"
},
"beer_remain": {
"default": "mdi:glass-mug-variant"
},
"battery_level": {
"default": "mdi:battery-medium"
},
"relative_to_start": {
"default": "mdi:clock-time-three-outline"
},
"relative_to_start_for_location": {
"default": "mdi:clock-time-three-outline"
},
"relative_to_start_wm": {
"default": "mdi:clock-time-three-outline"
},
"relative_to_start_wm_for_location": {
"default": "mdi:clock-time-three-outline"
},
"relative_to_stop": {
"default": "mdi:clock-time-three-outline"
},
"relative_to_stop_for_location": {
"default": "mdi:clock-time-three-outline"
},
"relative_to_stop_wm": {
"default": "mdi:clock-time-three-outline"
},
"relative_to_stop_wm_for_location": {
"default": "mdi:clock-time-three-outline"
},
"sleep_timer_relative_to_stop": {
"default": "mdi:bed-clock"
},
"sleep_timer_relative_to_stop_for_location": {
"default": "mdi:bed-clock"
},
"absolute_to_start": {
"default": "mdi:clock-time-three-outline"
},
"absolute_to_start_for_location": {
"default": "mdi:clock-time-three-outline"
},
"absolute_to_stop": {
"default": "mdi:clock-time-three-outline"
},
"absolute_to_stop_for_location": {
"default": "mdi:clock-time-three-outline"
},
"remain": {
"default": "mdi:timer-sand"
},
"remain_for_location": {
"default": "mdi:timer-sand"
},
"running": {
"default": "mdi:timer-play-outline"
},
"running_for_location": {
"default": "mdi:timer-play-outline"
},
"total": {
"default": "mdi:timer-play-outline"
},
"total_for_location": {
"default": "mdi:timer-play-outline"
},
"target": {
"default": "mdi:clock-time-three-outline"
},
"target_for_location": {
"default": "mdi:clock-time-three-outline"
},
"light_start": {
"default": "mdi:clock-time-three-outline"
},
"light_start_for_location": {
"default": "mdi:clock-time-three-outline"
},
"power_level": {
"default": "mdi:radiator"
},
"power_level_for_location": {
"default": "mdi:radiator"
}
}
}
}

View File

@ -0,0 +1,11 @@
{
"domain": "lg_thinq",
"name": "LG ThinQ",
"codeowners": ["@LG-ThinQ-Integration"],
"config_flow": true,
"dependencies": [],
"documentation": "https://www.home-assistant.io/integrations/lg_thinq/",
"iot_class": "cloud_push",
"loggers": ["thinqconnect"],
"requirements": ["thinqconnect==0.9.8"]
}

View File

@ -0,0 +1,186 @@
"""Support for LG ThinQ Connect API."""
from __future__ import annotations
import asyncio
from datetime import datetime
import json
import logging
from typing import Any
from thinqconnect import (
DeviceType,
ThinQApi,
ThinQAPIErrorCodes,
ThinQAPIException,
ThinQMQTTClient,
)
from homeassistant.core import Event, HomeAssistant
from .const import DEVICE_PUSH_MESSAGE, DEVICE_STATUS_MESSAGE
from .coordinator import DeviceDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
class ThinQMQTT:
"""A class that implements MQTT connection."""
def __init__(
self,
hass: HomeAssistant,
thinq_api: ThinQApi,
client_id: str,
coordinators: dict[str, DeviceDataUpdateCoordinator],
) -> None:
"""Initialize a mqtt."""
self.hass = hass
self.thinq_api = thinq_api
self.client_id = client_id
self.coordinators = coordinators
self.client: ThinQMQTTClient | None = None
async def async_connect(self) -> bool:
"""Create a mqtt client and then try to connect."""
try:
self.client = await ThinQMQTTClient(
self.thinq_api, self.client_id, self.on_message_received
)
if self.client is None:
return False
# Connect to server and create certificate.
return await self.client.async_prepare_mqtt()
except (ThinQAPIException, TypeError, ValueError):
_LOGGER.exception("Failed to connect")
return False
async def async_disconnect(self, event: Event | None = None) -> None:
"""Unregister client and disconnects handlers."""
await self.async_end_subscribes()
if self.client is not None:
try:
await self.client.async_disconnect()
except (ThinQAPIException, TypeError, ValueError):
_LOGGER.exception("Failed to disconnect")
def _get_failed_device_count(
self, results: list[dict | BaseException | None]
) -> int:
"""Check if there exists errors while performing tasks and then return count."""
# Note that result code '1207' means 'Already subscribed push'
# and is not actually fail.
return sum(
isinstance(result, (TypeError, ValueError))
or (
isinstance(result, ThinQAPIException)
and result.code != ThinQAPIErrorCodes.ALREADY_SUBSCRIBED_PUSH
)
for result in results
)
async def async_refresh_subscribe(self, now: datetime | None = None) -> None:
"""Update event subscribes."""
_LOGGER.debug("async_refresh_subscribe: now=%s", now)
tasks = [
self.hass.async_create_task(
self.thinq_api.async_post_event_subscribe(coordinator.device_id)
)
for coordinator in self.coordinators.values()
]
if tasks:
results = await asyncio.gather(*tasks, return_exceptions=True)
if (count := self._get_failed_device_count(results)) > 0:
_LOGGER.error("Failed to refresh subscription on %s devices", count)
async def async_start_subscribes(self) -> None:
"""Start push/event subscribes."""
_LOGGER.debug("async_start_subscribes")
if self.client is None:
_LOGGER.error("Failed to start subscription: No client")
return
tasks = [
self.hass.async_create_task(
self.thinq_api.async_post_push_subscribe(coordinator.device_id)
)
for coordinator in self.coordinators.values()
]
tasks.extend(
self.hass.async_create_task(
self.thinq_api.async_post_event_subscribe(coordinator.device_id)
)
for coordinator in self.coordinators.values()
)
if tasks:
results = await asyncio.gather(*tasks, return_exceptions=True)
if (count := self._get_failed_device_count(results)) > 0:
_LOGGER.error("Failed to start subscription on %s devices", count)
await self.client.async_connect_mqtt()
async def async_end_subscribes(self) -> None:
"""Start push/event unsubscribes."""
_LOGGER.debug("async_end_subscribes")
tasks = [
self.hass.async_create_task(
self.thinq_api.async_delete_push_subscribe(coordinator.device_id)
)
for coordinator in self.coordinators.values()
]
tasks.extend(
self.hass.async_create_task(
self.thinq_api.async_delete_event_subscribe(coordinator.device_id)
)
for coordinator in self.coordinators.values()
)
if tasks:
results = await asyncio.gather(*tasks, return_exceptions=True)
if (count := self._get_failed_device_count(results)) > 0:
_LOGGER.error("Failed to end subscription on %s devices", count)
def on_message_received(
self,
topic: str,
payload: bytes,
dup: bool,
qos: Any,
retain: bool,
**kwargs: dict,
) -> None:
"""Handle the received message that matching the topic."""
decoded = payload.decode()
try:
message = json.loads(decoded)
except ValueError:
_LOGGER.error("Failed to parse message: payload=%s", decoded)
return
asyncio.run_coroutine_threadsafe(
self.async_handle_device_event(message), self.hass.loop
).result()
async def async_handle_device_event(self, message: dict) -> None:
"""Handle received mqtt message."""
_LOGGER.debug("async_handle_device_event: message=%s", message)
unique_id = (
f"{message["deviceId"]}_{list(message["report"].keys())[0]}"
if message["deviceType"] == DeviceType.WASHTOWER
else message["deviceId"]
)
coordinator = self.coordinators.get(unique_id)
if coordinator is None:
_LOGGER.error("Failed to handle device event: No device")
return
push_type = message.get("pushType")
if push_type == DEVICE_STATUS_MESSAGE:
coordinator.handle_update_status(message.get("report", {}))
elif push_type == DEVICE_PUSH_MESSAGE:
coordinator.handle_notification_message(message.get("pushCode"))

View File

@ -0,0 +1,214 @@
"""Support for number entities."""
from __future__ import annotations
import logging
from thinqconnect import DeviceType
from thinqconnect.devices.const import Property as ThinQProperty
from thinqconnect.integration import ActiveMode, TimerProperty
from homeassistant.components.number import (
NumberDeviceClass,
NumberEntity,
NumberEntityDescription,
NumberMode,
)
from homeassistant.const import PERCENTAGE, UnitOfTemperature, UnitOfTime
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import ThinqConfigEntry
from .entity import ThinQEntity
NUMBER_DESC: dict[ThinQProperty, NumberEntityDescription] = {
ThinQProperty.FAN_SPEED: NumberEntityDescription(
key=ThinQProperty.FAN_SPEED,
translation_key=ThinQProperty.FAN_SPEED,
),
ThinQProperty.LAMP_BRIGHTNESS: NumberEntityDescription(
key=ThinQProperty.LAMP_BRIGHTNESS,
translation_key=ThinQProperty.LAMP_BRIGHTNESS,
),
ThinQProperty.LIGHT_STATUS: NumberEntityDescription(
key=ThinQProperty.LIGHT_STATUS,
native_unit_of_measurement=PERCENTAGE,
translation_key=ThinQProperty.LIGHT_STATUS,
),
ThinQProperty.TARGET_HUMIDITY: NumberEntityDescription(
key=ThinQProperty.TARGET_HUMIDITY,
device_class=NumberDeviceClass.HUMIDITY,
native_unit_of_measurement=PERCENTAGE,
),
ThinQProperty.TARGET_TEMPERATURE: NumberEntityDescription(
key=ThinQProperty.TARGET_TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
translation_key=ThinQProperty.TARGET_TEMPERATURE,
),
ThinQProperty.WIND_TEMPERATURE: NumberEntityDescription(
key=ThinQProperty.WIND_TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
translation_key=ThinQProperty.WIND_TEMPERATURE,
),
}
TIMER_NUMBER_DESC: dict[ThinQProperty, NumberEntityDescription] = {
ThinQProperty.RELATIVE_HOUR_TO_START: NumberEntityDescription(
key=ThinQProperty.RELATIVE_HOUR_TO_START,
native_unit_of_measurement=UnitOfTime.HOURS,
translation_key=ThinQProperty.RELATIVE_HOUR_TO_START,
),
TimerProperty.RELATIVE_HOUR_TO_START_WM: NumberEntityDescription(
key=ThinQProperty.RELATIVE_HOUR_TO_START,
native_min_value=0,
native_unit_of_measurement=UnitOfTime.HOURS,
translation_key=TimerProperty.RELATIVE_HOUR_TO_START_WM,
),
ThinQProperty.RELATIVE_HOUR_TO_STOP: NumberEntityDescription(
key=ThinQProperty.RELATIVE_HOUR_TO_STOP,
native_unit_of_measurement=UnitOfTime.HOURS,
translation_key=ThinQProperty.RELATIVE_HOUR_TO_STOP,
),
TimerProperty.RELATIVE_HOUR_TO_STOP_WM: NumberEntityDescription(
key=ThinQProperty.RELATIVE_HOUR_TO_STOP,
native_min_value=0,
native_unit_of_measurement=UnitOfTime.HOURS,
translation_key=TimerProperty.RELATIVE_HOUR_TO_STOP_WM,
),
ThinQProperty.SLEEP_TIMER_RELATIVE_HOUR_TO_STOP: NumberEntityDescription(
key=ThinQProperty.SLEEP_TIMER_RELATIVE_HOUR_TO_STOP,
native_unit_of_measurement=UnitOfTime.HOURS,
translation_key=ThinQProperty.SLEEP_TIMER_RELATIVE_HOUR_TO_STOP,
),
}
WASHER_NUMBERS: tuple[NumberEntityDescription, ...] = (
TIMER_NUMBER_DESC[TimerProperty.RELATIVE_HOUR_TO_START_WM],
TIMER_NUMBER_DESC[TimerProperty.RELATIVE_HOUR_TO_STOP_WM],
)
DEVICE_TYPE_NUMBER_MAP: dict[DeviceType, tuple[NumberEntityDescription, ...]] = {
DeviceType.AIR_CONDITIONER: (
TIMER_NUMBER_DESC[ThinQProperty.RELATIVE_HOUR_TO_START],
TIMER_NUMBER_DESC[ThinQProperty.RELATIVE_HOUR_TO_STOP],
TIMER_NUMBER_DESC[ThinQProperty.SLEEP_TIMER_RELATIVE_HOUR_TO_STOP],
),
DeviceType.AIR_PURIFIER_FAN: (
NUMBER_DESC[ThinQProperty.WIND_TEMPERATURE],
TIMER_NUMBER_DESC[ThinQProperty.SLEEP_TIMER_RELATIVE_HOUR_TO_STOP],
),
DeviceType.DRYER: WASHER_NUMBERS,
DeviceType.HOOD: (
NUMBER_DESC[ThinQProperty.LAMP_BRIGHTNESS],
NUMBER_DESC[ThinQProperty.FAN_SPEED],
),
DeviceType.HUMIDIFIER: (
NUMBER_DESC[ThinQProperty.TARGET_HUMIDITY],
TIMER_NUMBER_DESC[ThinQProperty.SLEEP_TIMER_RELATIVE_HOUR_TO_STOP],
),
DeviceType.MICROWAVE_OVEN: (
NUMBER_DESC[ThinQProperty.LAMP_BRIGHTNESS],
NUMBER_DESC[ThinQProperty.FAN_SPEED],
),
DeviceType.OVEN: (NUMBER_DESC[ThinQProperty.TARGET_TEMPERATURE],),
DeviceType.REFRIGERATOR: (NUMBER_DESC[ThinQProperty.TARGET_TEMPERATURE],),
DeviceType.STYLER: (TIMER_NUMBER_DESC[TimerProperty.RELATIVE_HOUR_TO_STOP_WM],),
DeviceType.WASHCOMBO_MAIN: WASHER_NUMBERS,
DeviceType.WASHCOMBO_MINI: WASHER_NUMBERS,
DeviceType.WASHER: WASHER_NUMBERS,
DeviceType.WASHTOWER_DRYER: WASHER_NUMBERS,
DeviceType.WASHTOWER: WASHER_NUMBERS,
DeviceType.WASHTOWER_WASHER: WASHER_NUMBERS,
DeviceType.WATER_HEATER: (NUMBER_DESC[ThinQProperty.TARGET_TEMPERATURE],),
DeviceType.WINE_CELLAR: (
NUMBER_DESC[ThinQProperty.LIGHT_STATUS],
NUMBER_DESC[ThinQProperty.TARGET_TEMPERATURE],
),
}
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
entry: ThinqConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up an entry for number platform."""
entities: list[ThinQNumberEntity] = []
for coordinator in entry.runtime_data.coordinators.values():
if (
descriptions := DEVICE_TYPE_NUMBER_MAP.get(
coordinator.api.device.device_type
)
) is not None:
for description in descriptions:
entities.extend(
ThinQNumberEntity(coordinator, description, property_id)
for property_id in coordinator.api.get_active_idx(
description.key, ActiveMode.READ_WRITE
)
)
if entities:
async_add_entities(entities)
class ThinQNumberEntity(ThinQEntity, NumberEntity):
"""Represent a thinq number platform."""
_attr_mode = NumberMode.BOX
def _update_status(self) -> None:
"""Update status itself."""
super()._update_status()
self._attr_native_value = self.data.value
# Update unit.
if (
unit_of_measurement := self._get_unit_of_measurement(self.data.unit)
) is not None:
self._attr_native_unit_of_measurement = unit_of_measurement
# Undate range.
if (
self.entity_description.native_min_value is None
and (min_value := self.data.min) is not None
):
self._attr_native_min_value = min_value
if (
self.entity_description.native_max_value is None
and (max_value := self.data.max) is not None
):
self._attr_native_max_value = max_value
if (
self.entity_description.native_step is None
and (step := self.data.step) is not None
):
self._attr_native_step = step
_LOGGER.debug(
"[%s:%s] update status: %s -> %s, unit:%s, min:%s, max:%s, step:%s",
self.coordinator.device_name,
self.property_id,
self.data.value,
self.native_value,
self.native_unit_of_measurement,
self.native_min_value,
self.native_max_value,
self.native_step,
)
async def async_set_native_value(self, value: float) -> None:
"""Change to new number value."""
if self.step.is_integer():
value = int(value)
_LOGGER.debug(
"[%s:%s] async_set_native_value: %s",
self.coordinator.device_name,
self.property_id,
value,
)
await self.async_call_api(self.coordinator.api.post(self.property_id, value))

View File

@ -0,0 +1,207 @@
"""Support for select entities."""
from __future__ import annotations
import logging
from thinqconnect import DeviceType
from thinqconnect.devices.const import Property as ThinQProperty
from thinqconnect.integration import ActiveMode
from homeassistant.components.select import SelectEntity, SelectEntityDescription
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import ThinqConfigEntry
from .coordinator import DeviceDataUpdateCoordinator
from .entity import ThinQEntity
SELECT_DESC: dict[ThinQProperty, SelectEntityDescription] = {
ThinQProperty.MONITORING_ENABLED: SelectEntityDescription(
key=ThinQProperty.MONITORING_ENABLED,
translation_key=ThinQProperty.MONITORING_ENABLED,
),
ThinQProperty.COOK_MODE: SelectEntityDescription(
key=ThinQProperty.COOK_MODE,
translation_key=ThinQProperty.COOK_MODE,
),
ThinQProperty.DISPLAY_LIGHT: SelectEntityDescription(
key=ThinQProperty.DISPLAY_LIGHT,
translation_key=ThinQProperty.DISPLAY_LIGHT,
),
ThinQProperty.CURRENT_JOB_MODE: SelectEntityDescription(
key=ThinQProperty.CURRENT_JOB_MODE,
translation_key=ThinQProperty.CURRENT_JOB_MODE,
),
ThinQProperty.FRESH_AIR_FILTER: SelectEntityDescription(
key=ThinQProperty.FRESH_AIR_FILTER,
translation_key=ThinQProperty.FRESH_AIR_FILTER,
),
}
AIR_FLOW_SELECT_DESC: dict[ThinQProperty, SelectEntityDescription] = {
ThinQProperty.WIND_STRENGTH: SelectEntityDescription(
key=ThinQProperty.WIND_STRENGTH,
translation_key=ThinQProperty.WIND_STRENGTH,
),
ThinQProperty.WIND_ANGLE: SelectEntityDescription(
key=ThinQProperty.WIND_ANGLE,
translation_key=ThinQProperty.WIND_ANGLE,
),
}
OPERATION_SELECT_DESC: dict[ThinQProperty, SelectEntityDescription] = {
ThinQProperty.AIR_CLEAN_OPERATION_MODE: SelectEntityDescription(
key=ThinQProperty.AIR_CLEAN_OPERATION_MODE,
translation_key="air_clean_operation_mode",
),
ThinQProperty.DISH_WASHER_OPERATION_MODE: SelectEntityDescription(
key=ThinQProperty.DISH_WASHER_OPERATION_MODE,
translation_key="operation_mode",
),
ThinQProperty.DRYER_OPERATION_MODE: SelectEntityDescription(
key=ThinQProperty.DRYER_OPERATION_MODE,
translation_key="operation_mode",
),
ThinQProperty.HYGIENE_DRY_MODE: SelectEntityDescription(
key=ThinQProperty.HYGIENE_DRY_MODE,
translation_key=ThinQProperty.HYGIENE_DRY_MODE,
),
ThinQProperty.LIGHT_BRIGHTNESS: SelectEntityDescription(
key=ThinQProperty.LIGHT_BRIGHTNESS,
translation_key=ThinQProperty.LIGHT_BRIGHTNESS,
),
ThinQProperty.OVEN_OPERATION_MODE: SelectEntityDescription(
key=ThinQProperty.OVEN_OPERATION_MODE,
translation_key="operation_mode",
),
ThinQProperty.STYLER_OPERATION_MODE: SelectEntityDescription(
key=ThinQProperty.STYLER_OPERATION_MODE,
translation_key="operation_mode",
),
ThinQProperty.WASHER_OPERATION_MODE: SelectEntityDescription(
key=ThinQProperty.WASHER_OPERATION_MODE,
translation_key="operation_mode",
),
}
DEVICE_TYPE_SELECT_MAP: dict[DeviceType, tuple[SelectEntityDescription, ...]] = {
DeviceType.AIR_CONDITIONER: (
SELECT_DESC[ThinQProperty.MONITORING_ENABLED],
OPERATION_SELECT_DESC[ThinQProperty.AIR_CLEAN_OPERATION_MODE],
),
DeviceType.AIR_PURIFIER_FAN: (
AIR_FLOW_SELECT_DESC[ThinQProperty.WIND_STRENGTH],
AIR_FLOW_SELECT_DESC[ThinQProperty.WIND_ANGLE],
SELECT_DESC[ThinQProperty.DISPLAY_LIGHT],
SELECT_DESC[ThinQProperty.CURRENT_JOB_MODE],
),
DeviceType.AIR_PURIFIER: (
AIR_FLOW_SELECT_DESC[ThinQProperty.WIND_STRENGTH],
SELECT_DESC[ThinQProperty.CURRENT_JOB_MODE],
),
DeviceType.DEHUMIDIFIER: (AIR_FLOW_SELECT_DESC[ThinQProperty.WIND_STRENGTH],),
DeviceType.DISH_WASHER: (
OPERATION_SELECT_DESC[ThinQProperty.DISH_WASHER_OPERATION_MODE],
),
DeviceType.DRYER: (OPERATION_SELECT_DESC[ThinQProperty.DRYER_OPERATION_MODE],),
DeviceType.HUMIDIFIER: (
AIR_FLOW_SELECT_DESC[ThinQProperty.WIND_STRENGTH],
SELECT_DESC[ThinQProperty.DISPLAY_LIGHT],
SELECT_DESC[ThinQProperty.CURRENT_JOB_MODE],
OPERATION_SELECT_DESC[ThinQProperty.HYGIENE_DRY_MODE],
),
DeviceType.OVEN: (
SELECT_DESC[ThinQProperty.COOK_MODE],
OPERATION_SELECT_DESC[ThinQProperty.OVEN_OPERATION_MODE],
),
DeviceType.REFRIGERATOR: (SELECT_DESC[ThinQProperty.FRESH_AIR_FILTER],),
DeviceType.STYLER: (OPERATION_SELECT_DESC[ThinQProperty.STYLER_OPERATION_MODE],),
DeviceType.WASHCOMBO_MAIN: (
OPERATION_SELECT_DESC[ThinQProperty.WASHER_OPERATION_MODE],
),
DeviceType.WASHCOMBO_MINI: (
OPERATION_SELECT_DESC[ThinQProperty.WASHER_OPERATION_MODE],
),
DeviceType.WASHER: (OPERATION_SELECT_DESC[ThinQProperty.WASHER_OPERATION_MODE],),
DeviceType.WASHTOWER_DRYER: (
OPERATION_SELECT_DESC[ThinQProperty.WASHER_OPERATION_MODE],
),
DeviceType.WASHTOWER: (
OPERATION_SELECT_DESC[ThinQProperty.DRYER_OPERATION_MODE],
OPERATION_SELECT_DESC[ThinQProperty.WASHER_OPERATION_MODE],
),
DeviceType.WASHTOWER_WASHER: (
OPERATION_SELECT_DESC[ThinQProperty.WASHER_OPERATION_MODE],
),
DeviceType.WATER_HEATER: (SELECT_DESC[ThinQProperty.CURRENT_JOB_MODE],),
DeviceType.WINE_CELLAR: (OPERATION_SELECT_DESC[ThinQProperty.LIGHT_BRIGHTNESS],),
}
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
entry: ThinqConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up an entry for select platform."""
entities: list[ThinQSelectEntity] = []
for coordinator in entry.runtime_data.coordinators.values():
if (
descriptions := DEVICE_TYPE_SELECT_MAP.get(
coordinator.api.device.device_type
)
) is not None:
for description in descriptions:
entities.extend(
ThinQSelectEntity(coordinator, description, property_id)
for property_id in coordinator.api.get_active_idx(
description.key, ActiveMode.WRITABLE
)
)
if entities:
async_add_entities(entities)
class ThinQSelectEntity(ThinQEntity, SelectEntity):
"""Represent a thinq select platform."""
def __init__(
self,
coordinator: DeviceDataUpdateCoordinator,
entity_description: SelectEntityDescription,
property_id: str,
) -> None:
"""Initialize a select entity."""
super().__init__(coordinator, entity_description, property_id)
self._attr_options = self.data.options if self.data.options is not None else []
def _update_status(self) -> None:
"""Update status itself."""
super()._update_status()
if self.data.value:
self._attr_current_option = str(self.data.value)
else:
self._attr_current_option = None
_LOGGER.debug(
"[%s:%s] update status: %s -> %s, options:%s",
self.coordinator.device_name,
self.property_id,
self.data.value,
self.current_option,
self.options,
)
async def async_select_option(self, option: str) -> None:
"""Change the selected option."""
_LOGGER.debug(
"[%s:%s] async_select_option: %s",
self.coordinator.device_name,
self.property_id,
option,
)
await self.async_call_api(self.coordinator.api.post(self.property_id, option))

View File

@ -0,0 +1,529 @@
"""Support for sensor entities."""
from __future__ import annotations
import logging
from thinqconnect import DeviceType
from thinqconnect.devices.const import Property as ThinQProperty
from thinqconnect.integration import ActiveMode, ThinQPropertyEx, TimerProperty
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import (
CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
PERCENTAGE,
UnitOfTemperature,
UnitOfTime,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import ThinqConfigEntry
from .coordinator import DeviceDataUpdateCoordinator
from .entity import ThinQEntity
AIR_QUALITY_SENSOR_DESC: dict[ThinQProperty, SensorEntityDescription] = {
ThinQProperty.PM1: SensorEntityDescription(
key=ThinQProperty.PM1,
device_class=SensorDeviceClass.PM1,
native_unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
state_class=SensorStateClass.MEASUREMENT,
),
ThinQProperty.PM2: SensorEntityDescription(
key=ThinQProperty.PM2,
device_class=SensorDeviceClass.PM25,
native_unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
state_class=SensorStateClass.MEASUREMENT,
),
ThinQProperty.PM10: SensorEntityDescription(
key=ThinQProperty.PM10,
device_class=SensorDeviceClass.PM10,
native_unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
state_class=SensorStateClass.MEASUREMENT,
),
ThinQProperty.HUMIDITY: SensorEntityDescription(
key=ThinQProperty.HUMIDITY,
device_class=SensorDeviceClass.HUMIDITY,
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
),
ThinQProperty.MONITORING_ENABLED: SensorEntityDescription(
key=ThinQProperty.MONITORING_ENABLED,
device_class=SensorDeviceClass.ENUM,
translation_key=ThinQProperty.MONITORING_ENABLED,
),
ThinQProperty.TEMPERATURE: SensorEntityDescription(
key=ThinQProperty.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
translation_key=ThinQProperty.TEMPERATURE,
),
ThinQProperty.ODOR_LEVEL: SensorEntityDescription(
key=ThinQProperty.ODOR_LEVEL,
device_class=SensorDeviceClass.ENUM,
translation_key=ThinQProperty.ODOR_LEVEL,
),
ThinQProperty.TOTAL_POLLUTION_LEVEL: SensorEntityDescription(
key=ThinQProperty.TOTAL_POLLUTION_LEVEL,
device_class=SensorDeviceClass.ENUM,
translation_key=ThinQProperty.TOTAL_POLLUTION_LEVEL,
),
}
BATTERY_SENSOR_DESC: dict[ThinQProperty, SensorEntityDescription] = {
ThinQProperty.BATTERY_PERCENT: SensorEntityDescription(
key=ThinQProperty.BATTERY_PERCENT,
translation_key=ThinQProperty.BATTERY_LEVEL,
),
}
DISH_WASHING_COURSE_SENSOR_DESC: dict[ThinQProperty, SensorEntityDescription] = {
ThinQProperty.CURRENT_DISH_WASHING_COURSE: SensorEntityDescription(
key=ThinQProperty.CURRENT_DISH_WASHING_COURSE,
device_class=SensorDeviceClass.ENUM,
translation_key=ThinQProperty.CURRENT_DISH_WASHING_COURSE,
)
}
FILTER_INFO_SENSOR_DESC: dict[ThinQProperty, SensorEntityDescription] = {
ThinQProperty.FILTER_LIFETIME: SensorEntityDescription(
key=ThinQProperty.FILTER_LIFETIME,
native_unit_of_measurement=UnitOfTime.HOURS,
translation_key=ThinQProperty.FILTER_LIFETIME,
),
}
HUMIDITY_SENSOR_DESC: dict[ThinQProperty, SensorEntityDescription] = {
ThinQProperty.CURRENT_HUMIDITY: SensorEntityDescription(
key=ThinQProperty.CURRENT_HUMIDITY,
device_class=SensorDeviceClass.HUMIDITY,
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
)
}
JOB_MODE_SENSOR_DESC: dict[ThinQProperty, SensorEntityDescription] = {
ThinQProperty.CURRENT_JOB_MODE: SensorEntityDescription(
key=ThinQProperty.CURRENT_JOB_MODE,
device_class=SensorDeviceClass.ENUM,
translation_key=ThinQProperty.CURRENT_JOB_MODE,
),
ThinQPropertyEx.CURRENT_JOB_MODE_STICK_CLEANER: SensorEntityDescription(
key=ThinQProperty.CURRENT_JOB_MODE,
device_class=SensorDeviceClass.ENUM,
translation_key=ThinQPropertyEx.CURRENT_JOB_MODE_STICK_CLEANER,
),
ThinQProperty.PERSONALIZATION_MODE: SensorEntityDescription(
key=ThinQProperty.PERSONALIZATION_MODE,
device_class=SensorDeviceClass.ENUM,
translation_key=ThinQProperty.PERSONALIZATION_MODE,
),
}
LIGHT_SENSOR_DESC: dict[ThinQProperty, SensorEntityDescription] = {
ThinQProperty.BRIGHTNESS: SensorEntityDescription(
key=ThinQProperty.BRIGHTNESS,
translation_key=ThinQProperty.BRIGHTNESS,
),
ThinQProperty.DURATION: SensorEntityDescription(
key=ThinQProperty.DURATION,
native_unit_of_measurement=UnitOfTime.HOURS,
translation_key=ThinQProperty.DURATION,
),
}
POWER_SENSOR_DESC: dict[ThinQProperty, SensorEntityDescription] = {
ThinQProperty.POWER_LEVEL: SensorEntityDescription(
key=ThinQProperty.POWER_LEVEL,
translation_key=ThinQProperty.POWER_LEVEL,
)
}
PREFERENCE_SENSOR_DESC: dict[ThinQProperty, SensorEntityDescription] = {
ThinQProperty.RINSE_LEVEL: SensorEntityDescription(
key=ThinQProperty.RINSE_LEVEL,
device_class=SensorDeviceClass.ENUM,
translation_key=ThinQProperty.RINSE_LEVEL,
),
ThinQProperty.SOFTENING_LEVEL: SensorEntityDescription(
key=ThinQProperty.SOFTENING_LEVEL,
device_class=SensorDeviceClass.ENUM,
translation_key=ThinQProperty.SOFTENING_LEVEL,
),
}
RECIPE_SENSOR_DESC: dict[ThinQProperty, SensorEntityDescription] = {
ThinQProperty.RECIPE_NAME: SensorEntityDescription(
key=ThinQProperty.RECIPE_NAME,
device_class=SensorDeviceClass.ENUM,
translation_key=ThinQProperty.RECIPE_NAME,
),
ThinQProperty.WORT_INFO: SensorEntityDescription(
key=ThinQProperty.WORT_INFO,
device_class=SensorDeviceClass.ENUM,
translation_key=ThinQProperty.WORT_INFO,
),
ThinQProperty.YEAST_INFO: SensorEntityDescription(
key=ThinQProperty.YEAST_INFO,
device_class=SensorDeviceClass.ENUM,
translation_key=ThinQProperty.YEAST_INFO,
),
ThinQProperty.HOP_OIL_INFO: SensorEntityDescription(
key=ThinQProperty.HOP_OIL_INFO,
translation_key=ThinQProperty.HOP_OIL_INFO,
),
ThinQProperty.FLAVOR_INFO: SensorEntityDescription(
key=ThinQProperty.FLAVOR_INFO,
translation_key=ThinQProperty.FLAVOR_INFO,
),
ThinQProperty.BEER_REMAIN: SensorEntityDescription(
key=ThinQProperty.BEER_REMAIN,
native_unit_of_measurement=PERCENTAGE,
translation_key=ThinQProperty.BEER_REMAIN,
),
}
REFRIGERATION_SENSOR_DESC: dict[ThinQProperty, SensorEntityDescription] = {
ThinQProperty.FRESH_AIR_FILTER: SensorEntityDescription(
key=ThinQProperty.FRESH_AIR_FILTER,
device_class=SensorDeviceClass.ENUM,
translation_key=ThinQProperty.FRESH_AIR_FILTER,
),
}
RUN_STATE_SENSOR_DESC: dict[ThinQProperty, SensorEntityDescription] = {
ThinQProperty.CURRENT_STATE: SensorEntityDescription(
key=ThinQProperty.CURRENT_STATE,
device_class=SensorDeviceClass.ENUM,
translation_key=ThinQProperty.CURRENT_STATE,
),
ThinQProperty.COCK_STATE: SensorEntityDescription(
key=ThinQProperty.COCK_STATE,
device_class=SensorDeviceClass.ENUM,
translation_key=ThinQProperty.COCK_STATE,
),
ThinQProperty.STERILIZING_STATE: SensorEntityDescription(
key=ThinQProperty.STERILIZING_STATE,
device_class=SensorDeviceClass.ENUM,
translation_key=ThinQProperty.STERILIZING_STATE,
),
ThinQProperty.GROWTH_MODE: SensorEntityDescription(
key=ThinQProperty.GROWTH_MODE,
device_class=SensorDeviceClass.ENUM,
translation_key=ThinQProperty.GROWTH_MODE,
),
ThinQProperty.WIND_VOLUME: SensorEntityDescription(
key=ThinQProperty.WIND_VOLUME,
device_class=SensorDeviceClass.WIND_SPEED,
translation_key=ThinQProperty.WIND_VOLUME,
),
}
TEMPERATURE_SENSOR_DESC: dict[ThinQProperty, SensorEntityDescription] = {
ThinQProperty.TARGET_TEMPERATURE: SensorEntityDescription(
key=ThinQProperty.TARGET_TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
translation_key=ThinQProperty.TARGET_TEMPERATURE,
),
ThinQProperty.DAY_TARGET_TEMPERATURE: SensorEntityDescription(
key=ThinQProperty.DAY_TARGET_TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
translation_key=ThinQProperty.DAY_TARGET_TEMPERATURE,
),
ThinQProperty.NIGHT_TARGET_TEMPERATURE: SensorEntityDescription(
key=ThinQProperty.NIGHT_TARGET_TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
translation_key=ThinQProperty.NIGHT_TARGET_TEMPERATURE,
),
ThinQProperty.TEMPERATURE_STATE: SensorEntityDescription(
key=ThinQProperty.TEMPERATURE_STATE,
device_class=SensorDeviceClass.ENUM,
translation_key=ThinQProperty.TEMPERATURE_STATE,
),
ThinQProperty.CURRENT_TEMPERATURE: SensorEntityDescription(
key=ThinQProperty.CURRENT_TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
translation_key=ThinQProperty.CURRENT_TEMPERATURE,
),
}
WATER_FILTER_INFO_SENSOR_DESC: dict[ThinQProperty, SensorEntityDescription] = {
ThinQProperty.USED_TIME: SensorEntityDescription(
key=ThinQProperty.USED_TIME,
native_unit_of_measurement=UnitOfTime.MONTHS,
translation_key=ThinQProperty.USED_TIME,
),
}
WATER_INFO_SENSOR_DESC: dict[ThinQProperty, SensorEntityDescription] = {
ThinQProperty.WATER_TYPE: SensorEntityDescription(
key=ThinQProperty.WATER_TYPE,
translation_key=ThinQProperty.WATER_TYPE,
),
}
TIMER_SENSOR_DESC: dict[ThinQProperty, SensorEntityDescription] = {
TimerProperty.RELATIVE_TO_START: SensorEntityDescription(
key=TimerProperty.RELATIVE_TO_START,
translation_key=TimerProperty.RELATIVE_TO_START,
),
TimerProperty.RELATIVE_TO_START_WM: SensorEntityDescription(
key=TimerProperty.RELATIVE_TO_START,
translation_key=TimerProperty.RELATIVE_TO_START_WM,
),
TimerProperty.RELATIVE_TO_STOP: SensorEntityDescription(
key=TimerProperty.RELATIVE_TO_STOP,
translation_key=TimerProperty.RELATIVE_TO_STOP,
),
TimerProperty.RELATIVE_TO_STOP_WM: SensorEntityDescription(
key=TimerProperty.RELATIVE_TO_STOP,
translation_key=TimerProperty.RELATIVE_TO_STOP_WM,
),
TimerProperty.SLEEP_TIMER_RELATIVE_TO_STOP: SensorEntityDescription(
key=TimerProperty.SLEEP_TIMER_RELATIVE_TO_STOP,
translation_key=TimerProperty.SLEEP_TIMER_RELATIVE_TO_STOP,
),
TimerProperty.ABSOLUTE_TO_START: SensorEntityDescription(
key=TimerProperty.ABSOLUTE_TO_START,
translation_key=TimerProperty.ABSOLUTE_TO_START,
),
TimerProperty.ABSOLUTE_TO_STOP: SensorEntityDescription(
key=TimerProperty.ABSOLUTE_TO_STOP,
translation_key=TimerProperty.ABSOLUTE_TO_STOP,
),
TimerProperty.REMAIN: SensorEntityDescription(
key=TimerProperty.REMAIN,
translation_key=TimerProperty.REMAIN,
),
TimerProperty.TARGET: SensorEntityDescription(
key=TimerProperty.TARGET,
translation_key=TimerProperty.TARGET,
),
TimerProperty.RUNNING: SensorEntityDescription(
key=TimerProperty.RUNNING,
translation_key=TimerProperty.RUNNING,
),
TimerProperty.TOTAL: SensorEntityDescription(
key=TimerProperty.TOTAL,
translation_key=TimerProperty.TOTAL,
),
TimerProperty.LIGHT_START: SensorEntityDescription(
key=TimerProperty.LIGHT_START,
translation_key=TimerProperty.LIGHT_START,
),
ThinQProperty.ELAPSED_DAY_STATE: SensorEntityDescription(
key=ThinQProperty.ELAPSED_DAY_STATE,
native_unit_of_measurement=UnitOfTime.DAYS,
translation_key=ThinQProperty.ELAPSED_DAY_STATE,
),
ThinQProperty.ELAPSED_DAY_TOTAL: SensorEntityDescription(
key=ThinQProperty.ELAPSED_DAY_TOTAL,
native_unit_of_measurement=UnitOfTime.DAYS,
translation_key=ThinQProperty.ELAPSED_DAY_TOTAL,
),
}
WASHER_SENSORS: tuple[SensorEntityDescription, ...] = (
RUN_STATE_SENSOR_DESC[ThinQProperty.CURRENT_STATE],
TIMER_SENSOR_DESC[TimerProperty.RELATIVE_TO_START_WM],
TIMER_SENSOR_DESC[TimerProperty.RELATIVE_TO_STOP_WM],
TIMER_SENSOR_DESC[TimerProperty.REMAIN],
TIMER_SENSOR_DESC[TimerProperty.TOTAL],
)
DEVICE_TYPE_SENSOR_MAP: dict[DeviceType, tuple[SensorEntityDescription, ...]] = {
DeviceType.AIR_CONDITIONER: (
AIR_QUALITY_SENSOR_DESC[ThinQProperty.PM1],
AIR_QUALITY_SENSOR_DESC[ThinQProperty.PM2],
AIR_QUALITY_SENSOR_DESC[ThinQProperty.PM10],
AIR_QUALITY_SENSOR_DESC[ThinQProperty.HUMIDITY],
AIR_QUALITY_SENSOR_DESC[ThinQProperty.ODOR_LEVEL],
AIR_QUALITY_SENSOR_DESC[ThinQProperty.TOTAL_POLLUTION_LEVEL],
FILTER_INFO_SENSOR_DESC[ThinQProperty.FILTER_LIFETIME],
TIMER_SENSOR_DESC[TimerProperty.RELATIVE_TO_START],
TIMER_SENSOR_DESC[TimerProperty.RELATIVE_TO_STOP],
TIMER_SENSOR_DESC[TimerProperty.SLEEP_TIMER_RELATIVE_TO_STOP],
),
DeviceType.AIR_PURIFIER_FAN: (
AIR_QUALITY_SENSOR_DESC[ThinQProperty.PM1],
AIR_QUALITY_SENSOR_DESC[ThinQProperty.PM2],
AIR_QUALITY_SENSOR_DESC[ThinQProperty.PM10],
AIR_QUALITY_SENSOR_DESC[ThinQProperty.HUMIDITY],
AIR_QUALITY_SENSOR_DESC[ThinQProperty.TEMPERATURE],
AIR_QUALITY_SENSOR_DESC[ThinQProperty.MONITORING_ENABLED],
AIR_QUALITY_SENSOR_DESC[ThinQProperty.ODOR_LEVEL],
AIR_QUALITY_SENSOR_DESC[ThinQProperty.TOTAL_POLLUTION_LEVEL],
TIMER_SENSOR_DESC[TimerProperty.SLEEP_TIMER_RELATIVE_TO_STOP],
),
DeviceType.AIR_PURIFIER: (
AIR_QUALITY_SENSOR_DESC[ThinQProperty.PM1],
AIR_QUALITY_SENSOR_DESC[ThinQProperty.PM2],
AIR_QUALITY_SENSOR_DESC[ThinQProperty.PM10],
AIR_QUALITY_SENSOR_DESC[ThinQProperty.HUMIDITY],
AIR_QUALITY_SENSOR_DESC[ThinQProperty.MONITORING_ENABLED],
AIR_QUALITY_SENSOR_DESC[ThinQProperty.ODOR_LEVEL],
AIR_QUALITY_SENSOR_DESC[ThinQProperty.TOTAL_POLLUTION_LEVEL],
JOB_MODE_SENSOR_DESC[ThinQProperty.CURRENT_JOB_MODE],
JOB_MODE_SENSOR_DESC[ThinQProperty.PERSONALIZATION_MODE],
),
DeviceType.COOKTOP: (
RUN_STATE_SENSOR_DESC[ThinQProperty.CURRENT_STATE],
POWER_SENSOR_DESC[ThinQProperty.POWER_LEVEL],
TIMER_SENSOR_DESC[TimerProperty.REMAIN],
),
DeviceType.DEHUMIDIFIER: (
JOB_MODE_SENSOR_DESC[ThinQProperty.CURRENT_JOB_MODE],
HUMIDITY_SENSOR_DESC[ThinQProperty.CURRENT_HUMIDITY],
),
DeviceType.DISH_WASHER: (
DISH_WASHING_COURSE_SENSOR_DESC[ThinQProperty.CURRENT_DISH_WASHING_COURSE],
PREFERENCE_SENSOR_DESC[ThinQProperty.RINSE_LEVEL],
PREFERENCE_SENSOR_DESC[ThinQProperty.SOFTENING_LEVEL],
RUN_STATE_SENSOR_DESC[ThinQProperty.CURRENT_STATE],
TIMER_SENSOR_DESC[TimerProperty.RELATIVE_TO_START_WM],
TIMER_SENSOR_DESC[TimerProperty.REMAIN],
TIMER_SENSOR_DESC[TimerProperty.TOTAL],
),
DeviceType.DRYER: WASHER_SENSORS,
DeviceType.HOME_BREW: (
RECIPE_SENSOR_DESC[ThinQProperty.RECIPE_NAME],
RECIPE_SENSOR_DESC[ThinQProperty.WORT_INFO],
RECIPE_SENSOR_DESC[ThinQProperty.YEAST_INFO],
RECIPE_SENSOR_DESC[ThinQProperty.HOP_OIL_INFO],
RECIPE_SENSOR_DESC[ThinQProperty.FLAVOR_INFO],
RECIPE_SENSOR_DESC[ThinQProperty.BEER_REMAIN],
RUN_STATE_SENSOR_DESC[ThinQProperty.CURRENT_STATE],
TIMER_SENSOR_DESC[ThinQProperty.ELAPSED_DAY_STATE],
TIMER_SENSOR_DESC[ThinQProperty.ELAPSED_DAY_TOTAL],
),
DeviceType.HOOD: (TIMER_SENSOR_DESC[TimerProperty.REMAIN],),
DeviceType.HUMIDIFIER: (
AIR_QUALITY_SENSOR_DESC[ThinQProperty.PM1],
AIR_QUALITY_SENSOR_DESC[ThinQProperty.PM2],
AIR_QUALITY_SENSOR_DESC[ThinQProperty.PM10],
AIR_QUALITY_SENSOR_DESC[ThinQProperty.HUMIDITY],
AIR_QUALITY_SENSOR_DESC[ThinQProperty.TEMPERATURE],
AIR_QUALITY_SENSOR_DESC[ThinQProperty.MONITORING_ENABLED],
AIR_QUALITY_SENSOR_DESC[ThinQProperty.TOTAL_POLLUTION_LEVEL],
TIMER_SENSOR_DESC[TimerProperty.ABSOLUTE_TO_START],
TIMER_SENSOR_DESC[TimerProperty.ABSOLUTE_TO_STOP],
TIMER_SENSOR_DESC[TimerProperty.SLEEP_TIMER_RELATIVE_TO_STOP],
),
DeviceType.KIMCHI_REFRIGERATOR: (
REFRIGERATION_SENSOR_DESC[ThinQProperty.FRESH_AIR_FILTER],
SensorEntityDescription(
key=ThinQProperty.TARGET_TEMPERATURE,
translation_key=ThinQProperty.TARGET_TEMPERATURE,
),
),
DeviceType.MICROWAVE_OVEN: (
RUN_STATE_SENSOR_DESC[ThinQProperty.CURRENT_STATE],
TIMER_SENSOR_DESC[TimerProperty.REMAIN],
),
DeviceType.OVEN: (
RUN_STATE_SENSOR_DESC[ThinQProperty.CURRENT_STATE],
TEMPERATURE_SENSOR_DESC[ThinQProperty.TARGET_TEMPERATURE],
TIMER_SENSOR_DESC[TimerProperty.REMAIN],
TIMER_SENSOR_DESC[TimerProperty.TARGET],
),
DeviceType.PLANT_CULTIVATOR: (
LIGHT_SENSOR_DESC[ThinQProperty.BRIGHTNESS],
LIGHT_SENSOR_DESC[ThinQProperty.DURATION],
RUN_STATE_SENSOR_DESC[ThinQProperty.CURRENT_STATE],
RUN_STATE_SENSOR_DESC[ThinQProperty.GROWTH_MODE],
RUN_STATE_SENSOR_DESC[ThinQProperty.WIND_VOLUME],
TEMPERATURE_SENSOR_DESC[ThinQProperty.DAY_TARGET_TEMPERATURE],
TEMPERATURE_SENSOR_DESC[ThinQProperty.NIGHT_TARGET_TEMPERATURE],
TEMPERATURE_SENSOR_DESC[ThinQProperty.TEMPERATURE_STATE],
TIMER_SENSOR_DESC[TimerProperty.LIGHT_START],
),
DeviceType.REFRIGERATOR: (
REFRIGERATION_SENSOR_DESC[ThinQProperty.FRESH_AIR_FILTER],
WATER_FILTER_INFO_SENSOR_DESC[ThinQProperty.USED_TIME],
),
DeviceType.ROBOT_CLEANER: (
RUN_STATE_SENSOR_DESC[ThinQProperty.CURRENT_STATE],
JOB_MODE_SENSOR_DESC[ThinQProperty.CURRENT_JOB_MODE],
TIMER_SENSOR_DESC[TimerProperty.RUNNING],
),
DeviceType.STICK_CLEANER: (
BATTERY_SENSOR_DESC[ThinQProperty.BATTERY_PERCENT],
JOB_MODE_SENSOR_DESC[ThinQPropertyEx.CURRENT_JOB_MODE_STICK_CLEANER],
RUN_STATE_SENSOR_DESC[ThinQProperty.CURRENT_STATE],
),
DeviceType.STYLER: WASHER_SENSORS,
DeviceType.WASHCOMBO_MAIN: WASHER_SENSORS,
DeviceType.WASHCOMBO_MINI: WASHER_SENSORS,
DeviceType.WASHER: WASHER_SENSORS,
DeviceType.WASHTOWER_DRYER: WASHER_SENSORS,
DeviceType.WASHTOWER: WASHER_SENSORS,
DeviceType.WASHTOWER_WASHER: WASHER_SENSORS,
DeviceType.WATER_HEATER: (
TEMPERATURE_SENSOR_DESC[ThinQProperty.CURRENT_TEMPERATURE],
),
DeviceType.WATER_PURIFIER: (
RUN_STATE_SENSOR_DESC[ThinQProperty.COCK_STATE],
RUN_STATE_SENSOR_DESC[ThinQProperty.STERILIZING_STATE],
WATER_INFO_SENSOR_DESC[ThinQProperty.WATER_TYPE],
),
}
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
entry: ThinqConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up an entry for sensor platform."""
entities: list[ThinQSensorEntity] = []
for coordinator in entry.runtime_data.coordinators.values():
if (
descriptions := DEVICE_TYPE_SENSOR_MAP.get(
coordinator.api.device.device_type
)
) is not None:
for description in descriptions:
entities.extend(
ThinQSensorEntity(coordinator, description, property_id)
for property_id in coordinator.api.get_active_idx(
description.key,
(
ActiveMode.READABLE
if coordinator.api.device.device_type == DeviceType.COOKTOP
else ActiveMode.READ_ONLY
),
)
)
if entities:
async_add_entities(entities)
class ThinQSensorEntity(ThinQEntity, SensorEntity):
"""Represent a thinq sensor platform."""
def __init__(
self,
coordinator: DeviceDataUpdateCoordinator,
entity_description: SensorEntityDescription,
property_id: str,
) -> None:
"""Initialize a sensor entity."""
super().__init__(coordinator, entity_description, property_id)
if entity_description.device_class == SensorDeviceClass.ENUM:
self._attr_options = self.data.options
def _update_status(self) -> None:
"""Update status itself."""
super()._update_status()
self._attr_native_value = self.data.value
if (data_unit := self._get_unit_of_measurement(self.data.unit)) is not None:
# For different from description's unit
self._attr_native_unit_of_measurement = data_unit
_LOGGER.debug(
"[%s:%s] update status: %s -> %s, options:%s, unit:%s",
self.coordinator.device_name,
self.property_id,
self.data.value,
self.native_value,
self.options,
self.native_unit_of_measurement,
)

View File

@ -0,0 +1,989 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]"
},
"error": {
"token_unauthorized": "The token is invalid or unauthorized."
},
"step": {
"user": {
"title": "Connect to ThinQ",
"description": "Please enter a ThinQ [PAT(Personal Access Token)]({pat_url}) created with your LG ThinQ account.",
"data": {
"access_token": "Personal Access Token",
"country": "Country"
}
}
}
},
"entity": {
"switch": {
"auto_mode": {
"name": "Auto mode"
},
"express_mode": {
"name": "Ice plus"
},
"hot_water_mode": {
"name": "Hot water"
},
"humidity_warm_mode": {
"name": "Warm mist"
},
"hygiene_dry_mode": {
"name": "Drying mode"
},
"mood_lamp_state": {
"name": "Mood light"
},
"operation_power": {
"name": "[%key:component::lg_thinq::entity::sensor::current_job_mode::state::high%]"
},
"optimal_humidity": {
"name": "Ventilation"
},
"power_save_enabled": {
"name": "Energy saving"
},
"rapid_freeze": {
"name": "Quick freeze"
},
"sleep_mode": {
"name": "Sleep mode"
},
"uv_nano": {
"name": "UVnano"
},
"warm_mode": {
"name": "Heating"
}
},
"binary_sensor": {
"eco_friendly_mode": {
"name": "Eco friendly"
},
"power_save_enabled": {
"name": "Power saving mode"
},
"remote_control_enabled": {
"name": "Remote start"
},
"remote_control_enabled_for_location": {
"name": "{location} remote start"
},
"rinse_refill": {
"name": "Rinse refill needed"
},
"sabbath_mode": {
"name": "Sabbath"
},
"machine_clean_reminder": {
"name": "Machine clean reminder"
},
"signal_level": {
"name": "Chime sound"
},
"clean_light_reminder": {
"name": "Clean indicator light"
},
"operation_mode": {
"name": "[%key:component::binary_sensor::entity_component::power::name%]"
},
"one_touch_filter": {
"name": "Fresh air filter"
}
},
"climate": {
"climate_air_conditioner": {
"state_attributes": {
"fan_mode": {
"state": {
"slow": "Slow",
"low": "Low",
"mid": "Medium",
"high": "High",
"power": "[%key:component::lg_thinq::entity::sensor::current_job_mode::state::high%]",
"auto": "[%key:component::lg_thinq::entity::sensor::growth_mode::state::standard%]"
}
},
"preset_mode": {
"state": {
"air_clean": "Air purify",
"aroma": "Aroma",
"energy_saving": "Energy saving"
}
}
}
}
},
"event": {
"error": {
"name": "Error",
"state_attributes": {
"event_type": {
"state": {
"block_error": "Cleaning has stopped. Check for obstacles",
"brush_error": "Moving brush has a problem",
"bubble_error": "Bubble error",
"child_lock_active_error": "Child lock",
"cliff_error": "Fall prevention sensor has an error",
"clutch_error": "Clutch error",
"compressor_error": "Compressor error",
"dispensing_error": "Dispensor error",
"door_close_error": "Door closed error",
"door_lock_error": "Door lock error",
"door_open_error": "Door open",
"door_sensor_error": "Door sensor error",
"drainmotor_error": "Drain error",
"dust_full_error": "Dust bin is full and needs to be emptied",
"empty_water_alert_error": "Empty water",
"fan_motor_error": "Fan lock error",
"filter_clogging_error": "Filter error",
"frozen_error": "Freezing detection error",
"heater_circuit_error": "Heater circuit failure",
"high_power_supply_error": "Power supply error",
"high_temperature_detection_error": "High-temperature error",
"inner_lid_open_error": "Lid open error",
"ir_sensor_error": "IR sensor error",
"le_error": "LE error",
"le2_error": "LE2 error",
"left_wheel_error": "Left wheel has a problem",
"locked_motor_error": "Driver motor error",
"mop_error": "Cannot operate properly without the mop attached",
"motor_error": "Motor trouble",
"motor_lock_error": "Motor lock error",
"move_error": "The wheels are not touching the floor",
"need_water_drain": "[%key:component::lg_thinq::entity::event::error::state_attributes::event_type::state::empty_water_alert_error%]",
"need_water_replenishment": "Fill water",
"no_battery_error": "Robot cleaner's battery is low",
"no_dust_bin_error": "Dust bin is not installed",
"no_filter_error": "[%key:component::lg_thinq::entity::event::error::state_attributes::event_type::state::filter_clogging_error%]",
"out_of_balance_error": "Out of balance load",
"overfill_error": "Overfill error",
"part_malfunction_error": "AIE error",
"power_code_connection_error": "Power cord connection error",
"power_fail_error": "Power failure",
"right_wheel_error": "Right wheel has a problem",
"stack_error": "Stacking error",
"steam_heat_error": "Steam heater error",
"suction_blocked_error": "Suction motor is clogged",
"temperature_sensor_error": "Thermistor error",
"time_to_run_the_tub_clean_cycle_error": "Tub clean recommendation",
"timeout_error": "Timeout error",
"turbidity_sensor_error": "turbidity sensor error",
"unable_to_lock_error": "Door lock error",
"unbalanced_load_error": "[%key:component::lg_thinq::entity::event::error::state_attributes::event_type::state::out_of_balance_error%]",
"unknown_error": "Product requires attention",
"vibration_sensor_error": "Vibration sensor error",
"water_drain_error": "Water drain error",
"water_leakage_error": "Water leakage problem",
"water_leaks_error": "[%key:component::lg_thinq::entity::event::error::state_attributes::event_type::state::water_leakage_error%]",
"water_level_sensor_error": "Water sensor error",
"water_supply_error": "Water supply error"
}
}
}
},
"notification": {
"name": "Notification",
"state_attributes": {
"event_type": {
"state": {
"charging_is_complete": "Charging is completed",
"cleaning_is_complete": "Cycle is finished",
"cleaning_is_completed": "Cleaning is completed",
"cleaning_is_failed": "Cleaning has failed",
"cooking_is_complete": "Turned off",
"door_is_open": "The door is open",
"drying_failed": "An error has occurred in the dryer",
"drying_is_complete": "Drying is completed",
"error_during_cleaning": "Cleaning stopped due to an error",
"error_during_washing": "An error has occurred in the washing machine",
"error_has_occurred": "An error has occurred",
"frozen_is_complete": "Ice plus is done",
"homeguard_is_stopped": "Home guard has stopped",
"lack_of_water": "There is no water in the water tank",
"motion_is_detected": "Photograph is sent as movement is detected during home guard",
"need_to_check_location": "Location check is required",
"pollution_is_high": "Air status is rapidly becoming bad",
"preheating_is_complete": "Preheating is done",
"rinse_is_not_enough": "Add rinse aid for better drying performance",
"salt_refill_is_needed": "Add salt for better softening performance",
"scheduled_cleaning_starts": "Scheduled cleaning starts",
"styling_is_complete": "Styling is completed",
"time_to_change_filter": "It is time to replace the filter",
"time_to_change_water_filter": "You need to replace water filter",
"time_to_clean": "Need to selfcleaning",
"time_to_clean_filter": "It is time to clean the filter",
"timer_is_complete": "Timer has been completed",
"washing_is_complete": "Washing is completed",
"water_is_full": "Water is full",
"water_leak_has_occurred": "The dishwasher has detected a water leak"
}
}
}
}
},
"number": {
"target_temperature": {
"name": "[%key:component::sensor::entity_component::temperature::name%]"
},
"target_temperature_for_location": {
"name": "{location} temperature"
},
"light_status": {
"name": "Light"
},
"fan_speed": {
"name": "Fan"
},
"lamp_brightness": {
"name": "[%key:component::lg_thinq::entity::number::light_status::name%]"
},
"wind_temperature": {
"name": "Wind temperature"
},
"relative_hour_to_start": {
"name": "Schedule turn-on"
},
"relative_hour_to_start_for_location": {
"name": "{location} schedule turn-on"
},
"relative_hour_to_start_wm": {
"name": "Delay starts in"
},
"relative_hour_to_start_wm_for_location": {
"name": "{location} delay starts in"
},
"relative_hour_to_stop": {
"name": "Schedule turn-off"
},
"relative_hour_to_stop_for_location": {
"name": "{location} schedule turn-off"
},
"relative_hour_to_stop_wm": {
"name": "Delay ends in"
},
"relative_hour_to_stop_wm_for_location": {
"name": "{location} delay ends in"
},
"sleep_timer_relative_hour_to_stop": {
"name": "Sleep timer"
},
"sleep_timer_relative_hour_to_stop_for_location": {
"name": "{location} sleep timer"
}
},
"sensor": {
"odor_level": {
"name": "Odor",
"state": {
"invalid": "Invalid",
"weak": "Weak",
"normal": "Normal",
"strong": "Strong",
"very_strong": "Very strong"
}
},
"current_temperature": {
"name": "Current temperature"
},
"temperature": {
"name": "Temperature"
},
"total_pollution_level": {
"name": "Overall air quality",
"state": {
"invalid": "Invalid",
"good": "Good",
"normal": "Moderate",
"bad": "Unhealthy",
"very_bad": "Poor"
}
},
"monitoring_enabled": {
"name": "Air quality sensor",
"state": {
"on_working": "Turns on with product",
"always": "Always on"
}
},
"growth_mode": {
"name": "Mode",
"state": {
"standard": "Auto",
"ext_leaf": "Vegetables",
"ext_herb": "Herbs",
"ext_flower": "Flowers",
"ext_expert": "Custom growing mode"
}
},
"growth_mode_for_location": {
"name": "{location} mode",
"state": {
"standard": "[%key:component::lg_thinq::entity::sensor::growth_mode::state::standard%]",
"ext_leaf": "[%key:component::lg_thinq::entity::sensor::growth_mode::state::ext_leaf%]",
"ext_herb": "[%key:component::lg_thinq::entity::sensor::growth_mode::state::ext_herb%]",
"ext_flower": "[%key:component::lg_thinq::entity::sensor::growth_mode::state::ext_flower%]",
"ext_expert": "[%key:component::lg_thinq::entity::sensor::growth_mode::state::ext_expert%]"
}
},
"wind_volume_for_location": {
"name": "{location} wind speed"
},
"brightness": {
"name": "Lighting intensity"
},
"brightness_for_location": {
"name": "{location} lighting intensity"
},
"duration": {
"name": "Lighting duration"
},
"duration_for_location": {
"name": "{location} lighting duration"
},
"day_target_temperature": {
"name": "Day growth temperature"
},
"day_target_temperature_for_location": {
"name": "{location} day growth temperature"
},
"night_target_temperature": {
"name": "Night growth temperature"
},
"night_target_temperature_for_location": {
"name": "{location} night growth temperature"
},
"temperature_state": {
"name": "[%key:component::sensor::entity_component::temperature::name%]",
"state": {
"high": "High",
"normal": "Good",
"low": "Low"
}
},
"temperature_state_for_location": {
"name": "[%key:component::lg_thinq::entity::number::target_temperature_for_location::name%]",
"state": {
"high": "[%key:component::lg_thinq::entity::sensor::temperature_state::state::high%]",
"normal": "[%key:component::lg_thinq::entity::sensor::temperature_state::state::normal%]",
"low": "[%key:component::lg_thinq::entity::sensor::temperature_state::state::low%]"
}
},
"current_state": {
"name": "Current status",
"state": {
"add_drain": "Filling",
"as_pop_up": "[%key:component::lg_thinq::entity::event::error::state_attributes::event_type::state::unknown_error%]",
"cancel": "Cancel",
"carbonation": "Carbonation",
"change_condition": "Settings Change",
"charging": "Charging",
"charging_complete": "Charging completed",
"checking_turbidity": "Detecting soil level",
"cleaning": "Cleaning",
"cleaning_is_done": "Cleaning is done",
"complete": "Done",
"cook": "Cooking",
"cook_complete": "[%key:component::lg_thinq::entity::sensor::current_state::state::complete%]",
"cooking_in_progress": "[%key:component::lg_thinq::entity::sensor::current_state::state::cook%]",
"cool_down": "Cool down",
"cooling": "Cooling",
"detecting": "Detecting",
"detergent_amount": "Providing the info about the amount of detergent",
"diagnosis": "Smart diagnosis is in progress",
"dispensing": "Auto dispensing",
"display_loadsize": "Load size",
"done": "[%key:component::lg_thinq::entity::sensor::current_state::state::complete%]",
"drying": "Drying",
"during_aging": "Aging",
"during_fermentation": "Fermentation",
"end": "Finished",
"end_cooling": "[%key:component::lg_thinq::entity::sensor::current_state::state::drying%]",
"error": "[%key:component::lg_thinq::entity::event::error::state_attributes::event_type::state::unknown_error%]",
"extracting_capsule": "Capsule brewing",
"extraction_mode": "Storing",
"firmware": "Updating firmware",
"fota": "Updating",
"frozen_prevent_initial": "Freeze protection standby",
"frozen_prevent_running": "Freeze protection in progress",
"frozen_prevent_pause": "Freeze protection paused",
"homing": "Moving",
"initial": "[%key:common::state::standby%]",
"initializing": "[%key:common::state::standby%]",
"lock": "Control lock",
"macrosector": "Remote is in use",
"melting": "Wort dissolving",
"monitoring_detecting": "HomeGuard is active",
"monitoring_moving": "Going to the starting point",
"monitoring_positioning": "Setting homeguard start point",
"night_dry": "Night dry",
"oven_setting": "Cooktop connected",
"pause": "[%key:common::state::paused%]",
"paused": "[%key:common::state::paused%]",
"power_fail": "Power fail",
"power_on": "[%key:common::state::on%]",
"power_off": "[%key:common::state::off%]",
"preference": "Setting",
"preheat": "Preheating",
"preheat_complete": "[%key:component::lg_thinq::entity::event::notification::state_attributes::event_type::state::preheating_is_complete%]",
"preheating": "[%key:component::lg_thinq::entity::sensor::current_state::state::preheat%]",
"preheating_is_done": "[%key:component::lg_thinq::entity::event::notification::state_attributes::event_type::state::preheating_is_complete%]",
"prepareing_fermentation": "Preparing now",
"presteam": "Ready to steam",
"prewash": "Prewashing",
"proofing": "Proofing",
"refreshing": "Refreshing",
"reservation": "[%key:component::lg_thinq::entity::sensor::current_state::state::macrosector%]",
"reserved": "Delay set",
"rinse_hold": "Waiting to rinse",
"rinsing": "Rinsing",
"running": "Running",
"running_end": "Complete",
"setdate": "[%key:component::lg_thinq::entity::sensor::current_state::state::macrosector%]",
"shoes_module": "Drying shoes",
"sleep": "In sleep mode",
"smart_grid_run": "Running smart grid",
"soaking": "Soak",
"softening": "Softener",
"spinning": "Spinning",
"stay": "Refresh",
"standby": "[%key:common::state::standby%]",
"steam": "Refresh",
"steam_softening": "Steam softening",
"sterilize": "Sterilize",
"temperature_stabilization": "Temperature adjusting",
"working": "[%key:component::lg_thinq::entity::sensor::current_state::state::cleaning%]",
"wrinkle_care": "Wrinkle care"
}
},
"current_state_for_location": {
"name": "{location} current status",
"state": {
"add_drain": "[%key:component::lg_thinq::entity::sensor::current_state::state::add_drain%]",
"as_pop_up": "[%key:component::lg_thinq::entity::event::error::state_attributes::event_type::state::unknown_error%]",
"cancel": "[%key:component::lg_thinq::entity::sensor::current_state::state::cancel%]",
"carbonation": "[%key:component::lg_thinq::entity::sensor::current_state::state::carbonation%]",
"change_condition": "[%key:component::lg_thinq::entity::sensor::current_state::state::change_condition%]",
"charging": "[%key:component::lg_thinq::entity::sensor::current_state::state::charging%]",
"charging_complete": "[%key:component::lg_thinq::entity::sensor::current_state::state::charging_complete%]",
"checking_turbidity": "[%key:component::lg_thinq::entity::sensor::current_state::state::checking_turbidity%]",
"cleaning": "[%key:component::lg_thinq::entity::sensor::current_state::state::cleaning%]",
"cleaning_is_done": "[%key:component::lg_thinq::entity::sensor::current_state::state::cleaning_is_done%]",
"complete": "[%key:component::lg_thinq::entity::sensor::current_state::state::complete%]",
"cook": "[%key:component::lg_thinq::entity::sensor::current_state::state::cook%]",
"cook_complete": "[%key:component::lg_thinq::entity::sensor::current_state::state::complete%]",
"cooking_in_progress": "[%key:component::lg_thinq::entity::sensor::current_state::state::cook%]",
"cool_down": "[%key:component::lg_thinq::entity::sensor::current_state::state::cool_down%]",
"cooling": "[%key:component::lg_thinq::entity::sensor::current_state::state::cooling%]",
"detecting": "[%key:component::lg_thinq::entity::sensor::current_state::state::detecting%]",
"detergent_amount": "[%key:component::lg_thinq::entity::sensor::current_state::state::detergent_amount%]",
"diagnosis": "[%key:component::lg_thinq::entity::sensor::current_state::state::diagnosis%]",
"dispensing": "[%key:component::lg_thinq::entity::sensor::current_state::state::dispensing%]",
"display_loadsize": "[%key:component::lg_thinq::entity::sensor::current_state::state::display_loadsize%]",
"done": "[%key:component::lg_thinq::entity::sensor::current_state::state::complete%]",
"drying": "[%key:component::lg_thinq::entity::sensor::current_state::state::drying%]",
"during_aging": "[%key:component::lg_thinq::entity::sensor::current_state::state::during_aging%]",
"during_fermentation": "[%key:component::lg_thinq::entity::sensor::current_state::state::during_fermentation%]",
"end": "[%key:component::lg_thinq::entity::sensor::current_state::state::end%]",
"end_cooling": "[%key:component::lg_thinq::entity::sensor::current_state::state::drying%]",
"error": "[%key:component::lg_thinq::entity::event::error::state_attributes::event_type::state::unknown_error%]",
"extracting_capsule": "[%key:component::lg_thinq::entity::sensor::current_state::state::extracting_capsule%]",
"extraction_mode": "[%key:component::lg_thinq::entity::sensor::current_state::state::extraction_mode%]",
"firmware": "[%key:component::lg_thinq::entity::sensor::current_state::state::firmware%]",
"fota": "[%key:component::lg_thinq::entity::sensor::current_state::state::fota%]",
"frozen_prevent_initial": "[%key:component::lg_thinq::entity::sensor::current_state::state::frozen_prevent_initial%]",
"frozen_prevent_running": "[%key:component::lg_thinq::entity::sensor::current_state::state::frozen_prevent_running%]",
"frozen_prevent_pause": "[%key:component::lg_thinq::entity::sensor::current_state::state::frozen_prevent_pause%]",
"homing": "[%key:component::lg_thinq::entity::sensor::current_state::state::homing%]",
"initial": "[%key:common::state::standby%]",
"initializing": "[%key:common::state::standby%]",
"lock": "[%key:component::lg_thinq::entity::sensor::current_state::state::lock%]",
"macrosector": "[%key:component::lg_thinq::entity::sensor::current_state::state::macrosector%]",
"melting": "[%key:component::lg_thinq::entity::sensor::current_state::state::melting%]",
"monitoring_detecting": "[%key:component::lg_thinq::entity::sensor::current_state::state::monitoring_detecting%]",
"monitoring_moving": "[%key:component::lg_thinq::entity::sensor::current_state::state::monitoring_moving%]",
"monitoring_positioning": "[%key:component::lg_thinq::entity::sensor::current_state::state::monitoring_positioning%]",
"night_dry": "[%key:component::lg_thinq::entity::sensor::current_state::state::night_dry%]",
"oven_setting": "[%key:component::lg_thinq::entity::sensor::current_state::state::oven_setting%]",
"pause": "[%key:common::state::paused%]",
"paused": "[%key:common::state::paused%]",
"power_fail": "[%key:component::lg_thinq::entity::sensor::current_state::state::power_fail%]",
"power_on": "[%key:common::state::on%]",
"power_off": "[%key:common::state::off%]",
"preference": "[%key:component::lg_thinq::entity::sensor::current_state::state::preference%]",
"preheat": "[%key:component::lg_thinq::entity::sensor::current_state::state::preheat%]",
"preheat_complete": "[%key:component::lg_thinq::entity::event::notification::state_attributes::event_type::state::preheating_is_complete%]",
"preheating": "[%key:component::lg_thinq::entity::sensor::current_state::state::preheat%]",
"preheating_is_done": "[%key:component::lg_thinq::entity::event::notification::state_attributes::event_type::state::preheating_is_complete%]",
"prepareing_fermentation": "[%key:component::lg_thinq::entity::sensor::current_state::state::prepareing_fermentation%]",
"presteam": "[%key:component::lg_thinq::entity::sensor::current_state::state::presteam%]",
"prewash": "[%key:component::lg_thinq::entity::sensor::current_state::state::prewash%]",
"proofing": "[%key:component::lg_thinq::entity::sensor::current_state::state::proofing%]",
"refreshing": "[%key:component::lg_thinq::entity::sensor::current_state::state::refreshing%]",
"reservation": "[%key:component::lg_thinq::entity::sensor::current_state::state::macrosector%]",
"reserved": "[%key:component::lg_thinq::entity::sensor::current_state::state::reserved%]",
"rinse_hold": "[%key:component::lg_thinq::entity::sensor::current_state::state::rinse_hold%]",
"rinsing": "[%key:component::lg_thinq::entity::sensor::current_state::state::rinsing%]",
"running": "[%key:component::lg_thinq::entity::sensor::current_state::state::running%]",
"running_end": "[%key:component::lg_thinq::entity::sensor::current_state::state::running_end%]",
"setdate": "[%key:component::lg_thinq::entity::sensor::current_state::state::macrosector%]",
"shoes_module": "[%key:component::lg_thinq::entity::sensor::current_state::state::shoes_module%]",
"sleep": "[%key:component::lg_thinq::entity::sensor::current_state::state::sleep%]",
"smart_grid_run": "[%key:component::lg_thinq::entity::sensor::current_state::state::smart_grid_run%]",
"soaking": "[%key:component::lg_thinq::entity::sensor::current_state::state::soaking%]",
"softening": "[%key:component::lg_thinq::entity::sensor::current_state::state::softening%]",
"spinning": "[%key:component::lg_thinq::entity::sensor::current_state::state::spinning%]",
"stay": "[%key:component::lg_thinq::entity::sensor::current_state::state::stay%]",
"standby": "[%key:common::state::standby%]",
"steam": "[%key:component::lg_thinq::entity::sensor::current_state::state::steam%]",
"steam_softening": "[%key:component::lg_thinq::entity::sensor::current_state::state::steam_softening%]",
"sterilize": "[%key:component::lg_thinq::entity::sensor::current_state::state::sterilize%]",
"temperature_stabilization": "[%key:component::lg_thinq::entity::sensor::current_state::state::temperature_stabilization%]",
"working": "[%key:component::lg_thinq::entity::sensor::current_state::state::cleaning%]",
"wrinkle_care": "[%key:component::lg_thinq::entity::sensor::current_state::state::wrinkle_care%]"
}
},
"fresh_air_filter": {
"name": "[%key:component::lg_thinq::entity::binary_sensor::one_touch_filter::name%]",
"state": {
"off": "[%key:common::state::off%]",
"auto": "[%key:component::lg_thinq::entity::sensor::growth_mode::state::standard%]",
"power": "[%key:component::lg_thinq::entity::sensor::current_job_mode::state::high%]",
"replace": "Replace filter",
"smart_power": "Smart safe storage",
"smart_off": "[%key:common::state::off%]",
"smart_on": "[%key:component::lg_thinq::entity::sensor::fresh_air_filter::state::smart_power%]"
}
},
"filter_lifetime": {
"name": "Filter remaining"
},
"used_time": {
"name": "Water filter used"
},
"current_job_mode": {
"name": "Operating mode",
"state": {
"air_clean": "Purify",
"auto": "[%key:component::lg_thinq::entity::sensor::growth_mode::state::standard%]",
"clothes_dry": "Laundry",
"edge": "Edge cleaning",
"heat_pump": "Heat pump",
"high": "Power",
"intensive_dry": "Spot",
"macro": "Custom mode",
"mop": "Mop",
"normal": "Normal",
"off": "[%key:common::state::off%]",
"quiet_humidity": "Silent",
"rapid_humidity": "Jet",
"sector_base": "Cell by cell",
"select": "My space",
"smart_humidity": "Smart",
"spot": "Spiral spot mode",
"turbo": "[%key:component::lg_thinq::entity::select::wind_strength::state::power%]",
"vacation": "Vacation",
"zigzag": "Zigzag"
}
},
"current_job_mode_stick_cleaner": {
"name": "Operating mode",
"state": {
"auto": "Low power",
"high": "[%key:component::lg_thinq::entity::sensor::current_job_mode::state::high%]",
"mop": "[%key:component::lg_thinq::entity::sensor::current_job_mode::state::mop%]",
"normal": "[%key:component::lg_thinq::entity::sensor::current_job_mode::state::normal%]",
"off": "[%key:common::state::off%]",
"turbo": "[%key:component::lg_thinq::entity::select::wind_strength::state::power%]"
}
},
"personalization_mode": {
"name": "Personal mode",
"state": {
"auto_inside": "[%key:component::lg_thinq::entity::switch::auto_mode::name%]",
"sleep": "Sleep mode",
"baby": "Baby care mode",
"sick_house": "New Home mode",
"auto_outside": "Interlocking mode",
"pet": "Pet mode",
"cooking": "Cooking mode",
"smoke": "Smoke mode",
"exercise": "Exercise mode",
"others": "Others"
}
},
"current_dish_washing_course": {
"name": "Current cycle",
"state": {
"auto": "[%key:component::lg_thinq::entity::sensor::growth_mode::state::standard%]",
"heavy": "Intensive",
"delicate": "Delicate",
"turbo": "[%key:component::lg_thinq::entity::select::wind_strength::state::power%]",
"normal": "Normal",
"rinse": "Rinse",
"refresh": "Refresh",
"express": "Express",
"machine_clean": "Machine clean",
"short_mode": "Short mode",
"download_cycle": "Download cycle",
"quick": "Quick",
"steam": "Steam care",
"spray": "Spray",
"eco": "Eco"
}
},
"rinse_level": {
"name": "Rinse aid dispenser level",
"state": {
"rinselevel_0": "0",
"rinselevel_1": "1",
"rinselevel_2": "2",
"rinselevel_3": "3",
"rinselevel_4": "4"
}
},
"softening_level": {
"name": "Softening level",
"state": {
"softeninglevel_0": "[%key:component::lg_thinq::entity::sensor::rinse_level::state::rinselevel_0%]",
"softeninglevel_1": "[%key:component::lg_thinq::entity::sensor::rinse_level::state::rinselevel_1%]",
"softeninglevel_2": "[%key:component::lg_thinq::entity::sensor::rinse_level::state::rinselevel_2%]",
"softeninglevel_3": "[%key:component::lg_thinq::entity::sensor::rinse_level::state::rinselevel_3%]",
"softeninglevel_4": "[%key:component::lg_thinq::entity::sensor::rinse_level::state::rinselevel_4%]"
}
},
"cock_state": {
"name": "[%key:component::lg_thinq::entity::switch::uv_nano::name%]",
"state": {
"cleaning": "In progress",
"normal": "[%key:common::state::standby%]"
}
},
"sterilizing_state": {
"name": "High-temp sterilization",
"state": {
"off": "[%key:common::state::off%]",
"on": "Sterilizing",
"cancel": "[%key:component::lg_thinq::entity::sensor::current_state::state::cancel%]"
}
},
"water_type": {
"name": "Type"
},
"target_temperature": {
"name": "[%key:component::sensor::entity_component::temperature::name%]",
"state": {
"kimchi": "Kimchi",
"off": "[%key:common::state::off%]",
"freezer": "Freezer",
"fridge": "Fridge",
"storage": "Storage",
"meat_fish": "Meat/Fish",
"rice_grain": "Rice/Grain",
"vegetable_fruit": "Vege/Fruit",
"temperature_number": "Number"
}
},
"target_temperature_for_location": {
"name": "[%key:component::lg_thinq::entity::number::target_temperature_for_location::name%]",
"state": {
"kimchi": "[%key:component::lg_thinq::entity::sensor::target_temperature::state::kimchi%]",
"off": "[%key:common::state::off%]",
"freezer": "[%key:component::lg_thinq::entity::sensor::target_temperature::state::freezer%]",
"fridge": "[%key:component::lg_thinq::entity::sensor::target_temperature::state::fridge%]",
"storage": "[%key:component::lg_thinq::entity::sensor::target_temperature::state::storage%]",
"meat_fish": "[%key:component::lg_thinq::entity::sensor::target_temperature::state::meat_fish%]",
"rice_grain": "[%key:component::lg_thinq::entity::sensor::target_temperature::state::rice_grain%]",
"vegetable_fruit": "[%key:component::lg_thinq::entity::sensor::target_temperature::state::vegetable_fruit%]",
"temperature_number": "[%key:component::lg_thinq::entity::sensor::target_temperature::state::temperature_number%]"
}
},
"elapsed_day_state": {
"name": "Brewing period"
},
"elapsed_day_total": {
"name": "Brewing duration"
},
"recipe_name": {
"name": "Homebrew recipe",
"state": {
"ipa": "IPA",
"pale_ale": "Pale ale",
"stout": "Stout",
"wheat": "Wheat",
"pilsner": "Pilsner",
"red_ale": "Red ale",
"my_recipe": "My recipe"
}
},
"wort_info": {
"name": "Wort",
"state": {
"hoppy": "Hoppy",
"deep_gold": "DeepGold",
"wheat": "Wheat",
"dark": "Dark"
}
},
"yeast_info": {
"name": "Yeast",
"state": {
"american_ale": "American ale",
"english_ale": "English ale",
"lager": "Lager",
"weizen": "Weizen"
}
},
"hop_oil_info": {
"name": "Hops"
},
"flavor_info": {
"name": "Flavor"
},
"beer_remain": {
"name": "Recipe progress"
},
"battery_level": {
"name": "Battery",
"state": {
"high": "Full",
"mid": "Medium",
"low": "Low",
"warning": "Empty"
}
},
"relative_to_start": {
"name": "[%key:component::lg_thinq::entity::number::relative_hour_to_start::name%]"
},
"relative_to_start_for_location": {
"name": "[%key:component::lg_thinq::entity::number::relative_hour_to_start_for_location::name%]"
},
"relative_to_start_wm": {
"name": "[%key:component::lg_thinq::entity::number::relative_hour_to_start_wm::name%]"
},
"relative_to_start_wm_for_location": {
"name": "[%key:component::lg_thinq::entity::number::relative_hour_to_start_wm_for_location::name%]"
},
"relative_to_stop": {
"name": "[%key:component::lg_thinq::entity::number::relative_hour_to_stop::name%]"
},
"relative_to_stop_for_location": {
"name": "[%key:component::lg_thinq::entity::number::relative_hour_to_stop_for_location::name%]"
},
"relative_to_stop_wm": {
"name": "[%key:component::lg_thinq::entity::number::relative_hour_to_stop_wm::name%]"
},
"relative_to_stop_wm_for_location": {
"name": "[%key:component::lg_thinq::entity::number::relative_hour_to_stop_wm_for_location::name%]"
},
"sleep_timer_relative_to_stop": {
"name": "[%key:component::lg_thinq::entity::number::sleep_timer_relative_hour_to_stop::name%]"
},
"sleep_timer_relative_to_stop_for_location": {
"name": "[%key:component::lg_thinq::entity::number::sleep_timer_relative_hour_to_stop_for_location::name%]"
},
"absolute_to_start": {
"name": "[%key:component::lg_thinq::entity::number::relative_hour_to_start::name%]"
},
"absolute_to_start_for_location": {
"name": "[%key:component::lg_thinq::entity::number::relative_hour_to_start_for_location::name%]"
},
"absolute_to_stop": {
"name": "[%key:component::lg_thinq::entity::number::relative_hour_to_stop::name%]"
},
"absolute_to_stop_for_location": {
"name": "[%key:component::lg_thinq::entity::number::relative_hour_to_stop_for_location::name%]"
},
"remain": {
"name": "Remaining time"
},
"remain_for_location": {
"name": "{location} remaining time"
},
"running": {
"name": "Running time"
},
"running_for_location": {
"name": "{location} running time"
},
"total": {
"name": "Total time"
},
"total_for_location": {
"name": "{location} total time"
},
"target": {
"name": "Cook time"
},
"target_for_location": {
"name": "{location} cook time"
},
"light_start": {
"name": "Lights on time"
},
"light_start_for_location": {
"name": "{location} lights on time"
},
"power_level": {
"name": "Power level"
},
"power_level_for_location": {
"name": "{location} power level"
}
},
"select": {
"wind_strength": {
"name": "Speed",
"state": {
"slow": "[%key:component::lg_thinq::entity::climate::climate_air_conditioner::state_attributes::fan_mode::state::slow%]",
"low": "Low",
"mid": "Medium",
"high": "High",
"power": "Turbo",
"turbo": "[%key:component::lg_thinq::entity::select::wind_strength::state::power%]",
"auto": "[%key:component::lg_thinq::entity::sensor::growth_mode::state::standard%]",
"wind_1": "Step 1",
"wind_2": "Step 2",
"wind_3": "Step 3",
"wind_4": "Step 4",
"wind_5": "Step 5",
"wind_6": "Step 6",
"wind_7": "Step 7",
"wind_8": "Step 8",
"wind_9": "Step 9",
"wind_10": "Step 10"
}
},
"monitoring_enabled": {
"name": "[%key:component::lg_thinq::entity::sensor::monitoring_enabled::name%]",
"state": {
"on_working": "[%key:component::lg_thinq::entity::sensor::monitoring_enabled::state::on_working%]",
"always": "[%key:component::lg_thinq::entity::sensor::monitoring_enabled::state::always%]"
}
},
"current_job_mode": {
"name": "Operating mode",
"state": {
"air_clean": "Purifying",
"auto": "[%key:component::lg_thinq::entity::sensor::growth_mode::state::standard%]",
"baby_care": "[%key:component::lg_thinq::entity::sensor::personalization_mode::state::baby%]",
"circulator": "Booster",
"clean": "Single",
"direct_clean": "Direct mode",
"dual_clean": "Dual",
"fast": "[%key:component::lg_thinq::entity::select::wind_strength::state::power%]",
"heat_pump": "[%key:component::lg_thinq::entity::sensor::current_job_mode::state::heat_pump%]",
"humidify": "Mist",
"humidify_and_air_clean": "Mist & purifying",
"humidity": "Humid",
"nature_clean": "Natural mode",
"pet_clean": "[%key:component::lg_thinq::entity::sensor::personalization_mode::state::pet%]",
"silent": "Silent",
"sleep": "Sleep",
"smart": "Smart mode",
"space_clean": "Diffusion mode",
"spot_clean": "Wide mode",
"turbo": "[%key:component::lg_thinq::entity::select::wind_strength::state::power%]",
"up_feature": "Additional mode",
"vacation": "Vacation"
}
},
"operation_mode": {
"name": "Operation",
"state": {
"cancel": "[%key:component::lg_thinq::entity::sensor::current_state::state::cancel%]",
"power_off": "Power off",
"preheating": "Preheating",
"start": "[%key:common::action::start%]",
"stop": "[%key:common::action::stop%]",
"wake_up": "Sleep mode off"
}
},
"operation_mode_for_location": {
"name": "{location} operation",
"state": {
"cancel": "[%key:component::lg_thinq::entity::sensor::current_state::state::cancel%]",
"power_off": "[%key:component::lg_thinq::entity::select::operation_mode::state::power_off%]",
"preheating": "[%key:component::lg_thinq::entity::select::operation_mode::state::preheating%]",
"start": "[%key:common::action::start%]",
"stop": "[%key:common::action::stop%]",
"wake_up": "[%key:component::lg_thinq::entity::select::operation_mode::state::wake_up%]"
}
},
"air_clean_operation_mode": {
"name": "[%key:component::lg_thinq::entity::climate::climate_air_conditioner::state_attributes::preset_mode::state::air_clean%]",
"state": {
"start": "[%key:common::action::start%]",
"stop": "[%key:common::action::stop%]"
}
},
"cook_mode": {
"name": "Cook mode",
"state": {
"bake": "Bake",
"convection_bake": "Convection bake",
"convection_roast": "Convection roast",
"roast": "Roast",
"crisp_convection": "Crisp convection"
}
},
"cook_mode_for_location": {
"name": "{location} cook mode",
"state": {
"bake": "[%key:component::lg_thinq::entity::select::cook_mode::state::bake%]",
"convection_bake": "[%key:component::lg_thinq::entity::select::cook_mode::state::convection_bake%]",
"convection_roast": "[%key:component::lg_thinq::entity::select::cook_mode::state::convection_roast%]",
"roast": "[%key:component::lg_thinq::entity::select::cook_mode::state::roast%]",
"crisp_convection": "[%key:component::lg_thinq::entity::select::cook_mode::state::crisp_convection%]"
}
},
"light_brightness": {
"name": "Light"
},
"wind_angle": {
"name": "Rotation",
"state": {
"off": "[%key:common::state::off%]",
"angle_45": "45°",
"angle_60": "60°",
"angle_90": "90°",
"angle_140": "140°"
}
},
"display_light": {
"name": "Display brightness",
"state": {
"off": "[%key:common::state::off%]",
"level_1": "Brightness 1",
"level_2": "Brightness 2",
"level_3": "Brightness 3"
}
},
"fresh_air_filter": {
"name": "[%key:component::lg_thinq::entity::binary_sensor::one_touch_filter::name%]",
"state": {
"off": "[%key:common::state::off%]",
"auto": "[%key:component::lg_thinq::entity::sensor::growth_mode::state::standard%]",
"power": "[%key:component::lg_thinq::entity::sensor::current_job_mode::state::high%]",
"replace": "[%key:component::lg_thinq::entity::sensor::fresh_air_filter::state::replace%]",
"smart_power": "[%key:component::lg_thinq::entity::sensor::fresh_air_filter::state::smart_power%]",
"smart_off": "[%key:common::state::off%]",
"smart_on": "[%key:component::lg_thinq::entity::sensor::fresh_air_filter::state::smart_power%]"
}
},
"hygiene_dry_mode": {
"name": "[%key:component::lg_thinq::entity::switch::hygiene_dry_mode::name%]",
"state": {
"off": "[%key:common::state::off%]",
"fast": "Fast",
"silent": "Silent",
"normal": "[%key:component::lg_thinq::entity::sensor::current_dish_washing_course::state::delicate%]"
}
}
}
}
}

View File

@ -0,0 +1,224 @@
"""Support for switch entities."""
from __future__ import annotations
from dataclasses import dataclass
import logging
from typing import Any
from thinqconnect import DeviceType
from thinqconnect.devices.const import Property as ThinQProperty
from thinqconnect.integration import ActiveMode
from homeassistant.components.switch import (
SwitchDeviceClass,
SwitchEntity,
SwitchEntityDescription,
)
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import ThinqConfigEntry
from .entity import ThinQEntity
@dataclass(frozen=True, kw_only=True)
class ThinQSwitchEntityDescription(SwitchEntityDescription):
"""Describes ThinQ switch entity."""
on_key: str | None = None
off_key: str | None = None
DEVICE_TYPE_SWITCH_MAP: dict[DeviceType, tuple[ThinQSwitchEntityDescription, ...]] = {
DeviceType.AIR_CONDITIONER: (
ThinQSwitchEntityDescription(
key=ThinQProperty.POWER_SAVE_ENABLED,
translation_key=ThinQProperty.POWER_SAVE_ENABLED,
on_key="true",
off_key="false",
),
),
DeviceType.AIR_PURIFIER_FAN: (
ThinQSwitchEntityDescription(
key=ThinQProperty.AIR_FAN_OPERATION_MODE, translation_key="operation_power"
),
ThinQSwitchEntityDescription(
key=ThinQProperty.UV_NANO,
translation_key=ThinQProperty.UV_NANO,
on_key="on",
off_key="off",
entity_category=EntityCategory.CONFIG,
),
ThinQSwitchEntityDescription(
key=ThinQProperty.WARM_MODE,
translation_key=ThinQProperty.WARM_MODE,
on_key="warm_on",
off_key="warm_off",
entity_category=EntityCategory.CONFIG,
),
),
DeviceType.AIR_PURIFIER: (
ThinQSwitchEntityDescription(
key=ThinQProperty.AIR_PURIFIER_OPERATION_MODE,
translation_key="operation_power",
),
),
DeviceType.DEHUMIDIFIER: (
ThinQSwitchEntityDescription(
key=ThinQProperty.DEHUMIDIFIER_OPERATION_MODE,
translation_key="operation_power",
),
),
DeviceType.HUMIDIFIER: (
ThinQSwitchEntityDescription(
key=ThinQProperty.HUMIDIFIER_OPERATION_MODE,
translation_key="operation_power",
),
ThinQSwitchEntityDescription(
key=ThinQProperty.WARM_MODE,
translation_key="humidity_warm_mode",
on_key="warm_on",
off_key="warm_off",
entity_category=EntityCategory.CONFIG,
),
ThinQSwitchEntityDescription(
key=ThinQProperty.MOOD_LAMP_STATE,
translation_key=ThinQProperty.MOOD_LAMP_STATE,
on_key="on",
off_key="off",
entity_category=EntityCategory.CONFIG,
),
ThinQSwitchEntityDescription(
key=ThinQProperty.AUTO_MODE,
translation_key=ThinQProperty.AUTO_MODE,
on_key="auto_on",
off_key="auto_off",
entity_category=EntityCategory.CONFIG,
),
ThinQSwitchEntityDescription(
key=ThinQProperty.SLEEP_MODE,
translation_key=ThinQProperty.SLEEP_MODE,
on_key="sleep_on",
off_key="sleep_off",
entity_category=EntityCategory.CONFIG,
),
),
DeviceType.REFRIGERATOR: (
ThinQSwitchEntityDescription(
key=ThinQProperty.EXPRESS_MODE,
translation_key=ThinQProperty.EXPRESS_MODE,
on_key="true",
off_key="false",
),
ThinQSwitchEntityDescription(
key=ThinQProperty.RAPID_FREEZE,
translation_key=ThinQProperty.RAPID_FREEZE,
on_key="true",
off_key="false",
entity_category=EntityCategory.CONFIG,
),
),
DeviceType.SYSTEM_BOILER: (
ThinQSwitchEntityDescription(
key=ThinQProperty.HOT_WATER_MODE,
translation_key=ThinQProperty.HOT_WATER_MODE,
on_key="on",
off_key="off",
),
),
DeviceType.WINE_CELLAR: (
ThinQSwitchEntityDescription(
key=ThinQProperty.OPTIMAL_HUMIDITY,
translation_key=ThinQProperty.OPTIMAL_HUMIDITY,
on_key="on",
off_key="off",
),
),
}
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
entry: ThinqConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up an entry for switch platform."""
entities: list[ThinQSwitchEntity] = []
for coordinator in entry.runtime_data.coordinators.values():
if (
descriptions := DEVICE_TYPE_SWITCH_MAP.get(
coordinator.api.device.device_type
)
) is not None:
for description in descriptions:
entities.extend(
ThinQSwitchEntity(coordinator, description, property_id)
for property_id in coordinator.api.get_active_idx(
description.key, ActiveMode.READ_WRITE
)
)
if entities:
async_add_entities(entities)
class ThinQSwitchEntity(ThinQEntity, SwitchEntity):
"""Represent a thinq switch platform."""
entity_description: ThinQSwitchEntityDescription
_attr_device_class = SwitchDeviceClass.SWITCH
def _update_status(self) -> None:
"""Update status itself."""
super()._update_status()
if (key := self.entity_description.on_key) is not None:
self._attr_is_on = self.data.value == key
else:
self._attr_is_on = self.data.is_on
_LOGGER.debug(
"[%s:%s] update status: %s -> %s",
self.coordinator.device_name,
self.property_id,
self.data.is_on,
self.is_on,
)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the switch."""
_LOGGER.debug(
"[%s:%s] async_turn_on id: %s",
self.coordinator.device_name,
self.name,
self.property_id,
)
if (on_command := self.entity_description.on_key) is not None:
await self.async_call_api(
self.coordinator.api.post(self.property_id, on_command)
)
else:
await self.async_call_api(
self.coordinator.api.async_turn_on(self.property_id)
)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the switch."""
_LOGGER.debug(
"[%s:%s] async_turn_off id: %s",
self.coordinator.device_name,
self.name,
self.property_id,
)
if (off_command := self.entity_description.off_key) is not None:
await self.async_call_api(
self.coordinator.api.post(self.property_id, off_command)
)
else:
await self.async_call_api(
self.coordinator.api.async_turn_off(self.property_id)
)

View File

@ -0,0 +1,172 @@
"""Support for vacuum entities."""
from __future__ import annotations
from enum import StrEnum
import logging
from thinqconnect import DeviceType
from thinqconnect.integration import ExtendedProperty
from homeassistant.components.vacuum import (
STATE_CLEANING,
STATE_DOCKED,
STATE_ERROR,
STATE_RETURNING,
StateVacuumEntity,
StateVacuumEntityDescription,
VacuumEntityFeature,
)
from homeassistant.const import STATE_IDLE, STATE_PAUSED
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import ThinqConfigEntry
from .entity import ThinQEntity
DEVICE_TYPE_VACUUM_MAP: dict[DeviceType, tuple[StateVacuumEntityDescription, ...]] = {
DeviceType.ROBOT_CLEANER: (
StateVacuumEntityDescription(
key=ExtendedProperty.VACUUM,
name=None,
),
),
}
class State(StrEnum):
"""State of device."""
HOMING = "homing"
PAUSE = "pause"
RESUME = "resume"
SLEEP = "sleep"
START = "start"
WAKE_UP = "wake_up"
ROBOT_STATUS_TO_HA = {
"charging": STATE_DOCKED,
"diagnosis": STATE_IDLE,
"homing": STATE_RETURNING,
"initializing": STATE_IDLE,
"macrosector": STATE_IDLE,
"monitoring_detecting": STATE_IDLE,
"monitoring_moving": STATE_IDLE,
"monitoring_positioning": STATE_IDLE,
"pause": STATE_PAUSED,
"reservation": STATE_IDLE,
"setdate": STATE_IDLE,
"sleep": STATE_IDLE,
"standby": STATE_IDLE,
"working": STATE_CLEANING,
"error": STATE_ERROR,
}
ROBOT_BATT_TO_HA = {
"moveless": 5,
"dock_level": 5,
"low": 30,
"mid": 50,
"high": 90,
"full": 100,
"over_charge": 100,
}
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
entry: ThinqConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up an entry for vacuum platform."""
entities: list[ThinQStateVacuumEntity] = []
for coordinator in entry.runtime_data.coordinators.values():
if (
descriptions := DEVICE_TYPE_VACUUM_MAP.get(
coordinator.api.device.device_type
)
) is not None:
for description in descriptions:
entities.extend(
ThinQStateVacuumEntity(coordinator, description, property_id)
for property_id in coordinator.api.get_active_idx(description.key)
)
if entities:
async_add_entities(entities)
class ThinQStateVacuumEntity(ThinQEntity, StateVacuumEntity):
"""Represent a thinq vacuum platform."""
_attr_supported_features = (
VacuumEntityFeature.SEND_COMMAND
| VacuumEntityFeature.STATE
| VacuumEntityFeature.BATTERY
| VacuumEntityFeature.START
| VacuumEntityFeature.PAUSE
| VacuumEntityFeature.RETURN_HOME
)
def _update_status(self) -> None:
"""Update status itself."""
super()._update_status()
# Update state.
self._attr_state = ROBOT_STATUS_TO_HA[self.data.current_state]
# Update battery.
if (level := self.data.battery) is not None:
self._attr_battery_level = (
level if isinstance(level, int) else ROBOT_BATT_TO_HA.get(level, 0)
)
_LOGGER.debug(
"[%s:%s] update status: %s -> %s (battery_level=%s)",
self.coordinator.device_name,
self.property_id,
self.data.current_state,
self.state,
self.battery_level,
)
async def async_start(self, **kwargs) -> None:
"""Start the device."""
if self.data.current_state == State.SLEEP:
value = State.WAKE_UP
elif self._attr_state == STATE_PAUSED:
value = State.RESUME
else:
value = State.START
_LOGGER.debug(
"[%s:%s] async_start", self.coordinator.device_name, self.property_id
)
await self.async_call_api(
self.coordinator.api.async_set_clean_operation_mode(self.property_id, value)
)
async def async_pause(self, **kwargs) -> None:
"""Pause the device."""
_LOGGER.debug(
"[%s:%s] async_pause", self.coordinator.device_name, self.property_id
)
await self.async_call_api(
self.coordinator.api.async_set_clean_operation_mode(
self.property_id, State.PAUSE
)
)
async def async_return_to_base(self, **kwargs) -> None:
"""Return device to dock."""
_LOGGER.debug(
"[%s:%s] async_return_to_base",
self.coordinator.device_name,
self.property_id,
)
await self.async_call_api(
self.coordinator.api.async_set_clean_operation_mode(
self.property_id, State.HOMING
)
)

View File

@ -327,6 +327,7 @@ FLOWS = {
"lektrico",
"lg_netcast",
"lg_soundbar",
"lg_thinq",
"lidarr",
"lifx",
"linear_garage_door",

View File

@ -3293,6 +3293,12 @@
}
}
},
"lg_thinq": {
"name": "LG ThinQ",
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_push"
},
"lidarr": {
"name": "Lidarr",
"integration_type": "service",

View File

@ -2824,6 +2824,9 @@ thermopro-ble==0.10.0
# homeassistant.components.thingspeak
thingspeak==1.0.0
# homeassistant.components.lg_thinq
thinqconnect==0.9.8
# homeassistant.components.tikteck
tikteck==0.4

View File

@ -2246,6 +2246,9 @@ thermobeacon-ble==0.7.0
# homeassistant.components.thermopro
thermopro-ble==0.10.0
# homeassistant.components.lg_thinq
thinqconnect==0.9.8
# homeassistant.components.tilt_ble
tilt-ble==0.2.3

View File

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

View File

@ -0,0 +1,86 @@
"""Configure tests for the LGThinQ integration."""
from collections.abc import Generator
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from thinqconnect import ThinQAPIException
from homeassistant.components.lg_thinq.const import CONF_CONNECT_CLIENT_ID, DOMAIN
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_COUNTRY
from .const import MOCK_CONNECT_CLIENT_ID, MOCK_COUNTRY, MOCK_PAT, MOCK_UUID
from tests.common import MockConfigEntry
def mock_thinq_api_response(
*,
status: int = 200,
body: dict | None = None,
error_code: str | None = None,
error_message: str | None = None,
) -> MagicMock:
"""Create a mock thinq api response."""
response = MagicMock()
response.status = status
response.body = body
response.error_code = error_code
response.error_message = error_message
return response
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Create a mock config entry."""
return MockConfigEntry(
domain=DOMAIN,
title=f"Test {DOMAIN}",
unique_id=MOCK_PAT,
data={
CONF_ACCESS_TOKEN: MOCK_PAT,
CONF_CONNECT_CLIENT_ID: MOCK_CONNECT_CLIENT_ID,
CONF_COUNTRY: MOCK_COUNTRY,
},
)
@pytest.fixture
def mock_uuid() -> Generator[AsyncMock]:
"""Mock a uuid."""
with (
patch("uuid.uuid4", autospec=True, return_value=MOCK_UUID) as mock_uuid,
patch(
"homeassistant.components.lg_thinq.config_flow.uuid.uuid4",
new=mock_uuid,
),
):
yield mock_uuid.return_value
@pytest.fixture
def mock_thinq_api() -> Generator[AsyncMock]:
"""Mock a thinq api."""
with (
patch("thinqconnect.ThinQApi", autospec=True) as mock_api,
patch(
"homeassistant.components.lg_thinq.config_flow.ThinQApi",
new=mock_api,
),
):
thinq_api = mock_api.return_value
thinq_api.async_get_device_list = AsyncMock(
return_value=mock_thinq_api_response(status=200, body={})
)
yield thinq_api
@pytest.fixture
def mock_invalid_thinq_api(mock_thinq_api: AsyncMock) -> AsyncMock:
"""Mock an invalid thinq api."""
mock_thinq_api.async_get_device_list = AsyncMock(
side_effect=ThinQAPIException(
code="1309", message="Not allowed api call", headers=None
)
)
return mock_thinq_api

View File

@ -0,0 +1,8 @@
"""Constants for lgthinq test."""
from typing import Final
MOCK_PAT: Final[str] = "123abc4567de8f90g123h4ij56klmn789012p345rst6uvw789xy"
MOCK_UUID: Final[str] = "1b3deabc-123d-456d-987d-2a1c7b3bdb67"
MOCK_CONNECT_CLIENT_ID: Final[str] = f"home-assistant-{MOCK_UUID}"
MOCK_COUNTRY: Final[str] = "KR"

View File

@ -0,0 +1,66 @@
"""Test the lgthinq config flow."""
from unittest.mock import AsyncMock
from homeassistant.components.lg_thinq.const import CONF_CONNECT_CLIENT_ID, DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_COUNTRY
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .const import MOCK_CONNECT_CLIENT_ID, MOCK_COUNTRY, MOCK_PAT
from tests.common import MockConfigEntry
async def test_config_flow(
hass: HomeAssistant, mock_thinq_api: AsyncMock, mock_uuid: AsyncMock
) -> None:
"""Test that an thinq entry is normally created."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_ACCESS_TOKEN: MOCK_PAT, CONF_COUNTRY: MOCK_COUNTRY},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {
CONF_ACCESS_TOKEN: MOCK_PAT,
CONF_COUNTRY: MOCK_COUNTRY,
CONF_CONNECT_CLIENT_ID: MOCK_CONNECT_CLIENT_ID,
}
mock_thinq_api.async_get_device_list.assert_called_once()
async def test_config_flow_invalid_pat(
hass: HomeAssistant, mock_invalid_thinq_api: AsyncMock
) -> None:
"""Test that an thinq flow should be aborted with an invalid PAT."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={CONF_ACCESS_TOKEN: MOCK_PAT, CONF_COUNTRY: MOCK_COUNTRY},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "token_unauthorized"}
mock_invalid_thinq_api.async_get_device_list.assert_called_once()
async def test_config_flow_already_configured(
hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_thinq_api: AsyncMock
) -> None:
"""Test that thinq flow should be aborted when already configured."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={CONF_ACCESS_TOKEN: MOCK_PAT, CONF_COUNTRY: MOCK_COUNTRY},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"