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

Bump thinqconnect to 0.9.6 (#125155)

Browse Source 
* Refactor LG ThinQ integration

* Rename ha_bridge_list to bridge_list

* Update for reviews

* Correct spells
Do not use mqtt related api

* Guarantee update status

* Update for reviews

* Update reviews

---------

Co-authored-by: jangwon.lee <jangwon.lee@lge.com>
This commit is contained in:
LG-ThinQ-Integration 2024-09-04 22:52:41 +09:00 committed by GitHub
parent b557e9e826
commit 1e1c3506fe
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 134 additions and 275 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -6,6 +6,7 @@ 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
@ -26,6 +27,8 @@ _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]
@ -55,29 +58,22 @@ async def async_setup_coordinators(
thinq_api: ThinQApi,
) -> None:
"""Set up coordinators and register devices."""
entry.runtime_data = {}
# Get a device list from the server.
# Get a list of ha bridge.
try:
device_list = await thinq_api.async_get_device_list()
bridge_list = await async_get_ha_bridge_list(thinq_api)
except ThinQAPIException as exc:
raise ConfigEntryNotReady(exc.message) from exc
if not device_list:
if not bridge_list:
return
# Setup coordinator per device.
coordinator_list: list[DeviceDataUpdateCoordinator] = []
task_list = [
hass.async_create_task(async_setup_device_coordinator(hass, thinq_api, device))
for device in device_list
hass.async_create_task(async_setup_device_coordinator(hass, bridge))
for bridge in bridge_list
]
task_result = await asyncio.gather(*task_list)
for coordinators in task_result:
if coordinators:
coordinator_list += coordinators
for coordinator in coordinator_list:
for coordinator in task_result:
entry.runtime_data[coordinator.unique_id] = coordinator

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

@ -2,9 +2,10 @@
from __future__ import annotations
from thinqconnect import PROPERTY_READABLE, DeviceType
import logging
from thinqconnect import DeviceType
from thinqconnect.devices.const import Property as ThinQProperty
from thinqconnect.integration.homeassistant.property import create_properties
from homeassistant.components.binary_sensor import (
BinarySensorEntity,
@ -71,6 +72,7 @@ DEVICE_TYPE_BINARY_SENSOR_MAP: dict[
),
DeviceType.WINE_CELLAR: (BINARY_SENSOR_DESC[ThinQProperty.SABBATH_MODE],),
}
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
@ -83,22 +85,13 @@ async def async_setup_entry(
for coordinator in entry.runtime_data.values():
if (
descriptions := DEVICE_TYPE_BINARY_SENSOR_MAP.get(
coordinator.device_api.device_type
coordinator.api.device.device_type
)
) is not None:
for description in descriptions:
properties = create_properties(
device_api=coordinator.device_api,
key=description.key,
children_keys=None,
rw_type=PROPERTY_READABLE,
)
if not properties:
continue
entities.extend(
ThinQBinarySensorEntity(coordinator, description, prop)
for prop in properties
ThinQBinarySensorEntity(coordinator, description, property_id)
for property_id in coordinator.api.get_active_idx(description.key)
)
if entities:
@ -112,4 +105,10 @@ class ThinQBinarySensorEntity(ThinQEntity, BinarySensorEntity):
"""Update status itself."""
super()._update_status()
self._attr_is_on = self.property.get_value_as_bool()
_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

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

@ -1,82 +1,12 @@
"""Constants for LG ThinQ."""
# Base component constants.
from typing import Final
from thinqconnect import (
AirConditionerDevice,
AirPurifierDevice,
AirPurifierFanDevice,
CeilingFanDevice,
CooktopDevice,
DehumidifierDevice,
DeviceType,
DishWasherDevice,
DryerDevice,
HomeBrewDevice,
HoodDevice,
HumidifierDevice,
KimchiRefrigeratorDevice,
MicrowaveOvenDevice,
OvenDevice,
PlantCultivatorDevice,
RefrigeratorDevice,
RobotCleanerDevice,
StickCleanerDevice,
StylerDevice,
SystemBoilerDevice,
WashcomboMainDevice,
WashcomboMiniDevice,
WasherDevice,
WashtowerDevice,
WashtowerDryerDevice,
WashtowerWasherDevice,
WaterHeaterDevice,
WaterPurifierDevice,
WineCellarDevice,
)
# Common
# 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"
# Config Flow
CLIENT_PREFIX: Final = "home-assistant"
CONF_CONNECT_CLIENT_ID: Final = "connect_client_id"
DEFAULT_COUNTRY: Final = "US"
THINQ_DEVICE_ADDED: Final = "thinq_device_added"
DEVICE_TYPE_API_MAP: Final = {
DeviceType.AIR_CONDITIONER: AirConditionerDevice,
DeviceType.AIR_PURIFIER_FAN: AirPurifierFanDevice,
DeviceType.AIR_PURIFIER: AirPurifierDevice,
DeviceType.CEILING_FAN: CeilingFanDevice,
DeviceType.COOKTOP: CooktopDevice,
DeviceType.DEHUMIDIFIER: DehumidifierDevice,
DeviceType.DISH_WASHER: DishWasherDevice,
DeviceType.DRYER: DryerDevice,
DeviceType.HOME_BREW: HomeBrewDevice,
DeviceType.HOOD: HoodDevice,
DeviceType.HUMIDIFIER: HumidifierDevice,
DeviceType.KIMCHI_REFRIGERATOR: KimchiRefrigeratorDevice,
DeviceType.MICROWAVE_OVEN: MicrowaveOvenDevice,
DeviceType.OVEN: OvenDevice,
DeviceType.PLANT_CULTIVATOR: PlantCultivatorDevice,
DeviceType.REFRIGERATOR: RefrigeratorDevice,
DeviceType.ROBOT_CLEANER: RobotCleanerDevice,
DeviceType.STICK_CLEANER: StickCleanerDevice,
DeviceType.STYLER: StylerDevice,
DeviceType.SYSTEM_BOILER: SystemBoilerDevice,
DeviceType.WASHER: WasherDevice,
DeviceType.WASHCOMBO_MAIN: WashcomboMainDevice,
DeviceType.WASHCOMBO_MINI: WashcomboMiniDevice,
DeviceType.WASHTOWER_DRYER: WashtowerDryerDevice,
DeviceType.WASHTOWER: WashtowerDevice,
DeviceType.WASHTOWER_WASHER: WashtowerWasherDevice,
DeviceType.WATER_HEATER: WaterHeaterDevice,
DeviceType.WATER_PURIFIER: WaterPurifierDevice,
DeviceType.WINE_CELLAR: WineCellarDevice,
}

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

@ -5,12 +5,13 @@ from __future__ import annotations
import logging
from typing import Any
from thinqconnect import ConnectBaseDevice, DeviceType, ThinQApi, ThinQAPIException
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 DEVICE_TYPE_API_MAP, DOMAIN
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
@ -18,125 +19,51 @@ _LOGGER = logging.getLogger(__name__)
class DeviceDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
"""LG Device's Data Update Coordinator."""
def __init__(
self,
hass: HomeAssistant,
device_api: ConnectBaseDevice,
*,
sub_id: str | None = None,
) -> None:
def __init__(self, hass: HomeAssistant, ha_bridge: HABridge) -> None:
"""Initialize data coordinator."""
super().__init__(
hass,
_LOGGER,
name=f"{DOMAIN}_{device_api.device_id}",
name=f"{DOMAIN}_{ha_bridge.device.device_id}",
)
# For washTower's washer or dryer
self.sub_id = sub_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"{device_api.alias} {self.sub_id}" if self.sub_id else device_api.alias
)
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"{device_api.device_id}_{self.sub_id}"
if self.sub_id
else device_api.device_id
f"{self.device_id}_{self.sub_id}" if self.sub_id else self.device_id
)
# Get the api instance.
self.device_api = device_api.get_sub_device(self.sub_id) or device_api
async def _async_update_data(self) -> dict[str, Any]:
"""Request to the server to update the status from full response data."""
try:
data = await self.device_api.thinq_api.async_get_device_status(
self.device_api.device_id
)
except ThinQAPIException as exc:
raise UpdateFailed(exc) from exc
return await self.api.fetch_data()
except ThinQAPIException as e:
raise UpdateFailed(e) from e
# Full response data into the device api.
self.device_api.set_status(data)
return data
def refresh_status(self) -> None:
"""Refresh current status."""
self.async_set_updated_data(self.data)
async def async_setup_device_coordinator(
hass: HomeAssistant, thinq_api: ThinQApi, device: dict[str, Any]
) -> list[DeviceDataUpdateCoordinator] | None:
hass: HomeAssistant, ha_bridge: HABridge
) -> DeviceDataUpdateCoordinator:
"""Create DeviceDataUpdateCoordinator and device_api per device."""
device_id = device["deviceId"]
device_info = device["deviceInfo"]
# Get an appropriate class constructor for the device type.
device_type = device_info.get("deviceType")
constructor = DEVICE_TYPE_API_MAP.get(device_type)
if constructor is None:
_LOGGER.error(
"Failed to setup device(%s): not supported device. type=%s",
device_id,
device_type,
)
return None
# Get a device profile from the server.
try:
profile = await thinq_api.async_get_device_profile(device_id)
except ThinQAPIException:
_LOGGER.warning("Failed to setup device(%s): no profile", device_id)
return None
device_group_id = device_info.get("groupId")
# Create new device api instance.
device_api: ConnectBaseDevice = (
constructor(
thinq_api=thinq_api,
device_id=device_id,
device_type=device_type,
model_name=device_info.get("modelName"),
alias=device_info.get("alias"),
group_id=device_group_id,
reportable=device_info.get("reportable"),
profile=profile,
)
if device_group_id
else constructor(
thinq_api=thinq_api,
device_id=device_id,
device_type=device_type,
model_name=device_info.get("modelName"),
alias=device_info.get("alias"),
reportable=device_info.get("reportable"),
profile=profile,
)
)
# Create a list of sub-devices from the profile.
# Note that some devices may have more than two device profiles.
# In this case we should create multiple lg device instance.
# e.g. 'WashTower-Single-Unit' = 'WashTower{dryer}' + 'WashTower{washer}'.
device_sub_ids = (
list(profile.keys())
if device_type == DeviceType.WASHTOWER and "property" not in profile
else [None]
)
# Create new device coordinator instances.
coordinator_list: list[DeviceDataUpdateCoordinator] = []
for sub_id in device_sub_ids:
coordinator = DeviceDataUpdateCoordinator(hass, device_api, sub_id=sub_id)
coordinator = DeviceDataUpdateCoordinator(hass, ha_bridge)
await coordinator.async_refresh()
# Finally add a device coordinator into the result list.
coordinator_list.append(coordinator)
_LOGGER.debug("Setup device's coordinator: %s", coordinator)
return coordinator_list
_LOGGER.debug("Setup device's coordinator: %s", coordinator.device_name)
return coordinator

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

@ -2,11 +2,13 @@
from __future__ import annotations
from collections.abc import Coroutine
import logging
from typing import Any
from thinqconnect import ThinQAPIException
from thinqconnect.integration.homeassistant.property import Property as ThinQProperty
from thinqconnect.devices.const import Location
from thinqconnect.integration import PropertyState
from homeassistant.core import callback
from homeassistant.exceptions import ServiceValidationError
@ -19,6 +21,8 @@ from .coordinator import DeviceDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
EMPTY_STATE = PropertyState()
class ThinQEntity(CoordinatorEntity[DeviceDataUpdateCoordinator]):
"""The base implementation of all lg thinq entities."""
@ -29,43 +33,36 @@ class ThinQEntity(CoordinatorEntity[DeviceDataUpdateCoordinator]):
self,
coordinator: DeviceDataUpdateCoordinator,
entity_description: EntityDescription,
property: ThinQProperty,
property_id: str,
) -> None:
"""Initialize an entity."""
super().__init__(coordinator)
self.entity_description = entity_description
self.property = property
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.device_api.model_name,
model=coordinator.api.device.model_name,
name=coordinator.device_name,
)
# Set the unique key. If there exist a location, add the prefix location name.
unique_key = (
f"{entity_description.key}"
if property.location is None
else f"{property.location}_{entity_description.key}"
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"
)
self._attr_unique_id = f"{coordinator.unique_id}_{unique_key}"
# Update initial status.
self._update_status()
async def async_post_value(self, value: Any) -> None:
"""Post the value of entity to server."""
try:
await self.property.async_post_value(value)
except ThinQAPIException as exc:
raise ServiceValidationError(
exc.message,
translation_domain=DOMAIN,
translation_key=exc.code,
) from exc
finally:
await self.coordinator.async_request_refresh()
@property
def data(self) -> PropertyState:
"""Return the state data of entity."""
return self.coordinator.data.get(self.property_id, EMPTY_STATE)
def _update_status(self) -> None:
"""Update status itself.
@ -78,3 +75,21 @@ class ThinQEntity(CoordinatorEntity[DeviceDataUpdateCoordinator]):
"""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]) -> None:
"""Call the given api and handle exception."""
try:
await target
except ThinQAPIException as exc:
raise ServiceValidationError(
exc.message,
translation_domain=DOMAIN,
translation_key=exc.code,
) from exc
finally:
await self.coordinator.async_request_refresh()

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

@ -15,6 +15,9 @@
"remote_control_enabled": {
"default": "mdi:remote"
},
"remote_control_enabled_for_location": {
"default": "mdi:remote"
},
"rinse_refill": {
"default": "mdi:tune-vertical-variant"
},

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

@ -7,5 +7,5 @@
"documentation": "https://www.home-assistant.io/integrations/lg_thinq/",
"iot_class": "cloud_push",
"loggers": ["thinqconnect"],
"requirements": ["thinqconnect==0.9.5"]
"requirements": ["thinqconnect==0.9.6"]
}

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

@ -34,6 +34,9 @@
"remote_control_enabled": {
"name": "Remote start"
},
"remote_control_enabled_for_location": {
"name": "{location} remote start"
},
"rinse_refill": {
"name": "Rinse refill needed"
},

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

@ -5,9 +5,8 @@ from __future__ import annotations
import logging
from typing import Any
from thinqconnect import PROPERTY_WRITABLE, DeviceType
from thinqconnect import DeviceType
from thinqconnect.devices.const import Property as ThinQProperty
from thinqconnect.integration.homeassistant.property import create_properties
from homeassistant.components.switch import (
SwitchDeviceClass,
@ -20,44 +19,34 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import ThinqConfigEntry
from .entity import ThinQEntity
OPERATION_SWITCH_DESC: dict[ThinQProperty, SwitchEntityDescription] = {
ThinQProperty.AIR_FAN_OPERATION_MODE: SwitchEntityDescription(
key=ThinQProperty.AIR_FAN_OPERATION_MODE,
translation_key="operation_power",
DEVICE_TYPE_SWITCH_MAP: dict[DeviceType, tuple[SwitchEntityDescription, ...]] = {
DeviceType.AIR_PURIFIER_FAN: (
SwitchEntityDescription(
key=ThinQProperty.AIR_FAN_OPERATION_MODE, translation_key="operation_power"
),
ThinQProperty.AIR_PURIFIER_OPERATION_MODE: SwitchEntityDescription(
),
DeviceType.AIR_PURIFIER: (
SwitchEntityDescription(
key=ThinQProperty.AIR_PURIFIER_OPERATION_MODE,
translation_key="operation_power",
),
ThinQProperty.BOILER_OPERATION_MODE: SwitchEntityDescription(
key=ThinQProperty.BOILER_OPERATION_MODE,
translation_key="operation_power",
),
ThinQProperty.DEHUMIDIFIER_OPERATION_MODE: SwitchEntityDescription(
DeviceType.DEHUMIDIFIER: (
SwitchEntityDescription(
key=ThinQProperty.DEHUMIDIFIER_OPERATION_MODE,
translation_key="operation_power",
),
ThinQProperty.HUMIDIFIER_OPERATION_MODE: SwitchEntityDescription(
),
DeviceType.HUMIDIFIER: (
SwitchEntityDescription(
key=ThinQProperty.HUMIDIFIER_OPERATION_MODE,
translation_key="operation_power",
),
}
DEVIE_TYPE_SWITCH_MAP: dict[DeviceType, tuple[SwitchEntityDescription, ...]] = {
DeviceType.AIR_PURIFIER_FAN: (
OPERATION_SWITCH_DESC[ThinQProperty.AIR_FAN_OPERATION_MODE],
),
DeviceType.AIR_PURIFIER: (
OPERATION_SWITCH_DESC[ThinQProperty.AIR_PURIFIER_OPERATION_MODE],
),
DeviceType.DEHUMIDIFIER: (
OPERATION_SWITCH_DESC[ThinQProperty.DEHUMIDIFIER_OPERATION_MODE],
),
DeviceType.HUMIDIFIER: (
OPERATION_SWITCH_DESC[ThinQProperty.HUMIDIFIER_OPERATION_MODE],
),
DeviceType.SYSTEM_BOILER: (
OPERATION_SWITCH_DESC[ThinQProperty.BOILER_OPERATION_MODE],
SwitchEntityDescription(
key=ThinQProperty.BOILER_OPERATION_MODE, translation_key="operation_power"
),
),
}
@ -73,23 +62,14 @@ async def async_setup_entry(
entities: list[ThinQSwitchEntity] = []
for coordinator in entry.runtime_data.values():
if (
descriptions := DEVIE_TYPE_SWITCH_MAP.get(
coordinator.device_api.device_type
descriptions := DEVICE_TYPE_SWITCH_MAP.get(
coordinator.api.device.device_type
)
) is not None:
for description in descriptions:
properties = create_properties(
device_api=coordinator.device_api,
key=description.key,
children_keys=None,
rw_type=PROPERTY_WRITABLE,
)
if not properties:
continue
entities.extend(
ThinQSwitchEntity(coordinator, description, prop)
for prop in properties
ThinQSwitchEntity(coordinator, description, property_id)
for property_id in coordinator.api.get_active_idx(description.key)
)
if entities:
@ -105,14 +85,20 @@ class ThinQSwitchEntity(ThinQEntity, SwitchEntity):
"""Update status itself."""
super()._update_status()
self._attr_is_on = self.property.get_value_as_bool()
_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_post_value("POWER_ON")
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_post_value("POWER_OFF")
await self.async_call_api(self.coordinator.api.async_turn_off(self.property_id))

2
requirements_all.txt
View File

@ -2798,7 +2798,7 @@ thermoworks-smoke==0.1.8
thingspeak==1.0.0
# homeassistant.components.lg_thinq
thinqconnect==0.9.5
thinqconnect==0.9.6
# homeassistant.components.tikteck
tikteck==0.4

2
requirements_test_all.txt
View File

@ -2211,7 +2211,7 @@ thermobeacon-ble==0.7.0
thermopro-ble==0.10.0
# homeassistant.components.lg_thinq
thinqconnect==0.9.5
thinqconnect==0.9.6
# homeassistant.components.tilt_ble
tilt-ble==0.2.3