File: /volume1/docker/homeassistant2025/custom_components/xiaomi_miot/sensor.py
"""Support for Xiaomi sensors."""
import logging
import time
import json
from typing import cast
from datetime import datetime, timedelta
from functools import cmp_to_key, cached_property
from homeassistant.const import STATE_UNKNOWN
from homeassistant.components.sensor import (
DOMAIN as ENTITY_DOMAIN,
SensorEntity as BaseEntity,
SensorDeviceClass,
SensorStateClass,
)
from homeassistant.helpers.restore_state import RestoreEntity, RestoredExtraData
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from . import (
DOMAIN,
CONF_MODEL,
XIAOMI_CONFIG_SCHEMA as PLATFORM_SCHEMA, # noqa: F401
HassEntry,
XEntity,
MiotEntity,
BaseSubEntity,
MiCoordinatorEntity,
MiotCloud,
async_setup_config_entry,
bind_services_to_entries,
)
from .core.miot_spec import (
MiotSpec,
MiotService,
)
from .core.utils import local_zone, get_translation
_LOGGER = logging.getLogger(__name__)
DATA_KEY = f'{ENTITY_DOMAIN}.{DOMAIN}'
SERVICE_TO_METHOD = {}
async def async_setup_entry(hass, config_entry, async_add_entities):
entry = HassEntry.init(hass, config_entry).new_adder(ENTITY_DOMAIN, async_add_entities)
cloud = await entry.get_cloud()
if cloud:
if not entry.get_config('disable_message'):
hass.data[DOMAIN]['accounts'].setdefault(cloud.user_id, {})
if not hass.data[DOMAIN]['accounts'][cloud.user_id].get('messenger'):
entity = MihomeMessageSensor(hass, cloud)
await entity.coordinator.async_config_entry_first_refresh()
hass.data[DOMAIN]['accounts'][cloud.user_id]['messenger'] = entity
async_add_entities([entity], update_before_add=False)
if not entry.get_config('disable_scene_history'):
homes = await cloud.async_get_homerooms()
for home in homes:
home_id = home.get('id')
if hass.data[DOMAIN]['accounts'][cloud.user_id].get(f'scene_history_{home_id}'):
continue
entity = MihomeSceneHistorySensor(hass, cloud, home_id, home.get('uid'))
await entity.coordinator.async_config_entry_first_refresh()
hass.data[DOMAIN]['accounts'][cloud.user_id][f'scene_history_{home_id}'] = entity
async_add_entities([entity], update_before_add=False)
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
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(
'switch_sensor',
'lock', 'door', 'bed', 'walking_pad', 'treadmill',
'fridge_chamber', 'germicidal_lamp', 'vital_signs',
'sterilizer', 'steriliser', 'chair', 'dryer', 'clothes_dryer',
):
if srv.name in ['lock']:
if not srv.get_property('operation_method', 'operation_id'):
continue
elif srv.name in ['door']:
if spec.get_service('lock'):
continue
elif not srv.mapping():
continue
entities.append(MiotSensorEntity(config, srv))
for entity in entities:
hass.data[DOMAIN]['entities'][entity.unique_id] = entity
async_add_entities(entities, update_before_add=False)
bind_services_to_entries(hass, SERVICE_TO_METHOD)
def datetime_with_tzinfo(value):
if isinstance(value, datetime):
pass
elif isinstance(value, str):
value = datetime.fromisoformat(value)
elif isinstance(value, (int, float)):
value = datetime.fromtimestamp(value)
else:
value = cast(datetime, value)
if value.tzinfo is None:
value = value.replace(tzinfo=local_zone())
return value
class SensorEntity(XEntity, BaseEntity, RestoreEntity):
def on_init(self):
self._attr_state_class = self.custom_config('state_class')
self._attr_native_unit_of_measurement = self.custom_config('unit_of_measurement')
if self._miot_property:
if not self._attr_icon:
self._attr_icon = self._miot_property.entity_icon
if not self._attr_device_class:
self._attr_device_class = self._miot_property.device_class
if not self._attr_state_class:
self._attr_state_class = self._miot_property.state_class
if not self._attr_native_unit_of_measurement:
self._attr_native_unit_of_measurement = self._miot_property.unit_of_measurement
def get_state(self) -> dict:
return {self.attr: self._attr_native_value}
def set_state(self, data: dict):
value = self.conv.value_from_dict(data)
prop = self._miot_property
if prop and prop.value_range:
if not prop.range_valid(value):
value = None
if value is not None:
try:
if ratio := self.custom_value_ratio:
value = round(float(value) * ratio, 3)
elif self.state_class:
value = round(float(value), 3)
except (TypeError, ValueError):
value = None
if self.device_class == SensorDeviceClass.TIMESTAMP:
value = datetime_with_tzinfo(value)
self._attr_native_value = value
@cached_property
def custom_value_ratio(self):
return self.custom_config_number('value_ratio') or 0
XEntity.CLS[ENTITY_DOMAIN] = SensorEntity
class MiotSensorEntity(MiotEntity, BaseEntity):
def __init__(self, config, miot_service: MiotService):
super().__init__(miot_service, config=config, logger=_LOGGER)
first_property = None
if len(miot_service.properties) > 0:
first_property = list(miot_service.properties.values() or [])[0].name
self._prop_state = miot_service.get_property(
'status', 'fault', first_property or 'status',
)
if miot_service.name in ['lock']:
self._prop_state = miot_service.get_property('operation_method') or self._prop_state
self._attr_icon = self._miot_service.entity_icon
self._attr_state_class = None
self._attr_native_unit_of_measurement = None
if self._prop_state:
self._name = f'{self.device_name} {self._prop_state.friendly_desc}'
self._attr_icon = self._prop_state.entity_icon or self._attr_icon
self._attr_state_class = self._prop_state.state_class
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
if self._prop_state:
self._attr_icon = self._prop_state.entity_icon
self._attr_device_class = self._prop_state.device_class
self._attr_native_unit_of_measurement = self._prop_state.unit_of_measurement
self._state_attrs.update({
'state_property': self._prop_state.full_name,
})
cls = self.custom_config('state_class')
if cls and cls in SensorStateClass:
self._attr_state_class = SensorStateClass(cls)
elif cls in ['', False]:
self._attr_state_class = None
if uom := self.custom_config('unit_of_measurement'):
self._attr_native_unit_of_measurement = uom
async def async_update(self):
await super().async_update()
if not self._available or not self._prop_state:
return
if self._miot_service.name in ['lock'] and self._prop_state.full_name not in self._state_attrs:
if how := self._state_attrs.get('lock_method'):
await self.async_update_attrs({
self._prop_state.full_name: get_translation(how, ['lock_method']),
})
elif edt := self._state_attrs.get('event.11', {}):
if isinstance(edt, dict):
await self.async_update_attrs({
self._prop_state.full_name: edt.get('method'),
})
self._prop_state.description_to_dict(self._state_attrs)
@property
def device_class(self):
"""Return the class of this entity."""
return self.get_device_class(SensorDeviceClass)
@property
def native_value(self):
prop = self._prop_state
if not prop:
return None
key = f'{prop.full_name}_desc'
if key in self._state_attrs:
return f'{self._state_attrs[key]}'.lower()
val = prop.from_device(self.device)
if not prop.range_valid(val):
val = None
return val
class BaseSensorSubEntity(BaseSubEntity, BaseEntity):
def __init__(self, parent, attr, option=None, **kwargs):
kwargs.setdefault('domain', ENTITY_DOMAIN)
self._attr_state_class = None
super().__init__(parent, attr, option, **kwargs)
@property
def native_value(self):
value = self._attr_state
if hasattr(self, '_attr_native_value') and self._attr_native_value is not None:
value = self._attr_native_value
value = get_translation(value, [self._attr])
if self.device_class == SensorDeviceClass.TIMESTAMP:
value = datetime_with_tzinfo(value)
return value
@property
def state_class(self):
return self._attr_state_class
async def async_added_to_hass(self):
await super().async_added_to_hass()
cls = self.custom_config('state_class')
if cls and cls in SensorStateClass:
self._attr_state_class = SensorStateClass(cls)
elif cls in ['', False]:
self._attr_state_class = None
if uom := self.custom_config('unit_of_measurement'):
self._attr_native_unit_of_measurement = uom
def update(self, data=None):
old_state = self._attr_state
super().update(data)
now = datetime.now(tz=local_zone(self.hass))
if self.state_class in ['total_increasing'] and old_state not in [None, '', STATE_UNKNOWN]:
ptm = self._extra_attrs.get('updated_time') or now
if now.strftime('%Y-%m-%d') == ptm.strftime('%Y-%m-%d'):
try:
if (self._attr_state or 0) < old_state:
self._attr_state = old_state
except (TypeError, ValueError) as exc:
_LOGGER.warning(
'%s: Total increasing sensor state error: %s',
self.name_model, [exc, self._attr_state, old_state],
)
if self._attr_state != old_state:
self._extra_attrs['updated_time'] = now
class MihomeMessageSensor(MiCoordinatorEntity, BaseEntity, RestoreEntity):
_filter_homes = None
_exclude_types = None
_has_none_message = False
def __init__(self, hass, cloud: MiotCloud):
self.hass = hass
self.cloud = cloud
self.message = {}
self.entity_id = f'{ENTITY_DOMAIN}.mi_{cloud.user_id}_message'
self._attr_unique_id = f'{DOMAIN}-mihome-message-{cloud.user_id}'
self._attr_name = f'Xiaomi {cloud.user_id} message'
self._attr_icon = 'mdi:message'
self._attr_should_poll = False
self._attr_native_value = None
self._attr_extra_state_attributes = {
'entity_class': self.__class__.__name__,
}
self.coordinator = DataUpdateCoordinator(
hass,
_LOGGER,
name=self._attr_unique_id,
update_method=self.fetch_latest_message,
update_interval=timedelta(seconds=15),
)
super().__init__(self.coordinator)
async def async_added_to_hass(self):
await super().async_added_to_hass()
self.hass.data[DOMAIN]['entities'][self.entity_id] = self
self._filter_homes = self.custom_config_list('filter_home') or []
self._exclude_types = list(map(lambda x: int(x), self.custom_config_list('exclude_type', [13]) or []))
if sec := self.custom_config_integer('interval_seconds'):
self.coordinator.update_interval = timedelta(seconds=sec)
if restored := await self.async_get_last_extra_data():
self._attr_native_value = restored.as_dict().get('state')
self._attr_extra_state_attributes.update(restored.as_dict().get('attrs', {}))
self._attr_extra_state_attributes['filter_homes'] = self._filter_homes
self._attr_extra_state_attributes['exclude_types'] = self._exclude_types
async def async_will_remove_from_hass(self):
"""Run when entity will be removed from hass.
To be extended by integrations.
"""
await super().async_will_remove_from_hass()
self.hass.data[DOMAIN]['accounts'].get(self.cloud.user_id, {}).pop('messenger', None)
@property
def extra_restore_state_data(self):
"""Return entity specific state data to be restored."""
return RestoredExtraData({
'state': self.native_value,
'attrs': self._attr_extra_state_attributes,
})
async def async_set_message(self, msg):
if msg == self.message:
return
if old := self._attr_native_value:
self._attr_extra_state_attributes['prev_message'] = old
con = msg.get('content')
if tit := msg.get('title'):
if con:
self._attr_native_value = f'{con}: {tit}'
else:
self._attr_native_value = tit
logger = _LOGGER.info if old != self._attr_native_value else _LOGGER.debug
logger('New xiaomi message for %s: %s', self.cloud.user_id, self._attr_native_value)
tim = msg.get('ctime')
params = msg.get('params', {})
body = params.get('body', {})
self._attr_entity_picture = msg.get('img_url') or None
self._attr_extra_state_attributes.update({
'msg_id': msg.get('msg_id'),
'is_new': msg.get('is_new'),
'type': msg.get('type'),
'title': tit,
'content': con,
'user_id': msg.get('uid'),
'ctime': tim,
'timestamp': datetime.fromtimestamp(tim, local_zone()) if tim else None,
'model': params.get('model', body.get('model')),
'device_id': msg.get('did', body.get('did')),
'home_name': msg.get('homeName'),
'room_name': msg.get('roomName'),
'event': body.get('event'),
'event_data': body.get('extra', body.get('value')),
})
async def fetch_latest_message(self):
res = await self.cloud.async_request_api('v2/message/v2/typelist', data={}) or {}
mls = (res.get('result') or {}).get('messages') or []
mls.sort(key=lambda x: x.get('ctime', 0), reverse=False)
prev_time = self._attr_extra_state_attributes.get('ctime')
prev_mid = self._attr_extra_state_attributes.get('msg_id')
msg = {}
for m in mls:
hre = m.get('params', {}).get('body', {}).get('homeRoomExtra', {})
home = hre.get('homeName')
if self._filter_homes and home and home not in self._filter_homes:
continue
typ = m.get('type', 0)
if self._exclude_types and typ in self._exclude_types:
continue
tim = m.get('ctime', 0)
mid = m.get('msg_id', 0)
if prev_time and tim < prev_time:
continue
if prev_mid and mid <= prev_mid:
continue
m['homeName'] = home
m['roomName'] = hre.get('roomName')
msg = m
break
if not mls:
if not self._has_none_message:
# Only raise a warning if there was a failure obtaining the xiaomi message
# Otherwise, a warning will show anytime that there are simply no messages waiting
code = res.get('code')
message = res.get('message')
if code == 0 and message == 'ok':
_LOGGER.debug('Get xiaomi message for %s failed: %s', self.cloud.user_id, res)
else:
_LOGGER.warning('Get xiaomi message for %s failed: %s', self.cloud.user_id, res)
self._has_none_message = True
if msg:
await self.async_set_message(msg)
self.message = msg
self._has_none_message = False
return msg
class MihomeSceneHistorySensor(MiCoordinatorEntity, BaseEntity, RestoreEntity):
MESSAGE_TIMEOUT = 60
UPDATE_INTERVAL = 15
_has_none_message = False
def __init__(self, hass, cloud: MiotCloud, home_id, owner_user_id):
self.hass = hass
self.cloud = cloud
self.home_id = int(home_id)
self.owner_user_id = int(owner_user_id)
self.entity_id = f'{ENTITY_DOMAIN}.mi_{cloud.user_id}_{home_id}_scene_history'
self._attr_unique_id = f'{DOMAIN}-mihome-scene-history-{cloud.user_id}_{home_id}'
self._attr_name = f'Xiaomi {cloud.user_id}_{home_id} Scene History'
self._attr_icon = 'mdi:message'
self._attr_should_poll = False
self._attr_native_value = None
self._attr_extra_state_attributes = {
'entity_class': self.__class__.__name__,
}
self.coordinator = DataUpdateCoordinator(
hass,
_LOGGER,
name=self._attr_unique_id,
update_method=self.fetch_latest_message,
update_interval=timedelta(seconds=self.UPDATE_INTERVAL),
)
super().__init__(self.coordinator)
async def async_added_to_hass(self):
await super().async_added_to_hass()
self.hass.data[DOMAIN]['entities'][self.entity_id] = self
if sec := self.custom_config_integer('interval_seconds'):
self.coordinator.update_interval = timedelta(seconds=sec)
if restored := await self.async_get_last_extra_data():
restored_dict = restored.as_dict()
attrs = restored_dict.get('attrs', {})
if ts := attrs.get('ts'):
attrs['timestamp'] = datetime.fromtimestamp(ts, local_zone()) if ts else None
_LOGGER.debug(
'xiaomi scene history %s %d, async_added_to_hass restore state: state= %s attrs= %s',
self.cloud.user_id, self.home_id, restored_dict.get('state'), attrs,
)
self._attr_native_value = restored_dict.get('state')
self._attr_extra_state_attributes.update(attrs)
async def async_will_remove_from_hass(self):
"""Run when entity will be removed from hass.
To be extended by integrations.
"""
await super().async_will_remove_from_hass()
self.hass.data[DOMAIN]['accounts'].get(self.cloud.user_id, {}).pop(f'scene_history_{self.home_id}', None)
@property
def extra_restore_state_data(self):
"""Return entity specific state data to be restored."""
return RestoredExtraData({
'state': self.native_value,
'attrs': self._attr_extra_state_attributes,
})
def trim_message(self, msg):
ts = msg.get('time') or int(time.time())
return {
"from": msg.get('from'),
"name": msg.get('name'),
"ts": ts,
'timestamp': datetime.fromtimestamp(ts, local_zone()),
"scene_id": str(msg.get('userSceneId')),
"targets": msg.get('msg', []),
}
@staticmethod
def _cmp_message(a, b):
a_ts, b_ts = a.get('ts', 0), b.get('ts', 0)
if a_ts != b_ts:
return a_ts - b_ts
a_scene_id, b_scene_id = a.get('scene_id', 0), b.get('scene_id', 0)
if a_scene_id < b_scene_id:
return -1
if a_scene_id > b_scene_id:
return 1
return 0
async def async_set_message(self, msg):
self._attr_native_value = msg.get('name')
_LOGGER.debug('New xiaomi scene history for %s %d: %s', self.cloud.user_id, self.home_id, self._attr_native_value)
old = self._attr_extra_state_attributes or {}
self._attr_extra_state_attributes.update({**msg, 'prev_value': old.get('name'), 'prev_scene_id': old.get('scene_id')})
async def fetch_latest_message(self):
res = await self.cloud.async_request_api('scene/history', data={
"home_id": self.home_id,
"uid": int(self.cloud.user_id),
"owner_uid": self.owner_user_id,
"command": "history",
"limit": 15,
}) or {}
messages = [self.trim_message(msg) for msg in (res.get('result') or {}).get('history') or []]
if not messages:
if not self._has_none_message:
_LOGGER.info('Get xiaomi scene history for %s %d failed: %s', self.cloud.user_id, self.home_id, res)
self._has_none_message = True
return {}
messages.sort(key=cmp_to_key(self._cmp_message), reverse=False)
_LOGGER.debug(
'Get xiaomi scene history for %s %d success: prev_timestamp= %d prev_scene_id= %s messages= %s,',
self.cloud.user_id, self.home_id,
self._attr_extra_state_attributes.get('ts') or 0,
self._attr_extra_state_attributes.get('scene_id') or '',
messages,
)
must_after = int(time.time()) - self.MESSAGE_TIMEOUT
for msg in messages:
if msg.get('ts') < must_after:
continue
if self._cmp_message(msg, self._attr_extra_state_attributes) <= 0:
continue
await self.async_set_message(msg)
self._has_none_message = False
return msg
return {}
class XiaoaiConversationSensor(MiCoordinatorEntity, BaseSensorSubEntity):
def __init__(self, parent, hass, option=None):
BaseSensorSubEntity.__init__(self, parent, 'conversation', option)
self.hass = hass
self.conversation = {}
self._available = True
self._attr_native_value = None
self._option.setdefault('icon', 'mdi:account-voice')
self.coordinator = DataUpdateCoordinator(
hass,
_LOGGER,
name=self.unique_id,
update_method=self.fetch_latest_message,
update_interval=timedelta(seconds=5),
)
super().__init__(self.coordinator)
async def async_added_to_hass(self):
await super().async_added_to_hass()
self.hass.data[DOMAIN]['entities'][self.entity_id] = self
await self.coordinator.async_refresh()
if sec := self.custom_config_integer('interval_seconds'):
self.coordinator.update_interval = timedelta(seconds=sec)
async def fetch_latest_message(self):
mic = self._parent.xiaoai_cloud
dvc = self._parent.xiaoai_device or {}
aid = dvc.get('deviceID')
if not isinstance(mic, MiotCloud) or not aid:
self._available = False
return
api = 'https://userprofile.mina.mi.com/device_profile/v2/conversation'
dat = {
'hardware': dvc.get('hardware', ''),
'timestamp': int(time.time() * 1000),
'limit': 3,
}
cks = {
'deviceId': aid,
}
try:
res = await mic.async_request_api(api, data=dat, method='GET', cookies=cks) or {}
rdt = res.get('data', {})
if not isinstance(rdt, dict):
rdt = json.loads(rdt) or {}
except (TypeError, ValueError, Exception) as exc:
rdt = {}
_LOGGER.warning(
'%s: Got exception while fetch xiaoai conversation: %s',
self.name_model, [aid, exc],
)
mls = rdt.get('records') or []
msg = mls.pop(0) if mls else {}
self.conversation = msg
old = self._attr_native_value
if con := msg.get('query'):
self._state = con
self._attr_native_value = con
logger = _LOGGER.info if old != self._attr_native_value else _LOGGER.debug
logger('%s: New xiaoai conversation: %s', self.name_model, self._attr_native_value)
tim = msg.get('time')
ans = []
for v in msg.get('answers', []):
if not isinstance(v, dict):
continue
typ = v.get('type', '').lower()
v.pop('bitSet', None)
v.get(typ, {}).pop('bitSet', None)
ans.append(v)
self._state_attrs.update({
'content': con,
'answers': ans,
'history': [
v.get('query')
for v in mls
],
'timestamp': datetime.fromtimestamp(tim / 1000, local_zone()) if tim else None,
})
return msg