custom_components/xiaomi_miot/binary_sensor.py

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