"""Support for Xiaomi binary sensors.""" import logging import time import json from datetime import datetime from functools import cached_property from homeassistant.components.binary_sensor import ( DOMAIN as ENTITY_DOMAIN, BinarySensorEntity as BaseEntity, BinarySensorDeviceClass, ) from homeassistant.helpers.restore_state import RestoreEntity from . import ( DOMAIN, CONF_MODEL, XIAOMI_CONFIG_SCHEMA as PLATFORM_SCHEMA, # noqa: F401 HassEntry, XEntity, MiotToggleEntity, async_setup_config_entry, bind_services_to_entries, ) from .core.miot_spec import ( MiotSpec, MiotService, ) from .core.xiaomi_cloud import MiotCloud from .core.utils import local_zone _LOGGER = logging.getLogger(__name__) DATA_KEY = f'{ENTITY_DOMAIN}.{DOMAIN}' SERVICE_TO_METHOD = {} async def async_setup_entry(hass, config_entry, async_add_entities): HassEntry.init(hass, config_entry).new_adder(ENTITY_DOMAIN, async_add_entities) await async_setup_config_entry(hass, config_entry, async_setup_platform, async_add_entities, ENTITY_DOMAIN) async def async_setup_platform(hass, config, async_add_entities, discovery_info=None): hass.data.setdefault(DATA_KEY, {}) hass.data[DOMAIN]['add_entities'][ENTITY_DOMAIN] = async_add_entities config['hass'] = hass did = str(config.get('miot_did') or '') model = str(config.get(CONF_MODEL) or '') spec = hass.data[DOMAIN]['miot_specs'].get(model) entities = [] if isinstance(spec, MiotSpec): for srv in spec.get_services( 'toilet', 'seat', 'motion_sensor', 'magnet_sensor', 'submersion_sensor', 'alertor', ): if spec.get_service('nobody_time'): # lumi.motion.agl02 # lumi.motion.agl04 pass elif model in ['lumi.sensor_wleak.aq1', 'htcx.alarm.dt210']: pass elif not srv.mapping(): continue if srv.name in ['toilet']: entities.append(MiotToiletEntity(config, srv)) elif srv.name in ['seat'] and spec.name in ['toilet']: if spec.get_service('toilet'): continue # tinymu.toiletlid.v1 entities.append(MiotToiletEntity(config, srv)) elif 'blt.' in did: entities.append(BleBinarySensorEntity(config, srv)) elif 'lumi.' in model: entities.append(LumiBinarySensorEntity(config, srv)) else: entities.append(MiotBinarySensorEntity(config, srv)) for entity in entities: hass.data[DOMAIN]['entities'][entity.unique_id] = entity async_add_entities(entities, update_before_add=True) bind_services_to_entries(hass, SERVICE_TO_METHOD) class BinarySensorEntity(XEntity, BaseEntity, RestoreEntity): def set_state(self, data: dict): val = self.conv.value_from_dict(data) if val is None: return if self.custom_reverse: self._attr_extra_state_attributes['reverse_state'] = True val = not val self._attr_is_on = val def get_state(self) -> dict: return {self.attr: not self._attr_is_on if self.custom_reverse else self._attr_is_on} @cached_property def custom_reverse(self): return self.custom_config_bool('reverse_state', False) XEntity.CLS[ENTITY_DOMAIN] = BinarySensorEntity class MiotBinarySensorEntity(MiotToggleEntity, BaseEntity): def __init__(self, config, miot_service: MiotService, **kwargs): kwargs.setdefault('logger', _LOGGER) super().__init__(miot_service, config=config, **kwargs) pls = [] if len(miot_service.properties) > 0: first_property = list(miot_service.properties.values() or [])[0].name if first_property: pls.append(first_property) self._prop_state = miot_service.get_property(*pls) if miot_service.name in ['motion_sensor']: self._prop_state = miot_service.get_property('motion_state', 'no_motion_duration') or self._prop_state if self._prop_state.name in ['illumination']: # cgllc.motion.cgpr1 self._prop_state = None self._attr_device_class = BinarySensorDeviceClass.MOTION if self._prop_state is None: srv = miot_service.spec.get_service('nobody_time') if srv: self._prop_state = srv.get_property('nobody_time') if miot_service.name in ['magnet_sensor']: self._prop_state = miot_service.get_property('contact_state') or self._prop_state if self._prop_state and self._prop_state.name in ['contact_state']: # https://github.com/al-one/hass-xiaomi-miot/issues/270 self._vars['reverse_state'] = True self._attr_device_class = BinarySensorDeviceClass.DOOR if miot_service.name in ['submersion_sensor']: self._prop_state = miot_service.get_property('submersion_state') or self._prop_state self._attr_device_class = BinarySensorDeviceClass.MOISTURE async def async_added_to_hass(self): await super().async_added_to_hass() if prop := self.custom_config('state_property'): self._prop_state = self._miot_service.get_property_by_full_name(prop) or self._prop_state self._state_attrs.update({ 'state_property': self._prop_state.full_name if self._prop_state else None, }) rev = self.custom_config_bool('reverse_state', None) if rev is not None: self._vars['reverse_state'] = rev @property def is_on(self): ret = self._state if self._prop_state: val = self._prop_state.from_device(self.device) if val is None: pass elif self._prop_state.name in ['no_motion_duration', 'nobody_time']: if self._prop_state.unit in ['minutes']: val *= 60 dur = self.custom_config_integer('motion_timeout') if dur is None and self._prop_state.value_range: stp = self._prop_state.range_step() if stp >= 10: dur = self._prop_state.range_min() + stp if dur is None: dur = 60 ret = val <= dur elif self._prop_state.value_list: # linp.magnet.m1 if des := self._prop_state.list_description(val): des = f'{des}'.lower() return des in ['open', 'opened'] else: ret = val if self._vars.get('reverse_state'): ret = not ret return ret @property def device_class(self): return self.get_device_class(BinarySensorDeviceClass) class BleBinarySensorEntity(MiotBinarySensorEntity): def __init__(self, config, miot_service: MiotService): super().__init__(config, miot_service) self._prop_illumination = miot_service.get_property('illumination') if not self._prop_illumination: if srv := miot_service.spec.get_service('illumination_sensor'): self._prop_illumination = srv.get_property('illumination') # https://iot.mi.com/new/doc/embedded-development/ble/object-definition self._props = [ 'event.15', # 0x000F motion&illumination 'prop.4103', # 0x1007 lux 'prop.4106', # 0x100A battery 'prop.4117', # 0x1015 smoke 'prop.4119', # 0x1017 no_motion_duration 'prop.4123', # 0x101B no_motion_timeout 'prop.4120', # 0x1018 illumination_level 'prop.4121', # 0x1019 magnet ] async def async_update_for_main_entity(self): if self.custom_config_bool('use_ble_object', True): await self.async_update_ble_data() await super().async_update_for_main_entity() async def async_update_ble_data(self): did = self.miot_did mic = self.miot_cloud if not did or not isinstance(mic, MiotCloud): return pms = { 'did': did, 'props': self._props, } rdt = await mic.async_request_api('device/batchdevicedatas', [pms]) or {} self.logger.debug('%s: Got miio cloud props: %s', self.name_model, rdt) props = (rdt.get('result') or {}).get(did, {}) sta = None adt = {} for k, v in props.items(): if v is None: continue ise = 'event.' in k evt = json.loads(v) if ise else {'value': [v]} tim = float(evt.get('timestamp') or 0) val = vlk = None if vls := evt.get('value'): val = vls[0] if val: try: val = int.from_bytes(bytes.fromhex(val), 'little') except (TypeError, ValueError): val = None self.logger.warning('%s: BLE object data invalid: %s (%s)', self.name_model, k, vls) if ise and not tim: continue # https://iot.mi.com/new/doc/embedded-development/ble/object-definition#%E6%9C%89%E4%BA%BA%E7%A7%BB%E5%8A%A8%E4%BA%8B%E4%BB%B6%EF%BC%88%E5%B8%A6%E5%85%89%E7%85%A7%EF%BC%89 if k == 'event.15': adt.update({ 'trigger_time': tim, 'trigger_at': datetime.fromtimestamp(tim, local_zone()), }) dif = time.time() - adt['trigger_time'] sta = dif <= (self.custom_config_integer('motion_timeout') or 60) vlk = 'illumination' if self._prop_illumination and self._prop_illumination.value_range: vlk = self._prop_illumination.full_name # https://iot.mi.com/new/doc/embedded-development/ble/object-definition#%E5%85%89%E7%85%A7%E5%BA%A6%E5%B1%9E%E6%80%A7 elif k == 'prop.4103': vlk = 'illumination' if self._prop_illumination and self._prop_illumination.value_range: vlk = self._prop_illumination.full_name # https://iot.mi.com/new/doc/embedded-development/ble/object-definition#%E6%97%A0%E4%BA%BA%E7%A7%BB%E5%8A%A8%E5%B1%9E%E6%80%A7 elif k == 'prop.4119': vlk = 'no_motion_seconds' # https://iot.mi.com/new/doc/embedded-development/ble/object-definition#%E5%85%89%E7%85%A7%E5%BC%BA%E5%BC%B1%E5%B1%9E%E6%80%A7 elif k == 'prop.4120': adt['light_strong'] = not not val vlk = 'illumination_level' val = 'strong' if val else 'weak' if self._prop_illumination and self._prop_illumination.value_list: vid = self._prop_illumination.list_value(val) if vid is not None: adt[self._prop_illumination.full_name] = vid # https://iot.mi.com/new/doc/embedded-development/ble/object-definition#%E9%97%A8%E7%A3%81%E5%B1%9E%E6%80%A7 elif k == 'prop.4121': sta = val != 2 if vlk is not None and val is not None: adt[vlk] = val if sta is not None: self._state = sta if adt: await self.async_update_attrs(adt) class MiotToiletEntity(MiotBinarySensorEntity): def __init__(self, config, miot_service: MiotService): super().__init__(config, miot_service) self._prop_state = None for s in miot_service.spec.get_services('toilet', 'seat'): if p := s.get_property('seating_state'): self._prop_state = p break if not self._prop_state: self._prop_state = miot_service.get_property( self._prop_state.name if self._prop_state else 'status', ) @property def icon(self): return 'mdi:toilet' class LumiBinarySensorEntity(MiotBinarySensorEntity): def __init__(self, config, miot_service: MiotService): super().__init__(config, miot_service) async def async_update(self): await super().async_update() if not self._available: return mic = self.miot_cloud pes = dlg = None if isinstance(mic, MiotCloud): now = int(time.time()) ofs = self.custom_config_integer('time_start_offset') or -86400 * 3 dlg = await mic.async_get_last_device_data(self.miot_did, 'device_log', time_start=now + ofs) pes = json.loads(dlg or '[]') adt = {} typ = None dif = time.time() if pes and len(pes) >= 2: typ = pes[1][0] adt['trigger_type'] = typ adt['trigger_time'] = int(pes[0] or 0) adt['trigger_at'] = f'{datetime.fromtimestamp(adt["trigger_time"])}' dif = time.time() - adt['trigger_time'] if typ == 'prop.illumination': prop = self._miot_service.get_property('illumination') if prop: adt[prop.full_name] = pes[1][1][0] self._state = None if typ == 'event.motion' or self._miot_service.name in ['motion_sensor']: self._state = dif <= (self.custom_config_integer('motion_timeout') or 60) elif typ in ['event.open', 'event.close']: self._state = typ == 'event.open' elif typ in ['event.leak', 'event.no_leak']: self._state = typ == 'event.leak' elif self._prop_state and self._prop_state.full_name in self._state_attrs: _LOGGER.info('%s: Get miio data failed: %s', self.name_model, dlg) else: _LOGGER.warning('%s: Get miio data failed: %s', self.name_model, dlg) if self._prop_state and self._state is not None: adt[self._prop_state.full_name] = self._state if adt: await self.async_update_attrs(adt)