249 lines
8.5 KiB
Python
249 lines
8.5 KiB
Python
"""Support for Xiaomi device tracker."""
|
|
import logging
|
|
import time
|
|
from datetime import timedelta
|
|
|
|
from homeassistant.components.device_tracker import (
|
|
DOMAIN as ENTITY_DOMAIN,
|
|
)
|
|
from homeassistant.components.device_tracker.const import SourceType
|
|
from homeassistant.components.device_tracker.config_entry import (
|
|
TrackerEntity as BaseTrackerEntity,
|
|
ScannerEntity as BaseScannerEntity,
|
|
)
|
|
from homeassistant.helpers.restore_state import RestoreEntity
|
|
|
|
from . import (
|
|
DOMAIN,
|
|
CONF_MODEL,
|
|
XIAOMI_CONFIG_SCHEMA as PLATFORM_SCHEMA, # noqa: F401
|
|
HassEntry,
|
|
XEntity,
|
|
MiotEntity,
|
|
async_setup_config_entry,
|
|
bind_services_to_entries,
|
|
)
|
|
from .core.miot_spec import (
|
|
MiotSpec,
|
|
MiotService,
|
|
)
|
|
from .core.coord_transform import gcj02_to_wgs84, bd09_to_wgs84
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
DATA_KEY = f'{ENTITY_DOMAIN}.{DOMAIN}'
|
|
SCAN_INTERVAL = timedelta(seconds=60)
|
|
|
|
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
|
|
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('watch', 'rearview_mirror', 'head_up_display'):
|
|
if 'xiaoxun.watch.' in model:
|
|
entities.append(XiaoxunWatchTrackerEntity(config, srv))
|
|
elif srv.get_property('latitude', 'longitude'):
|
|
entities.append(MiotTrackerEntity(config, srv))
|
|
if not entities and ('xiaoxun.watch.' in model or 'xiaoxun.tracker.' in model):
|
|
# xiaoxun.watch.sw763
|
|
# xiaoxun.tracker.v1
|
|
entities.append(XiaoxunWatchTrackerEntity(config))
|
|
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 ScannerEntity(XEntity, BaseScannerEntity, RestoreEntity):
|
|
@property
|
|
def device_info(self):
|
|
return self._attr_device_info
|
|
|
|
@property
|
|
def unique_id(self):
|
|
return self._attr_unique_id
|
|
|
|
@property
|
|
def is_connected(self):
|
|
"""Return true if the device is connected to the network."""
|
|
return True if self._attr_state else False
|
|
|
|
@property
|
|
def source_type(self):
|
|
"""Return the source type, eg gps or router, of the device."""
|
|
return SourceType.ROUTER
|
|
|
|
def get_state(self) -> dict:
|
|
return {self.attr: self._attr_state}
|
|
|
|
|
|
XEntity.CLS[ENTITY_DOMAIN] = ScannerEntity
|
|
XEntity.CLS['scanner'] = ScannerEntity
|
|
|
|
|
|
class MiotTrackerEntity(MiotEntity, BaseTrackerEntity):
|
|
_attr_latitude = None
|
|
_attr_longitude = None
|
|
_attr_location_name = None
|
|
_attr_location_accuracy = 0
|
|
_disable_location_name = False
|
|
|
|
def __init__(self, config, miot_service: MiotService = None):
|
|
super().__init__(miot_service, config=config, logger=_LOGGER)
|
|
|
|
async def async_added_to_hass(self):
|
|
await super().async_added_to_hass()
|
|
self._disable_location_name = self.custom_config_bool('disable_location_name')
|
|
|
|
async def async_update(self):
|
|
await super().async_update()
|
|
if not self._available or not self._miot_service:
|
|
return
|
|
|
|
if prop := self._miot_service.get_property('latitude'):
|
|
self._attr_latitude = prop.from_device(self.device)
|
|
if prop := self._miot_service.get_property('longitude'):
|
|
self._attr_longitude = prop.from_device(self.device)
|
|
if prop := self._miot_service.get_property('current_address'):
|
|
self._attr_location_name = prop.from_device(self.device)
|
|
await self.transform_coord()
|
|
|
|
async def transform_coord(self, default=None):
|
|
if not (self._attr_latitude or self._attr_longitude):
|
|
return
|
|
typ = self.custom_config('coord_type') or default
|
|
if not typ:
|
|
return
|
|
typ = f'{typ}'.lower()
|
|
if typ == 'gcj02':
|
|
self._attr_longitude, self._attr_latitude = gcj02_to_wgs84(self._attr_longitude, self._attr_latitude)
|
|
if typ == 'bd09':
|
|
self._attr_longitude, self._attr_latitude = bd09_to_wgs84(self._attr_longitude, self._attr_latitude)
|
|
|
|
@property
|
|
def should_poll(self):
|
|
"""No polling for entities that have location pushed."""
|
|
return True
|
|
|
|
@property
|
|
def source_type(self):
|
|
"""Return the source type, eg gps or router, of the device."""
|
|
return SourceType.GPS
|
|
|
|
@property
|
|
def latitude(self):
|
|
"""Return latitude value of the device."""
|
|
return self._attr_latitude
|
|
|
|
@property
|
|
def longitude(self):
|
|
"""Return longitude value of the device."""
|
|
return self._attr_longitude
|
|
|
|
@property
|
|
def location_name(self):
|
|
"""Return a location name for the current location of the device."""
|
|
if self._disable_location_name:
|
|
return None
|
|
return self._attr_location_name
|
|
|
|
@property
|
|
def location_accuracy(self):
|
|
"""Return the location accuracy of the device.
|
|
Value in meters.
|
|
"""
|
|
return self._attr_location_accuracy
|
|
|
|
@property
|
|
def battery_level(self):
|
|
"""Return the battery level of the device."""
|
|
if not self._miot_service:
|
|
return None
|
|
sls = [self._miot_service, *self._miot_service.spec.get_services('battery')]
|
|
for srv in sls:
|
|
prop = srv.get_property('battery_level')
|
|
if prop:
|
|
return prop.from_device(self.device)
|
|
return None
|
|
|
|
|
|
class XiaoxunWatchTrackerEntity(MiotTrackerEntity):
|
|
def __init__(self, config, miot_service: MiotService = None, miot_spec: MiotSpec = None):
|
|
self._miot_spec = miot_spec
|
|
super().__init__(config=config, miot_service=miot_service)
|
|
|
|
@property
|
|
def device_eid(self):
|
|
did = f'{self.miot_did}'
|
|
return did.replace('xiaoxun.', '')
|
|
|
|
async def async_update(self):
|
|
await super().async_update()
|
|
await self.update_location()
|
|
|
|
async def update_location(self):
|
|
did = f'{self.miot_did}'
|
|
mic = self.xiaomi_cloud
|
|
if not did or not mic:
|
|
return
|
|
pms = {
|
|
'app_id': '10025',
|
|
'dids': [did],
|
|
'params': {
|
|
'CID': 50031,
|
|
'model': self.model,
|
|
'SN': int(time.time() / 1000),
|
|
'PL': {
|
|
'Size': 1,
|
|
'Key': '78999898989898998',
|
|
'EID': self.device_eid,
|
|
},
|
|
},
|
|
}
|
|
rdt = await mic.async_request_api('third/api', pms) or {}
|
|
loc = {}
|
|
for v in (rdt.get('result') or {}).get('PL', {}).get('List', {}).values():
|
|
if loc := v.get('result') or {}:
|
|
loc.setdefault('device', v)
|
|
loc.setdefault('timestamp', v.get('timestamp', ''))
|
|
break
|
|
if not loc:
|
|
self.logger.warning('%s: Got xiaoxun watch location faild: %s', self.name_model, rdt)
|
|
return
|
|
self.logger.debug('%s: Got xiaoxun watch location: %s', self.name_model, rdt)
|
|
dvc = loc.get('device') or {}
|
|
gps = f"{loc.get('location', '')},".split(',')
|
|
self._attr_latitude = float(gps[1])
|
|
self._attr_longitude = float(gps[0])
|
|
self._attr_location_name = loc.get('desc')
|
|
self._attr_location_accuracy = int(loc.get('radius') or 0)
|
|
await self.transform_coord(default='gcj02')
|
|
self.update_attrs({
|
|
'sos': dvc.get('SOS', 0),
|
|
'steps': dvc.get('steps', 0),
|
|
'home_wifi': dvc.get('home_wifi', 0),
|
|
'imei': dvc.get('imei'),
|
|
'adcode': loc.get('adcode'),
|
|
'country': loc.get('country'),
|
|
'province': loc.get('province'),
|
|
'city': loc.get('province'),
|
|
'district': loc.get('district'),
|
|
'township': loc.get('township'),
|
|
'road': loc.get('road'),
|
|
'street': loc.get('street'),
|
|
})
|
|
if tim := loc.get('timestamp', ''):
|
|
self.update_attrs({
|
|
'timestamp': f'{tim[0:4]}-{tim[4:6]}-{tim[6:8]} {tim[8:10]}:{tim[10:12]}:{tim[12:14]}',
|
|
})
|