2a3c94bad0
* split entity into base and entity * add initial light group support * add dispatching of groups to light * added zha group object * add group event listener * add and remove group members * get group by name * fix rebase * fix rebase * use group_id for unique_id * get entities from registry * use group name * update entity domain * update zha storage to handle groups * dispatch group entities * update light group * fix group remove and dispatch light group entities * allow picking the domain for group entities * beginning - auto determine entity domain * move methods to helpers so they can be shared * fix rebase * remove double init groups... again * cleanup startup * use asyncio create task * group entity discovery * add logging and fix group name * add logging and update group after probe if needed * test add group via gateway * add method to get group entity ids * update storage * test get group by name * update storage on remove * test group with single member * add light group tests * test some light group logic * type hints * fix tests and cleanup * revert init changes except for create task * remove group entity domain changing for now * add missing import * tricky code saving * review comments * clean up class defs * cleanup * fix rebase because I cant read * make pylint happy
144 lines
4.9 KiB
Python
144 lines
4.9 KiB
Python
"""Group for Zigbee Home Automation."""
|
|
import asyncio
|
|
import logging
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from zigpy.types.named import EUI64
|
|
|
|
from homeassistant.core import callback
|
|
from homeassistant.helpers.entity_registry import async_entries_for_device
|
|
from homeassistant.helpers.typing import HomeAssistantType
|
|
|
|
from .helpers import LogMixin
|
|
from .typing import ZhaDeviceType, ZhaGatewayType, ZigpyEndpointType, ZigpyGroupType
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
class ZHAGroup(LogMixin):
|
|
"""ZHA Zigbee group object."""
|
|
|
|
def __init__(
|
|
self,
|
|
hass: HomeAssistantType,
|
|
zha_gateway: ZhaGatewayType,
|
|
zigpy_group: ZigpyGroupType,
|
|
):
|
|
"""Initialize the group."""
|
|
self.hass: HomeAssistantType = hass
|
|
self._zigpy_group: ZigpyGroupType = zigpy_group
|
|
self._zha_gateway: ZhaGatewayType = zha_gateway
|
|
self._entity_domain: str = None
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
"""Return group name."""
|
|
return self._zigpy_group.name
|
|
|
|
@property
|
|
def group_id(self) -> int:
|
|
"""Return group name."""
|
|
return self._zigpy_group.group_id
|
|
|
|
@property
|
|
def endpoint(self) -> ZigpyEndpointType:
|
|
"""Return the endpoint for this group."""
|
|
return self._zigpy_group.endpoint
|
|
|
|
@property
|
|
def entity_domain(self) -> Optional[str]:
|
|
"""Return the domain that will be used for the entity representing this group."""
|
|
return self._entity_domain
|
|
|
|
@entity_domain.setter
|
|
def entity_domain(self, domain: Optional[str]) -> None:
|
|
"""Set the domain that will be used for the entity representing this group."""
|
|
self._entity_domain = domain
|
|
|
|
@property
|
|
def members(self) -> List[ZhaDeviceType]:
|
|
"""Return the ZHA devices that are members of this group."""
|
|
return [
|
|
self._zha_gateway.devices.get(member_ieee[0])
|
|
for member_ieee in self._zigpy_group.members.keys()
|
|
if member_ieee[0] in self._zha_gateway.devices
|
|
]
|
|
|
|
async def async_add_members(self, member_ieee_addresses: List[EUI64]) -> None:
|
|
"""Add members to this group."""
|
|
if len(member_ieee_addresses) > 1:
|
|
tasks = []
|
|
for ieee in member_ieee_addresses:
|
|
tasks.append(
|
|
self._zha_gateway.devices[ieee].async_add_to_group(self.group_id)
|
|
)
|
|
await asyncio.gather(*tasks)
|
|
else:
|
|
await self._zha_gateway.devices[
|
|
member_ieee_addresses[0]
|
|
].async_add_to_group(self.group_id)
|
|
|
|
async def async_remove_members(self, member_ieee_addresses: List[EUI64]) -> None:
|
|
"""Remove members from this group."""
|
|
if len(member_ieee_addresses) > 1:
|
|
tasks = []
|
|
for ieee in member_ieee_addresses:
|
|
tasks.append(
|
|
self._zha_gateway.devices[ieee].async_remove_from_group(
|
|
self.group_id
|
|
)
|
|
)
|
|
await asyncio.gather(*tasks)
|
|
else:
|
|
await self._zha_gateway.devices[
|
|
member_ieee_addresses[0]
|
|
].async_remove_from_group(self.group_id)
|
|
|
|
@property
|
|
def member_entity_ids(self) -> List[str]:
|
|
"""Return the ZHA entity ids for all entities for the members of this group."""
|
|
all_entity_ids: List[str] = []
|
|
for device in self.members:
|
|
entities = async_entries_for_device(
|
|
self._zha_gateway.ha_entity_registry, device.device_id
|
|
)
|
|
for entity in entities:
|
|
all_entity_ids.append(entity.entity_id)
|
|
return all_entity_ids
|
|
|
|
@property
|
|
def domain_entity_ids(self) -> List[str]:
|
|
"""Return entity ids from the entity domain for this group."""
|
|
if self.entity_domain is None:
|
|
return
|
|
domain_entity_ids: List[str] = []
|
|
for device in self.members:
|
|
entities = async_entries_for_device(
|
|
self._zha_gateway.ha_entity_registry, device.device_id
|
|
)
|
|
domain_entity_ids.extend(
|
|
[
|
|
entity.entity_id
|
|
for entity in entities
|
|
if entity.domain == self.entity_domain
|
|
]
|
|
)
|
|
return domain_entity_ids
|
|
|
|
@callback
|
|
def async_get_info(self) -> Dict[str, Any]:
|
|
"""Get ZHA group info."""
|
|
group_info: Dict[str, Any] = {}
|
|
group_info["group_id"] = self.group_id
|
|
group_info["entity_domain"] = self.entity_domain
|
|
group_info["name"] = self.name
|
|
group_info["members"] = [
|
|
zha_device.async_get_info() for zha_device in self.members
|
|
]
|
|
return group_info
|
|
|
|
def log(self, level: int, msg: str, *args):
|
|
"""Log a message."""
|
|
msg = f"[%s](%s): {msg}"
|
|
args = (self.name, self.group_id) + args
|
|
_LOGGER.log(level, msg, *args)
|