ha-core/homeassistant/components/homekit_controller/__init__.py

"""Support for Homekit device discovery."""
import logging

from homeassistant.core import callback
from homeassistant.helpers.entity import Entity
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr

# We need an import from .config_flow, without it .config_flow is never loaded.
from .config_flow import HomekitControllerFlowHandler  # noqa: F401
from .connection import get_accessory_information, HKDevice
from .const import CONTROLLER, ENTITY_MAP, KNOWN_DEVICES
from .const import DOMAIN  # noqa: pylint: disable=unused-import
from .storage import EntityMapStorage

_LOGGER = logging.getLogger(__name__)


def escape_characteristic_name(char_name):
    """Escape any dash or dots in a characteristics name."""
    return char_name.replace("-", "_").replace(".", "_")


class HomeKitEntity(Entity):
    """Representation of a Home Assistant HomeKit device."""

    def __init__(self, accessory, devinfo):
        """Initialise a generic HomeKit device."""
        self._accessory = accessory
        self._aid = devinfo["aid"]
        self._iid = devinfo["iid"]
        self._features = 0
        self._chars = {}
        self.setup()

        self._signals = []

    async def async_added_to_hass(self):
        """Entity added to hass."""
        self._signals.append(
            self.hass.helpers.dispatcher.async_dispatcher_connect(
                self._accessory.signal_state_updated, self.async_state_changed
            )
        )

        self._accessory.add_pollable_characteristics(self.pollable_characteristics)

    async def async_will_remove_from_hass(self):
        """Prepare to be removed from hass."""
        self._accessory.remove_pollable_characteristics(self._aid)

        for signal_remove in self._signals:
            signal_remove()
        self._signals.clear()

    @property
    def should_poll(self) -> bool:
        """Return False.

        Data update is triggered from HKDevice.
        """
        return False

    def setup(self):
        """Configure an entity baed on its HomeKit characterstics metadata."""
        # pylint: disable=import-error
        from homekit.model.characteristics import CharacteristicsTypes

        accessories = self._accessory.accessories

        get_uuid = CharacteristicsTypes.get_uuid
        characteristic_types = [get_uuid(c) for c in self.get_characteristic_types()]

        self.pollable_characteristics = []
        self._chars = {}
        self._char_names = {}

        for accessory in accessories:
            if accessory["aid"] != self._aid:
                continue
            self._accessory_info = get_accessory_information(accessory)
            for service in accessory["services"]:
                if service["iid"] != self._iid:
                    continue
                for char in service["characteristics"]:
                    try:
                        uuid = CharacteristicsTypes.get_uuid(char["type"])
                    except KeyError:
                        # If a KeyError is raised its a non-standard
                        # characteristic. We must ignore it in this case.
                        continue
                    if uuid not in characteristic_types:
                        continue
                    self._setup_characteristic(char)

    def _setup_characteristic(self, char):
        """Configure an entity based on a HomeKit characteristics metadata."""
        # pylint: disable=import-error
        from homekit.model.characteristics import CharacteristicsTypes

        # Build up a list of (aid, iid) tuples to poll on update()
        self.pollable_characteristics.append((self._aid, char["iid"]))

        # Build a map of ctype -> iid
        short_name = CharacteristicsTypes.get_short(char["type"])
        self._chars[short_name] = char["iid"]
        self._char_names[char["iid"]] = short_name

        # Callback to allow entity to configure itself based on this
        # characteristics metadata (valid values, value ranges, features, etc)
        setup_fn_name = escape_characteristic_name(short_name)
        setup_fn = getattr(self, "_setup_{}".format(setup_fn_name), None)
        if not setup_fn:
            return
        # pylint: disable=not-callable
        setup_fn(char)

    @callback
    def async_state_changed(self):
        """Collect new data from bridge and update the entity state in hass."""
        accessory_state = self._accessory.current_state.get(self._aid, {})
        for iid, result in accessory_state.items():
            # No value so dont process this result
            if "value" not in result:
                continue

            # Unknown iid - this is probably for a sibling service that is part
            # of the same physical accessory. Ignore it.
            if iid not in self._char_names:
                continue

            # Callback to update the entity with this characteristic value
            char_name = escape_characteristic_name(self._char_names[iid])
            update_fn = getattr(self, "_update_{}".format(char_name), None)
            if not update_fn:
                continue

            # pylint: disable=not-callable
            update_fn(result["value"])

        self.async_write_ha_state()

    @property
    def unique_id(self):
        """Return the ID of this device."""
        serial = self._accessory_info["serial-number"]
        return "homekit-{}-{}".format(serial, self._iid)

    @property
    def name(self):
        """Return the name of the device if any."""
        return self._accessory_info.get("name")

    @property
    def available(self) -> bool:
        """Return True if entity is available."""
        return self._accessory.available

    @property
    def device_info(self):
        """Return the device info."""
        accessory_serial = self._accessory_info["serial-number"]

        device_info = {
            "identifiers": {(DOMAIN, "serial-number", accessory_serial)},
            "name": self._accessory_info["name"],
            "manufacturer": self._accessory_info.get("manufacturer", ""),
            "model": self._accessory_info.get("model", ""),
            "sw_version": self._accessory_info.get("firmware.revision", ""),
        }

        # Some devices only have a single accessory - we don't add a
        # via_device otherwise it would be self referential.
        bridge_serial = self._accessory.connection_info["serial-number"]
        if accessory_serial != bridge_serial:
            device_info["via_device"] = (DOMAIN, "serial-number", bridge_serial)

        return device_info

    def get_characteristic_types(self):
        """Define the homekit characteristics the entity cares about."""
        raise NotImplementedError


async def async_setup_entry(hass, entry):
    """Set up a HomeKit connection on a config entry."""
    conn = HKDevice(hass, entry, entry.data)
    hass.data[KNOWN_DEVICES][conn.unique_id] = conn

    if not await conn.async_setup():
        del hass.data[KNOWN_DEVICES][conn.unique_id]
        raise ConfigEntryNotReady

    conn_info = conn.connection_info

    device_registry = await dr.async_get_registry(hass)
    device_registry.async_get_or_create(
        config_entry_id=entry.entry_id,
        identifiers={
            (DOMAIN, "serial-number", conn_info["serial-number"]),
            (DOMAIN, "accessory-id", conn.unique_id),
        },
        name=conn.name,
        manufacturer=conn_info.get("manufacturer"),
        model=conn_info.get("model"),
        sw_version=conn_info.get("firmware.revision"),
    )

    return True


async def async_setup(hass, config):
    """Set up for Homekit devices."""
    # pylint: disable=import-error
    import homekit

    map_storage = hass.data[ENTITY_MAP] = EntityMapStorage(hass)
    await map_storage.async_initialize()

    hass.data[CONTROLLER] = homekit.Controller()
    hass.data[KNOWN_DEVICES] = {}

    return True


async def async_unload_entry(hass, entry):
    """Disconnect from HomeKit devices before unloading entry."""
    hkid = entry.data["AccessoryPairingID"]

    if hkid in hass.data[KNOWN_DEVICES]:
        connection = hass.data[KNOWN_DEVICES][hkid]
        await connection.async_unload()

    return True


async def async_remove_entry(hass, entry):
    """Cleanup caches before removing config entry."""
    hkid = entry.data["AccessoryPairingID"]
    hass.data[ENTITY_MAP].async_delete_map(hkid)