evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
hass-core/homeassistant/components/xiaomi_miio/fan.py
Ville Skyttä b4bac0f7a0
Exception chaining and wrapping improvements (#39320) 
* Remove unnecessary exception re-wraps

* Preserve exception chains on re-raise

We slap "from cause" to almost all possible cases here. In some cases it
could conceivably be better to do "from None" if we really want to hide
the cause. However those should be in the minority, and "from cause"
should be an improvement over the corresponding raise without a "from"
in all cases anyway.

The only case where we raise from None here is in plex, where the
exception for an original invalid SSL cert is not the root cause for
failure to validate a newly fetched one.

Follow local convention on exception variable names if there is a
consistent one, otherwise `err` to match with majority of codebase.

* Fix mistaken re-wrap in homematicip_cloud/hap.py

Missed the difference between HmipConnectionError and
HmipcConnectionError.

* Do not hide original error on plex new cert validation error

Original is not the cause for the new one, but showing old in the
traceback is useful nevertheless.
2020-08-28 13:50:32 +02:00

1220 lines
39 KiB
Python
Raw Blame History

"""Support for Xiaomi Mi Air Purifier and Xiaomi Mi Air Humidifier."""
import asyncio
from enum import Enum
from functools import partial
import logging
from miio import ( # pylint: disable=import-error
AirFresh,
AirHumidifier,
AirPurifier,
AirPurifierMiot,
Device,
DeviceException,
)
from miio.airfresh import ( # pylint: disable=import-error, import-error
LedBrightness as AirfreshLedBrightness,
OperationMode as AirfreshOperationMode,
)
from miio.airhumidifier import ( # pylint: disable=import-error, import-error
LedBrightness as AirhumidifierLedBrightness,
OperationMode as AirhumidifierOperationMode,
)
from miio.airpurifier import ( # pylint: disable=import-error, import-error
LedBrightness as AirpurifierLedBrightness,
OperationMode as AirpurifierOperationMode,
)
from miio.airpurifier_miot import ( # pylint: disable=import-error, import-error
LedBrightness as AirpurifierMiotLedBrightness,
OperationMode as AirpurifierMiotOperationMode,
)
import voluptuous as vol
from homeassistant.components.fan import PLATFORM_SCHEMA, SUPPORT_SET_SPEED, FanEntity
from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_MODE,
CONF_HOST,
CONF_NAME,
CONF_TOKEN,
)
from homeassistant.exceptions import PlatformNotReady
import homeassistant.helpers.config_validation as cv
from .const import (
DOMAIN,
SERVICE_RESET_FILTER,
SERVICE_SET_AUTO_DETECT_OFF,
SERVICE_SET_AUTO_DETECT_ON,
SERVICE_SET_BUZZER_OFF,
SERVICE_SET_BUZZER_ON,
SERVICE_SET_CHILD_LOCK_OFF,
SERVICE_SET_CHILD_LOCK_ON,
SERVICE_SET_DRY_OFF,
SERVICE_SET_DRY_ON,
SERVICE_SET_EXTRA_FEATURES,
SERVICE_SET_FAN_LED_OFF,
SERVICE_SET_FAN_LED_ON,
SERVICE_SET_FAN_LEVEL,
SERVICE_SET_FAVORITE_LEVEL,
SERVICE_SET_LEARN_MODE_OFF,
SERVICE_SET_LEARN_MODE_ON,
SERVICE_SET_LED_BRIGHTNESS,
SERVICE_SET_TARGET_HUMIDITY,
SERVICE_SET_VOLUME,
)
_LOGGER = logging.getLogger(__name__)
DEFAULT_NAME = "Xiaomi Miio Device"
DATA_KEY = "fan.xiaomi_miio"
CONF_MODEL = "model"
MODEL_AIRPURIFIER_V1 = "zhimi.airpurifier.v1"
MODEL_AIRPURIFIER_V2 = "zhimi.airpurifier.v2"
MODEL_AIRPURIFIER_V3 = "zhimi.airpurifier.v3"
MODEL_AIRPURIFIER_V5 = "zhimi.airpurifier.v5"
MODEL_AIRPURIFIER_PRO = "zhimi.airpurifier.v6"
MODEL_AIRPURIFIER_PRO_V7 = "zhimi.airpurifier.v7"
MODEL_AIRPURIFIER_M1 = "zhimi.airpurifier.m1"
MODEL_AIRPURIFIER_M2 = "zhimi.airpurifier.m2"
MODEL_AIRPURIFIER_MA1 = "zhimi.airpurifier.ma1"
MODEL_AIRPURIFIER_MA2 = "zhimi.airpurifier.ma2"
MODEL_AIRPURIFIER_SA1 = "zhimi.airpurifier.sa1"
MODEL_AIRPURIFIER_SA2 = "zhimi.airpurifier.sa2"
MODEL_AIRPURIFIER_2S = "zhimi.airpurifier.mc1"
MODEL_AIRPURIFIER_3 = "zhimi.airpurifier.ma4"
MODEL_AIRPURIFIER_3H = "zhimi.airpurifier.mb3"
MODEL_AIRHUMIDIFIER_V1 = "zhimi.humidifier.v1"
MODEL_AIRHUMIDIFIER_CA1 = "zhimi.humidifier.ca1"
MODEL_AIRHUMIDIFIER_CB1 = "zhimi.humidifier.cb1"
MODEL_AIRFRESH_VA2 = "zhimi.airfresh.va2"
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_TOKEN): vol.All(cv.string, vol.Length(min=32, max=32)),
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_MODEL): vol.In(
[
MODEL_AIRPURIFIER_V1,
MODEL_AIRPURIFIER_V2,
MODEL_AIRPURIFIER_V3,
MODEL_AIRPURIFIER_V5,
MODEL_AIRPURIFIER_PRO,
MODEL_AIRPURIFIER_PRO_V7,
MODEL_AIRPURIFIER_M1,
MODEL_AIRPURIFIER_M2,
MODEL_AIRPURIFIER_MA1,
MODEL_AIRPURIFIER_MA2,
MODEL_AIRPURIFIER_SA1,
MODEL_AIRPURIFIER_SA2,
MODEL_AIRPURIFIER_2S,
MODEL_AIRPURIFIER_3,
MODEL_AIRPURIFIER_3H,
MODEL_AIRHUMIDIFIER_V1,
MODEL_AIRHUMIDIFIER_CA1,
MODEL_AIRHUMIDIFIER_CB1,
MODEL_AIRFRESH_VA2,
]
),
}
)
ATTR_MODEL = "model"
# Air Purifier
ATTR_TEMPERATURE = "temperature"
ATTR_HUMIDITY = "humidity"
ATTR_AIR_QUALITY_INDEX = "aqi"
ATTR_FILTER_HOURS_USED = "filter_hours_used"
ATTR_FILTER_LIFE = "filter_life_remaining"
ATTR_FAVORITE_LEVEL = "favorite_level"
ATTR_BUZZER = "buzzer"
ATTR_CHILD_LOCK = "child_lock"
ATTR_LED = "led"
ATTR_LED_BRIGHTNESS = "led_brightness"
ATTR_MOTOR_SPEED = "motor_speed"
ATTR_AVERAGE_AIR_QUALITY_INDEX = "average_aqi"
ATTR_PURIFY_VOLUME = "purify_volume"
ATTR_BRIGHTNESS = "brightness"
ATTR_LEVEL = "level"
ATTR_FAN_LEVEL = "fan_level"
ATTR_MOTOR2_SPEED = "motor2_speed"
ATTR_ILLUMINANCE = "illuminance"
ATTR_FILTER_RFID_PRODUCT_ID = "filter_rfid_product_id"
ATTR_FILTER_RFID_TAG = "filter_rfid_tag"
ATTR_FILTER_TYPE = "filter_type"
ATTR_LEARN_MODE = "learn_mode"
ATTR_SLEEP_TIME = "sleep_time"
ATTR_SLEEP_LEARN_COUNT = "sleep_mode_learn_count"
ATTR_EXTRA_FEATURES = "extra_features"
ATTR_FEATURES = "features"
ATTR_TURBO_MODE_SUPPORTED = "turbo_mode_supported"
ATTR_AUTO_DETECT = "auto_detect"
ATTR_SLEEP_MODE = "sleep_mode"
ATTR_VOLUME = "volume"
ATTR_USE_TIME = "use_time"
ATTR_BUTTON_PRESSED = "button_pressed"
# Air Humidifier
ATTR_TARGET_HUMIDITY = "target_humidity"
ATTR_TRANS_LEVEL = "trans_level"
ATTR_HARDWARE_VERSION = "hardware_version"
# Air Humidifier CA
# ATTR_MOTOR_SPEED = "motor_speed"
ATTR_DEPTH = "depth"
ATTR_DRY = "dry"
# Air Fresh
ATTR_CO2 = "co2"
PURIFIER_MIOT = [MODEL_AIRPURIFIER_3, MODEL_AIRPURIFIER_3H]
# Map attributes to properties of the state object
AVAILABLE_ATTRIBUTES_AIRPURIFIER_COMMON = {
ATTR_TEMPERATURE: "temperature",
ATTR_HUMIDITY: "humidity",
ATTR_AIR_QUALITY_INDEX: "aqi",
ATTR_MODE: "mode",
ATTR_FILTER_HOURS_USED: "filter_hours_used",
ATTR_FILTER_LIFE: "filter_life_remaining",
ATTR_FAVORITE_LEVEL: "favorite_level",
ATTR_CHILD_LOCK: "child_lock",
ATTR_LED: "led",
ATTR_MOTOR_SPEED: "motor_speed",
ATTR_AVERAGE_AIR_QUALITY_INDEX: "average_aqi",
ATTR_LEARN_MODE: "learn_mode",
ATTR_EXTRA_FEATURES: "extra_features",
ATTR_TURBO_MODE_SUPPORTED: "turbo_mode_supported",
ATTR_BUTTON_PRESSED: "button_pressed",
}
AVAILABLE_ATTRIBUTES_AIRPURIFIER = {
**AVAILABLE_ATTRIBUTES_AIRPURIFIER_COMMON,
ATTR_PURIFY_VOLUME: "purify_volume",
ATTR_SLEEP_TIME: "sleep_time",
ATTR_SLEEP_LEARN_COUNT: "sleep_mode_learn_count",
ATTR_AUTO_DETECT: "auto_detect",
ATTR_USE_TIME: "use_time",
ATTR_BUZZER: "buzzer",
ATTR_LED_BRIGHTNESS: "led_brightness",
ATTR_SLEEP_MODE: "sleep_mode",
}
AVAILABLE_ATTRIBUTES_AIRPURIFIER_PRO = {
**AVAILABLE_ATTRIBUTES_AIRPURIFIER_COMMON,
ATTR_PURIFY_VOLUME: "purify_volume",
ATTR_USE_TIME: "use_time",
ATTR_FILTER_RFID_PRODUCT_ID: "filter_rfid_product_id",
ATTR_FILTER_RFID_TAG: "filter_rfid_tag",
ATTR_FILTER_TYPE: "filter_type",
ATTR_ILLUMINANCE: "illuminance",
ATTR_MOTOR2_SPEED: "motor2_speed",
ATTR_VOLUME: "volume",
# perhaps supported but unconfirmed
ATTR_AUTO_DETECT: "auto_detect",
ATTR_SLEEP_TIME: "sleep_time",
ATTR_SLEEP_LEARN_COUNT: "sleep_mode_learn_count",
}
AVAILABLE_ATTRIBUTES_AIRPURIFIER_PRO_V7 = {
**AVAILABLE_ATTRIBUTES_AIRPURIFIER_COMMON,
ATTR_FILTER_RFID_PRODUCT_ID: "filter_rfid_product_id",
ATTR_FILTER_RFID_TAG: "filter_rfid_tag",
ATTR_FILTER_TYPE: "filter_type",
ATTR_ILLUMINANCE: "illuminance",
ATTR_MOTOR2_SPEED: "motor2_speed",
ATTR_VOLUME: "volume",
}
AVAILABLE_ATTRIBUTES_AIRPURIFIER_2S = {
**AVAILABLE_ATTRIBUTES_AIRPURIFIER_COMMON,
ATTR_BUZZER: "buzzer",
ATTR_FILTER_RFID_PRODUCT_ID: "filter_rfid_product_id",
ATTR_FILTER_RFID_TAG: "filter_rfid_tag",
ATTR_FILTER_TYPE: "filter_type",
ATTR_ILLUMINANCE: "illuminance",
}
AVAILABLE_ATTRIBUTES_AIRPURIFIER_3 = {
ATTR_TEMPERATURE: "temperature",
ATTR_HUMIDITY: "humidity",
ATTR_AIR_QUALITY_INDEX: "aqi",
ATTR_MODE: "mode",
ATTR_FILTER_HOURS_USED: "filter_hours_used",
ATTR_FILTER_LIFE: "filter_life_remaining",
ATTR_FAVORITE_LEVEL: "favorite_level",
ATTR_CHILD_LOCK: "child_lock",
ATTR_LED: "led",
ATTR_MOTOR_SPEED: "motor_speed",
ATTR_AVERAGE_AIR_QUALITY_INDEX: "average_aqi",
ATTR_PURIFY_VOLUME: "purify_volume",
ATTR_USE_TIME: "use_time",
ATTR_BUZZER: "buzzer",
ATTR_LED_BRIGHTNESS: "led_brightness",
ATTR_FILTER_RFID_PRODUCT_ID: "filter_rfid_product_id",
ATTR_FILTER_RFID_TAG: "filter_rfid_tag",
ATTR_FILTER_TYPE: "filter_type",
ATTR_FAN_LEVEL: "fan_level",
}
AVAILABLE_ATTRIBUTES_AIRPURIFIER_V3 = {
# Common set isn't used here. It's a very basic version of the device.
ATTR_AIR_QUALITY_INDEX: "aqi",
ATTR_MODE: "mode",
ATTR_LED: "led",
ATTR_BUZZER: "buzzer",
ATTR_CHILD_LOCK: "child_lock",
ATTR_ILLUMINANCE: "illuminance",
ATTR_FILTER_HOURS_USED: "filter_hours_used",
ATTR_FILTER_LIFE: "filter_life_remaining",
ATTR_MOTOR_SPEED: "motor_speed",
# perhaps supported but unconfirmed
ATTR_AVERAGE_AIR_QUALITY_INDEX: "average_aqi",
ATTR_VOLUME: "volume",
ATTR_MOTOR2_SPEED: "motor2_speed",
ATTR_FILTER_RFID_PRODUCT_ID: "filter_rfid_product_id",
ATTR_FILTER_RFID_TAG: "filter_rfid_tag",
ATTR_FILTER_TYPE: "filter_type",
ATTR_PURIFY_VOLUME: "purify_volume",
ATTR_LEARN_MODE: "learn_mode",
ATTR_SLEEP_TIME: "sleep_time",
ATTR_SLEEP_LEARN_COUNT: "sleep_mode_learn_count",
ATTR_EXTRA_FEATURES: "extra_features",
ATTR_AUTO_DETECT: "auto_detect",
ATTR_USE_TIME: "use_time",
ATTR_BUTTON_PRESSED: "button_pressed",
}
AVAILABLE_ATTRIBUTES_AIRHUMIDIFIER_COMMON = {
ATTR_TEMPERATURE: "temperature",
ATTR_HUMIDITY: "humidity",
ATTR_MODE: "mode",
ATTR_BUZZER: "buzzer",
ATTR_CHILD_LOCK: "child_lock",
ATTR_TARGET_HUMIDITY: "target_humidity",
ATTR_LED_BRIGHTNESS: "led_brightness",
ATTR_USE_TIME: "use_time",
ATTR_HARDWARE_VERSION: "hardware_version",
}
AVAILABLE_ATTRIBUTES_AIRHUMIDIFIER = {
**AVAILABLE_ATTRIBUTES_AIRHUMIDIFIER_COMMON,
ATTR_TRANS_LEVEL: "trans_level",
ATTR_BUTTON_PRESSED: "button_pressed",
}
AVAILABLE_ATTRIBUTES_AIRHUMIDIFIER_CA_AND_CB = {
**AVAILABLE_ATTRIBUTES_AIRHUMIDIFIER_COMMON,
ATTR_MOTOR_SPEED: "motor_speed",
ATTR_DEPTH: "depth",
ATTR_DRY: "dry",
}
AVAILABLE_ATTRIBUTES_AIRFRESH = {
ATTR_TEMPERATURE: "temperature",
ATTR_AIR_QUALITY_INDEX: "aqi",
ATTR_AVERAGE_AIR_QUALITY_INDEX: "average_aqi",
ATTR_CO2: "co2",
ATTR_HUMIDITY: "humidity",
ATTR_MODE: "mode",
ATTR_LED: "led",
ATTR_LED_BRIGHTNESS: "led_brightness",
ATTR_BUZZER: "buzzer",
ATTR_CHILD_LOCK: "child_lock",
ATTR_FILTER_LIFE: "filter_life_remaining",
ATTR_FILTER_HOURS_USED: "filter_hours_used",
ATTR_USE_TIME: "use_time",
ATTR_MOTOR_SPEED: "motor_speed",
ATTR_EXTRA_FEATURES: "extra_features",
}
OPERATION_MODES_AIRPURIFIER = ["Auto", "Silent", "Favorite", "Idle"]
OPERATION_MODES_AIRPURIFIER_PRO = ["Auto", "Silent", "Favorite"]
OPERATION_MODES_AIRPURIFIER_PRO_V7 = OPERATION_MODES_AIRPURIFIER_PRO
OPERATION_MODES_AIRPURIFIER_2S = ["Auto", "Silent", "Favorite"]
OPERATION_MODES_AIRPURIFIER_3 = ["Auto", "Silent", "Favorite", "Fan"]
OPERATION_MODES_AIRPURIFIER_V3 = [
"Auto",
"Silent",
"Favorite",
"Idle",
"Medium",
"High",
"Strong",
]
OPERATION_MODES_AIRFRESH = ["Auto", "Silent", "Interval", "Low", "Middle", "Strong"]
SUCCESS = ["ok"]
FEATURE_SET_BUZZER = 1
FEATURE_SET_LED = 2
FEATURE_SET_CHILD_LOCK = 4
FEATURE_SET_LED_BRIGHTNESS = 8
FEATURE_SET_FAVORITE_LEVEL = 16
FEATURE_SET_AUTO_DETECT = 32
FEATURE_SET_LEARN_MODE = 64
FEATURE_SET_VOLUME = 128
FEATURE_RESET_FILTER = 256
FEATURE_SET_EXTRA_FEATURES = 512
FEATURE_SET_TARGET_HUMIDITY = 1024
FEATURE_SET_DRY = 2048
FEATURE_SET_FAN_LEVEL = 4096
FEATURE_FLAGS_AIRPURIFIER = (
FEATURE_SET_BUZZER
| FEATURE_SET_CHILD_LOCK
| FEATURE_SET_LED
| FEATURE_SET_LED_BRIGHTNESS
| FEATURE_SET_FAVORITE_LEVEL
| FEATURE_SET_LEARN_MODE
| FEATURE_RESET_FILTER
| FEATURE_SET_EXTRA_FEATURES
)
FEATURE_FLAGS_AIRPURIFIER_PRO = (
FEATURE_SET_CHILD_LOCK
| FEATURE_SET_LED
| FEATURE_SET_FAVORITE_LEVEL
| FEATURE_SET_AUTO_DETECT
| FEATURE_SET_VOLUME
)
FEATURE_FLAGS_AIRPURIFIER_PRO_V7 = (
FEATURE_SET_CHILD_LOCK
| FEATURE_SET_LED
| FEATURE_SET_FAVORITE_LEVEL
| FEATURE_SET_VOLUME
)
FEATURE_FLAGS_AIRPURIFIER_2S = (
FEATURE_SET_BUZZER
| FEATURE_SET_CHILD_LOCK
| FEATURE_SET_LED
| FEATURE_SET_FAVORITE_LEVEL
)
FEATURE_FLAGS_AIRPURIFIER_3 = (
FEATURE_SET_BUZZER
| FEATURE_SET_CHILD_LOCK
| FEATURE_SET_LED
| FEATURE_SET_FAVORITE_LEVEL
| FEATURE_SET_FAN_LEVEL
| FEATURE_SET_LED_BRIGHTNESS
)
FEATURE_FLAGS_AIRPURIFIER_V3 = (
FEATURE_SET_BUZZER | FEATURE_SET_CHILD_LOCK | FEATURE_SET_LED
)
FEATURE_FLAGS_AIRHUMIDIFIER = (
FEATURE_SET_BUZZER
| FEATURE_SET_CHILD_LOCK
| FEATURE_SET_LED
| FEATURE_SET_LED_BRIGHTNESS
| FEATURE_SET_TARGET_HUMIDITY
)
FEATURE_FLAGS_AIRHUMIDIFIER_CA_AND_CB = FEATURE_FLAGS_AIRHUMIDIFIER | FEATURE_SET_DRY
FEATURE_FLAGS_AIRFRESH = (
FEATURE_SET_BUZZER
| FEATURE_SET_CHILD_LOCK
| FEATURE_SET_LED
| FEATURE_SET_LED_BRIGHTNESS
| FEATURE_RESET_FILTER
| FEATURE_SET_EXTRA_FEATURES
)
AIRPURIFIER_SERVICE_SCHEMA = vol.Schema({vol.Optional(ATTR_ENTITY_ID): cv.entity_ids})
SERVICE_SCHEMA_LED_BRIGHTNESS = AIRPURIFIER_SERVICE_SCHEMA.extend(
{vol.Required(ATTR_BRIGHTNESS): vol.All(vol.Coerce(int), vol.Clamp(min=0, max=2))}
)
SERVICE_SCHEMA_FAVORITE_LEVEL = AIRPURIFIER_SERVICE_SCHEMA.extend(
{vol.Required(ATTR_LEVEL): vol.All(vol.Coerce(int), vol.Clamp(min=0, max=17))}
)
SERVICE_SCHEMA_FAN_LEVEL = AIRPURIFIER_SERVICE_SCHEMA.extend(
{vol.Required(ATTR_LEVEL): vol.All(vol.Coerce(int), vol.Clamp(min=1, max=3))}
)
SERVICE_SCHEMA_VOLUME = AIRPURIFIER_SERVICE_SCHEMA.extend(
{vol.Required(ATTR_VOLUME): vol.All(vol.Coerce(int), vol.Clamp(min=0, max=100))}
)
SERVICE_SCHEMA_EXTRA_FEATURES = AIRPURIFIER_SERVICE_SCHEMA.extend(
{vol.Required(ATTR_FEATURES): vol.All(vol.Coerce(int), vol.Range(min=0))}
)
SERVICE_SCHEMA_TARGET_HUMIDITY = AIRPURIFIER_SERVICE_SCHEMA.extend(
{
vol.Required(ATTR_HUMIDITY): vol.All(
vol.Coerce(int), vol.In([30, 40, 50, 60, 70, 80])
)
}
)
SERVICE_TO_METHOD = {
SERVICE_SET_BUZZER_ON: {"method": "async_set_buzzer_on"},
SERVICE_SET_BUZZER_OFF: {"method": "async_set_buzzer_off"},
SERVICE_SET_FAN_LED_ON: {"method": "async_set_led_on"},
SERVICE_SET_FAN_LED_OFF: {"method": "async_set_led_off"},
SERVICE_SET_CHILD_LOCK_ON: {"method": "async_set_child_lock_on"},
SERVICE_SET_CHILD_LOCK_OFF: {"method": "async_set_child_lock_off"},
SERVICE_SET_AUTO_DETECT_ON: {"method": "async_set_auto_detect_on"},
SERVICE_SET_AUTO_DETECT_OFF: {"method": "async_set_auto_detect_off"},
SERVICE_SET_LEARN_MODE_ON: {"method": "async_set_learn_mode_on"},
SERVICE_SET_LEARN_MODE_OFF: {"method": "async_set_learn_mode_off"},
SERVICE_RESET_FILTER: {"method": "async_reset_filter"},
SERVICE_SET_LED_BRIGHTNESS: {
"method": "async_set_led_brightness",
"schema": SERVICE_SCHEMA_LED_BRIGHTNESS,
},
SERVICE_SET_FAVORITE_LEVEL: {
"method": "async_set_favorite_level",
"schema": SERVICE_SCHEMA_FAVORITE_LEVEL,
},
SERVICE_SET_FAN_LEVEL: {
"method": "async_set_fan_level",
"schema": SERVICE_SCHEMA_FAN_LEVEL,
},
SERVICE_SET_VOLUME: {"method": "async_set_volume", "schema": SERVICE_SCHEMA_VOLUME},
SERVICE_SET_EXTRA_FEATURES: {
"method": "async_set_extra_features",
"schema": SERVICE_SCHEMA_EXTRA_FEATURES,
},
SERVICE_SET_TARGET_HUMIDITY: {
"method": "async_set_target_humidity",
"schema": SERVICE_SCHEMA_TARGET_HUMIDITY,
},
SERVICE_SET_DRY_ON: {"method": "async_set_dry_on"},
SERVICE_SET_DRY_OFF: {"method": "async_set_dry_off"},
}
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up the miio fan device from config."""
if DATA_KEY not in hass.data:
hass.data[DATA_KEY] = {}
host = config[CONF_HOST]
token = config[CONF_TOKEN]
name = config[CONF_NAME]
model = config.get(CONF_MODEL)
_LOGGER.info("Initializing with host %s (token %s...)", host, token[:5])
unique_id = None
if model is None:
try:
miio_device = Device(host, token)
device_info = await hass.async_add_executor_job(miio_device.info)
model = device_info.model
unique_id = f"{model}-{device_info.mac_address}"
_LOGGER.info(
"%s %s %s detected",
model,
device_info.firmware_version,
device_info.hardware_version,
)
except DeviceException as ex:
raise PlatformNotReady from ex
if model in PURIFIER_MIOT:
air_purifier = AirPurifierMiot(host, token)
device = XiaomiAirPurifierMiot(name, air_purifier, model, unique_id)
elif model.startswith("zhimi.airpurifier."):
air_purifier = AirPurifier(host, token)
device = XiaomiAirPurifier(name, air_purifier, model, unique_id)
elif model.startswith("zhimi.humidifier."):
air_humidifier = AirHumidifier(host, token, model=model)
device = XiaomiAirHumidifier(name, air_humidifier, model, unique_id)
elif model.startswith("zhimi.airfresh."):
air_fresh = AirFresh(host, token)
device = XiaomiAirFresh(name, air_fresh, model, unique_id)
else:
_LOGGER.error(
"Unsupported device found! Please create an issue at "
"https://github.com/syssi/xiaomi_airpurifier/issues "
"and provide the following data: %s",
model,
)
return False
hass.data[DATA_KEY][host] = device
async_add_entities([device], update_before_add=True)
async def async_service_handler(service):
"""Map services to methods on XiaomiAirPurifier."""
method = SERVICE_TO_METHOD.get(service.service)
params = {
key: value for key, value in service.data.items() if key != ATTR_ENTITY_ID
}
entity_ids = service.data.get(ATTR_ENTITY_ID)
if entity_ids:
devices = [
device
for device in hass.data[DATA_KEY].values()
if device.entity_id in entity_ids
]
else:
devices = hass.data[DATA_KEY].values()
update_tasks = []
for device in devices:
if not hasattr(device, method["method"]):
continue
await getattr(device, method["method"])(**params)
update_tasks.append(device.async_update_ha_state(True))
if update_tasks:
await asyncio.wait(update_tasks)
for air_purifier_service in SERVICE_TO_METHOD:
schema = SERVICE_TO_METHOD[air_purifier_service].get(
"schema", AIRPURIFIER_SERVICE_SCHEMA
)
hass.services.async_register(
DOMAIN, air_purifier_service, async_service_handler, schema=schema
)
class XiaomiGenericDevice(FanEntity):
"""Representation of a generic Xiaomi device."""
def __init__(self, name, device, model, unique_id):
"""Initialize the generic Xiaomi device."""
self._name = name
self._device = device
self._model = model
self._unique_id = unique_id
self._available = False
self._state = None
self._state_attrs = {ATTR_MODEL: self._model}
self._device_features = FEATURE_SET_CHILD_LOCK
self._skip_update = False
@property
def supported_features(self):
"""Flag supported features."""
return SUPPORT_SET_SPEED
@property
def should_poll(self):
"""Poll the device."""
return True
@property
def unique_id(self):
"""Return an unique ID."""
return self._unique_id
@property
def name(self):
"""Return the name of the device if any."""
return self._name
@property
def available(self):
"""Return true when state is known."""
return self._available
@property
def device_state_attributes(self):
"""Return the state attributes of the device."""
return self._state_attrs
@property
def is_on(self):
"""Return true if device is on."""
return self._state
@staticmethod
def _extract_value_from_attribute(state, attribute):
value = getattr(state, attribute)
if isinstance(value, Enum):
return value.value
return value
async def _try_command(self, mask_error, func, *args, **kwargs):
"""Call a miio device command handling error messages."""
try:
result = await self.hass.async_add_executor_job(
partial(func, *args, **kwargs)
)
_LOGGER.debug("Response received from miio device: %s", result)
return result == SUCCESS
except DeviceException as exc:
_LOGGER.error(mask_error, exc)
self._available = False
return False
async def async_turn_on(self, speed: str = None, **kwargs) -> None:
"""Turn the device on."""
if speed:
# If operation mode was set the device must not be turned on.
result = await self.async_set_speed(speed)
else:
result = await self._try_command(
"Turning the miio device on failed.", self._device.on
)
if result:
self._state = True
self._skip_update = True
async def async_turn_off(self, **kwargs) -> None:
"""Turn the device off."""
result = await self._try_command(
"Turning the miio device off failed.", self._device.off
)
if result:
self._state = False
self._skip_update = True
async def async_set_buzzer_on(self):
"""Turn the buzzer on."""
if self._device_features & FEATURE_SET_BUZZER == 0:
return
await self._try_command(
"Turning the buzzer of the miio device on failed.",
self._device.set_buzzer,
True,
)
async def async_set_buzzer_off(self):
"""Turn the buzzer off."""
if self._device_features & FEATURE_SET_BUZZER == 0:
return
await self._try_command(
"Turning the buzzer of the miio device off failed.",
self._device.set_buzzer,
False,
)
async def async_set_child_lock_on(self):
"""Turn the child lock on."""
if self._device_features & FEATURE_SET_CHILD_LOCK == 0:
return
await self._try_command(
"Turning the child lock of the miio device on failed.",
self._device.set_child_lock,
True,
)
async def async_set_child_lock_off(self):
"""Turn the child lock off."""
if self._device_features & FEATURE_SET_CHILD_LOCK == 0:
return
await self._try_command(
"Turning the child lock of the miio device off failed.",
self._device.set_child_lock,
False,
)
class XiaomiAirPurifier(XiaomiGenericDevice):
"""Representation of a Xiaomi Air Purifier."""
def __init__(self, name, device, model, unique_id):
"""Initialize the plug switch."""
super().__init__(name, device, model, unique_id)
if self._model == MODEL_AIRPURIFIER_PRO:
self._device_features = FEATURE_FLAGS_AIRPURIFIER_PRO
self._available_attributes = AVAILABLE_ATTRIBUTES_AIRPURIFIER_PRO
self._speed_list = OPERATION_MODES_AIRPURIFIER_PRO
elif self._model == MODEL_AIRPURIFIER_PRO_V7:
self._device_features = FEATURE_FLAGS_AIRPURIFIER_PRO_V7
self._available_attributes = AVAILABLE_ATTRIBUTES_AIRPURIFIER_PRO_V7
self._speed_list = OPERATION_MODES_AIRPURIFIER_PRO_V7
elif self._model == MODEL_AIRPURIFIER_2S:
self._device_features = FEATURE_FLAGS_AIRPURIFIER_2S
self._available_attributes = AVAILABLE_ATTRIBUTES_AIRPURIFIER_2S
self._speed_list = OPERATION_MODES_AIRPURIFIER_2S
elif self._model == MODEL_AIRPURIFIER_3 or self._model == MODEL_AIRPURIFIER_3H:
self._device_features = FEATURE_FLAGS_AIRPURIFIER_3
self._available_attributes = AVAILABLE_ATTRIBUTES_AIRPURIFIER_3
self._speed_list = OPERATION_MODES_AIRPURIFIER_3
elif self._model == MODEL_AIRPURIFIER_V3:
self._device_features = FEATURE_FLAGS_AIRPURIFIER_V3
self._available_attributes = AVAILABLE_ATTRIBUTES_AIRPURIFIER_V3
self._speed_list = OPERATION_MODES_AIRPURIFIER_V3
else:
self._device_features = FEATURE_FLAGS_AIRPURIFIER
self._available_attributes = AVAILABLE_ATTRIBUTES_AIRPURIFIER
self._speed_list = OPERATION_MODES_AIRPURIFIER
self._state_attrs.update(
{attribute: None for attribute in self._available_attributes}
)
async def async_update(self):
"""Fetch state from the device."""
# On state change the device doesn't provide the new state immediately.
if self._skip_update:
self._skip_update = False
return
try:
state = await self.hass.async_add_executor_job(self._device.status)
_LOGGER.debug("Got new state: %s", state)
self._available = True
self._state = state.is_on
self._state_attrs.update(
{
key: self._extract_value_from_attribute(state, value)
for key, value in self._available_attributes.items()
}
)
except DeviceException as ex:
self._available = False
_LOGGER.error("Got exception while fetching the state: %s", ex)
@property
def speed_list(self) -> list:
"""Get the list of available speeds."""
return self._speed_list
@property
def speed(self):
"""Return the current speed."""
if self._state:
return AirpurifierOperationMode(self._state_attrs[ATTR_MODE]).name
return None
async def async_set_speed(self, speed: str) -> None:
"""Set the speed of the fan."""
if self.supported_features & SUPPORT_SET_SPEED == 0:
return
_LOGGER.debug("Setting the operation mode to: %s", speed)
await self._try_command(
"Setting operation mode of the miio device failed.",
self._device.set_mode,
AirpurifierOperationMode[speed.title()],
)
async def async_set_led_on(self):
"""Turn the led on."""
if self._device_features & FEATURE_SET_LED == 0:
return
await self._try_command(
"Turning the led of the miio device off failed.", self._device.set_led, True
)
async def async_set_led_off(self):
"""Turn the led off."""
if self._device_features & FEATURE_SET_LED == 0:
return
await self._try_command(
"Turning the led of the miio device off failed.",
self._device.set_led,
False,
)
async def async_set_led_brightness(self, brightness: int = 2):
"""Set the led brightness."""
if self._device_features & FEATURE_SET_LED_BRIGHTNESS == 0:
return
await self._try_command(
"Setting the led brightness of the miio device failed.",
self._device.set_led_brightness,
AirpurifierLedBrightness(brightness),
)
async def async_set_favorite_level(self, level: int = 1):
"""Set the favorite level."""
if self._device_features & FEATURE_SET_FAVORITE_LEVEL == 0:
return
await self._try_command(
"Setting the favorite level of the miio device failed.",
self._device.set_favorite_level,
level,
)
async def async_set_fan_level(self, level: int = 1):
"""Set the favorite level."""
if self._device_features & FEATURE_SET_FAN_LEVEL == 0:
return
await self._try_command(
"Setting the fan level of the miio device failed.",
self._device.set_fan_level,
level,
)
async def async_set_auto_detect_on(self):
"""Turn the auto detect on."""
if self._device_features & FEATURE_SET_AUTO_DETECT == 0:
return
await self._try_command(
"Turning the auto detect of the miio device on failed.",
self._device.set_auto_detect,
True,
)
async def async_set_auto_detect_off(self):
"""Turn the auto detect off."""
if self._device_features & FEATURE_SET_AUTO_DETECT == 0:
return
await self._try_command(
"Turning the auto detect of the miio device off failed.",
self._device.set_auto_detect,
False,
)
async def async_set_learn_mode_on(self):
"""Turn the learn mode on."""
if self._device_features & FEATURE_SET_LEARN_MODE == 0:
return
await self._try_command(
"Turning the learn mode of the miio device on failed.",
self._device.set_learn_mode,
True,
)
async def async_set_learn_mode_off(self):
"""Turn the learn mode off."""
if self._device_features & FEATURE_SET_LEARN_MODE == 0:
return
await self._try_command(
"Turning the learn mode of the miio device off failed.",
self._device.set_learn_mode,
False,
)
async def async_set_volume(self, volume: int = 50):
"""Set the sound volume."""
if self._device_features & FEATURE_SET_VOLUME == 0:
return
await self._try_command(
"Setting the sound volume of the miio device failed.",
self._device.set_volume,
volume,
)
async def async_set_extra_features(self, features: int = 1):
"""Set the extra features."""
if self._device_features & FEATURE_SET_EXTRA_FEATURES == 0:
return
await self._try_command(
"Setting the extra features of the miio device failed.",
self._device.set_extra_features,
features,
)
async def async_reset_filter(self):
"""Reset the filter lifetime and usage."""
if self._device_features & FEATURE_RESET_FILTER == 0:
return
await self._try_command(
"Resetting the filter lifetime of the miio device failed.",
self._device.reset_filter,
)
class XiaomiAirPurifierMiot(XiaomiAirPurifier):
"""Representation of a Xiaomi Air Purifier (MiOT protocol)."""
@property
def speed(self):
"""Return the current speed."""
if self._state:
return AirpurifierMiotOperationMode(self._state_attrs[ATTR_MODE]).name
return None
async def async_set_speed(self, speed: str) -> None:
"""Set the speed of the fan."""
if self.supported_features & SUPPORT_SET_SPEED == 0:
return
_LOGGER.debug("Setting the operation mode to: %s", speed)
await self._try_command(
"Setting operation mode of the miio device failed.",
self._device.set_mode,
AirpurifierMiotOperationMode[speed.title()],
)
async def async_set_led_brightness(self, brightness: int = 2):
"""Set the led brightness."""
if self._device_features & FEATURE_SET_LED_BRIGHTNESS == 0:
return
await self._try_command(
"Setting the led brightness of the miio device failed.",
self._device.set_led_brightness,
AirpurifierMiotLedBrightness(brightness),
)
class XiaomiAirHumidifier(XiaomiGenericDevice):
"""Representation of a Xiaomi Air Humidifier."""
def __init__(self, name, device, model, unique_id):
"""Initialize the plug switch."""
super().__init__(name, device, model, unique_id)
if self._model in [MODEL_AIRHUMIDIFIER_CA1, MODEL_AIRHUMIDIFIER_CB1]:
self._device_features = FEATURE_FLAGS_AIRHUMIDIFIER_CA_AND_CB
self._available_attributes = AVAILABLE_ATTRIBUTES_AIRHUMIDIFIER_CA_AND_CB
self._speed_list = [
mode.name
for mode in AirhumidifierOperationMode
if mode is not AirhumidifierOperationMode.Strong
]
else:
self._device_features = FEATURE_FLAGS_AIRHUMIDIFIER
self._available_attributes = AVAILABLE_ATTRIBUTES_AIRHUMIDIFIER
self._speed_list = [
mode.name
for mode in AirhumidifierOperationMode
if mode is not AirhumidifierOperationMode.Auto
]
self._state_attrs.update(
{attribute: None for attribute in self._available_attributes}
)
async def async_update(self):
"""Fetch state from the device."""
# On state change the device doesn't provide the new state immediately.
if self._skip_update:
self._skip_update = False
return
try:
state = await self.hass.async_add_executor_job(self._device.status)
_LOGGER.debug("Got new state: %s", state)
self._available = True
self._state = state.is_on
self._state_attrs.update(
{
key: self._extract_value_from_attribute(state, value)
for key, value in self._available_attributes.items()
}
)
except DeviceException as ex:
self._available = False
_LOGGER.error("Got exception while fetching the state: %s", ex)
@property
def speed_list(self) -> list:
"""Get the list of available speeds."""
return self._speed_list
@property
def speed(self):
"""Return the current speed."""
if self._state:
return AirhumidifierOperationMode(self._state_attrs[ATTR_MODE]).name
return None
async def async_set_speed(self, speed: str) -> None:
"""Set the speed of the fan."""
if self.supported_features & SUPPORT_SET_SPEED == 0:
return
_LOGGER.debug("Setting the operation mode to: %s", speed)
await self._try_command(
"Setting operation mode of the miio device failed.",
self._device.set_mode,
AirhumidifierOperationMode[speed.title()],
)
async def async_set_led_brightness(self, brightness: int = 2):
"""Set the led brightness."""
if self._device_features & FEATURE_SET_LED_BRIGHTNESS == 0:
return
await self._try_command(
"Setting the led brightness of the miio device failed.",
self._device.set_led_brightness,
AirhumidifierLedBrightness(brightness),
)
async def async_set_target_humidity(self, humidity: int = 40):
"""Set the target humidity."""
if self._device_features & FEATURE_SET_TARGET_HUMIDITY == 0:
return
await self._try_command(
"Setting the target humidity of the miio device failed.",
self._device.set_target_humidity,
humidity,
)
async def async_set_dry_on(self):
"""Turn the dry mode on."""
if self._device_features & FEATURE_SET_DRY == 0:
return
await self._try_command(
"Turning the dry mode of the miio device off failed.",
self._device.set_dry,
True,
)
async def async_set_dry_off(self):
"""Turn the dry mode off."""
if self._device_features & FEATURE_SET_DRY == 0:
return
await self._try_command(
"Turning the dry mode of the miio device off failed.",
self._device.set_dry,
False,
)
class XiaomiAirFresh(XiaomiGenericDevice):
"""Representation of a Xiaomi Air Fresh."""
def __init__(self, name, device, model, unique_id):
"""Initialize the miio device."""
super().__init__(name, device, model, unique_id)
self._device_features = FEATURE_FLAGS_AIRFRESH
self._available_attributes = AVAILABLE_ATTRIBUTES_AIRFRESH
self._speed_list = OPERATION_MODES_AIRFRESH
self._state_attrs.update(
{attribute: None for attribute in self._available_attributes}
)
async def async_update(self):
"""Fetch state from the device."""
# On state change the device doesn't provide the new state immediately.
if self._skip_update:
self._skip_update = False
return
try:
state = await self.hass.async_add_executor_job(self._device.status)
_LOGGER.debug("Got new state: %s", state)
self._available = True
self._state = state.is_on
self._state_attrs.update(
{
key: self._extract_value_from_attribute(state, value)
for key, value in self._available_attributes.items()
}
)
except DeviceException as ex:
self._available = False
_LOGGER.error("Got exception while fetching the state: %s", ex)
@property
def speed_list(self) -> list:
"""Get the list of available speeds."""
return self._speed_list
@property
def speed(self):
"""Return the current speed."""
if self._state:
return AirfreshOperationMode(self._state_attrs[ATTR_MODE]).name
return None
async def async_set_speed(self, speed: str) -> None:
"""Set the speed of the fan."""
if self.supported_features & SUPPORT_SET_SPEED == 0:
return
_LOGGER.debug("Setting the operation mode to: %s", speed)
await self._try_command(
"Setting operation mode of the miio device failed.",
self._device.set_mode,
AirfreshOperationMode[speed.title()],
)
async def async_set_led_on(self):
"""Turn the led on."""
if self._device_features & FEATURE_SET_LED == 0:
return
await self._try_command(
"Turning the led of the miio device off failed.", self._device.set_led, True
)
async def async_set_led_off(self):
"""Turn the led off."""
if self._device_features & FEATURE_SET_LED == 0:
return
await self._try_command(
"Turning the led of the miio device off failed.",
self._device.set_led,
False,
)
async def async_set_led_brightness(self, brightness: int = 2):
"""Set the led brightness."""
if self._device_features & FEATURE_SET_LED_BRIGHTNESS == 0:
return
await self._try_command(
"Setting the led brightness of the miio device failed.",
self._device.set_led_brightness,
AirfreshLedBrightness(brightness),
)
async def async_set_extra_features(self, features: int = 1):
"""Set the extra features."""
if self._device_features & FEATURE_SET_EXTRA_FEATURES == 0:
return
await self._try_command(
"Setting the extra features of the miio device failed.",
self._device.set_extra_features,
features,
)
async def async_reset_filter(self):
"""Reset the filter lifetime and usage."""
if self._device_features & FEATURE_RESET_FILTER == 0:
return
await self._try_command(
"Resetting the filter lifetime of the miio device failed.",
self._device.reset_filter,
)
Reference in a new issue View git blame Copy permalink