"""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)