custom_components/xiaomi_miot/__init__.py

"""Support for Xiaomi Miot."""
import logging
import asyncio
import json
import os
import re
import traceback
from datetime import timedelta
import voluptuous as vol

from homeassistant import (
    core as hass_core,
    config_entries,
)
from homeassistant.const import (
    ATTR_ENTITY_ID,
    CONF_HOST,
    CONF_NAME,
    CONF_DEVICE,
    CONF_PASSWORD,
    CONF_SCAN_INTERVAL,
    CONF_TOKEN,
    CONF_USERNAME,
    STATE_UNKNOWN,
    SERVICE_RELOAD,
)
from homeassistant.helpers.entity import ToggleEntity, EntityCategory
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from homeassistant.helpers.reload import async_integration_yaml_config
from homeassistant.helpers.service import async_register_admin_service
import homeassistant.helpers.device_registry as dr
import homeassistant.helpers.config_validation as cv

from .core.const import *
from .core.utils import DeviceException, slugify_object_id, wildcard_models
from .core import HassEntry, BasicEntity, XEntity # noqa
from .core.device import Device, AsyncMiIO
from .core.miot_spec import (
    MiotService,
    MiotProperty,
    MiotResult,
    MiotResults,
)
from .core.xiaomi_cloud import (
    MiotCloud,
    MiCloudException,
    MiCloudAccessDenied,
)
from .core.templates import CUSTOM_TEMPLATES

_LOGGER = logging.getLogger(__name__)

SCAN_INTERVAL = timedelta(seconds=60)

XIAOMI_CONFIG_SCHEMA = cv.PLATFORM_SCHEMA_BASE.extend(
    {
        vol.Optional(CONF_HOST): cv.string,
        vol.Optional(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, default=''): cv.string,
    }
)

XIAOMI_MIIO_SERVICE_SCHEMA = vol.Schema(
    {
        vol.Optional(ATTR_ENTITY_ID): cv.entity_ids,
    },
)

SERVICE_TO_METHOD_BASE = {
    'send_command': {
        'method': 'async_miio_command',
        'schema': XIAOMI_MIIO_SERVICE_SCHEMA.extend(
            {
                vol.Required('method'): cv.string,
                vol.Optional('params', default=[]): cv.ensure_list,
            },
        ),
    },
    'set_property': {
        'method': 'async_set_property',
        'schema': XIAOMI_MIIO_SERVICE_SCHEMA.extend(
            {
                vol.Required('field'): cv.string,
                vol.Required('value'): cv.match_all,
            },
        ),
    },
    'set_miot_property': {
        'method': 'async_set_miot_property',
        'schema': XIAOMI_MIIO_SERVICE_SCHEMA.extend(
            {
                vol.Optional('did'): cv.string,
                vol.Required('siid'): int,
                vol.Required('piid'): int,
                vol.Required('value'): cv.match_all,
                vol.Optional('throw', default=False): cv.boolean,
            },
        ),
    },
    'get_properties': {
        'method': 'async_get_properties',
        'schema': XIAOMI_MIIO_SERVICE_SCHEMA.extend(
            {
                vol.Required('mapping'): vol.Any(dict, list),
                vol.Optional('update_entity', default=False): cv.boolean,
                vol.Optional('throw', default=False): cv.boolean,
            },
        ),
    },
    'call_action': {
        'method': 'async_call_action',
        'schema': XIAOMI_MIIO_SERVICE_SCHEMA.extend(
            {
                vol.Required('siid'): int,
                vol.Required('aiid'): int,
                vol.Optional('did'): cv.string,
                vol.Optional('params', default=[]): cv.ensure_list,
                vol.Optional('force_params', default=False): cv.boolean,
                vol.Optional('throw', default=False): cv.boolean,
            },
        ),
    },
    'get_device_data': {
        'method': 'async_get_device_data',
        'schema': XIAOMI_MIIO_SERVICE_SCHEMA.extend(
            {
                vol.Optional('type', default='prop'): cv.string,
                vol.Required('key'): cv.string,
                vol.Optional('did'): cv.string,
                vol.Optional('time_start'): int,
                vol.Optional('time_end'): int,
                vol.Optional('limit'): int,
                vol.Optional('group'): cv.string,
                vol.Optional('throw', default=False): cv.boolean,
            },
        ),
    },
    'get_bindkey': {
        'method': 'async_get_bindkey',
        'schema': XIAOMI_MIIO_SERVICE_SCHEMA.extend(
            {
                vol.Optional('did', default=''): cv.string,
                vol.Optional('throw', default=False): cv.boolean,
            },
        ),
    },
    'request_xiaomi_api': {
        'method': 'async_request_xiaomi_api',
        'schema': XIAOMI_MIIO_SERVICE_SCHEMA.extend(
            {
                vol.Required('api'): cv.string,
                vol.Optional('data', default={}): vol.Any(dict, list),
                vol.Optional('method', default='POST'): cv.string,
                vol.Optional('crypt', default=True): cv.boolean,
                vol.Optional('sid', default=None): vol.Any(cv.string, None),
                vol.Optional('throw', default=False): cv.boolean,
            },
        ),
    },
}

CONFIG_SCHEMA = vol.Schema(
    {
        DOMAIN: vol.Schema(
            {
                vol.Optional(CONF_USERNAME): cv.string,
                vol.Optional(CONF_PASSWORD): cv.string,
                vol.Optional(CONF_SERVER_COUNTRY): cv.string,
            },
            extra=vol.ALLOW_EXTRA,
        ),
    },
    extra=vol.ALLOW_EXTRA,
)


async def async_setup(hass, hass_config: dict):
    init_integration_data(hass)
    config = hass_config.get(DOMAIN) or {}
    await async_reload_integration_config(hass, config)

    def extend_miot_specs():
        with open(os.path.dirname(__file__) + '/core/miot_specs_extend.json') as file:
            try:
                models = json.load(file) or {}
            except ValueError as exc:
                models = {}
                _LOGGER.exception('Error parsing miot_specs_extend.json: %s', exc)
            for m, specs in models.items():
                DEVICE_CUSTOMIZES.setdefault(m, {})
                DEVICE_CUSTOMIZES[m]['extend_miot_specs'] = specs

    await hass.async_add_executor_job(extend_miot_specs)

    component = EntityComponent(_LOGGER, DOMAIN, hass, SCAN_INTERVAL)
    hass.data[DOMAIN]['component'] = component
    await component.async_setup(config)
    await async_setup_component_services(hass)
    bind_services_to_entries(hass, SERVICE_TO_METHOD_BASE)

    if config.get(CONF_USERNAME) and config.get(CONF_PASSWORD):
        try:
            mic = MiotCloud(
                hass,
                config.get(CONF_USERNAME),
                config.get(CONF_PASSWORD),
                config.get(CONF_SERVER_COUNTRY),
            )
            if not await mic.async_login():
                raise MiCloudException('Login failed')
            hass.data[DOMAIN][CONF_XIAOMI_CLOUD] = mic
            hass.data[DOMAIN]['devices_by_mac'] = await mic.async_get_devices_by_key('mac') or {}
            hass.data[DOMAIN]['accounts'].setdefault(mic.user_id, {CONF_XIAOMI_CLOUD: mic})
            cnt = len(hass.data[DOMAIN]['devices_by_mac'])
            _LOGGER.debug('Setup xiaomi cloud for user: %s, %s devices', config.get(CONF_USERNAME), cnt)
        except (MiCloudException, MiCloudAccessDenied) as exc:
            _LOGGER.warning('Setup xiaomi cloud for user: %s failed: %s', config.get(CONF_USERNAME), exc)

    await _handle_device_registry_event(hass)
    return True


async def async_setup_entry(hass: hass_core.HomeAssistant, config_entry: config_entries.ConfigEntry):
    hass.data.setdefault(DOMAIN, {})
    entry_id = config_entry.entry_id

    if config_entry.data.get('customizing_entity') or config_entry.data.get('customizing_device'):
        await async_setup_customizes(hass, config_entry)
    elif config_entry.data.get(CONF_USERNAME):
        await async_setup_xiaomi_cloud(hass, config_entry)
    else:
        entry = HassEntry.init(hass, config_entry)
        config = {**entry.get_config()}
        device = await entry.new_device(config)
        config[CONF_DEVICE] = device
        config[CONF_MODEL] = device.model
        config['miot_type'] = await device.get_urn()
        config['config_entry'] = config_entry
        config['miot_local'] = True
        config[CONF_CONN_MODE] = 'local'
        hass.data[DOMAIN][entry_id] = config
        _LOGGER.debug('Xiaomi Miot setup config entry: %s', {
            'entry_id': entry_id,
            'config': config,
        })

    if not config_entry.update_listeners:
        config_entry.add_update_listener(async_update_options)

    await hass.config_entries.async_forward_entry_setups(config_entry, SUPPORTED_DOMAINS)
    return True


async def async_setup_xiaomi_cloud(hass: hass_core.HomeAssistant, config_entry: config_entries.ConfigEntry):
    entry_id = config_entry.entry_id
    entry = HassEntry.init(hass, config_entry)
    entry_config = entry.get_config()
    username = entry_config.get(CONF_USERNAME)
    config = {
        'entry_id': entry_id,
        'config_entry': config_entry,
        'configs': [],
    }
    try:
        cloud = await entry.get_cloud(check=True)
        config[CONF_XIAOMI_CLOUD] = cloud
        devices = await entry.get_cloud_devices()
    except (MiCloudException, MiCloudAccessDenied) as exc:
        _LOGGER.error('Setup xiaomi cloud for user: %s failed: %s', username, exc)
        return False
    if not devices:
        _LOGGER.warning('None device in xiaomi cloud: %s', username)
    else:
        _LOGGER.debug('Setup xiaomi cloud for user: %s, %s devices', username, len(devices))
    for d in devices.values():
        device = await entry.new_device(d)
        if not device.spec:
            _LOGGER.warning('%s: Device has no spec %s', device.name_model, device.info.urn)
            continue
        conn = device.conn_mode
        cfg = {
            CONF_DEVICE: device,
            CONF_NAME: device.name,
            CONF_HOST: device.info.host,
            CONF_TOKEN: device.info.token,
            CONF_MODEL: device.info.model,
            'miot_did': device.info.did,
            'miot_type': await device.get_urn(),
            'miio_info': device.info.miio_info,
            CONF_CONN_MODE: conn,
            'miot_local': conn == 'local',
            'miot_cloud': conn != 'local',
            'home_name': device.info.home_name,
            'room_name': device.info.room_name,
            'entry_id': entry_id,
            CONF_CONFIG_VERSION: entry_config.get(CONF_CONFIG_VERSION) or 0,
        }
        if conn == 'auto' and device.info.model in MIOT_LOCAL_MODELS:
            cfg['miot_local'] = True
            cfg['miot_cloud'] = False
        config['configs'].append(cfg)
        _LOGGER.debug('Xiaomi cloud device: %s', {**cfg, CONF_TOKEN: '****'})
    hass.data[DOMAIN][entry_id] = config
    hass.data[DOMAIN]['accounts'].setdefault(cloud.user_id, {CONF_XIAOMI_CLOUD: cloud})
    return True


async def async_setup_customizes(hass: hass_core.HomeAssistant, config_entry: config_entries.ConfigEntry):
    entry_data = {**config_entry.data, **config_entry.options}
    if cus := entry_data.get('customizing_entity'):
        hass.data[DOMAIN][DATA_CUSTOMIZE] = cus
    if cus := entry_data.get('customizing_device'):
        for m, cfg in cus.items():
            if not isinstance(cfg, dict):
                continue
            DEVICE_CUSTOMIZES.setdefault(m, {})
            DEVICE_CUSTOMIZES[m].update(cfg)
    if entry_data:
        _LOGGER.info('Customizing via config flow: %s', entry_data)


async def async_update_options(hass: hass_core.HomeAssistant, config_entry: config_entries.ConfigEntry):
    entry = {**config_entry.data, **config_entry.options}
    entry.pop(CONF_TOKEN, None)
    entry.pop(CONF_PASSWORD, None)
    entry.pop('service_token', None)
    entry.pop('ssecurity', None)
    _LOGGER.debug('Xiaomi Miot update options: %s', entry)
    hass.data[DOMAIN]['sub_entities'] = {}
    await hass.config_entries.async_reload(config_entry.entry_id)


async def async_unload_entry(hass: hass_core.HomeAssistant, config_entry: config_entries.ConfigEntry):
    unload_ok = await HassEntry.init(hass, config_entry).async_unload()
    if unload_ok:
        hass.data[DOMAIN].pop(config_entry.entry_id, None)
        hass.data[DOMAIN]['sub_entities'] = {}
    return unload_ok


def init_integration_data(hass):
    hass.data.setdefault(DOMAIN, {})
    hass.data[DOMAIN].setdefault('entries', {})
    hass.data[DOMAIN].setdefault('configs', {})
    hass.data[DOMAIN].setdefault('entities', {})
    hass.data[DOMAIN].setdefault('accounts', {})
    hass.data[DOMAIN].setdefault('sessions', {})
    hass.data[DOMAIN].setdefault('miot_specs', {})
    hass.data[DOMAIN].setdefault('add_entities', {})
    hass.data[DOMAIN].setdefault('sub_entities', {})


def bind_services_to_entries(hass, services):
    async def async_service_handler(service) -> ServiceResponse:
        result = None
        method = services.get(service.service)
        fun = method['method']
        params = {
            key: value
            for key, value in service.data.items()
            if key != ATTR_ENTITY_ID
        }
        target_entities = []
        entity_ids = service.data.get(ATTR_ENTITY_ID)
        if entity_ids:
            target_entities = [
                hass.data[DOMAIN]['entities'][eid]
                for eid in entity_ids
                if eid in hass.data[DOMAIN].get('entities', {})
            ]
        if not target_entities:
            _LOGGER.warning('Call service failed: Entities not found for %s', entity_ids)
        else:
            _LOGGER.debug('Xiaomi Miot service handler: %s', {
                'targets': [ent.entity_id for ent in target_entities],
                'method': fun,
                'params': params,
            })
        update_tasks = []
        for ent in target_entities:
            if parent := getattr(ent, 'parent_entity', None):
                ent = parent
            if not hasattr(ent, fun):
                _LOGGER.warning('Call service failed: Entity %s have no method: %s', ent.entity_id, fun)
                continue
            try:
                result = await getattr(ent, fun)(**params)
                update_tasks.append(ent.async_update_ha_state(True))
            except Exception as exc:
                _LOGGER.warning('Call service failed', exc_info=True)
                result = {
                    'error': str(exc),
                    'trace': traceback.format_exc(),
                }
        if update_tasks:
            await asyncio.gather(*update_tasks)
        if isinstance(result, (MiotResult, MiotResults)):
            result = result.to_json()
        if not isinstance(result, dict):
            result = {'result': result}
        return result

    for srv, obj in services.items():
        kws = {
            'schema': obj.get('schema', XIAOMI_MIIO_SERVICE_SCHEMA),
        }
        if SupportsResponse:
            kws['supports_response'] = SupportsResponse.OPTIONAL
        hass.services.async_register(DOMAIN, srv, async_service_handler, **kws)


async def async_reload_integration_config(hass, config):
    hass.data[DOMAIN]['config'] = config

    if lang := config.get('language'):
        dic = TRANSLATION_LANGUAGES.get(lang)
        if isinstance(dic, dict):
            TRANSLATION_LANGUAGES.update(dic)
    dic = config.get('translations') or {}
    if dic and isinstance(dic, dict):
        TRANSLATION_LANGUAGES.update(dic)

    dcs = config.get('device_customizes')
    if dcs and isinstance(dcs, dict):
        for m, cus in dcs.items():
            if not isinstance(cus, dict):
                continue
            DEVICE_CUSTOMIZES.setdefault(m, {})
            DEVICE_CUSTOMIZES[m].update(cus)
    return config


async def async_setup_component_services(hass):

    async def async_get_token(call) -> ServiceResponse:
        nam = call.data.get('name')
        kwd = f'{nam}'.strip().lower()
        cnt = 0
        lst = []
        dls = {}
        beaconkey = miio_info = None
        for cld in MiotCloud.all_clouds(hass):
            dvs = await cld.async_get_devices() or []
            for d in dvs:
                if not isinstance(d, dict):
                    continue
                did = d.get('did') or ''
                if dls.get(did):
                    continue
                dnm = f"{d.get('name') or ''}"
                dip = d.get('localip') or ''
                dmd = d.get('model') or ''
                tok = d.get('token') or ''
                if kwd in [did, dip] or kwd in dnm.lower() or kwd in dmd:
                    row = {
                        'did': did,
                        CONF_NAME: dnm,
                        CONF_HOST: dip,
                        CONF_MODEL: dmd,
                        CONF_TOKEN: tok,
                    }
                    if not beaconkey and 'blt.' in did:
                        beaconkey = await cld.async_get_beaconkey(did)
                        row['beaconkey'] = (beaconkey or {}).get('beaconkey', beaconkey)
                        row.pop(CONF_TOKEN, None)
                    elif dip and tok:
                        row['miio_cmd'] = f'miiocli device --ip {dip} --token {tok} info'
                        if not miio_info:
                            try:
                                miio = AsyncMiIO(dip, tok)
                                miio_info = await miio.info()
                            except Exception as exc:
                                miio_info = {'error': str(exc)}
                            row['miio_info'] = miio_info
                    lst.append(row)
                dls[did] = 1
                cnt += 1
        if not lst:
            lst = [f'Not Found "{nam}" in {cnt} devices.']
        return {
            'list': lst,
        }

    kws = {
        'schema': XIAOMI_MIIO_SERVICE_SCHEMA.extend({
            vol.Required('name', default=''): cv.string,
        }),
    }
    if SupportsResponse:
        kws['supports_response'] = SupportsResponse.OPTIONAL,
    hass.services.async_register(
        DOMAIN, 'get_token', async_get_token, **kws,
    )

    async def async_renew_devices(call):
        nam = call.data.get('username')
        for cld in MiotCloud.all_clouds(hass):
            if nam and str(nam) not in [cld.user_id, cld.username]:
                continue
            dvs = await cld.async_renew_devices()
            cnt = len(dvs)
            _LOGGER.info('Renew xiaomi devices for %s. Got %s devices.', cld.username, cnt)
        return True

    hass.services.async_register(
        DOMAIN, 'renew_devices', async_renew_devices,
        schema=vol.Schema({
            vol.Optional('username', default=''): cv.string,
        }),
    )

    async def _handle_reload_config(service):
        config = await async_integration_yaml_config(hass, DOMAIN)
        if not config or DOMAIN not in config:
            return
        await async_reload_integration_config(hass, config.get(DOMAIN) or {})
        current_entries = hass.config_entries.async_entries(DOMAIN)
        reload_tasks = [
            hass.config_entries.async_reload(entry.entry_id)
            for entry in current_entries
        ]
        await asyncio.gather(*reload_tasks)

    async_register_admin_service(
        hass,
        DOMAIN,
        SERVICE_RELOAD,
        _handle_reload_config,
    )


async def async_setup_config_entry(hass, config_entry, async_setup_platform, async_add_entities, domain=None):
    eid = config_entry.entry_id
    cfg = hass.data[DOMAIN].get(eid) or {}
    if not cfg:
        hass.data[DOMAIN].setdefault(eid, {})
    if domain:
        hass.data[DOMAIN][eid].setdefault('add_entities', {})
        hass.data[DOMAIN][eid]['add_entities'][domain] = async_add_entities
    cls = cfg.get('configs')
    if not cls:
        cls = [
            hass.data[DOMAIN].get(eid, dict(config_entry.data)),
        ]
    for c in cls:
        await async_setup_platform(hass, c, async_add_entities)
    return cls


async def _handle_device_registry_event(hass: hass_core.HomeAssistant):
    async def updated(event: hass_core.Event):
        action = event.data['action']
        registry = dr.async_get(hass)
        device_id = event.data.get('device_id')
        if device_id not in registry.devices:
            return
        device = registry.async_get(device_id)
        if not device or not device.identifiers:
            return
        identifier = next(iter(device.identifiers))
        if identifier[0] != DOMAIN:
            return
        miot_device = None
        for entry_id in device.config_entries:
            entry = HassEntry.ALL.get(entry_id)
            if not entry:
                continue
            for d in entry.devices.values():
                if d.identifiers == device.identifiers:
                    miot_device = d
                    break
        if action == 'update' and device.name_by_user in ['delete', 'remove', '删除']:
            # remove from Hass
            if miot_device:
                await miot_device.async_unload()
            registry.async_remove_device(device.id)
            return
        if not miot_device:
            return
        miot_device.log.info('Device registry updated: %s', [action, identifier, device.disabled])
        if device.disabled and miot_device.coordinators:
            await miot_device.async_unload()
        if not device.disabled and not miot_device.coordinators:
            await miot_device.init_coordinators()
    hass.bus.async_listen(dr.EVENT_DEVICE_REGISTRY_UPDATED, updated)


async def async_remove_config_entry_device(hass: hass_core.HomeAssistant, config_entry: ConfigEntry, device: dr.DeviceEntry):
    """Supported from Hass v2022.3"""
    entry = HassEntry.init(hass, config_entry)
    cloud_device = None
    identifier = next(iter(device.identifiers))
    if len(identifier) >= 2 and identifier[0] == DOMAIN:
        mac = identifier[1].split('-')[0]
        if mac:
            cloud_device = await entry.get_cloud_device(mac=mac.upper())
    data = {**entry.entry.data}
    for typ in (['did'] if cloud_device else []):
        filter_typ = data.get(f'filter_{typ}')
        filter_val = cloud_device.get(typ)
        if not filter_val or not filter_typ:
            continue
        lst = data.get(f'{typ}_list') or []
        if filter_typ == 'exclude':
            lst = list({*lst, filter_val})
        else:
            lst = list({*lst}.difference({filter_val}))
        data[f'{typ}_list'] = lst
        hass.config_entries.async_update_entry(config_entry, data=data)
        _LOGGER.info('Remove miot device: %s', cloud_device)

    dr.async_get(hass).async_remove_device(device.id)
    return True


class BaseEntity(BasicEntity):
    device: Device = None
    _config = None
    _model = None
    _unique_did = None
    _attr_device_class = None
    _attr_entity_category = None
    _attr_translation_key = None

    async def async_added_to_hass(self):
        await super().async_added_to_hass()
        if self.hass:
            self.hass.data[DOMAIN]['entities'][self.entity_id] = self

    @property
    def entity_category(self):
        cat = super().entity_category
        if isinstance(cat, EntityCategory):
            return cat
        if isinstance(cat, str) and cat in EntityCategory:
            return EntityCategory(cat)
        return None

    def get_device_class(self, enum):
        cls = self._attr_device_class
        if isinstance(cls, enum):
            return cls
        if isinstance(cls, str) and cls in enum:
            return enum(cls)
        return None

    @property
    def model(self):
        if self.device:
            return self.device.info.model
        return self._model

    @property
    def name_model(self):
        return f'{self.name}({self.model})'

    @property
    def conn_mode(self):
        return self._config.get(CONF_CONN_MODE)

    @property
    def local_only(self):
        return self.conn_mode == 'local'

    @property
    def cloud_only(self):
        return self.conn_mode == 'cloud'

    @property
    def entry_config_version(self):
        return self._config.get(CONF_CONFIG_VERSION) or 0

    def entry_config(self, key=None, default=None):
        if not self.hass:
            return default
        cfg = self.hass.data[DOMAIN] or {}
        eid = None
        if self._config:
            eid = self._config.get('entry_id')
        if not eid and self.platform.config_entry:
            eid = self.platform.config_entry.entry_id
        if eid:
            cfg = {**cfg, **(self.hass.data[DOMAIN].get(eid) or {})}
        return cfg if key is None else cfg.get(key, default)

    def update_custom_scan_interval(self, only_custom=False):
        if not self.platform:
            return
        sec = self.custom_config('interval_seconds')
        if not sec and not only_custom:
            sec = self.entry_config(CONF_SCAN_INTERVAL)
        try:
            sec = int(sec or 0)
        except (TypeError, ValueError):
            sec = 0
        tim = timedelta(seconds=sec)
        if sec > 0 and tim != self.platform.scan_interval:
            self.platform.scan_interval = tim
            if hasattr(self.platform, 'scan_interval_seconds'):
                self.platform.scan_interval_seconds = tim.total_seconds()            
            _LOGGER.debug('%s: Update custom scan interval: %s', self.name_model, tim)

    def update_custom_parallel_updates(self):
        if not self.hass:
            return False
        if not hasattr(self, '_unique_did'):
            return False
        num = self.custom_config_integer('parallel_updates', 0)
        if not num:
            return False
        did = self._unique_did
        self.hass.data[DOMAIN].setdefault(did, {})
        dcs = self.hass.data[DOMAIN].get(did, {})
        pus = dcs.get('parallel_updates')
        if not pus:
            pus = asyncio.Semaphore(num)
            self.hass.data[DOMAIN][did]['parallel_updates'] = pus
            _LOGGER.debug('%s: Update custom parallel updates: %s', self.name_model, num)
        self.parallel_updates = pus
        return pus


class MiCoordinatorEntity(CoordinatorEntity, BaseEntity):
    async def async_added_to_hass(self):
        await super().async_added_to_hass()


class MiioEntity(BaseEntity):
    def __init__(self, name, device, **kwargs):
        self._device = device
        self._config = dict(kwargs.get('config') or {})
        self.device = self._config.get(CONF_DEVICE)
        self.hass = self.device.hass
        self.log = self.device.log
        self.logger = self.device.log
        self._miio_info = self.device.info.miio_info
        self._unique_did = self.unique_did
        self._unique_id = self._unique_did
        self._name = name
        self._model = self.device.info.model
        self._state = None
        self._available = False
        self._state_attrs = {}
        self._attr_device_info = self.device.hass_device_info
        self._supported_features = 0
        self._props = []
        self._success_result = ['ok']
        self._add_entities = {}
        self._vars = {}
        self._subs = {}

        self._vars['is_main_entity'] = not self.device.miot_entity
        self.device.miot_entity = self

    @property
    def unique_id(self):
        return self._unique_id

    @property
    def unique_mac(self):
        mac = self.device.info.mac
        if not mac:
            mac = self.device.info.did
        return mac

    @property
    def unique_did(self):
        return self.device.unique_id

    @property
    def name(self):
        return self._name

    @property
    def name_model(self):
        return self.device.name_model

    @property
    def device_name(self):
        return self.device.name

    @property
    def device_host(self):
        return self.device.info.host

    @property
    def available(self):
        return self._available

    @property
    def is_on(self):
        return self._state

    @property
    def state_attrs(self):
        return self._state_attrs

    @property
    def extra_state_attributes(self):
        ext = self.state_attributes or {}
        esa = {**self._state_attrs, **ext}
        return self.filter_state_attributes(esa)

    @property
    def supported_features(self):
        return self._supported_features

    async def async_added_to_hass(self):
        await super().async_added_to_hass()
        if self.platform:
            self.update_custom_scan_interval()
            self.update_custom_parallel_updates()
            if self.platform.config_entry:
                eid = self.platform.config_entry.entry_id
                self._add_entities = self.hass.data[DOMAIN][eid].get('add_entities') or {}

    async def async_miio_command(self, method, params=None, **kwargs):
        return await self.device.local.async_send(method, params, **kwargs)

    async def async_update(self):
        if not self._props:
            return
        try:
            attrs = await self.device.local.async_get_prop(self._props)
        except DeviceException as ex:
            self._available = False
            self.logger.error('%s: Got exception while fetching the state %s: %s', self.name_model, self._props, ex)
            return
        attrs = dict(zip(self._props, attrs))
        self.logger.debug('%s: Got new state: %s', self.name_model, attrs)
        self._available = True
        self._state = attrs.get('power') == 'on'
        await self.async_update_attrs(attrs)

    def update_attrs(self, attrs: dict, update_parent=False):
        self._state_attrs.update(attrs or {})
        if update_parent and hasattr(self, '_parent'):
            if self._parent and hasattr(self._parent, 'update_attrs'):
                getattr(self._parent, 'update_attrs')(attrs or {}, update_parent=False)
        return self._state_attrs

    async def async_update_attrs(self, attrs: dict, update_subs=True):
        self._state_attrs.update(attrs or {})
        if update_subs:
            if self.hass and self.platform:
                tps = cv.ensure_list(self.custom_config('attributes_template'))
                for tpl in tps:
                    if not tpl:
                        continue
                    tpl = CUSTOM_TEMPLATES.get(tpl, tpl)
                    tpl = cv.template(tpl)
                    tpl.hass = self.hass
                    adt = tpl.async_render({'data': self._state_attrs}) or {}
                    if isinstance(adt, dict):
                        if adt.pop('_override', False):
                            self._state_attrs = adt
                        else:
                            self._state_attrs.update(adt)
        return self._state_attrs


class MiotEntityInterface:
    _miot_service = None
    _model = ''
    _state_attrs: dict
    _supported_features = 0

    def update_attrs(self, *args, **kwargs):
        raise NotImplementedError()


class MiotEntity(MiioEntity):
    def __init__(self, miot_service=None, device=None, **kwargs):
        self._config = dict(kwargs.get('config') or {})
        name = kwargs.get(CONF_NAME) or self._config.get(CONF_NAME) or ''
        self._miot_service = miot_service if isinstance(miot_service, MiotService) else None
        if self._miot_service:
            name = f'{name} {self._miot_service.friendly_desc}'.strip()
            kwargs['miot_service'] = self._miot_service
        super().__init__(name, device, **kwargs)

        self._local_state = None
        self._miio2miot = self.device.miio2miot
        self._miot_mapping = dict(kwargs.get('mapping') or {})
        if self._miot_service:
            if not self._miot_mapping:
                self._miot_mapping = miot_service.mapping(
                    excludes=self.device._exclude_miot_properties,
                    unreadable_properties=self.device._unreadable_properties,
                ) or {}
            self._unique_id = f'{self._unique_id}-{self._miot_service.iid}'
            self.entity_id = self._miot_service.generate_entity_id(self)
            self._attr_translation_key = self._miot_service.name
        if not self.entity_id and self.model:
            mls = f'{self.model}..'.split('.')
            mac = re.sub(r'[\W_]+', '', self.unique_mac)
            obj = f'{mls[0]}_{mls[2]}_{mac[-4:]}_{mls[1]}'
            self.entity_id = f'{DOMAIN}.{slugify_object_id(obj)}'
        self._success_code = 0
        self.logger.info('%s: Initializing miot device with mapping: %s', self.name_model, self._miot_mapping)

    async def async_added_to_hass(self):
        await super().async_added_to_hass()
        if not self._miot_service:
            return
        self._vars['ignore_offline'] = self.custom_config_bool('ignore_offline')
        self.logger.debug('%s: Added to hass: %s', self.name_model, [self.custom_config()])

    @property
    def miot_device(self):
        if not self._device:
            device = self.device.local
            if device:
                self._device = device
        return self._device

    @property
    def miot_did(self):
        did = self.custom_config('miot_did') or self._config.get('miot_did')
        if did:
            return did
        return self.device.did

    @property
    def xiaomi_cloud(self):
        return self.device.cloud

    @property
    def miot_cloud(self):
        isc = False
        if self.miot_local:
            isc = False
        elif self._config.get('miot_cloud'):
            isc = True
        elif self.custom_config_bool('miot_cloud'):
            isc = True
        if isc and self.miot_did:
            return self.xiaomi_cloud
        return None

    @property
    def miot_cloud_write(self):
        isc = False
        if self.custom_config_bool('miot_cloud_write'):
            isc = True
        if isc and self.hass and self.miot_did:
            return self.entry_config(CONF_XIAOMI_CLOUD)
        return self.miot_cloud

    @property
    def miot_cloud_action(self):
        isc = False
        if self.custom_config_bool('miot_cloud_action'):
            isc = True
        if isc and self.hass and self.miot_did:
            return self.entry_config(CONF_XIAOMI_CLOUD)
        return self.miot_cloud

    @property
    def miot_local(self):
        if self.custom_config_bool('miot_local') or self._config.get('miot_local'):
            return self.miot_device
        return None

    @property
    def is_main_entity(self):
        return self._vars.get('is_main_entity')

    @property
    def miot_config(self):
        return self._config or {}

    @property
    def entity_id_prefix(self):
        if not self._miot_service:
            return None
        return self._miot_service.spec.generate_entity_id(self)

    async def async_update_from_device(self):
        self._available = self.device.available
        if self.is_main_entity:
            attrs = self.device.props
            attrs['state_updater'] = self.device.data.get('updater')
            await self.async_update_for_main_entity()
        else:
            attrs = {
                k: v
                for k, v in self.device.props.items()
                if k in self._miot_mapping
            }
        self._state_attrs = attrs
        await self.async_update_attrs(attrs, update_subs=True)
        return attrs

    async def async_update(self):
        if self._vars.get('delay_update'):
            await asyncio.sleep(self._vars.get('delay_update'))
            self._vars.pop('delay_update', 0)
        await self.device.update_main_status()
        attrs = await self.async_update_from_device()
        self.logger.debug('%s: Got new state: %s', self.name, attrs)

    async def async_update_for_main_entity(self):
        pass

    async def async_get_device_data(self, key, did=None, **kwargs):
        if did is None:
            did = self.miot_did
        mic = self.xiaomi_cloud
        if not isinstance(mic, MiotCloud):
            return None
        result = await mic.async_get_user_device_data(did, key, raw=True, **kwargs)
        _LOGGER.info('%s: Xiaomi device data: %s', self.name_model, result)
        return result

    async def async_get_bindkey(self, did=None):
        mic = self.xiaomi_cloud
        if not isinstance(mic, MiotCloud):
            return None
        result = await mic.async_get_beaconkey(did or self.miot_did)
        _LOGGER.info('%s: Xiaomi device bindkey/beaconkey: %s', self.name_model, result)
        return result

    async def async_request_xiaomi_api(self, api, data=None, method='POST', crypt=True, **kwargs):
        mic = self.xiaomi_cloud
        if not isinstance(mic, MiotCloud):
            return None
        sid = kwargs.pop('sid', None) or 'xiaomiio'
        if sid != mic.sid:
            mic = await mic.async_change_sid(sid)
        pms = kwargs.pop('params', None)
        dat = data or pms
        result = await mic.async_request_api(api, data=dat, method=method, crypt=crypt, **kwargs)
        _LOGGER.debug('Xiaomi Api %s: %s', api, result)
        return result


class MiotToggleEntity(MiotEntity, ToggleEntity):
    _reverse_state = None

    def __init__(self, miot_service=None, device=None, **kwargs):
        super().__init__(miot_service, device, **kwargs)
        self._prop_power = None
        if miot_service:
            self._prop_power = miot_service.get_property('on', 'power', 'switch')

    async def async_added_to_hass(self):
        await super().async_added_to_hass()
        self._reverse_state = self.custom_config_bool('reverse_state', None)

    @property
    def is_on(self):
        val = None
        if self._prop_power:
            val = not not self._state_attrs.get(self._prop_power.full_name)
            if self._reverse_state:
                val = not val
        return val

    async def async_turn_on(self, **kwargs):
        if self._prop_power:
            val = True
            if self._prop_power.value_range:
                val = self._prop_power.range_max() or 1
            elif self._reverse_state:
                val = not val
            return await self.async_set_property(self._prop_power, val)
        return False

    async def async_turn_off(self, **kwargs):
        if self._prop_power:
            val = False
            if self._prop_power.value_range:
                val = self._prop_power.range_min() or 0
            elif self._reverse_state:
                val = not val
            return await self.async_set_property(self._prop_power, val)
        act = self._miot_service.get_action('stop_working', 'power_off')
        if act:
            return await self.async_call_action(self._miot_service.iid, act.iid)
        return False


class MiirToggleEntity(MiotEntity, ToggleEntity):
    def __init__(self, miot_service=None, device=None, **kwargs):
        super().__init__(miot_service, device, **kwargs)
        self._available = True
        self._miot_actions = []
        for a in miot_service.actions.values():
            if a.ins:
                continue
            self._miot_actions.append(a.friendly_desc)
        self._attr_is_on = None
        self._act_turn_on = miot_service.get_action('turn_on')
        self._act_turn_off = miot_service.get_action('turn_off')
        self._attr_should_poll = False

    async def async_added_to_hass(self):
        await super().async_added_to_hass()
        if self.is_main_entity:
            await self.async_update_for_main_entity()

    @property
    def is_on(self):
        """Return True if entity is on."""
        return self._attr_is_on

    async def async_turn_on(self, **kwargs):
        """Turn the entity on."""
        if not self._act_turn_on:
            raise NotImplementedError()
        return await self.async_call_action(self._act_turn_on)

    async def async_turn_off(self, **kwargs):
        """Turn the entity off."""
        if not self._act_turn_off:
            raise NotImplementedError()
        return await self.async_call_action(self._act_turn_off)


class BaseSubEntity(BaseEntity):
    def __init__(self, parent, attr, option=None, **kwargs):
        self.hass = parent.hass
        self.device = parent.device
        self._unique_id = f'{parent.unique_id}-{attr}'
        self._name = f'{parent.name} {attr}'
        self._state = STATE_UNKNOWN
        self._attr_state = None
        self._available = False
        self._parent = parent
        self._attr = attr
        self._model = parent.device_info.get('model', '')
        self._option = dict(option or {})
        self._dict_key = self._option.get('dict_key')
        if self._dict_key:
            self._unique_id = f'{self._unique_id}-{self._dict_key}'
            self._name = f'{self._name} {self._dict_key}'
        if self._option.get('unique_id'):
            self._unique_id = self._option.get('unique_id')
        if self._option.get('name'):
            self._name = self._option.get('name')
        self._option['domain'] = kwargs.get('domain')
        self.generate_entity_id()
        self._supported_features = int(self._option.get('supported_features', 0))
        self._attr_entity_category = self.custom_config('entity_category', self._option.get('entity_category'))
        self._attr_native_unit_of_measurement = self._option.get('unit')
        self._attr_translation_key = self.custom_config('translation_key') or attr
        self._extra_attrs = {
            'entity_class': self.__class__.__name__,
            'parent_entity_id': parent.entity_id,
        }
        self._state_attrs = {}
        self._parent_attrs = {}

    def generate_entity_id(self, domain=None):
        entity_id = None
        if self._option.get('entity_id'):
            entity_id = self._option.get('entity_id')
        elif not hasattr(self._parent, 'entity_id_prefix'):
            pass
        elif eip := self._parent.entity_id_prefix:
            suf = self._attr
            if self._dict_key:
                suf = f'{suf}_{self._dict_key}'
            entity_id = f'{eip}_{suf}'
        if not domain:
            domain = self._option.get('domain') or DOMAIN
        if entity_id is None:
            pass
        elif f'{domain}.' in entity_id:
            obj = hass_core.split_entity_id(entity_id)[1]
            self.entity_id = f'{domain}.{slugify_object_id(obj)}'
        else:
            if '.' in entity_id:
                entity_id = hass_core.split_entity_id(entity_id)[1]
            self.entity_id = f'{domain}.{slugify_object_id(entity_id)}'

    @property
    def unique_id(self):
        return self._unique_id

    @property
    def unique_mac(self):
        return self._parent.unique_mac

    @property
    def name(self):
        return self._name

    @property
    def device_name(self):
        return self._parent.device_name

    @property
    def name_model(self):
        return f'{self.device_name}({self.model})'

    def format_name_by_property(self, prop: MiotProperty):
        return f'{self.device_name} {prop.friendly_desc}'.strip()

    @property
    def available(self):
        return self._available and self._parent.available

    @property
    def supported_features(self):
        return self._supported_features

    @property
    def parent_entity(self):
        return self._parent

    @property
    def parent_attributes(self):
        return self.device.props or {}

    @property
    def extra_state_attributes(self):
        esa = {
            **self._extra_attrs,
            **self._state_attrs,
        }
        return self.filter_state_attributes(esa)

    @property
    def device_class(self):
        return self._option.get('device_class', self._option.get('class'))

    @property
    def device_info(self):
        return self._parent.device_info

    @property
    def icon(self):
        return self._option.get('icon')

    @property
    def miot_cloud(self):
        mic = self._parent.miot_cloud
        if not isinstance(mic, MiotCloud):
            raise RuntimeError('The parent entity of %s does not have Mi Cloud.', self.name)
        return mic

    @property
    def customize_keys(self):
        mar = []
        for mod in wildcard_models(self.model):
            if self._dict_key:
                mar.append(f'{mod}:{self._attr}:{self._dict_key}')
            elif self._attr:
                mar.append(f'{mod}:{self._attr}')
            if hasattr(self, '_miot_property'):
                prop = getattr(self, '_miot_property', None)
                if prop:
                    mar.append(f'{mod}:{prop.full_name}')
                    mar.append(f'{mod}:{prop.name}')
        return mar

    async def async_added_to_hass(self):
        await super().async_added_to_hass()
        if self.platform:
            self.update_custom_scan_interval(only_custom=True)
        self._option['icon'] = self.custom_config('icon', self.icon)
        self._option['device_class'] = self.custom_config('device_class', self.device_class)
        if uom := self.custom_config('unit_of_measurement'):
            self._attr_native_unit_of_measurement = uom

    def update_from_parent(self):
        self.update()
        if self.platform:
            self.schedule_update_ha_state()

    def update(self, data=None):
        attrs = self.parent_attributes
        self._parent_attrs = attrs
        if self._attr in attrs:
            self._available = True
            self._attr_state = attrs.get(self._attr)
            if self._dict_key and isinstance(self._attr_state, dict):
                self._attr_state = self._attr_state.get(self._dict_key)
            svd = self.custom_config_number('value_ratio') or 0
            if svd:
                try:
                    self._attr_state = round(float(self._attr_state) * svd, 3)
                except (TypeError, ValueError):
                    pass
        keys = self._option.get('keys', [])
        if isinstance(keys, list) and self._attr not in keys:
            keys.append(self._attr)
        self._state_attrs = {}.update(attrs) if keys is True else {
            k: v
            for k, v in attrs.items()
            if k in keys
        }
        if data:
            self.update_attrs(data, update_parent=False)

    async def async_update(self):
        await self.hass.async_add_executor_job(self.update)

    def update_attrs(self, attrs: dict, update_parent=True):
        self._state_attrs.update(attrs or {})
        if update_parent:
            if self._parent and hasattr(self._parent, 'update_attrs'):
                getattr(self._parent, 'update_attrs')(attrs or {}, update_parent=False)
        if self.hass and self.platform:
            # don't set state before added to hass
            self.schedule_update_ha_state()
        return self._state_attrs

    async def async_call_parent(self, method, *args, **kwargs):
        ret = None
        if fun := getattr(self, method, None):
            ret = await fun(*args, **kwargs)
        if ret:
            await self.async_update()
        return ret

    def call_parent(self, method, *args, **kwargs):
        ret = None
        if fun := getattr(self, method, None):
            ret = fun(*args, **kwargs)
        if ret:
            self.update()
        return ret


class ToggleSubEntity(BaseSubEntity, ToggleEntity):
    def __init__(self, parent, attr='power', option=None, **kwargs):
        self._prop_power = None
        self._reverse_state = None
        super().__init__(parent, attr, option, **kwargs)

    async def async_added_to_hass(self):
        await super().async_added_to_hass()
        self._reverse_state = self.custom_config_bool('reverse_state', None)

    def update(self, data=None):
        super().update(data)
        if not self._available:
            return
        try:
            self._state = cv.boolean(self._state_attrs.get(self._attr))
        except vol.Invalid:
            self._state = None

    @property
    def is_on(self):
        if self._reverse_state and self._state is not None:
            return not self._state
        return self._state

    async def async_turn_on(self, **kwargs):
        if self._prop_power:
            ret = await self.async_call_parent('async_set_property', self._prop_power, True)
            if ret:
                self._state = True
            return ret
        return await self.async_call_parent('async_turn_on', **kwargs)

    async def async_turn_off(self, **kwargs):
        if self._prop_power:
            ret = await self.async_call_parent('async_set_property', self._prop_power, False)
            if ret:
                self._state = False
            return ret
        return await self.async_call_parent('async_turn_off', **kwargs)