Add new integration Midea LAN (#151930)

Co-authored-by: Copilot <copilot@github.com>
This commit is contained in:
Simone Chemelli
2026-07-20 23:50:13 +02:00
committed by GitHub
co-authored by Copilot
parent 0c552ec037
commit 3f30346e19
21 changed files with 6084 additions and 0 deletions
Generated
+2
View File
@@ -1134,6 +1134,8 @@ CLAUDE.md @home-assistant/core
/tests/components/metoffice/ @MrHarcombe @avee87
/homeassistant/components/microbees/ @microBeesTech
/tests/components/microbees/ @microBeesTech
/homeassistant/components/midea_lan/ @chemelli74 @rokam @wuwentao
/tests/components/midea_lan/ @chemelli74 @rokam @wuwentao
/homeassistant/components/miele/ @astrandb
/tests/components/miele/ @astrandb
/homeassistant/components/mikrotik/ @engrbm87 @chemelli74
@@ -0,0 +1,76 @@
"""The Midea LAN integration."""
from midealocal.const import ProtocolVersion
from midealocal.devices import device_selector
from homeassistant.const import (
CONF_DEVICE_ID,
CONF_IP_ADDRESS,
CONF_MODEL,
CONF_NAME,
CONF_PORT,
CONF_PROTOCOL,
CONF_TOKEN,
CONF_TYPE,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryError, ConfigEntryNotReady
from .const import CONF_KEY, CONF_SUBTYPE
from .entity import MideaLanConfigEntry
_PLATFORMS: list[Platform] = [Platform.CLIMATE]
async def async_setup_entry(hass: HomeAssistant, entry: MideaLanConfigEntry) -> bool:
"""Set up Midea LAN from a config entry."""
data = entry.data
device_id: int = data[CONF_DEVICE_ID]
device = await hass.async_add_executor_job(
device_selector,
data[CONF_NAME],
device_id,
data[CONF_TYPE],
data[CONF_IP_ADDRESS],
data[CONF_PORT],
data[CONF_TOKEN],
data[CONF_KEY],
ProtocolVersion(data[CONF_PROTOCOL]),
data[CONF_MODEL],
data[CONF_SUBTYPE],
"",
)
if device is None:
raise ConfigEntryError("Unable to initialize device")
connected = await hass.async_add_executor_job(device.connect, True)
if not connected:
# connect() swallows AuthException/SocketException internally and can
# leave the socket open even though it reports failure, so it must be
# closed explicitly here to avoid a ResourceWarning.
await hass.async_add_executor_job(device.close_socket)
raise ConfigEntryNotReady(f"Unable to connect to device {device_id}")
# The library's reconnect loop keeps retrying with a growing backoff
# (up to 600s) without checking for a stop request while sleeping, so
# device.close() alone cannot guarantee the background thread exits
# promptly when offline. Marking it a daemon thread ensures it can
# never block Home Assistant shutdown as a zombie thread.
device.daemon = True
await hass.async_add_executor_job(device.open)
entry.runtime_data = device
async def _close_device() -> None:
await hass.async_add_executor_job(device.close)
entry.async_on_unload(_close_device)
await hass.config_entries.async_forward_entry_setups(entry, _PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: MideaLanConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, _PLATFORMS)
@@ -0,0 +1,725 @@
"""Midea Climate entries."""
from dataclasses import dataclass
import logging
from typing import Any, cast, override
from midealocal.const import DeviceType
from midealocal.devices.ac import DeviceAttributes as ACAttributes, MideaACDevice
from midealocal.devices.c3 import MideaC3Device
from midealocal.devices.c3.const import DeviceAttributes as C3Attributes
from midealocal.devices.cc import DeviceAttributes as CCAttributes, MideaCCDevice
from midealocal.devices.cf import DeviceAttributes as CFAttributes, MideaCFDevice
from midealocal.devices.fb import DeviceAttributes as FBAttributes, MideaFBDevice
from homeassistant.components.climate import (
ATTR_HVAC_MODE,
FAN_AUTO,
FAN_HIGH,
FAN_LOW,
FAN_MEDIUM,
PRESET_AWAY,
PRESET_BOOST,
PRESET_COMFORT,
PRESET_ECO,
PRESET_NONE,
PRESET_SLEEP,
SWING_BOTH,
SWING_HORIZONTAL,
SWING_OFF,
SWING_ON,
SWING_VERTICAL,
ClimateEntity,
ClimateEntityDescription,
ClimateEntityFeature,
HVACMode,
)
from homeassistant.const import (
ATTR_TEMPERATURE,
PRECISION_HALVES,
PRECISION_WHOLE,
UnitOfTemperature,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import DOMAIN, FanSpeed
from .entity import MideaEntity, MideaLanConfigEntry
_LOGGER = logging.getLogger(__name__)
PARALLEL_UPDATES = 0
TEMPERATURE_MAX = 30
TEMPERATURE_MIN = 16
TEMPERATURE_MAX_C3 = 60
TEMPERATURE_MIN_C3 = 5
FAN_SILENT = "silent"
FAN_FULL_SPEED = "full"
FEATURES_TARGET_AND_POWER = (
ClimateEntityFeature.TARGET_TEMPERATURE
| ClimateEntityFeature.TURN_OFF
| ClimateEntityFeature.TURN_ON
)
type MideaClimateDevice = (
MideaACDevice | MideaCCDevice | MideaCFDevice | MideaC3Device | MideaFBDevice
)
@dataclass(kw_only=True, frozen=True)
class MideaClimateEntityDescription(ClimateEntityDescription):
"""Description for a Midea climate entity."""
models: list[DeviceType]
zone: int | None = None
CLIMATE_ENTITIES: list[MideaClimateEntityDescription] = [
MideaClimateEntityDescription(
key="climate",
models=[DeviceType.AC, DeviceType.CC, DeviceType.CF, DeviceType.FB],
),
MideaClimateEntityDescription(
key="climate_zone1",
models=[DeviceType.C3],
translation_key="climate_zone1",
zone=0,
),
MideaClimateEntityDescription(
key="climate_zone2",
models=[DeviceType.C3],
translation_key="climate_zone2",
zone=1,
entity_registry_enabled_default=False,
),
]
_PRESET_TO_ATTR: dict[str, str] = {
PRESET_AWAY: "frost_protect",
PRESET_COMFORT: "comfort_mode",
PRESET_SLEEP: "sleep_mode",
PRESET_ECO: "eco_mode",
PRESET_BOOST: "boost_mode",
}
_ATTR_TO_PRESET: dict[str, str] = {v: k for k, v in _PRESET_TO_ATTR.items()}
_SWING_MODE_MAP: dict[str, tuple[bool, bool]] = {
SWING_OFF: (False, False),
SWING_VERTICAL: (True, False),
SWING_HORIZONTAL: (False, True),
SWING_BOTH: (True, True),
}
_SWING_STATE_MAP: dict[tuple[bool, bool], str] = {
v: k for k, v in _SWING_MODE_MAP.items()
}
async def async_setup_entry(
hass: HomeAssistant,
config_entry: MideaLanConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up climate entries."""
device = config_entry.runtime_data
entities: list[MideaClimate] = []
for description in CLIMATE_ENTITIES:
if device.device_type not in description.models:
continue
if device.device_type == DeviceType.AC:
entities.append(MideaACClimate(cast(MideaACDevice, device), description))
elif device.device_type == DeviceType.CC:
entities.append(MideaCCClimate(cast(MideaCCDevice, device), description))
elif device.device_type == DeviceType.CF:
entities.append(MideaCFClimate(cast(MideaCFDevice, device), description))
elif device.device_type == DeviceType.C3 and description.zone is not None:
entities.append(
MideaC3Climate(
cast(MideaC3Device, device), description, description.zone
)
)
elif device.device_type == DeviceType.FB:
entities.append(MideaFBClimate(cast(MideaFBDevice, device), description))
async_add_entities(entities)
class MideaClimate(MideaEntity, ClimateEntity):
"""Midea Climate Entries Base Class."""
_device: MideaClimateDevice
_attr_supported_features = (
ClimateEntityFeature.TARGET_TEMPERATURE
| ClimateEntityFeature.FAN_MODE
| ClimateEntityFeature.PRESET_MODE
| ClimateEntityFeature.SWING_MODE
| ClimateEntityFeature.TURN_OFF
| ClimateEntityFeature.TURN_ON
)
_attr_max_temp = TEMPERATURE_MAX
_attr_min_temp = TEMPERATURE_MIN
_attr_temperature_unit = UnitOfTemperature.CELSIUS
_zone: int | None = None
def __init__(
self,
device: MideaClimateDevice,
description: MideaClimateEntityDescription,
) -> None:
"""Midea Climate entity init."""
super().__init__(device, description.key)
self.entity_description = description
def _float_attribute(self, attr: str) -> float | None:
"""Return a device attribute as float, if convertible."""
value = self._device.get_attribute(attr)
if not isinstance(value, (int, float, str)):
return None
return float(value)
@property
@override
def hvac_mode(self) -> HVACMode | None:
"""Midea Climate hvac mode."""
power = self._device.get_attribute(attr="power")
if not isinstance(power, bool):
return None
if not power:
return HVACMode.OFF
mode = self._device.get_attribute("mode")
if isinstance(mode, int):
return self._protocol_mode_to_hvac(mode)
return None
def _protocol_mode_to_hvac(self, mode: int) -> HVACMode | None:
"""Convert protocol mode value to Home Assistant HVAC mode."""
if 1 <= mode < len(self.hvac_modes):
return self.hvac_modes[mode]
return None
def _hvac_to_protocol_mode(self, hvac_mode: HVACMode) -> int:
"""Convert Home Assistant HVAC mode to protocol mode value."""
return self.hvac_modes.index(hvac_mode)
@property
@override
def target_temperature(self) -> float | None:
"""Midea Climate target temperature."""
return self._float_attribute("target_temperature")
@property
@override
def current_temperature(self) -> float | None:
"""Midea Climate current temperature."""
return self._float_attribute("indoor_temperature")
@property
@override
def preset_mode(self) -> str | None:
"""Midea Climate preset mode."""
for attr, preset in _ATTR_TO_PRESET.items():
if self._device.get_attribute(attr):
return preset
return PRESET_NONE
@override
def turn_on(self, **kwargs: Any) -> None:
"""Midea Climate turn on."""
self._device.set_attribute(attr="power", value=True)
@override
def turn_off(self, **kwargs: Any) -> None:
"""Midea Climate turn off."""
self._device.set_attribute(attr="power", value=False)
@override
def set_temperature(self, **kwargs: Any) -> None:
"""Midea Climate set temperature."""
if ATTR_TEMPERATURE not in kwargs:
return
temperature = kwargs[ATTR_TEMPERATURE]
hvac_mode = kwargs.get(ATTR_HVAC_MODE)
if hvac_mode == HVACMode.OFF:
self.turn_off()
else:
mode = None
if hvac_mode:
if hvac_mode not in self.hvac_modes:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="unsupported_hvac_mode",
translation_placeholders={"hvac_mode": hvac_mode},
)
mode = self.hvac_modes.index(hvac_mode)
self._device.set_target_temperature(
target_temperature=temperature,
mode=mode,
zone=self._zone,
)
@override
def set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Midea Climate set hvac mode."""
if hvac_mode == HVACMode.OFF:
self.turn_off()
else:
self._device.set_attribute(
attr="mode",
value=self._hvac_to_protocol_mode(hvac_mode),
)
@override
def set_preset_mode(self, preset_mode: str) -> None:
"""Midea Climate set preset mode."""
if new_attr := _PRESET_TO_ATTR.get(preset_mode):
self._device.set_attribute(attr=new_attr, value=True)
return
old_mode = self.preset_mode
old_attr = _PRESET_TO_ATTR.get(old_mode) if isinstance(old_mode, str) else None
if old_attr:
self._device.set_attribute(attr=old_attr, value=False)
class MideaACClimate(MideaClimate):
"""Midea AC Climate Entries."""
_device: MideaACDevice
_fan_thresholds: tuple[tuple[int, str], ...] = (
(FanSpeed.AUTO, FAN_AUTO),
(FanSpeed.FULL_SPEED, FAN_FULL_SPEED),
(FanSpeed.HIGH, FAN_HIGH),
(FanSpeed.MEDIUM, FAN_MEDIUM),
(FanSpeed.LOW, FAN_LOW),
)
_fan_speeds: dict[str, int] = {
FAN_SILENT: 20,
FAN_LOW: 40,
FAN_MEDIUM: 60,
FAN_HIGH: 80,
FAN_FULL_SPEED: 100,
FAN_AUTO: 102,
}
_attr_fan_modes: list[str] = [
FAN_SILENT,
FAN_LOW,
FAN_MEDIUM,
FAN_HIGH,
FAN_FULL_SPEED,
FAN_AUTO,
]
_attr_hvac_modes = [
HVACMode.OFF,
HVACMode.AUTO,
HVACMode.COOL,
HVACMode.DRY,
HVACMode.HEAT,
HVACMode.FAN_ONLY,
]
_attr_swing_modes: list[str] = [
SWING_OFF,
SWING_VERTICAL,
SWING_HORIZONTAL,
SWING_BOTH,
]
_attr_preset_modes = [
PRESET_NONE,
PRESET_COMFORT,
PRESET_ECO,
PRESET_BOOST,
PRESET_SLEEP,
PRESET_AWAY,
]
def __init__(
self,
device: MideaACDevice,
description: MideaClimateEntityDescription,
) -> None:
"""Midea AC Climate entity init."""
super().__init__(device, description)
self._attr_target_temperature_step = float(
PRECISION_WHOLE if self._device.temperature_step == 1 else PRECISION_HALVES,
)
@property
@override
def min_temp(self) -> float:
"""Midea AC Climate min temperature."""
min_temperature = self._float_attribute(ACAttributes.min_temperature)
if min_temperature is None:
return float(TEMPERATURE_MIN)
return min_temperature
@property
@override
def max_temp(self) -> float:
"""Midea AC Climate max temperature."""
max_temperature = self._float_attribute(ACAttributes.max_temperature)
if max_temperature is None:
return float(TEMPERATURE_MAX)
return max_temperature
@property
@override
def fan_mode(self) -> str | None:
"""Midea AC Climate fan mode."""
fan_speed = self._device.get_attribute(ACAttributes.fan_speed)
if not isinstance(fan_speed, int):
return None
for threshold, mode in self._fan_thresholds:
if fan_speed > threshold:
return mode
return FAN_SILENT
@property
@override
def swing_mode(self) -> str | None:
"""Midea AC Climate swing mode."""
vertical = bool(self._device.get_attribute(ACAttributes.swing_vertical))
horizontal = bool(self._device.get_attribute(ACAttributes.swing_horizontal))
return _SWING_STATE_MAP.get((vertical, horizontal))
@property
@override
def current_humidity(self) -> float | None:
"""Return the current indoor humidity, or None if unavailable."""
raw = self._device.get_attribute(ACAttributes.indoor_humidity)
# Some devices report invalid values (0 or 0xFF) for this sensor
# so filter those out and return None instead.
if isinstance(raw, (int, float)) and raw not in {0, 0xFF}:
return float(raw)
return None
@override
def set_fan_mode(self, fan_mode: str) -> None:
"""Midea AC Climate set fan mode."""
fan_speed = self._fan_speeds[fan_mode]
self._device.set_attribute(attr=ACAttributes.fan_speed, value=fan_speed)
@override
def set_swing_mode(self, swing_mode: str) -> None:
"""Midea AC Climate set swing mode."""
swing_vertical, swing_horizontal = _SWING_MODE_MAP.get(
swing_mode, (False, False)
)
self._device.set_swing(
swing_vertical=swing_vertical,
swing_horizontal=swing_horizontal,
)
class MideaCCClimate(MideaClimate):
"""Midea CC Climate Entries."""
_device: MideaCCDevice
_attr_hvac_modes = [
HVACMode.OFF,
HVACMode.FAN_ONLY,
HVACMode.DRY,
HVACMode.HEAT,
HVACMode.COOL,
HVACMode.AUTO,
]
_attr_swing_modes = [SWING_OFF, SWING_ON]
_attr_preset_modes = [PRESET_NONE, PRESET_SLEEP, PRESET_ECO]
@property
@override
def fan_modes(self) -> list[str] | None:
"""Midea CC Climate fan modes."""
return self._device.fan_modes
@property
@override
def fan_mode(self) -> str | None:
"""Midea CC Climate fan mode."""
fan_mode = self._device.get_attribute(CCAttributes.fan_speed)
if not isinstance(fan_mode, str):
return None
return fan_mode
@property
@override
def target_temperature_step(self) -> float | None:
"""Midea CC Climate target temperature step."""
return self._float_attribute(CCAttributes.temperature_precision)
@property
@override
def swing_mode(self) -> str | None:
"""Midea CC Climate swing mode."""
swing = self._device.get_attribute(CCAttributes.swing)
if not isinstance(swing, bool):
return None
return SWING_ON if swing else SWING_OFF
@override
def set_fan_mode(self, fan_mode: str) -> None:
"""Midea CC Climate set fan mode."""
self._device.set_attribute(attr=CCAttributes.fan_speed, value=fan_mode)
@override
def set_swing_mode(self, swing_mode: str) -> None:
"""Midea CC Climate set swing mode."""
self._device.set_attribute(
attr=CCAttributes.swing,
value=swing_mode == SWING_ON,
)
class MideaCFClimate(MideaClimate):
"""Midea CF Climate Entries."""
_device: MideaCFDevice
_attr_hvac_modes = [
HVACMode.OFF,
HVACMode.AUTO,
HVACMode.COOL,
HVACMode.HEAT,
]
_attr_target_temperature_step: float | None = PRECISION_WHOLE
_attr_supported_features = FEATURES_TARGET_AND_POWER
@override
def set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Midea CF Climate set hvac mode."""
if hvac_mode == HVACMode.OFF:
self.turn_off()
else:
target_temperature = self.target_temperature or self.min_temp
self._device.set_target_temperature(
target_temperature=target_temperature,
mode=self._hvac_to_protocol_mode(hvac_mode),
)
@property
@override
def min_temp(self) -> float:
"""Midea CF Climate min temperature."""
min_temperature = self._float_attribute(CFAttributes.min_temperature)
if min_temperature is None:
return float(TEMPERATURE_MIN)
return min_temperature
@property
@override
def max_temp(self) -> float:
"""Midea CF Climate max temperature."""
max_temperature = self._float_attribute(CFAttributes.max_temperature)
if max_temperature is None:
return float(TEMPERATURE_MAX)
return max_temperature
@property
@override
def current_temperature(self) -> float | None:
"""Midea CF Climate current temperature."""
return self._float_attribute(CFAttributes.current_temperature)
class MideaC3Climate(MideaClimate):
"""Midea C3 Climate Entries."""
_device: MideaC3Device
_zone: int
_powers: tuple[C3Attributes, ...] = (
C3Attributes.zone1_power,
C3Attributes.zone2_power,
)
_attr_hvac_modes = [
HVACMode.OFF,
HVACMode.AUTO,
HVACMode.COOL,
HVACMode.HEAT,
]
def __init__(
self,
device: MideaC3Device,
description: MideaClimateEntityDescription,
zone: int,
) -> None:
"""Midea C3 Climate entity init."""
super().__init__(device, description)
self._zone = zone
self._power_attr = MideaC3Climate._powers[zone]
def _temperature(self, *, minimum: bool) -> list[float]:
"""Midea C3 Climate temperature."""
value = (
C3Attributes.temperature_min if minimum else C3Attributes.temperature_max
)
temperatures = self._device.get_attribute(value)
fallback = float(TEMPERATURE_MIN_C3 if minimum else TEMPERATURE_MAX_C3)
if not isinstance(temperatures, list):
return [fallback, fallback]
parsed_temperatures = [float(temperature) for temperature in temperatures]
if len(parsed_temperatures) < 2:
return [fallback, fallback]
return [
fallback if temperature == 0.0 else temperature
for temperature in parsed_temperatures
]
_attr_supported_features = FEATURES_TARGET_AND_POWER
@property
@override
def target_temperature_step(self) -> float:
"""Midea C3 Climate target temperature step."""
zone_temp_type = self._device.get_attribute(C3Attributes.zone_temp_type)
if not isinstance(zone_temp_type, list) or len(zone_temp_type) <= self._zone:
return float(PRECISION_HALVES)
return float(
PRECISION_WHOLE if zone_temp_type[self._zone] else PRECISION_HALVES,
)
@property
@override
def min_temp(self) -> float:
"""Midea C3 Climate min temperature."""
return self._temperature(minimum=True)[self._zone]
@property
@override
def max_temp(self) -> float:
"""Midea C3 Climate max temperature."""
return self._temperature(minimum=False)[self._zone]
@override
def turn_on(self, **kwargs: Any) -> None:
"""Midea C3 Climate turn on."""
self._device.set_attribute(attr=self._power_attr, value=True)
@override
def turn_off(self, **kwargs: Any) -> None:
"""Midea C3 Climate turn off."""
self._device.set_attribute(attr=self._power_attr, value=False)
@property
@override
def hvac_mode(self) -> HVACMode | None:
"""Midea C3 Climate hvac mode."""
power = self._device.get_attribute(self._power_attr)
if not isinstance(power, bool):
return None
if not power:
return HVACMode.OFF
mode = self._device.get_attribute(C3Attributes.mode)
if isinstance(mode, int):
return self._protocol_mode_to_hvac(mode)
return None
@property
@override
def target_temperature(self) -> float | None:
"""Midea C3 Climate target temperature."""
target_temperature = self._device.get_attribute(C3Attributes.target_temperature)
if (
not isinstance(target_temperature, list)
or len(target_temperature) <= self._zone
):
return None
return float(target_temperature[self._zone])
@property
@override
def current_temperature(self) -> float | None:
"""Midea C3 Climate current temperature."""
return self._float_attribute(C3Attributes.temp_tw_out)
@override
def set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Midea C3 Climate set hvac mode."""
if hvac_mode == HVACMode.OFF:
self.turn_off()
else:
self._device.set_mode(self._zone, self._hvac_to_protocol_mode(hvac_mode))
class MideaFBClimate(MideaClimate):
"""Midea FB Climate Entries."""
_device: MideaFBDevice
_attr_hvac_modes = [HVACMode.OFF, HVACMode.HEAT]
_attr_max_temp = 35
_attr_min_temp = 5
_attr_supported_features = (
ClimateEntityFeature.TARGET_TEMPERATURE
| ClimateEntityFeature.PRESET_MODE
| ClimateEntityFeature.TURN_OFF
| ClimateEntityFeature.TURN_ON
)
_attr_target_temperature_step = PRECISION_WHOLE
def __init__(
self,
device: MideaFBDevice,
description: MideaClimateEntityDescription,
) -> None:
"""Midea FB Climate entity init."""
super().__init__(device, description)
self._attr_preset_modes: list[str] = self._device.modes
@property
@override
def preset_mode(self) -> str | None:
"""Midea FB Climate preset mode."""
preset_mode = self._device.get_attribute(attr=FBAttributes.mode)
if not isinstance(preset_mode, str):
return None
return preset_mode
@property
@override
def hvac_mode(self) -> HVACMode | None:
"""Midea FB Climate hvac mode."""
hvac_mode = self._device.get_attribute(attr=FBAttributes.power)
if not isinstance(hvac_mode, bool):
return None
return HVACMode.HEAT if hvac_mode else HVACMode.OFF
@property
@override
def current_temperature(self) -> float | None:
"""Midea FB Climate current temperature."""
return self._float_attribute(FBAttributes.current_temperature)
@override
def set_temperature(self, **kwargs: Any) -> None:
"""Midea FB Climate set temperature."""
wants_heat = kwargs.get(ATTR_HVAC_MODE) == HVACMode.HEAT
if wants_heat and self.hvac_mode == HVACMode.OFF:
self.turn_on()
super().set_temperature(**kwargs)
@override
def set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Midea FB Climate set hvac mode."""
if hvac_mode == HVACMode.OFF:
self.turn_off()
else:
self.turn_on()
@override
def set_preset_mode(self, preset_mode: str) -> None:
"""Midea FB Climate set preset mode."""
self._device.set_attribute(attr=FBAttributes.mode, value=preset_mode)
@@ -0,0 +1,708 @@
"""Config flow for Midea LAN."""
from operator import itemgetter
from typing import Any, override
from midealocal.cloud import (
MideaCloud,
get_default_cloud,
get_midea_cloud,
get_preset_account_cloud,
)
from midealocal.const import DeviceType, ProtocolVersion
from midealocal.device import MideaDevice
from midealocal.discover import discover
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import (
CONF_DEVICE,
CONF_DEVICE_ID,
CONF_IP_ADDRESS,
CONF_MODEL,
CONF_NAME,
CONF_PASSWORD,
CONF_PORT,
CONF_PROTOCOL,
CONF_TOKEN,
CONF_TYPE,
)
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.selector import SelectSelector, SelectSelectorConfig
from .const import _LOGGER, CONF_ACCOUNT, CONF_KEY, CONF_SERVER, CONF_SUBTYPE, DOMAIN
from .device_catalog import MIDEA_DEVICE_NAMES
DEFAULT_CLOUD: str = get_default_cloud()
LOGIN_MODE_PRESET = "preset"
LOGIN_MODE_ACCOUNT = "account"
def _connect_and_close(dm: MideaDevice) -> bool:
"""Connect to the device, always closing the socket afterwards."""
try:
return dm.connect(check_protocol=True)
finally:
dm.close_socket()
class MideaLanConfigFlow(ConfigFlow, domain=DOMAIN):
"""Define current integration setup steps.
Use ConfigFlow handle to support config entries
ConfigFlow will manage the creation of entries from user input, discovery
"""
VERSION = 1
MINOR_VERSION = 1
def __init__(self) -> None:
"""MideaLanConfigFlow class."""
self.available_device: dict = {}
self.devices: dict = {}
self.found_device: dict[str, Any] = {}
self.supports: dict = {}
self.cloud: MideaCloud | None = None
self._login_data: dict[str, str] | None = None
unsorted = dict(MIDEA_DEVICE_NAMES)
# sort and assign supports
self.supports = dict(sorted(unsorted.items(), key=itemgetter(1)))
# Try the preset account first, as it is usually enough to retrieve most data.
# Users registered on a different server may not be able to retrieve the
# required key with their own credentials.
# If this fails, fall back to user-provided credentials.
preset_account = get_preset_account_cloud()
self.preset_account: str = preset_account["username"]
self.preset_password: str = preset_account["password"]
self.preset_cloud_name: str = preset_account["cloud_name"]
def _clear_login_state(self) -> None:
"""Clear flow-scoped credentials and cloud."""
self._login_data = None
self.cloud = None
def _already_configured(self, device_id: str, ip_address: str) -> bool:
"""Check device from json with device_id or ip address."""
for entry in self._async_current_entries():
if str(device_id) == str(
entry.data.get(CONF_DEVICE_ID)
) or ip_address == entry.data.get(CONF_IP_ADDRESS):
return True
return False
@override
async def async_step_user(
self,
user_input: dict[str, Any] | None = None,
) -> ConfigFlowResult:
"""Start a user flow."""
return self.async_show_menu(
step_id="user",
menu_options=["search", "manually", "list"],
)
async def async_step_login_credentials(
self,
user_input: dict[str, Any] | None = None,
error: str | None = None,
) -> ConfigFlowResult:
"""User login steps."""
# get cloud servers configs
cloud_servers = await MideaCloud.get_cloud_servers()
cloud_server_options = list(cloud_servers.values())
if not cloud_server_options:
cloud_server_options = [DEFAULT_CLOUD]
default_server = next(
(server for server in cloud_server_options if server == DEFAULT_CLOUD),
cloud_server_options[0],
)
# user input data exist
if user_input is not None:
cloud_server = user_input[CONF_SERVER]
account = user_input[CONF_ACCOUNT]
password = user_input[CONF_PASSWORD]
# cloud login MUST pass with user input or preset account
if await self._check_cloud_login(
cloud_name=cloud_server,
account=account,
password=password,
force_login=True,
):
self._login_data = {
CONF_ACCOUNT: account,
CONF_PASSWORD: password,
CONF_SERVER: cloud_server,
}
# resume device processing with the already selected device
return await self.async_step_auto(
user_input={CONF_DEVICE: self.found_device[CONF_DEVICE_ID]},
)
# return error with login failed
_LOGGER.debug(
"Failed to login to %s cloud with user credentials",
cloud_server,
)
return self._show_login_credentials_form(
cloud_server_options,
default_server,
user_input=user_input,
error="login_failed",
)
# user not login, show login form in UI
return self._show_login_credentials_form(
cloud_server_options,
default_server,
user_input=None,
error=error,
)
def _show_login_credentials_form(
self,
cloud_server_options: list[str],
default_server: str,
user_input: dict[str, Any] | None,
error: str | None = None,
) -> ConfigFlowResult:
"""Show the login form, retaining any previously entered values."""
schema = vol.Schema(
{
vol.Required(CONF_ACCOUNT): str,
vol.Required(CONF_PASSWORD): str,
vol.Required(
CONF_SERVER,
default=default_server,
): SelectSelector(
SelectSelectorConfig(
options=cloud_server_options,
)
),
},
)
if user_input is not None:
schema = self.add_suggested_values_to_schema(schema, user_input)
return self.async_show_form(
step_id="login_credentials",
data_schema=schema,
errors={"base": error} if error else None,
)
async def async_step_auth_method(
self,
user_input: dict[str, Any] | None = None,
error: str | None = None,
) -> ConfigFlowResult:
"""Select how to authenticate."""
if user_input is not None:
if user_input["login_mode"] == LOGIN_MODE_ACCOUNT:
return await self.async_step_login_credentials()
# preset selected
if await self._check_cloud_login(force_login=True):
self._login_data = {
CONF_SERVER: DEFAULT_CLOUD,
CONF_ACCOUNT: self.preset_account,
CONF_PASSWORD: self.preset_password,
}
# resume device processing with the already selected device
return await self.async_step_auto(
user_input={CONF_DEVICE: self.found_device[CONF_DEVICE_ID]},
)
return await self.async_step_auth_method(
error="preset_login_failed",
)
return self.async_show_form(
step_id="auth_method",
data_schema=vol.Schema(
{
vol.Required(
"login_mode",
default=LOGIN_MODE_PRESET,
): SelectSelector(
SelectSelectorConfig(
options=[
LOGIN_MODE_PRESET,
LOGIN_MODE_ACCOUNT,
],
translation_key="login_mode",
)
),
}
),
errors={"base": error} if error else None,
)
async def async_step_list(
self,
user_input: dict[str, Any] | None = None,
error: str | None = None,
) -> ConfigFlowResult:
"""List all devices and show device info in web UI."""
if user_input is not None:
return await self.async_step_user()
# get all devices list
all_devices = await self.hass.async_add_executor_job(discover)
# available devices exist
if len(all_devices) > 0:
table = (
"Appliance code|Type|IP address|SN|Supported\n:--:|:--:|:--:|:--:|:--:"
)
for device_id, device in all_devices.items():
supported = device.get(CONF_TYPE) in self.supports
table += (
f"\n{device_id}|{f'{device.get(CONF_TYPE):02X}'}|"
f"{device.get(CONF_IP_ADDRESS)}|"
f"{device.get('sn')}|"
f"{'YES' if supported else 'NO'}"
)
# no available device
else:
table = "Not found"
# show devices list result in UI
return self.async_show_form(
step_id="list",
description_placeholders={"table": table},
errors={"base": error} if error else None,
)
async def async_step_search(
self,
user_input: dict[str, Any] | None = None,
error: str | None = None,
) -> ConfigFlowResult:
"""Search device with auto mode or ip address."""
# input is not None, using ip_address to discovery device
if user_input is not None:
# auto mode, ip_address is None
if user_input[CONF_IP_ADDRESS].lower() == "auto":
ip_address = None
# ip exist
else:
ip_address = user_input[CONF_IP_ADDRESS]
# use midea-local discover() to get devices list with ip_address
self.devices = await self.hass.async_add_executor_job(
lambda: discover(list(self.supports.keys()), ip_address=ip_address),
)
self.available_device = {}
for device_id, device in self.devices.items():
# remove exist devices and only return new devices
if not self._already_configured(
str(device_id),
device[CONF_IP_ADDRESS],
):
# fmt: off
self.available_device[device_id] = (
f"{device_id} ({self.supports.get(device.get(CONF_TYPE))})"
)
# fmt: on
if len(self.available_device) > 0:
return await self.async_step_auto()
return await self.async_step_search(error="no_devices")
# show discovery device input form with auto or ip address in web UI
return self.async_show_form(
step_id="search",
data_schema=vol.Schema(
{vol.Required(CONF_IP_ADDRESS, default="auto"): str},
),
errors={"base": error} if error else None,
)
async def _check_cloud_login(
self,
cloud_name: str | None = None,
account: str | None = None,
password: str | None = None,
force_login: bool = False,
) -> bool:
"""Check cloud login."""
# default to preset account
if cloud_name is None or account is None or password is None:
cloud_name = self.preset_cloud_name
account = self.preset_account
password = self.preset_password
session = async_get_clientsession(self.hass)
# init cloud object or force reinit with new one
if self.cloud is None or force_login:
self.cloud = get_midea_cloud(
cloud_name,
session,
account,
password,
)
# check cloud login after self.cloud exist
if await self.cloud.login():
_LOGGER.debug(
"Cloud login succeeded for %s",
cloud_name,
)
return True
_LOGGER.debug(
"Unable to login to %s cloud",
cloud_name,
)
return False
async def _check_key_from_cloud(
self,
appliance_id: int,
default_key: bool = True,
) -> dict[str, Any]:
"""Use preset DEFAULT_CLOUD account to get v3 device token and key."""
device = self.devices[appliance_id]
# _check_cloud_login always succeeds before this is called, setting self.cloud
assert self.cloud is not None
# get device token/key from cloud, plus the well-known default keys
keys = await self.cloud.get_cloud_keys(appliance_id)
if default_key:
keys = {**keys, **(await MideaCloud.get_default_keys())}
# use token/key to connect device and confirm token result
for k, value in keys.items():
dm = MideaDevice(
name="",
device_id=appliance_id,
device_type=device.get(CONF_TYPE),
ip_address=device.get(CONF_IP_ADDRESS),
port=device.get(CONF_PORT),
token=value["token"],
key=value["key"],
device_protocol=ProtocolVersion.V3,
model=device.get(CONF_MODEL),
subtype=device.get(CONF_SUBTYPE, 0),
attributes={},
)
connected = await self.hass.async_add_executor_job(_connect_and_close, dm)
if connected:
return value
# return debug log with failed key
_LOGGER.debug(
"Connect device using method %s token/key failed",
k,
)
_LOGGER.debug(
"Unable to connect device with all the token/key",
)
return {"error": "connect_error"}
async def async_step_auto(
self,
user_input: dict[str, Any] | None = None,
error: str | None = None,
) -> ConfigFlowResult:
"""Discovery device detail info."""
# input device exist
if user_input is not None:
device_id = user_input[CONF_DEVICE]
device = self.devices[device_id]
self.found_device = {
CONF_DEVICE_ID: device_id,
CONF_NAME: self.supports.get(device.get(CONF_TYPE), str(device_id)),
CONF_TYPE: device.get(CONF_TYPE),
CONF_PROTOCOL: device.get(CONF_PROTOCOL),
CONF_IP_ADDRESS: device.get(CONF_IP_ADDRESS),
CONF_PORT: device.get(CONF_PORT),
CONF_MODEL: device.get(CONF_MODEL),
}
# MUST get a auth passed token/key for v3 device, disable add before pass
if device.get(CONF_PROTOCOL) == ProtocolVersion.V3:
# check login cache, show login web if no cache
if self._login_data is None or self.cloud is None:
return await self.async_step_auth_method()
# get subtype from cloud
if device_info := await self.cloud.get_device_info(device_id):
# set subtype with model_number
if cloud_name := device_info.get("name"):
self.found_device[CONF_NAME] = cloud_name
self.found_device[CONF_SUBTYPE] = device_info.get("model_number")
# phase 1, try with user input login data
keys = await self._check_key_from_cloud(device_id)
# no available key, continue the phase 2
if not keys.get("token") or not keys.get("key"):
_LOGGER.debug(
"Can't get valid token using user credentials on %s",
self._login_data[CONF_SERVER],
)
# get key phase 2: reinit cloud with preset account
if not await self._check_cloud_login(force_login=True):
self._clear_login_state()
return await self.async_step_auto(
error="preset_login_failed",
)
# try to get a passed key, without default_key
keys = await self._check_key_from_cloud(
device_id,
default_key=False,
)
# phase 2 got no available token/key, disable device add
if not keys.get("token") or not keys.get("key"):
_LOGGER.debug(
"Can't get available token from Midea server for device %s",
device_id,
)
self._clear_login_state()
return await self.async_step_auto(
error="token_unavailable",
)
# get key pass
self.found_device[CONF_TOKEN] = keys["token"]
self.found_device[CONF_KEY] = keys["key"]
self._clear_login_state()
return await self._async_create_midea_entry(
self._found_device_to_user_input(),
)
# v1/v2 device add without token/key, no cloud interaction needed
self._clear_login_state()
return await self._async_create_midea_entry(
self._found_device_to_user_input(),
)
# show available device list in UI
return self.async_show_form(
step_id="auto",
data_schema=vol.Schema(
{
vol.Required(
CONF_DEVICE,
default=next(iter(self.available_device.keys())),
): vol.In(self.available_device),
},
),
errors={"base": error} if error else None,
)
def _found_device_to_user_input(self) -> dict[str, Any]:
"""Build a manual-step-shaped user_input from the found device."""
return {
CONF_DEVICE_ID: self.found_device[CONF_DEVICE_ID],
CONF_TYPE: self.found_device[CONF_TYPE],
CONF_IP_ADDRESS: self.found_device[CONF_IP_ADDRESS],
CONF_PORT: self.found_device[CONF_PORT],
CONF_PROTOCOL: self.found_device[CONF_PROTOCOL],
CONF_MODEL: self.found_device[CONF_MODEL],
CONF_SUBTYPE: self.found_device.get(CONF_SUBTYPE) or 0,
CONF_TOKEN: self.found_device.get(CONF_TOKEN) or "",
CONF_KEY: self.found_device.get(CONF_KEY) or "",
}
async def _async_create_midea_entry(
self,
user_input: dict[str, Any],
) -> ConfigFlowResult:
"""Validate device connection with all the input and create the entry."""
device_id = user_input[CONF_DEVICE_ID]
# check unique_id before attempting a connection, so a re-add of an
# already configured but currently offline device aborts (already_configured)
# instead of failing with device_auth_failed
await self.async_set_unique_id(str(device_id))
self._abort_if_unique_id_configured()
dm = MideaDevice(
name="",
device_id=device_id,
device_type=user_input[CONF_TYPE],
ip_address=user_input[CONF_IP_ADDRESS],
port=user_input[CONF_PORT],
token=user_input[CONF_TOKEN],
key=user_input[CONF_KEY],
device_protocol=user_input[CONF_PROTOCOL],
model=user_input[CONF_MODEL],
subtype=user_input[CONF_SUBTYPE],
attributes={},
)
connected = await self.hass.async_add_executor_job(_connect_and_close, dm)
if connected:
device_type = user_input[CONF_TYPE]
found_name = self.found_device.get(CONF_NAME)
if isinstance(found_name, str) and found_name:
name = found_name
else:
name = self.supports.get(device_type, str(device_id))
data = {
CONF_NAME: name,
CONF_DEVICE_ID: device_id,
CONF_TYPE: device_type,
CONF_PROTOCOL: user_input[CONF_PROTOCOL],
CONF_IP_ADDRESS: user_input[CONF_IP_ADDRESS],
CONF_PORT: user_input[CONF_PORT],
CONF_MODEL: user_input[CONF_MODEL],
CONF_SUBTYPE: user_input[CONF_SUBTYPE],
CONF_TOKEN: user_input[CONF_TOKEN],
CONF_KEY: user_input[CONF_KEY],
}
return self.async_create_entry(
title=name,
data=data,
)
return self._show_manually_form(user_input, error="device_auth_failed")
async def async_step_manually(
self,
user_input: dict[str, Any] | None = None,
error: str | None = None,
) -> ConfigFlowResult:
"""Add device with device detail info."""
if user_input is not None:
try:
bytearray.fromhex(user_input[CONF_TOKEN])
bytearray.fromhex(user_input[CONF_KEY])
except ValueError:
return self._show_manually_form(user_input, error="invalid_token")
device_id = user_input[CONF_DEVICE_ID]
# (re)discover whenever the requested device isn't already known,
# so correcting the IP/device_id and resubmitting can succeed
if device_id not in self.devices:
ip = user_input[CONF_IP_ADDRESS]
# discover device
self.devices = await self.hass.async_add_executor_job(
lambda: discover(list(self.supports.keys()), ip_address=ip),
)
# discover result MUST exist
if len(self.devices) != 1:
return self._show_manually_form(
user_input, error="invalid_device_ip"
)
# check all the input, disable error add
device_id = next(iter(self.devices.keys()))
# check if device_id is correctly set for that IP
if user_input[CONF_DEVICE_ID] != device_id:
return self._show_manually_form(
user_input,
error="invalid_device_id_for_ip",
)
device = self.devices[device_id]
if user_input[CONF_IP_ADDRESS] != device.get(CONF_IP_ADDRESS):
return self._show_manually_form(
user_input,
error="ip_address_mismatch",
)
if user_input[CONF_PROTOCOL] != device.get(CONF_PROTOCOL):
return self._show_manually_form(
user_input,
error="protocol_mismatch",
)
if user_input[CONF_TYPE] != device.get(CONF_TYPE):
return self._show_manually_form(
user_input,
error="type_mismatch",
)
# try to get token/key with preset account
if user_input[CONF_PROTOCOL] == ProtocolVersion.V3 and (
len(user_input[CONF_TOKEN]) == 0 or len(user_input[CONF_KEY]) == 0
):
# init cloud with preset account
result = await self._check_cloud_login()
if not result:
return self._show_manually_form(
user_input,
error="preset_login_failed",
)
# try to get a passed key
keys = await self._check_key_from_cloud(int(user_input[CONF_DEVICE_ID]))
# no available token/key, disable device add
if not keys.get("token") or not keys.get("key"):
_LOGGER.debug(
"Can't get a valid token from Midea server for device %s",
user_input[CONF_DEVICE_ID],
)
return self._show_manually_form(
user_input,
error="token_unavailable",
)
# set token/key from preset account
user_input[CONF_KEY] = keys["key"]
user_input[CONF_TOKEN] = keys["token"]
self.found_device = {
CONF_DEVICE_ID: user_input[CONF_DEVICE_ID],
CONF_NAME: self.found_device.get(CONF_NAME),
CONF_TYPE: user_input[CONF_TYPE],
CONF_PROTOCOL: user_input[CONF_PROTOCOL],
CONF_IP_ADDRESS: user_input[CONF_IP_ADDRESS],
CONF_PORT: user_input[CONF_PORT],
CONF_MODEL: user_input[CONF_MODEL],
CONF_TOKEN: user_input[CONF_TOKEN],
CONF_KEY: user_input[CONF_KEY],
}
return await self._async_create_midea_entry(user_input)
return self._show_manually_form(user_input, error)
def _show_manually_form(
self,
user_input: dict[str, Any] | None,
error: str | None = None,
) -> ConfigFlowResult:
"""Show the manual step form, retaining any previously entered values."""
protocol = self.found_device.get(CONF_PROTOCOL)
schema = vol.Schema(
{
vol.Required(
CONF_DEVICE_ID,
default=self.found_device.get(CONF_DEVICE_ID),
): int,
vol.Required(
CONF_TYPE,
default=(self.found_device.get(CONF_TYPE) or DeviceType.AC),
): vol.In(self.supports),
vol.Required(
CONF_IP_ADDRESS,
default=self.found_device.get(CONF_IP_ADDRESS),
): str,
vol.Required(
CONF_PORT,
default=(self.found_device.get(CONF_PORT) or 6444),
): int,
vol.Required(
CONF_PROTOCOL,
default=(protocol or ProtocolVersion.V3),
): vol.In(
[protocol] if protocol else ProtocolVersion,
),
vol.Required(
CONF_MODEL,
default=(self.found_device.get(CONF_MODEL) or "Unknown"),
): str,
vol.Required(
CONF_SUBTYPE,
default=(self.found_device.get(CONF_SUBTYPE) or 0),
): int,
vol.Optional(
CONF_TOKEN,
default=(self.found_device.get(CONF_TOKEN) or ""),
): str,
vol.Optional(
CONF_KEY,
default=(self.found_device.get(CONF_KEY) or ""),
): str,
},
)
if user_input is not None:
schema = self.add_suggested_values_to_schema(schema, user_input)
return self.async_show_form(
step_id="manually",
data_schema=schema,
errors={"base": error} if error else None,
)
@@ -0,0 +1,24 @@
"""Constants for the Midea LAN integration."""
from enum import IntEnum
import logging
_LOGGER = logging.getLogger(__package__)
DOMAIN = "midea_lan"
CONF_KEY = "key"
CONF_SUBTYPE = "subtype"
CONF_ACCOUNT = "account"
CONF_SERVER = "server"
class FanSpeed(IntEnum):
"""FanSpeed reference values."""
LOW = 20
MEDIUM = 40
HIGH = 60
FULL_SPEED = 80
AUTO = 100
@@ -0,0 +1,11 @@
"""Helpers for Midea device names and entity definitions."""
from midealocal.const import DeviceType
MIDEA_DEVICE_NAMES: dict[DeviceType, str] = {
DeviceType.AC: "Air Conditioner",
DeviceType.C3: "Heat Pump Wi-Fi Controller",
DeviceType.CC: "MDV Wi-Fi Controller",
DeviceType.CF: "Heat Pump",
DeviceType.FB: "Electric Heater",
}
@@ -0,0 +1,79 @@
"""Base entity for Midea Lan."""
import logging
from typing import Any, override
from midealocal.device import MideaDevice
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity import Entity
from .const import DOMAIN
from .device_catalog import MIDEA_DEVICE_NAMES
_LOGGER = logging.getLogger(__name__)
type MideaLanConfigEntry = ConfigEntry[MideaDevice]
class MideaEntity(Entity):
"""Base Midea entity."""
_attr_has_entity_name = True
_attr_should_poll = False
def __init__(self, device: MideaDevice, entity_key: str) -> None:
"""Initialize Midea base entity."""
self._device = device
self._unique_id = f"{self._device.device_id}_{entity_key}"
self._device_name = self._device.name
@override
async def async_added_to_hass(self) -> None:
"""Register update callback when entity is added."""
self._device.register_update(self.update_state)
@override
async def async_will_remove_from_hass(self) -> None:
"""Unregister update callback when entity is removed."""
self._device.unregister_update(self.update_state)
@property
@override
def device_info(self) -> DeviceInfo:
"""Return device info."""
return DeviceInfo(
manufacturer="Midea",
# Map the device type (numeric ID) to a human-readable model name.
model=MIDEA_DEVICE_NAMES.get(self._device.device_type, "Unknown"),
identifiers={(DOMAIN, str(self._device.device_id))},
name=self._device_name,
model_id=str(self._device.device_type),
hw_version=str(self._device.subtype),
)
@property
@override
def unique_id(self) -> str:
"""Return entity unique id."""
return self._unique_id
@property
@override
def available(self) -> bool:
"""Return entity availability."""
return bool(self._device.available)
def update_state(self, status: Any) -> None:
"""Update entity state."""
if self.hass.is_stopping:
_LOGGER.debug(
"MideaEntity update_state for %s [%s] with status %s: HASS is stopping",
self.name,
type(self),
status,
)
return
self.schedule_update_ha_state()
@@ -0,0 +1,12 @@
{
"domain": "midea_lan",
"name": "Midea LAN",
"codeowners": ["@chemelli74", "@rokam", "@wuwentao"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/midea_lan",
"integration_type": "device",
"iot_class": "local_polling",
"loggers": ["midealocal"],
"quality_scale": "bronze",
"requirements": ["midea-local==6.10.0"]
}
@@ -0,0 +1,70 @@
rules:
# Bronze
action-setup:
status: exempt
comment: no action
appropriate-polling: done
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: no actions
docs-conditions:
status: exempt
comment: This integration does not have any conditions.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
docs-triggers:
status: exempt
comment: This integration does not have any triggers.
entity-event-setup: done
entity-unique-id: done
has-entity-name: done
runtime-data: done
test-before-configure: done
test-before-setup: done
unique-config-entry: done
# Silver
action-exceptions: todo
config-entry-unloading: done
docs-configuration-parameters: done
docs-installation-parameters: done
entity-unavailable: todo
integration-owner: done
log-when-unavailable: todo
parallel-updates: done
reauthentication-flow: todo
test-coverage: todo
# Gold
devices: todo
diagnostics: todo
discovery-update-info: todo
discovery: todo
docs-data-update: done
docs-examples: todo
docs-known-limitations: done
docs-supported-devices: done
docs-supported-functions: done
docs-troubleshooting: todo
docs-use-cases: todo
dynamic-devices: todo
entity-category: todo
entity-device-class: todo
entity-disabled-by-default: todo
entity-translations: todo
exception-translations: todo
icon-translations: todo
reconfiguration-flow: todo
repair-issues: todo
stale-devices: todo
# Platinum
async-dependency: todo
inject-websession: todo
strict-typing: todo
@@ -0,0 +1,123 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
},
"error": {
"device_auth_failed": "Could not connect with the provided configuration",
"invalid_device_id_for_ip": "The device ID does not match the selected IP address",
"invalid_device_ip": "Could not find a supported device at this IP address",
"invalid_token": "Token and key must be valid hexadecimal strings",
"ip_address_mismatch": "The IP address does not match the discovered device",
"login_failed": "Could not log in to the selected cloud server",
"no_devices": "No devices found",
"preset_login_failed": "Could not log in with the preset account",
"protocol_mismatch": "The protocol does not match the discovered device",
"token_unavailable": "Could not get a valid token and key from the cloud",
"type_mismatch": "The type does not match the discovered device"
},
"step": {
"auth_method": {
"data": {
"login_mode": "Login mode"
},
"data_description": {
"login_mode": "How to authenticate with the Midea cloud"
},
"description": "Choose how you want to authenticate.",
"title": "Authentication"
},
"auto": {
"data": {
"device": "Device"
},
"data_description": {
"device": "Select the discovered device to configure"
},
"title": "Select a discovered device"
},
"list": {
"description": "{table}",
"title": "Discovered appliances"
},
"login_credentials": {
"data": {
"account": "Account",
"password": "[%key:common::config_flow::data::password%]",
"server": "Server"
},
"data_description": {
"account": "Your Midea cloud account username or email address",
"password": "Your Midea cloud account password",
"server": "The Midea cloud server for your region"
},
"title": "Cloud login"
},
"manually": {
"data": {
"device_id": "Device ID",
"ip_address": "[%key:common::config_flow::data::ip%]",
"key": "Key",
"model": "Model",
"port": "[%key:common::config_flow::data::port%]",
"protocol": "Protocol",
"subtype": "Subtype",
"token": "Token",
"type": "Type"
},
"data_description": {
"device_id": "The device ID of your Midea appliance",
"ip_address": "The local IP address of your Midea appliance",
"key": "The key of your Midea appliance (hexadecimal string)",
"model": "The model of your Midea appliance",
"port": "The local port used by your Midea appliance",
"protocol": "The protocol used by your Midea appliance",
"subtype": "The subtype of your Midea appliance",
"token": "The token of your Midea appliance (hexadecimal string)",
"type": "The type of your Midea appliance"
},
"title": "Configure manually"
},
"search": {
"data": {
"ip_address": "[%key:common::config_flow::data::ip%]"
},
"data_description": {
"ip_address": "Enter 'auto' to scan your network, or enter a specific IP address"
},
"title": "Search devices"
},
"user": {
"menu_options": {
"list": "List all appliances only",
"manually": "Configure manually",
"search": "Search automatically"
},
"title": "Set up Midea LAN"
}
}
},
"entity": {
"climate": {
"climate_zone1": {
"name": "Zone 1 thermostat"
},
"climate_zone2": {
"name": "Zone 2 thermostat"
}
}
},
"exceptions": {
"unsupported_hvac_mode": {
"message": "HVAC mode {hvac_mode} is not supported by this device."
}
},
"selector": {
"login_mode": {
"options": {
"account": "Your account credentials",
"preset": "Preset cloud credentials"
}
}
}
}
+1
View File
@@ -472,6 +472,7 @@ FLOWS = {
"meteoclimatic",
"metoffice",
"microbees",
"midea_lan",
"miele",
"mikrotik",
"mill",
@@ -4341,6 +4341,12 @@
}
}
},
"midea_lan": {
"name": "Midea LAN",
"integration_type": "device",
"config_flow": true,
"iot_class": "local_polling"
},
"miele": {
"name": "Miele",
"integration_type": "hub",
+3
View File
@@ -1588,6 +1588,9 @@ micloud==0.5
# homeassistant.components.microbees
microBeesPy==0.3.5
# homeassistant.components.midea_lan
midea-local==6.10.0
# homeassistant.components.mill
mill-local==0.5.0
+24
View File
@@ -0,0 +1,24 @@
"""Tests for the Midea LAN integration."""
from unittest.mock import patch
from homeassistant.core import HomeAssistant
from .conftest import DummyDevice
from tests.common import MockConfigEntry
async def setup_integration(
hass: HomeAssistant, config_entry: MockConfigEntry, device: DummyDevice
) -> None:
"""Set up a Midea LAN config entry backed by a fake device."""
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.midea_lan.device_selector",
return_value=device,
):
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
+170
View File
@@ -0,0 +1,170 @@
"""Fixtures for Midea LAN tests."""
from collections.abc import Callable, Generator
from typing import Any
from unittest.mock import AsyncMock, patch
from midealocal.const import DeviceType
import pytest
from homeassistant.components.midea_lan.const import CONF_KEY, CONF_SUBTYPE, DOMAIN
from homeassistant.const import CONF_NAME, CONF_TOKEN, CONF_TYPE
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from .const import (
BASE_DATA,
TEST_DEVICE_ID,
TEST_KEY,
TEST_MODEL,
TEST_NAME,
TEST_SUBTYPE,
TEST_TOKEN,
)
from tests.common import MockConfigEntry
class DummyDevice:
"""Shared fake Midea device for tests."""
def __init__(
self,
device_type: int,
*,
attributes: dict | None = None,
) -> None:
"""Initialize fake device."""
self.device_type = device_type
self.device_id = TEST_DEVICE_ID
self.name = TEST_NAME
self.model = TEST_MODEL
self.subtype = TEST_SUBTYPE
self.available = False
self.attributes = attributes or {}
self._callbacks: list[Callable] = []
self.calls: list[tuple] = []
self.temperature_step = 1
self.fan_modes = ["Low", "Medium", "High", "Auto"]
self.modes = [
"Auto",
"ECO",
"Sleep",
"Anti-freezing",
"Comfort",
"Constant-temperature",
"Normal",
"Fast-heating",
"Standby",
]
def register_update(self, callback: Callable) -> None:
"""Record update callback registration."""
self._callbacks.append(callback)
def unregister_update(self, callback: Callable) -> None:
"""Record update callback removal."""
self._callbacks.remove(callback)
def notify_update(self, status: dict[str, Any]) -> None:
"""Notify all registered callbacks with new state."""
for callback in self._callbacks.copy():
callback(status)
def get_attribute(self, attr: str) -> Any:
"""Return attribute value."""
return self.attributes.get(attr)
def set_attribute(self, attr: str, value: Any) -> None:
"""Record set attribute call."""
self.calls.append(("set_attribute", attr, value))
def set_target_temperature(self, **kwargs: Any) -> None:
"""Record set target temperature call."""
self.calls.append(("set_target_temperature", kwargs))
def set_swing(self, **kwargs: Any) -> None:
"""Record set swing call."""
self.calls.append(("set_swing", kwargs))
def set_mode(self, zone: int, mode: int) -> None:
"""Record set mode call."""
self.calls.append(("set_mode", zone, mode))
def connect(self, check_protocol: bool = False) -> bool:
"""Record connect call and mirror midealocal's availability handling."""
self.calls.append(("connect", check_protocol))
self.available = check_protocol
return check_protocol
def open(self) -> None:
"""Record open call."""
self.calls.append(("open",))
def close(self) -> None:
"""Record close call."""
self.calls.append(("close",))
def close_socket(self) -> None:
"""Record close_socket call."""
self.calls.append(("close_socket",))
def default_ac_device() -> DummyDevice:
"""Return a default AC device for tests."""
return DummyDevice(
DeviceType.AC,
attributes={
"power": True,
"mode": 1,
"target_temperature": 22.0,
"indoor_temperature": 21.0,
"fan_speed": 103,
"swing_vertical": True,
"swing_horizontal": True,
"indoor_humidity": 50,
},
)
def entity_entries(
hass: HomeAssistant, entry: MockConfigEntry
) -> dict[str, er.RegistryEntry]:
"""Return entity registry entries for a config entry, keyed by unique id."""
entity_registry = er.async_get(hass)
return {
entity_entry.unique_id: entity_entry
for entity_entry in er.async_entries_for_config_entry(
entity_registry, entry.entry_id
)
}
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Prevent loading the integration during config flow tests."""
with patch(
"homeassistant.components.midea_lan.async_setup_entry",
return_value=True,
) as mock_entry:
yield mock_entry
@pytest.fixture
def mock_config_entry() -> Callable[[DummyDevice], MockConfigEntry]:
"""Return a function that creates a mock config entry for a given device."""
def _create(device: DummyDevice) -> MockConfigEntry:
return MockConfigEntry(
domain=DOMAIN,
data={
**BASE_DATA,
CONF_TYPE: device.device_type,
CONF_NAME: TEST_NAME,
CONF_TOKEN: TEST_TOKEN,
CONF_KEY: TEST_KEY,
CONF_SUBTYPE: TEST_SUBTYPE,
},
)
return _create
+49
View File
@@ -0,0 +1,49 @@
"""Constants for Midea LAN tests."""
from midealocal.const import ProtocolVersion
from homeassistant.components.midea_lan.const import CONF_KEY, CONF_SUBTYPE
from homeassistant.components.midea_lan.device_catalog import MIDEA_DEVICE_NAMES
from homeassistant.const import (
CONF_DEVICE_ID,
CONF_IP_ADDRESS,
CONF_MODEL,
CONF_PORT,
CONF_PROTOCOL,
CONF_TOKEN,
CONF_TYPE,
)
TEST_DEVICE_ID = 12345678
TEST_IP_ADDRESS = "1.1.1.1"
TEST_KEY = "bb" * 16
TEST_MODEL = "MSAGBU-09HRFN8"
TEST_NAME = "Bedroom AC"
TEST_PORT = 6444
TEST_PROTOCOL = ProtocolVersion.V3
TEST_SUBTYPE = 0
TEST_TOKEN = "aa" * 16
TEST_TYPE = next(iter(MIDEA_DEVICE_NAMES))
BASE_DATA = {
CONF_DEVICE_ID: TEST_DEVICE_ID,
CONF_IP_ADDRESS: TEST_IP_ADDRESS,
CONF_PORT: TEST_PORT,
CONF_MODEL: TEST_MODEL,
CONF_PROTOCOL: TEST_PROTOCOL,
}
DISCOVERY_RESULT = {
TEST_DEVICE_ID: {
**BASE_DATA,
CONF_TYPE: TEST_TYPE,
}
}
EXTENDED_DATA = {
**BASE_DATA,
CONF_TYPE: TEST_TYPE,
CONF_SUBTYPE: TEST_SUBTYPE,
CONF_TOKEN: TEST_TOKEN,
CONF_KEY: TEST_KEY,
}
@@ -0,0 +1,541 @@
# serializer version: 1
# name: test_climate_state_snapshot[ac][climate.bedroom_ac-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<ClimateEntityCapabilityAttribute.FAN_MODES: 'fan_modes'>: list([
'silent',
'low',
'medium',
'high',
'full',
'auto',
]),
<ClimateEntityCapabilityAttribute.HVAC_MODES: 'hvac_modes'>: list([
<HVACMode.OFF: 'off'>,
<HVACMode.AUTO: 'auto'>,
<HVACMode.COOL: 'cool'>,
<HVACMode.DRY: 'dry'>,
<HVACMode.HEAT: 'heat'>,
<HVACMode.FAN_ONLY: 'fan_only'>,
]),
<ClimateEntityCapabilityAttribute.MAX_TEMP: 'max_temp'>: 30.0,
<ClimateEntityCapabilityAttribute.MIN_TEMP: 'min_temp'>: 16.0,
<ClimateEntityCapabilityAttribute.PRESET_MODES: 'preset_modes'>: list([
'none',
'comfort',
'eco',
'boost',
'sleep',
'away',
]),
<ClimateEntityCapabilityAttribute.SWING_MODES: 'swing_modes'>: list([
'off',
'vertical',
'horizontal',
'both',
]),
<ClimateEntityCapabilityAttribute.TARGET_TEMP_STEP: 'target_temp_step'>: 1.0,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'climate',
'entity_category': None,
'entity_id': 'climate.bedroom_ac',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': None,
'platform': 'midea_lan',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <ClimateEntityFeature: 441>,
'translation_key': None,
'unique_id': '12345678_climate',
'unit_of_measurement': None,
})
# ---
# name: test_climate_state_snapshot[ac][climate.bedroom_ac-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<ClimateEntityStateAttribute.CURRENT_HUMIDITY: 'current_humidity'>: 50.0,
<ClimateEntityStateAttribute.CURRENT_TEMPERATURE: 'current_temperature'>: 21.0,
<ClimateEntityStateAttribute.FAN_MODE: 'fan_mode'>: 'auto',
<ClimateEntityCapabilityAttribute.FAN_MODES: 'fan_modes'>: list([
'silent',
'low',
'medium',
'high',
'full',
'auto',
]),
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Bedroom AC',
<ClimateEntityCapabilityAttribute.HVAC_MODES: 'hvac_modes'>: list([
<HVACMode.OFF: 'off'>,
<HVACMode.AUTO: 'auto'>,
<HVACMode.COOL: 'cool'>,
<HVACMode.DRY: 'dry'>,
<HVACMode.HEAT: 'heat'>,
<HVACMode.FAN_ONLY: 'fan_only'>,
]),
<ClimateEntityCapabilityAttribute.MAX_TEMP: 'max_temp'>: 30.0,
<ClimateEntityCapabilityAttribute.MIN_TEMP: 'min_temp'>: 16.0,
<ClimateEntityStateAttribute.PRESET_MODE: 'preset_mode'>: 'none',
<ClimateEntityCapabilityAttribute.PRESET_MODES: 'preset_modes'>: list([
'none',
'comfort',
'eco',
'boost',
'sleep',
'away',
]),
<EntityStateAttribute.SUPPORTED_FEATURES: 'supported_features'>: <ClimateEntityFeature: 441>,
<ClimateEntityStateAttribute.SWING_MODE: 'swing_mode'>: 'both',
<ClimateEntityCapabilityAttribute.SWING_MODES: 'swing_modes'>: list([
'off',
'vertical',
'horizontal',
'both',
]),
<ClimateEntityCapabilityAttribute.TARGET_TEMP_STEP: 'target_temp_step'>: 1.0,
<ClimateEntityStateAttribute.TEMPERATURE: 'temperature'>: 22.0,
}),
'context': <ANY>,
'entity_id': 'climate.bedroom_ac',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'auto',
})
# ---
# name: test_climate_state_snapshot[c3][climate.bedroom_ac_zone_1_thermostat-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<ClimateEntityCapabilityAttribute.HVAC_MODES: 'hvac_modes'>: list([
<HVACMode.OFF: 'off'>,
<HVACMode.AUTO: 'auto'>,
<HVACMode.COOL: 'cool'>,
<HVACMode.HEAT: 'heat'>,
]),
<ClimateEntityCapabilityAttribute.MAX_TEMP: 'max_temp'>: 30.0,
<ClimateEntityCapabilityAttribute.MIN_TEMP: 'min_temp'>: 16.0,
<ClimateEntityCapabilityAttribute.TARGET_TEMP_STEP: 'target_temp_step'>: 1.0,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'climate',
'entity_category': None,
'entity_id': 'climate.bedroom_ac_zone_1_thermostat',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Zone 1 thermostat',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Zone 1 thermostat',
'platform': 'midea_lan',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <ClimateEntityFeature: 385>,
'translation_key': 'climate_zone1',
'unique_id': '12345678_climate_zone1',
'unit_of_measurement': None,
})
# ---
# name: test_climate_state_snapshot[c3][climate.bedroom_ac_zone_1_thermostat-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<ClimateEntityStateAttribute.CURRENT_TEMPERATURE: 'current_temperature'>: 21.5,
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Bedroom AC Zone 1 thermostat',
<ClimateEntityCapabilityAttribute.HVAC_MODES: 'hvac_modes'>: list([
<HVACMode.OFF: 'off'>,
<HVACMode.AUTO: 'auto'>,
<HVACMode.COOL: 'cool'>,
<HVACMode.HEAT: 'heat'>,
]),
<ClimateEntityCapabilityAttribute.MAX_TEMP: 'max_temp'>: 30.0,
<ClimateEntityCapabilityAttribute.MIN_TEMP: 'min_temp'>: 16.0,
<EntityStateAttribute.SUPPORTED_FEATURES: 'supported_features'>: <ClimateEntityFeature: 385>,
<ClimateEntityCapabilityAttribute.TARGET_TEMP_STEP: 'target_temp_step'>: 1.0,
<ClimateEntityStateAttribute.TEMPERATURE: 'temperature'>: 22.0,
}),
'context': <ANY>,
'entity_id': 'climate.bedroom_ac_zone_1_thermostat',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'auto',
})
# ---
# name: test_climate_state_snapshot[c3][climate.bedroom_ac_zone_2_thermostat-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<ClimateEntityCapabilityAttribute.HVAC_MODES: 'hvac_modes'>: list([
<HVACMode.OFF: 'off'>,
<HVACMode.AUTO: 'auto'>,
<HVACMode.COOL: 'cool'>,
<HVACMode.HEAT: 'heat'>,
]),
<ClimateEntityCapabilityAttribute.MAX_TEMP: 'max_temp'>: 29.0,
<ClimateEntityCapabilityAttribute.MIN_TEMP: 'min_temp'>: 17.0,
<ClimateEntityCapabilityAttribute.TARGET_TEMP_STEP: 'target_temp_step'>: 0.5,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'climate',
'entity_category': None,
'entity_id': 'climate.bedroom_ac_zone_2_thermostat',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Zone 2 thermostat',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Zone 2 thermostat',
'platform': 'midea_lan',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <ClimateEntityFeature: 385>,
'translation_key': 'climate_zone2',
'unique_id': '12345678_climate_zone2',
'unit_of_measurement': None,
})
# ---
# name: test_climate_state_snapshot[c3][climate.bedroom_ac_zone_2_thermostat-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<ClimateEntityStateAttribute.CURRENT_TEMPERATURE: 'current_temperature'>: 21.5,
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Bedroom AC Zone 2 thermostat',
<ClimateEntityCapabilityAttribute.HVAC_MODES: 'hvac_modes'>: list([
<HVACMode.OFF: 'off'>,
<HVACMode.AUTO: 'auto'>,
<HVACMode.COOL: 'cool'>,
<HVACMode.HEAT: 'heat'>,
]),
<ClimateEntityCapabilityAttribute.MAX_TEMP: 'max_temp'>: 29.0,
<ClimateEntityCapabilityAttribute.MIN_TEMP: 'min_temp'>: 17.0,
<EntityStateAttribute.SUPPORTED_FEATURES: 'supported_features'>: <ClimateEntityFeature: 385>,
<ClimateEntityCapabilityAttribute.TARGET_TEMP_STEP: 'target_temp_step'>: 0.5,
<ClimateEntityStateAttribute.TEMPERATURE: 'temperature'>: 23.0,
}),
'context': <ANY>,
'entity_id': 'climate.bedroom_ac_zone_2_thermostat',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_climate_state_snapshot[cc][climate.bedroom_ac-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<ClimateEntityCapabilityAttribute.FAN_MODES: 'fan_modes'>: list([
'Low',
'Medium',
'High',
'Auto',
]),
<ClimateEntityCapabilityAttribute.HVAC_MODES: 'hvac_modes'>: list([
<HVACMode.OFF: 'off'>,
<HVACMode.FAN_ONLY: 'fan_only'>,
<HVACMode.DRY: 'dry'>,
<HVACMode.HEAT: 'heat'>,
<HVACMode.COOL: 'cool'>,
<HVACMode.AUTO: 'auto'>,
]),
<ClimateEntityCapabilityAttribute.MAX_TEMP: 'max_temp'>: 30,
<ClimateEntityCapabilityAttribute.MIN_TEMP: 'min_temp'>: 16,
<ClimateEntityCapabilityAttribute.PRESET_MODES: 'preset_modes'>: list([
'none',
'sleep',
'eco',
]),
<ClimateEntityCapabilityAttribute.SWING_MODES: 'swing_modes'>: list([
'off',
'on',
]),
<ClimateEntityCapabilityAttribute.TARGET_TEMP_STEP: 'target_temp_step'>: 0.5,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'climate',
'entity_category': None,
'entity_id': 'climate.bedroom_ac',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': None,
'platform': 'midea_lan',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <ClimateEntityFeature: 441>,
'translation_key': None,
'unique_id': '12345678_climate',
'unit_of_measurement': None,
})
# ---
# name: test_climate_state_snapshot[cc][climate.bedroom_ac-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<ClimateEntityStateAttribute.CURRENT_TEMPERATURE: 'current_temperature'>: None,
<ClimateEntityStateAttribute.FAN_MODE: 'fan_mode'>: 'High',
<ClimateEntityCapabilityAttribute.FAN_MODES: 'fan_modes'>: list([
'Low',
'Medium',
'High',
'Auto',
]),
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Bedroom AC',
<ClimateEntityCapabilityAttribute.HVAC_MODES: 'hvac_modes'>: list([
<HVACMode.OFF: 'off'>,
<HVACMode.FAN_ONLY: 'fan_only'>,
<HVACMode.DRY: 'dry'>,
<HVACMode.HEAT: 'heat'>,
<HVACMode.COOL: 'cool'>,
<HVACMode.AUTO: 'auto'>,
]),
<ClimateEntityCapabilityAttribute.MAX_TEMP: 'max_temp'>: 30,
<ClimateEntityCapabilityAttribute.MIN_TEMP: 'min_temp'>: 16,
<ClimateEntityStateAttribute.PRESET_MODE: 'preset_mode'>: 'none',
<ClimateEntityCapabilityAttribute.PRESET_MODES: 'preset_modes'>: list([
'none',
'sleep',
'eco',
]),
<EntityStateAttribute.SUPPORTED_FEATURES: 'supported_features'>: <ClimateEntityFeature: 441>,
<ClimateEntityStateAttribute.SWING_MODE: 'swing_mode'>: 'on',
<ClimateEntityCapabilityAttribute.SWING_MODES: 'swing_modes'>: list([
'off',
'on',
]),
<ClimateEntityCapabilityAttribute.TARGET_TEMP_STEP: 'target_temp_step'>: 0.5,
<ClimateEntityStateAttribute.TEMPERATURE: 'temperature'>: None,
}),
'context': <ANY>,
'entity_id': 'climate.bedroom_ac',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'auto',
})
# ---
# name: test_climate_state_snapshot[cf][climate.bedroom_ac-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<ClimateEntityCapabilityAttribute.HVAC_MODES: 'hvac_modes'>: list([
<HVACMode.OFF: 'off'>,
<HVACMode.AUTO: 'auto'>,
<HVACMode.COOL: 'cool'>,
<HVACMode.HEAT: 'heat'>,
]),
<ClimateEntityCapabilityAttribute.MAX_TEMP: 'max_temp'>: 30.0,
<ClimateEntityCapabilityAttribute.MIN_TEMP: 'min_temp'>: 16.0,
<ClimateEntityCapabilityAttribute.TARGET_TEMP_STEP: 'target_temp_step'>: 1,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'climate',
'entity_category': None,
'entity_id': 'climate.bedroom_ac',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': None,
'platform': 'midea_lan',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <ClimateEntityFeature: 385>,
'translation_key': None,
'unique_id': '12345678_climate',
'unit_of_measurement': None,
})
# ---
# name: test_climate_state_snapshot[cf][climate.bedroom_ac-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<ClimateEntityStateAttribute.CURRENT_TEMPERATURE: 'current_temperature'>: 22.0,
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Bedroom AC',
<ClimateEntityCapabilityAttribute.HVAC_MODES: 'hvac_modes'>: list([
<HVACMode.OFF: 'off'>,
<HVACMode.AUTO: 'auto'>,
<HVACMode.COOL: 'cool'>,
<HVACMode.HEAT: 'heat'>,
]),
<ClimateEntityCapabilityAttribute.MAX_TEMP: 'max_temp'>: 30.0,
<ClimateEntityCapabilityAttribute.MIN_TEMP: 'min_temp'>: 16.0,
<EntityStateAttribute.SUPPORTED_FEATURES: 'supported_features'>: <ClimateEntityFeature: 385>,
<ClimateEntityCapabilityAttribute.TARGET_TEMP_STEP: 'target_temp_step'>: 1,
<ClimateEntityStateAttribute.TEMPERATURE: 'temperature'>: None,
}),
'context': <ANY>,
'entity_id': 'climate.bedroom_ac',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'cool',
})
# ---
# name: test_climate_state_snapshot[fb][climate.bedroom_ac-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<ClimateEntityCapabilityAttribute.HVAC_MODES: 'hvac_modes'>: list([
<HVACMode.OFF: 'off'>,
<HVACMode.HEAT: 'heat'>,
]),
<ClimateEntityCapabilityAttribute.MAX_TEMP: 'max_temp'>: 35,
<ClimateEntityCapabilityAttribute.MIN_TEMP: 'min_temp'>: 5,
<ClimateEntityCapabilityAttribute.PRESET_MODES: 'preset_modes'>: list([
'Auto',
'ECO',
'Sleep',
'Anti-freezing',
'Comfort',
'Constant-temperature',
'Normal',
'Fast-heating',
'Standby',
]),
<ClimateEntityCapabilityAttribute.TARGET_TEMP_STEP: 'target_temp_step'>: 1,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'climate',
'entity_category': None,
'entity_id': 'climate.bedroom_ac',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': None,
'platform': 'midea_lan',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <ClimateEntityFeature: 401>,
'translation_key': None,
'unique_id': '12345678_climate',
'unit_of_measurement': None,
})
# ---
# name: test_climate_state_snapshot[fb][climate.bedroom_ac-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<ClimateEntityStateAttribute.CURRENT_TEMPERATURE: 'current_temperature'>: 20.0,
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Bedroom AC',
<ClimateEntityCapabilityAttribute.HVAC_MODES: 'hvac_modes'>: list([
<HVACMode.OFF: 'off'>,
<HVACMode.HEAT: 'heat'>,
]),
<ClimateEntityCapabilityAttribute.MAX_TEMP: 'max_temp'>: 35,
<ClimateEntityCapabilityAttribute.MIN_TEMP: 'min_temp'>: 5,
<ClimateEntityStateAttribute.PRESET_MODE: 'preset_mode'>: 'Comfort',
<ClimateEntityCapabilityAttribute.PRESET_MODES: 'preset_modes'>: list([
'Auto',
'ECO',
'Sleep',
'Anti-freezing',
'Comfort',
'Constant-temperature',
'Normal',
'Fast-heating',
'Standby',
]),
<EntityStateAttribute.SUPPORTED_FEATURES: 'supported_features'>: <ClimateEntityFeature: 401>,
<ClimateEntityCapabilityAttribute.TARGET_TEMP_STEP: 'target_temp_step'>: 1,
<ClimateEntityStateAttribute.TEMPERATURE: 'temperature'>: None,
}),
'context': <ANY>,
'entity_id': 'climate.bedroom_ac',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'heat',
})
# ---
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
+101
View File
@@ -0,0 +1,101 @@
"""Tests for midea_lan entity behavior via loaded platforms."""
from collections.abc import Callable
from midealocal.devices.ac import DeviceAttributes as ACAttributes
import pytest
from homeassistant.core import CoreState, HomeAssistant
from . import setup_integration
from .conftest import DummyDevice, default_ac_device, entity_entries
from .const import TEST_DEVICE_ID
from tests.common import MockConfigEntry
@pytest.mark.parametrize(
(
"update",
"status",
"availability",
"expected_current_temp",
"expected_unavailable",
),
[
pytest.param(
{ACAttributes.indoor_temperature: 24.0},
{"available": True},
True,
24.0,
False,
id="temperature_update",
),
pytest.param(
{},
{"available": False},
False,
None,
True,
id="availability_update",
),
pytest.param(
{ACAttributes.indoor_temperature: 24.0},
{"power": True, ACAttributes.indoor_temperature: 24.0},
True,
24.0,
False,
id="attribute_update_without_available_key",
),
],
)
async def test_entity_updates_from_device_callback(
hass: HomeAssistant,
mock_config_entry: Callable[[DummyDevice], MockConfigEntry],
update: dict[str, float],
status: dict[str, bool | float],
availability: bool,
expected_current_temp: float | None,
expected_unavailable: bool,
) -> None:
"""Test entity callback updates state and availability."""
device = default_ac_device()
config_entry = mock_config_entry(device)
await setup_integration(hass, config_entry, device)
entity_entry = entity_entries(hass, config_entry)[f"{TEST_DEVICE_ID}_climate"]
assert (state := hass.states.get(entity_entry.entity_id))
assert state.attributes["current_temperature"] == 21.0
assert state.state != "unavailable"
device.attributes.update(update)
device.available = availability
device.notify_update(status)
await hass.async_block_till_done()
assert (state := hass.states.get(entity_entry.entity_id))
assert state.attributes.get("current_temperature") == expected_current_temp
assert (state.state == "unavailable") is expected_unavailable
async def test_entity_callback_ignored_while_hass_stopping(
hass: HomeAssistant,
mock_config_entry: Callable[[DummyDevice], MockConfigEntry],
) -> None:
"""Test update callback does not schedule updates while Home Assistant stops."""
device = default_ac_device()
config_entry = mock_config_entry(device)
await setup_integration(hass, config_entry, device)
entity_entry = entity_entries(hass, config_entry)[f"{TEST_DEVICE_ID}_climate"]
assert hass.states.get(entity_entry.entity_id) is not None
device.attributes[ACAttributes.indoor_temperature] = 25.0
hass.set_state(CoreState.stopping)
device.notify_update({"available": True})
await hass.async_block_till_done()
assert (state := hass.states.get(entity_entry.entity_id))
assert state.attributes["current_temperature"] == 21.0
+109
View File
@@ -0,0 +1,109 @@
"""Tests for midea_lan __init__.py."""
from unittest.mock import patch
from midealocal.const import DeviceType, ProtocolVersion
from homeassistant.components.midea_lan.const import CONF_KEY, CONF_SUBTYPE, DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import (
CONF_DEVICE_ID,
CONF_IP_ADDRESS,
CONF_MODEL,
CONF_NAME,
CONF_PORT,
CONF_PROTOCOL,
CONF_TOKEN,
CONF_TYPE,
)
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from .conftest import DummyDevice
from .const import TEST_DEVICE_ID
from tests.common import MockConfigEntry
_ENTRY_DATA = {
CONF_DEVICE_ID: TEST_DEVICE_ID,
CONF_NAME: "m",
CONF_TYPE: DeviceType.AC,
CONF_IP_ADDRESS: "1.1.1.1",
CONF_PORT: 6444,
CONF_MODEL: "m",
CONF_PROTOCOL: ProtocolVersion.V2,
CONF_TOKEN: "",
CONF_KEY: "",
CONF_SUBTYPE: 0,
}
async def test_async_setup(hass: HomeAssistant) -> None:
"""Test the midea_lan domain can be set up without any config entries."""
assert await async_setup_component(hass, DOMAIN, {})
async def test_unload_entry(hass: HomeAssistant) -> None:
"""Test async_unload_entry unloads platforms and closes the device."""
entry = MockConfigEntry(domain=DOMAIN, data=_ENTRY_DATA)
entry.add_to_hass(hass)
device = DummyDevice(DeviceType.AC)
with patch(
"homeassistant.components.midea_lan.device_selector",
return_value=device,
):
await hass.config_entries.async_setup(entry.entry_id)
assert entry.state is ConfigEntryState.LOADED
assert device.daemon is True
assert await hass.config_entries.async_unload(entry.entry_id)
assert entry.state is ConfigEntryState.NOT_LOADED
assert ("close",) in device.calls
async def test_async_setup_entry_paths(hass: HomeAssistant) -> None:
"""Test async_setup_entry for success and no-device return."""
entry = MockConfigEntry(domain=DOMAIN, data=_ENTRY_DATA)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.midea_lan.device_selector",
return_value=DummyDevice(DeviceType.AC),
):
await hass.config_entries.async_setup(entry.entry_id)
assert entry.state is ConfigEntryState.LOADED
entry2 = MockConfigEntry(
domain=DOMAIN,
data={**_ENTRY_DATA, CONF_DEVICE_ID: TEST_DEVICE_ID + 1},
)
entry2.add_to_hass(hass)
with patch(
"homeassistant.components.midea_lan.device_selector",
return_value=None,
):
await hass.config_entries.async_setup(entry2.entry_id)
assert entry2.state is ConfigEntryState.SETUP_ERROR
async def test_setup_entry_not_ready_on_connect_failure(
hass: HomeAssistant,
) -> None:
"""Test async_setup_entry raises ConfigEntryNotReady when connect returns False.
The real device.connect() already catches SocketException/AuthException
internally and reports failure by returning False; it never raises them.
It can also leave the socket open in that case (e.g. when authentication
fails), so the socket must be closed explicitly to avoid a ResourceWarning.
"""
entry = MockConfigEntry(domain=DOMAIN, data=_ENTRY_DATA)
entry.add_to_hass(hass)
device = DummyDevice(DeviceType.AC)
with (
patch(
"homeassistant.components.midea_lan.device_selector",
return_value=device,
),
patch.object(device, "connect", return_value=False),
):
await hass.config_entries.async_setup(entry.entry_id)
assert entry.state is ConfigEntryState.SETUP_RETRY
assert ("close_socket",) in device.calls