Add LG ThinQ integration

CODEOWNERS

@@ -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

homeassistant/components/lg_thinq/__init__.py

@@ -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)

homeassistant/components/lg_thinq/binary_sensor.py

@@ -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,
+        )