245 lines
7.6 KiB
Python
245 lines
7.6 KiB
Python
"""Utilities for Midea Air Appliances integration"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
from copy import deepcopy
|
|
from typing import Any, Tuple, cast, final
|
|
|
|
import homeassistant.components.logger as hass_logger
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.const import (
|
|
CONF_DEVICES,
|
|
CONF_ID,
|
|
CONF_INCLUDE,
|
|
CONF_PASSWORD,
|
|
CONF_TOKEN,
|
|
CONF_UNIQUE_ID,
|
|
CONF_USERNAME,
|
|
)
|
|
from homeassistant.core import HomeAssistant
|
|
import midea_beautiful as midea_beautiful_api
|
|
from midea_beautiful.appliance import AirConditionerAppliance, DehumidifierAppliance
|
|
from midea_beautiful.cloud import MideaCloud
|
|
from midea_beautiful.lan import LanDevice
|
|
from midea_beautiful.midea import (
|
|
APPLIANCE_TYPE_AIRCON,
|
|
APPLIANCE_TYPE_DEHUMIDIFIER,
|
|
)
|
|
from midea_beautiful.util import very_verbose
|
|
|
|
from custom_components.midea_dehumidifier_lan.const import (
|
|
_ALWAYS_CREATE,
|
|
CONF_MOBILE_APP,
|
|
CONF_TOKEN_KEY,
|
|
UNKNOWN_IP,
|
|
)
|
|
|
|
_SUPPORTABLE_APPLIANCES = {
|
|
APPLIANCE_TYPE_AIRCON: AirConditionerAppliance.supported,
|
|
APPLIANCE_TYPE_DEHUMIDIFIER: DehumidifierAppliance.supported,
|
|
}
|
|
|
|
|
|
def _redact(data: dict[str, Any], key: str, char="*", length: int = 0) -> None:
|
|
"""Redacts/obfuscates key in disctionary"""
|
|
if data.get(key) is not None:
|
|
to_redact = str(data[key])
|
|
if length <= 0 or length >= len(to_redact):
|
|
data[key] = char * len(to_redact)
|
|
else:
|
|
data[key] = to_redact[:-length] + char * length
|
|
|
|
|
|
def _redact_device_conf(device) -> None:
|
|
_redact(device, CONF_TOKEN)
|
|
_redact(device, CONF_TOKEN_KEY)
|
|
_redact(device, CONF_UNIQUE_ID, length=8)
|
|
_redact(device, CONF_ID, length=4)
|
|
|
|
|
|
class RedactedConf:
|
|
"""Outputs redacted configuration dictionary by removing or masking
|
|
confidential data."""
|
|
|
|
def __init__(self, data: dict[str, Any]) -> None:
|
|
"""Remove sensitive information from configuration"""
|
|
self.conf = data
|
|
|
|
@property
|
|
def __dict__(self) -> dict[str, Any]:
|
|
conf = deepcopy(self.conf)
|
|
_redact(conf, CONF_USERNAME)
|
|
_redact(conf, CONF_PASSWORD)
|
|
_redact_device_conf(conf)
|
|
if conf.get(CONF_DEVICES) and isinstance(conf.get(CONF_DEVICES), list):
|
|
for device in conf[CONF_DEVICES]:
|
|
if device and isinstance(device, dict):
|
|
_redact_device_conf(device)
|
|
return conf
|
|
|
|
def __str__(self) -> str:
|
|
"""Remove sensitive information from configuration"""
|
|
|
|
return str(self.__dict__)
|
|
|
|
|
|
def is_enabled_by_capabilities(capabilities: dict[str, Any], capability: str) -> bool:
|
|
"""Returns True if given capability is enabled"""
|
|
if capability in _ALWAYS_CREATE:
|
|
return True
|
|
if not capabilities or capabilities.get(capability, False):
|
|
return True
|
|
return False
|
|
|
|
|
|
def is_climate(appliance: LanDevice) -> bool:
|
|
"""True if appliance is air conditioner"""
|
|
return AirConditionerAppliance.supported(appliance.type)
|
|
|
|
|
|
def is_dehumidifier(appliance: LanDevice) -> bool:
|
|
"""True if appliance is dehumidifier"""
|
|
return DehumidifierAppliance.supported(appliance.type)
|
|
|
|
|
|
def supported_appliance(conf: dict, appliance: LanDevice) -> bool:
|
|
"""Checks if appliance is supported by integration"""
|
|
included = conf.get(CONF_INCLUDE, [])
|
|
for type_id, check in _SUPPORTABLE_APPLIANCES.items():
|
|
if type_id in included and check(appliance.type):
|
|
return True
|
|
return False
|
|
|
|
|
|
class ApplianceCoordinator(ABC): # pylint: disable=too-few-public-methods
|
|
"""Abstract interface for Appliance update coordinators"""
|
|
|
|
appliance: LanDevice
|
|
available: bool
|
|
device: dict[str, Any]
|
|
|
|
def is_climate(self) -> bool:
|
|
"""True if appliance is air conditioner"""
|
|
return is_climate(self.appliance)
|
|
|
|
def is_dehumidifier(self) -> bool:
|
|
"""True if appliance is dehumidifier"""
|
|
return is_dehumidifier(self.appliance)
|
|
|
|
@final
|
|
def dehumidifier(self) -> DehumidifierAppliance:
|
|
"""Returns state as dehumidifier"""
|
|
return cast(DehumidifierAppliance, self.appliance.state)
|
|
|
|
@final
|
|
def airconditioner(self) -> AirConditionerAppliance:
|
|
"""Returns state as air conditioner"""
|
|
return cast(AirConditionerAppliance, self.appliance.state)
|
|
|
|
|
|
class AbstractHub(ABC):
|
|
"""Interface for central class for interacting with appliances"""
|
|
|
|
coordinators: list[ApplianceCoordinator]
|
|
config: dict[str, Any]
|
|
errors: dict[str, Any]
|
|
|
|
def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry) -> None:
|
|
self.client = MideaClient(hass)
|
|
self.cloud: MideaCloud | None = None
|
|
self.hass = hass
|
|
self.config_entry = config_entry
|
|
|
|
@abstractmethod
|
|
async def async_discover_device(
|
|
self, device: dict[str, Any], initial_discovery=False
|
|
) -> Tuple[bool, LanDevice | None]:
|
|
"""Finds device on local network or cloud"""
|
|
return False, None
|
|
|
|
@abstractmethod
|
|
async def async_update_config(self) -> None:
|
|
"""Updates config entry from Hub's data"""
|
|
|
|
|
|
class MideaClient:
|
|
"""Delegate to midea API"""
|
|
|
|
def __init__(self, hass: HomeAssistant) -> None:
|
|
self.hass = hass
|
|
|
|
async def async_debug_mode(self, activate: bool) -> None:
|
|
"""Activated advanced debug mode."""
|
|
very_verbose(activate)
|
|
if activate:
|
|
await self.hass.services.async_call(
|
|
domain=hass_logger.DOMAIN,
|
|
service=hass_logger.SERVICE_SET_LEVEL,
|
|
service_data={"midea_beautiful": "DEBUG"},
|
|
)
|
|
|
|
async def async_connect_to_cloud(self, conf: dict[str, Any]) -> MideaCloud:
|
|
"""Delegate to midea_beautiful_api.connect_to_cloud"""
|
|
return await self.hass.async_add_executor_job(
|
|
self.connect_to_cloud,
|
|
conf,
|
|
)
|
|
|
|
# pylint: disable=no-self-use
|
|
def connect_to_cloud(self, conf: dict[str, Any]) -> MideaCloud:
|
|
"""Delegate to midea_beautiful_api.connect_to_cloud"""
|
|
return midea_beautiful_api.connect_to_cloud(
|
|
account=conf[CONF_USERNAME],
|
|
password=conf[CONF_PASSWORD],
|
|
appname=conf[CONF_MOBILE_APP],
|
|
)
|
|
|
|
def appliance_state( # pylint: disable=too-many-arguments,no-self-use
|
|
self,
|
|
address: str = None,
|
|
token: str = None,
|
|
key: str = None,
|
|
cloud: MideaCloud = None,
|
|
use_cloud: bool = False,
|
|
appliance_id: str = None,
|
|
):
|
|
"""Delegate to midea_beautiful_api.appliance_state"""
|
|
return midea_beautiful_api.appliance_state(
|
|
address=address,
|
|
token=token,
|
|
key=key,
|
|
cloud=cloud,
|
|
use_cloud=use_cloud,
|
|
appliance_id=appliance_id,
|
|
retries=5,
|
|
cloud_timeout=6,
|
|
)
|
|
|
|
def find_appliances( # pylint: disable=too-many-arguments,no-self-use
|
|
self,
|
|
cloud: MideaCloud = None,
|
|
addresses: list[str] = None,
|
|
retries: int = 3,
|
|
timeout: int = 3,
|
|
) -> list[LanDevice]:
|
|
"""Delegate to midea_beautiful_api.find_appliances"""
|
|
return midea_beautiful_api.find_appliances(
|
|
cloud=cloud,
|
|
addresses=addresses,
|
|
retries=retries,
|
|
timeout=timeout,
|
|
)
|
|
|
|
# pylint: disable=no-self-use
|
|
async def async_list_appliances(self, cloud: MideaCloud) -> list:
|
|
"""Delegate to midea_beautiful_api.connect_to_cloud"""
|
|
return await self.hass.async_add_executor_job(
|
|
cloud.list_appliances,
|
|
)
|
|
|
|
|
|
def address_ok(address: str | None) -> bool:
|
|
"""Returns True if address is not known"""
|
|
return address is not None and address != UNKNOWN_IP
|