197 lines
6.4 KiB
Python
197 lines
6.4 KiB
Python
"""Reusable utilities for the Bond component."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any, cast
|
|
|
|
from aiohttp import ClientResponseError
|
|
from bond_api import Action, Bond
|
|
|
|
from homeassistant.util.async_ import gather_with_concurrency
|
|
|
|
from .const import BRIDGE_MAKE
|
|
|
|
MAX_REQUESTS = 6
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
class BondDevice:
|
|
"""Helper device class to hold ID and attributes together."""
|
|
|
|
def __init__(
|
|
self, device_id: str, attrs: dict[str, Any], props: dict[str, Any]
|
|
) -> None:
|
|
"""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) -> str:
|
|
"""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 cast(str, self._attrs["name"])
|
|
|
|
@property
|
|
def type(self) -> str:
|
|
"""Get the type of this device."""
|
|
return cast(str, self._attrs["type"])
|
|
|
|
@property
|
|
def location(self) -> str | None:
|
|
"""Get the location of this device."""
|
|
return self._attrs.get("location")
|
|
|
|
@property
|
|
def template(self) -> str | None:
|
|
"""Return this model template."""
|
|
return self._attrs.get("template")
|
|
|
|
@property
|
|
def branding_profile(self) -> str | None:
|
|
"""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 _has_any_action(self, actions: set[str]) -> bool:
|
|
"""Check to see if the device supports any of the actions."""
|
|
supported_actions: list[str] = self._attrs["actions"]
|
|
for action in supported_actions:
|
|
if action in actions:
|
|
return True
|
|
return False
|
|
|
|
def supports_speed(self) -> bool:
|
|
"""Return True if this device supports any of the speed related commands."""
|
|
return self._has_any_action({Action.SET_SPEED})
|
|
|
|
def supports_direction(self) -> bool:
|
|
"""Return True if this device supports any of the direction related commands."""
|
|
return self._has_any_action({Action.SET_DIRECTION})
|
|
|
|
def supports_light(self) -> bool:
|
|
"""Return True if this device supports any of the light related commands."""
|
|
return self._has_any_action({Action.TURN_LIGHT_ON, Action.TURN_LIGHT_OFF})
|
|
|
|
def supports_up_light(self) -> bool:
|
|
"""Return true if the device has an up light."""
|
|
return self._has_any_action({Action.TURN_UP_LIGHT_ON, Action.TURN_UP_LIGHT_OFF})
|
|
|
|
def supports_down_light(self) -> bool:
|
|
"""Return true if the device has a down light."""
|
|
return self._has_any_action(
|
|
{Action.TURN_DOWN_LIGHT_ON, Action.TURN_DOWN_LIGHT_OFF}
|
|
)
|
|
|
|
def supports_set_brightness(self) -> bool:
|
|
"""Return True if this device supports setting a light brightness."""
|
|
return self._has_any_action({Action.SET_BRIGHTNESS})
|
|
|
|
|
|
class BondHub:
|
|
"""Hub device representing Bond Bridge."""
|
|
|
|
def __init__(self, bond: Bond) -> None:
|
|
"""Initialize Bond Hub."""
|
|
self.bond: Bond = bond
|
|
self._bridge: dict[str, Any] = {}
|
|
self._version: dict[str, Any] = {}
|
|
self._devices: list[BondDevice] = []
|
|
|
|
async def setup(self, max_devices: int | None = None) -> 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 = []
|
|
setup_device_ids = []
|
|
tasks = []
|
|
for idx, device_id in enumerate(device_ids):
|
|
if max_devices is not None and idx >= max_devices:
|
|
break
|
|
setup_device_ids.append(device_id)
|
|
tasks.extend(
|
|
[self.bond.device(device_id), self.bond.device_properties(device_id)]
|
|
)
|
|
|
|
responses = await gather_with_concurrency(MAX_REQUESTS, *tasks)
|
|
response_idx = 0
|
|
for device_id in setup_device_ids:
|
|
self._devices.append(
|
|
BondDevice(
|
|
device_id, responses[response_idx], responses[response_idx + 1]
|
|
)
|
|
)
|
|
response_idx += 2
|
|
|
|
_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) -> str | None:
|
|
"""Return unique Bond ID for this hub."""
|
|
# Old firmwares are missing the bondid
|
|
return self._version.get("bondid")
|
|
|
|
@property
|
|
def target(self) -> str | None:
|
|
"""Return this hub target."""
|
|
return self._version.get("target")
|
|
|
|
@property
|
|
def model(self) -> str | None:
|
|
"""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) -> str:
|
|
"""Get the name of this bridge."""
|
|
if not self.is_bridge and self._devices:
|
|
return self._devices[0].name
|
|
return cast(str, self._bridge["name"])
|
|
|
|
@property
|
|
def location(self) -> str | None:
|
|
"""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 | None:
|
|
"""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)
|