core/homeassistant/components/bond/utils.py

"""Reusable utilities for the Bond component."""
import asyncio
import logging
from typing import List, Optional

from aiohttp import ClientResponseError
from bond_api import Action, Bond

from .const import BRIDGE_MAKE

_LOGGER = logging.getLogger(__name__)


class BondDevice:
    """Helper device class to hold ID and attributes together."""

    def __init__(self, device_id: str, attrs: dict, props: dict):
        """Create a helper device from ID and attributes returned by API."""
        self.device_id = device_id
        self.props = props
        self._attrs = attrs

    def __repr__(self):
        """Return readable representation of a bond device."""
        return {
            "device_id": self.device_id,
            "props": self.props,
            "attrs": self._attrs,
        }.__repr__()

    @property
    def name(self) -> str:
        """Get the name of this device."""
        return self._attrs["name"]

    @property
    def type(self) -> str:
        """Get the type of this device."""
        return self._attrs["type"]

    @property
    def location(self) -> str:
        """Get the location of this device."""
        return self._attrs.get("location")

    @property
    def template(self) -> str:
        """Return this model template."""
        return self._attrs.get("template")

    @property
    def branding_profile(self) -> str:
        """Return this branding profile."""
        return self.props.get("branding_profile")

    @property
    def trust_state(self) -> bool:
        """Check if Trust State is turned on."""
        return self.props.get("trust_state", False)

    def supports_speed(self) -> bool:
        """Return True if this device supports any of the speed related commands."""
        actions: List[str] = self._attrs["actions"]
        return bool([action for action in actions if action in [Action.SET_SPEED]])

    def supports_direction(self) -> bool:
        """Return True if this device supports any of the direction related commands."""
        actions: List[str] = self._attrs["actions"]
        return bool([action for action in actions if action in [Action.SET_DIRECTION]])

    def supports_light(self) -> bool:
        """Return True if this device supports any of the light related commands."""
        actions: List[str] = self._attrs["actions"]
        return bool(
            [
                action
                for action in actions
                if action in [Action.TURN_LIGHT_ON, Action.TURN_LIGHT_OFF]
            ]
        )

    def supports_set_brightness(self) -> bool:
        """Return True if this device supports setting a light brightness."""
        actions: List[str] = self._attrs["actions"]
        return bool([action for action in actions if action in [Action.SET_BRIGHTNESS]])


class BondHub:
    """Hub device representing Bond Bridge."""

    def __init__(self, bond: Bond):
        """Initialize Bond Hub."""
        self.bond: Bond = bond
        self._bridge: Optional[dict] = None
        self._version: Optional[dict] = None
        self._devices: Optional[List[BondDevice]] = None

    async def setup(self, max_devices=None):
        """Read hub version information."""
        self._version = await self.bond.version()
        _LOGGER.debug("Bond reported the following version info: %s", self._version)
        # Fetch all available devices using Bond API.
        device_ids = await self.bond.devices()
        self._devices = []
        for idx, device_id in enumerate(device_ids):
            if max_devices is not None and idx >= max_devices:
                break

            device, props = await asyncio.gather(
                self.bond.device(device_id), self.bond.device_properties(device_id)
            )

            self._devices.append(BondDevice(device_id, device, props))

        _LOGGER.debug("Discovered Bond devices: %s", self._devices)
        try:
            # Smart by bond devices do not have a bridge api call
            self._bridge = await self.bond.bridge()
        except ClientResponseError:
            self._bridge = {}
        _LOGGER.debug("Bond reported the following bridge info: %s", self._bridge)

    @property
    def bond_id(self) -> Optional[str]:
        """Return unique Bond ID for this hub."""
        # Old firmwares are missing the bondid
        return self._version.get("bondid")

    @property
    def target(self) -> str:
        """Return this hub target."""
        return self._version.get("target")

    @property
    def model(self) -> str:
        """Return this hub model."""
        return self._version.get("model")

    @property
    def make(self) -> str:
        """Return this hub make."""
        return self._version.get("make", BRIDGE_MAKE)

    @property
    def name(self) -> Optional[str]:
        """Get the name of this bridge."""
        if not self.is_bridge and self._devices:
            return self._devices[0].name
        return self._bridge.get("name")

    @property
    def location(self) -> Optional[str]:
        """Get the location of this bridge."""
        if not self.is_bridge and self._devices:
            return self._devices[0].location
        return self._bridge.get("location")

    @property
    def fw_ver(self) -> str:
        """Return this hub firmware version."""
        return self._version.get("fw_ver")

    @property
    def devices(self) -> List[BondDevice]:
        """Return a list of all devices controlled by this hub."""
        return self._devices

    @property
    def is_bridge(self) -> bool:
        """Return if the Bond is a Bond Bridge."""
        return bool(self._bridge)