jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-16 01:37:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove LG Thinq (#125900)

Browse Source
This commit is contained in:
Joost Lekkerkerker 2024-09-18 16:11:29 +02:00 committed by GitHub
parent 139765995e
commit ac93570476
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
20 changed files with 1 additions and 983 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CODEOWNERS
View File

@ -817,8 +817,6 @@ 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

2
homeassistant/brands/lg.json
View File

@ -1,5 +1,5 @@
{
"domain": "lg",
"name": "LG",
"integrations": ["lg_netcast", "lg_thinq", "lg_soundbar", "webostv"]
"integrations": ["lg_netcast", "lg_soundbar", "webostv"]
}

101
homeassistant/components/lg_thinq/__init__.py
View File

@ -1,101 +0,0 @@
"""Support for LG ThinQ Connect device."""
from __future__ import annotations
import asyncio
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, 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 .const import CONF_CONNECT_CLIENT_ID
from .coordinator import DeviceDataUpdateCoordinator, async_setup_device_coordinator
type ThinqConfigEntry = ConfigEntry[dict[str, DeviceDataUpdateCoordinator]]
PLATFORMS = [Platform.BINARY_SENSOR, Platform.SWITCH]
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ThinqConfigEntry) -> bool:
"""Set up an entry."""
entry.runtime_data = {}
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)
# 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[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.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_unload_entry(hass: HomeAssistant, entry: ThinqConfigEntry) -> bool:
"""Unload the entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

181
homeassistant/components/lg_thinq/binary_sensor.py
View File

@ -1,181 +0,0 @@
"""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.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,
)

103
homeassistant/components/lg_thinq/config_flow.py
View File

@ -1,103 +0,0 @@
"""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,
)

12
homeassistant/components/lg_thinq/const.py
View File

@ -1,12 +0,0 @@
"""Constants for LG ThinQ."""
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"

69
homeassistant/components/lg_thinq/coordinator.py
View File

@ -1,69 +0,0 @@
"""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)
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

115
homeassistant/components/lg_thinq/entity.py
View File

@ -1,115 +0,0 @@
"""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
finally:
await self.coordinator.async_request_refresh()

44
homeassistant/components/lg_thinq/icons.json
View File

@ -1,44 +0,0 @@
{
"entity": {
"switch": {
"operation_power": {
"default": "mdi:power"
}
},
"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"
}
}
}
}

11
homeassistant/components/lg_thinq/manifest.json
View File

@ -1,11 +0,0 @@
{
"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.7"]
}

63
homeassistant/components/lg_thinq/strings.json
View File

@ -1,63 +0,0 @@
{
"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": {
"operation_power": {
"name": "Power"
}
},
"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"
}
}
}
}

107
homeassistant/components/lg_thinq/switch.py
View File

@ -1,107 +0,0 @@
"""Support for switch entities."""
from __future__ import annotations
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.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import ThinqConfigEntry
from .entity import ThinQEntity
DEVICE_TYPE_SWITCH_MAP: dict[DeviceType, tuple[SwitchEntityDescription, ...]] = {
DeviceType.AIR_PURIFIER_FAN: (
SwitchEntityDescription(
key=ThinQProperty.AIR_FAN_OPERATION_MODE, translation_key="operation_power"
),
),
DeviceType.AIR_PURIFIER: (
SwitchEntityDescription(
key=ThinQProperty.AIR_PURIFIER_OPERATION_MODE,
translation_key="operation_power",
),
),
DeviceType.DEHUMIDIFIER: (
SwitchEntityDescription(
key=ThinQProperty.DEHUMIDIFIER_OPERATION_MODE,
translation_key="operation_power",
),
),
DeviceType.HUMIDIFIER: (
SwitchEntityDescription(
key=ThinQProperty.HUMIDIFIER_OPERATION_MODE,
translation_key="operation_power",
),
),
DeviceType.SYSTEM_BOILER: (
SwitchEntityDescription(
key=ThinQProperty.BOILER_OPERATION_MODE, translation_key="operation_power"
),
),
}
_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.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."""
_attr_device_class = SwitchDeviceClass.SWITCH
def _update_status(self) -> None:
"""Update status itself."""
super()._update_status()
_LOGGER.debug(
"[%s:%s] update status: %s",
self.coordinator.device_name,
self.property_id,
self.data.is_on,
)
self._attr_is_on = self.data.is_on
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the switch."""
_LOGGER.debug("[%s] async_turn_on", self.name)
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] async_turn_off", self.name)
await self.async_call_api(self.coordinator.api.async_turn_off(self.property_id))

1
homeassistant/generated/config_flows.py
View File

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

6
homeassistant/generated/integrations.json
View File

@ -3262,12 +3262,6 @@
"iot_class": "local_polling",
"name": "LG Netcast"
},
"lg_thinq": {
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_push",
"name": "LG ThinQ"
},
"lg_soundbar": {
"integration_type": "hub",
"config_flow": true,

3
requirements_all.txt
View File

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

3
requirements_test_all.txt
View File

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

1
tests/components/lg_thinq/__init__.py
View File

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

86
tests/components/lg_thinq/conftest.py
View File

@ -1,86 +0,0 @@
"""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

8
tests/components/lg_thinq/const.py
View File

@ -1,8 +0,0 @@
"""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"

66
tests/components/lg_thinq/test_config_flow.py
View File

@ -1,66 +0,0 @@
"""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"