344 lines
14 KiB
Python
344 lines
14 KiB
Python
"""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)
|