diff --git a/homeassistant/components/zha/__init__.py b/homeassistant/components/zha/__init__.py index f4f17f960e1c..3763793727e2 100644 --- a/homeassistant/components/zha/__init__.py +++ b/homeassistant/components/zha/__init__.py @@ -99,6 +99,7 @@ PLATFORMS = ( Platform.DEVICE_TRACKER, Platform.EVENT, Platform.FAN, + Platform.INFRARED, Platform.LIGHT, Platform.LOCK, Platform.NUMBER, diff --git a/homeassistant/components/zha/helpers.py b/homeassistant/components/zha/helpers.py index fd43351a0c63..720c0e0f9905 100644 --- a/homeassistant/components/zha/helpers.py +++ b/homeassistant/components/zha/helpers.py @@ -1292,7 +1292,7 @@ def async_cluster_exists(hass: HomeAssistant, cluster_id, skip_coordinator=True) @callback def async_add_entities( _async_add_entities: AddEntitiesCallback, - entity_class: type[ZHAEntity], + entity_class: Callable[[EntityData], ZHAEntity], entities: list[EntityData], **kwargs, ) -> None: diff --git a/homeassistant/components/zha/infrared.py b/homeassistant/components/zha/infrared.py new file mode 100644 index 000000000000..26a06e3beee4 --- /dev/null +++ b/homeassistant/components/zha/infrared.py @@ -0,0 +1,107 @@ +"""Support for ZHA infrared emitters and receivers.""" + +import functools +from typing import override + +from zha.application.platforms.infrared import ( + BaseInfraredEmitter, + BaseInfraredReceiver, + EntityInfraredSignalReceivedEvent, + InfraredSignal, +) + +from homeassistant.components.infrared import ( + InfraredCommand, + InfraredEmitterEntity, + InfraredReceivedSignal, + InfraredReceiverEntity, +) +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import Platform +from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers.dispatcher import async_dispatcher_connect +from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback + +from .entity import ZHAEntity +from .helpers import ( + SIGNAL_ADD_ENTITIES, + EntityData, + async_add_entities as zha_async_add_entities, + convert_zha_error_to_ha_error, + get_zha_data, +) + + +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddConfigEntryEntitiesCallback, +) -> None: + """Set up the Zigbee Home Automation infrared from config entry.""" + zha_data = get_zha_data(hass) + entities_to_create = zha_data.platforms[Platform.INFRARED] + + unsub = async_dispatcher_connect( + hass, + SIGNAL_ADD_ENTITIES, + functools.partial( + zha_async_add_entities, + async_add_entities, + _make_infrared_entity, + entities_to_create, + ), + ) + config_entry.async_on_unload(unsub) + + +def _make_infrared_entity(entity_data: EntityData) -> ZHAEntity: + """Create the HA entity matching the ZHA infrared entity type.""" + if isinstance(entity_data.entity, BaseInfraredEmitter): + return ZHAInfraredEmitter(entity_data) + + if isinstance(entity_data.entity, BaseInfraredReceiver): + return ZHAInfraredReceiver(entity_data) + + raise TypeError(f"Unknown infrared entity: {entity_data.entity!r}") + + +class ZHAInfraredEmitter(ZHAEntity, InfraredEmitterEntity): + """Representation of a ZHA infrared emitter.""" + + @convert_zha_error_to_ha_error() + @override + async def async_send_command(self, command: InfraredCommand) -> None: + """Send an IR command.""" + await self.entity_data.entity.async_send_command( + InfraredSignal( + timings=command.get_raw_timings(), + modulation=command.modulation, + ) + ) + + +class ZHAInfraredReceiver(ZHAEntity, InfraredReceiverEntity): + """Representation of a ZHA infrared receiver.""" + + @override + async def async_added_to_hass(self) -> None: + """Subscribe to signals captured by the ZHA entity.""" + await super().async_added_to_hass() + self._unsubs.append( + self.entity_data.entity.on_event( + EntityInfraredSignalReceivedEvent.event, + self._handle_zha_received_signal, + ) + ) + + @callback + def _handle_zha_received_signal( + self, event: EntityInfraredSignalReceivedEvent + ) -> None: + """Handle a signal captured by the ZHA entity.""" + self._handle_received_signal( + InfraredReceivedSignal( + timings=event.signal.timings, + modulation=event.signal.modulation, + ) + ) diff --git a/tests/components/zha/test_infrared.py b/tests/components/zha/test_infrared.py new file mode 100644 index 000000000000..f2f26205a66c --- /dev/null +++ b/tests/components/zha/test_infrared.py @@ -0,0 +1,275 @@ +"""Test ZHA infrared.""" + +from collections.abc import Callable, Coroutine, Generator +from typing import Any +from unittest.mock import patch + +from infrared_protocols.commands.nec import NECCommand +import pytest +from zha.application.platforms import ENTITY_REGISTRY, ClusterMatch +from zha.application.platforms.infrared import ( + BaseInfraredEmitter, + BaseInfraredReceiver, + InfraredSignal, +) +from zha.exceptions import ZHAException +from zigpy.const import SIG_EP_INPUT, SIG_EP_OUTPUT, SIG_EP_PROFILE, SIG_EP_TYPE +from zigpy.device import ZigbeeDevice +from zigpy.profiles import zha +from zigpy.zcl.clusters import general + +from homeassistant.components import infrared +from homeassistant.components.infrared import ( + InfraredDeviceClass, + InfraredReceivedSignal, +) +from homeassistant.components.zha.const import DOMAIN +from homeassistant.components.zha.helpers import ( + ZHADeviceProxy, + ZHAGatewayProxy, + get_zha_gateway, + get_zha_gateway_proxy, +) +from homeassistant.const import ATTR_DEVICE_CLASS, STATE_UNKNOWN, Platform +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers import entity_registry as er + + +class FakeEmitter(BaseInfraredEmitter): + """Emitter that records what it was asked to transmit.""" + + _unique_id_suffix = "fake_emitter" + _cluster_match = ClusterMatch( + client_clusters=frozenset({general.OnOff.cluster_id}), + ) + + def __init__(self, *args: Any, **kwargs: Any) -> None: + """Initialize the fake emitter.""" + super().__init__(*args, **kwargs) + self.sent: list[InfraredSignal] = [] + + async def async_send_command(self, signal: InfraredSignal) -> None: + """Record the signal instead of transmitting it.""" + self.sent.append(signal) + + +class FakeFailingEmitter(FakeEmitter): + """Emitter whose transmissions fail.""" + + async def async_send_command(self, signal: InfraredSignal) -> None: + """Fail to transmit the signal.""" + raise ZHAException("Failed to send IR command") + + +class FakeReceiver(BaseInfraredReceiver): + """Receiver that captures signals on demand.""" + + _unique_id_suffix = "fake_receiver" + _cluster_match = ClusterMatch( + client_clusters=frozenset({general.OnOff.cluster_id}), + ) + + def receive(self, signal: InfraredSignal) -> None: + """Capture a signal, as a concrete subclass would.""" + self._handle_received_signal(signal) + + +@pytest.fixture(autouse=True) +def infrared_platform_only() -> Generator[None]: + """Only set up the infrared and required base platforms to speed up tests.""" + with patch( + "homeassistant.components.zha.PLATFORMS", + (Platform.INFRARED, Platform.SENSOR), + ): + yield + + +@pytest.fixture +def speed_up_radio_mgr() -> Generator[None]: + """Remove radio manager delays, which never elapse with time frozen.""" + with ( + patch("homeassistant.components.zha.radio_manager.CONNECT_DELAY_S", 0), + patch("zha.application.gateway.SHUT_DOWN_DELAY_S", 0), + ): + yield + + +@pytest.fixture +def emitter_class() -> type[FakeEmitter]: + """Return the fake emitter entity class to discover.""" + return FakeEmitter + + +@pytest.fixture(autouse=True) +def register_fake_infrared(emitter_class: type[FakeEmitter]) -> Generator[None]: + """Make zha discover the fake infrared entities on the OnOff client cluster.""" + with patch.dict( + ENTITY_REGISTRY, + { + general.OnOff.cluster_id: [ + *ENTITY_REGISTRY[general.OnOff.cluster_id], + emitter_class, + FakeReceiver, + ] + }, + ): + yield + + +async def _setup_device( + hass: HomeAssistant, + setup_zha: Callable[..., Coroutine[None]], + zigpy_device_mock: Callable[..., ZigbeeDevice], +) -> ZHADeviceProxy: + """Join a remote with an OnOff client cluster.""" + await setup_zha() + + gateway = get_zha_gateway(hass) + gateway_proxy: ZHAGatewayProxy = get_zha_gateway_proxy(hass) + + zigpy_device = zigpy_device_mock( + { + 1: { + SIG_EP_PROFILE: zha.PROFILE_ID, + SIG_EP_TYPE: zha.DeviceType.REMOTE_CONTROL, + SIG_EP_INPUT: [general.Basic.cluster_id], + SIG_EP_OUTPUT: [general.OnOff.cluster_id], + } + }, + ) + + gateway.get_or_create_device(zigpy_device) + await gateway.async_device_initialized(zigpy_device) + await hass.async_block_till_done(wait_background_tasks=True) + + return gateway_proxy.get_device_proxy(zigpy_device.ieee) + + +def _get_entity[T: FakeEmitter | FakeReceiver]( + hass: HomeAssistant, zha_device_proxy: ZHADeviceProxy, entity_class: type[T] +) -> tuple[str, T]: + """Return the HA entity ID and the zha entity of the given class.""" + zha_entity = next( + entity + for entity in zha_device_proxy.device.platform_entities.values() + if isinstance(entity, entity_class) + ) + entity_id = er.async_get(hass).async_get_entity_id( + Platform.INFRARED, DOMAIN, zha_entity.unique_id + ) + assert entity_id is not None + + return entity_id, zha_entity + + +async def test_infrared_entities( + hass: HomeAssistant, + setup_zha: Callable[..., Coroutine[None]], + zigpy_device_mock: Callable[..., ZigbeeDevice], +) -> None: + """Test ZHA infrared emitters and receivers are created.""" + zha_device_proxy = await _setup_device(hass, setup_zha, zigpy_device_mock) + emitter_id, _ = _get_entity(hass, zha_device_proxy, FakeEmitter) + receiver_id, _ = _get_entity(hass, zha_device_proxy, FakeReceiver) + + assert infrared.async_get_emitters(hass) == [emitter_id] + assert infrared.async_get_receivers(hass) == [receiver_id] + + emitter_state = hass.states.get(emitter_id) + assert emitter_state + assert emitter_state.state == STATE_UNKNOWN + assert emitter_state.attributes[ATTR_DEVICE_CLASS] == InfraredDeviceClass.EMITTER + + receiver_state = hass.states.get(receiver_id) + assert receiver_state + assert receiver_state.state == STATE_UNKNOWN + assert receiver_state.attributes[ATTR_DEVICE_CLASS] == InfraredDeviceClass.RECEIVER + + +@pytest.mark.freeze_time("2026-09-29 12:00:00+00:00") +async def test_infrared_send_command( + hass: HomeAssistant, + setup_zha: Callable[..., Coroutine[None]], + zigpy_device_mock: Callable[..., ZigbeeDevice], +) -> None: + """Test IR commands are sent through the zha emitter as raw timings.""" + zha_device_proxy = await _setup_device(hass, setup_zha, zigpy_device_mock) + entity_id, zha_entity = _get_entity(hass, zha_device_proxy, FakeEmitter) + + command = NECCommand(address=0x04, command=0x08, modulation=38000) + await infrared.async_send_command(hass, entity_id, command) + + assert zha_entity.sent == [ + InfraredSignal(timings=command.get_raw_timings(), modulation=38000) + ] + assert hass.states.get(entity_id).state == "2026-09-29T12:00:00.000+00:00" + + +@pytest.mark.parametrize("emitter_class", [FakeFailingEmitter]) +async def test_infrared_send_command_failure( + hass: HomeAssistant, + setup_zha: Callable[..., Coroutine[None]], + zigpy_device_mock: Callable[..., ZigbeeDevice], +) -> None: + """Test ZHA errors are raised as Home Assistant errors.""" + zha_device_proxy = await _setup_device(hass, setup_zha, zigpy_device_mock) + entity_id, _ = _get_entity(hass, zha_device_proxy, FakeFailingEmitter) + + command = NECCommand(address=0x04, command=0x08, modulation=38000) + with pytest.raises(HomeAssistantError, match="Failed to send IR command"): + await infrared.async_send_command(hass, entity_id, command) + + assert hass.states.get(entity_id).state == STATE_UNKNOWN + + +@pytest.mark.freeze_time("2026-09-29 12:00:00+00:00") +async def test_infrared_received_signal( + hass: HomeAssistant, + setup_zha: Callable[..., Coroutine[None]], + zigpy_device_mock: Callable[..., ZigbeeDevice], +) -> None: + """Test signals captured by the zha receiver are dispatched to subscribers.""" + zha_device_proxy = await _setup_device(hass, setup_zha, zigpy_device_mock) + entity_id, zha_entity = _get_entity(hass, zha_device_proxy, FakeReceiver) + + received_signals: list[InfraredReceivedSignal] = [] + unsub = infrared.async_subscribe_receiver(hass, entity_id, received_signals.append) + + zha_entity.receive(InfraredSignal(timings=[9000, -4500], modulation=38000)) + zha_entity.receive(InfraredSignal(timings=[560, -1690])) + await hass.async_block_till_done() + + assert received_signals == [ + InfraredReceivedSignal(timings=[9000, -4500], modulation=38000), + InfraredReceivedSignal(timings=[560, -1690]), + ] + assert hass.states.get(entity_id).state == "2026-09-29T12:00:00.000+00:00" + + unsub() + zha_entity.receive(InfraredSignal(timings=[9000, -4500])) + await hass.async_block_till_done() + assert len(received_signals) == 2 + + +async def test_infrared_receiver_removed( + hass: HomeAssistant, + entity_registry: er.EntityRegistry, + setup_zha: Callable[..., Coroutine[None]], + zigpy_device_mock: Callable[..., ZigbeeDevice], +) -> None: + """Test removing the receiver stops listening to the zha entity.""" + zha_device_proxy = await _setup_device(hass, setup_zha, zigpy_device_mock) + entity_id, zha_entity = _get_entity(hass, zha_device_proxy, FakeReceiver) + + received_signals: list[InfraredReceivedSignal] = [] + infrared.async_subscribe_receiver(hass, entity_id, received_signals.append) + + entity_registry.async_remove(entity_id) + await hass.async_block_till_done() + assert hass.states.get(entity_id) is None + + # Subscribers live on the removed entity, so they only fire if the listener leaked + zha_entity.receive(InfraredSignal(timings=[9000, -4500])) + await hass.async_block_till_done() + assert received_signals == []