custom_components/xiaomi_miot/device_tracker.py

"""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]}',
            })