diff --git a/homeassistant/components/broadlink/const.py b/homeassistant/components/broadlink/const.py index e5865168fdf30..d866e701cce3f 100644 --- a/homeassistant/components/broadlink/const.py +++ b/homeassistant/components/broadlink/const.py @@ -6,6 +6,7 @@ DOMAINS_AND_TYPES = { Platform.CLIMATE: {"HYS"}, + Platform.INFRARED: {"RM4MINI", "RM4PRO", "RMMINI", "RMMINIB", "RMPRO"}, Platform.LIGHT: {"LB1", "LB2"}, Platform.RADIO_FREQUENCY: {"RM4PRO", "RMPRO"}, Platform.REMOTE: {"RM4MINI", "RM4PRO", "RMMINI", "RMMINIB", "RMPRO"}, diff --git a/homeassistant/components/broadlink/infrared.py b/homeassistant/components/broadlink/infrared.py new file mode 100644 index 0000000000000..29238f08e1bd6 --- /dev/null +++ b/homeassistant/components/broadlink/infrared.py @@ -0,0 +1,69 @@ +"""Infrared platform for Broadlink remotes.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from broadlink.exceptions import BroadlinkException +from broadlink.remote import pulses_to_data as _bl_pulses_to_data + +from homeassistant.components.infrared import InfraredCommand, InfraredEntity +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback + +from .const import DOMAIN +from .entity import BroadlinkEntity + +if TYPE_CHECKING: + from .device import BroadlinkDevice + +PARALLEL_UPDATES = 1 + + +def _timings_to_broadlink_packet(timings: list[int]) -> bytes: + """Convert signed microsecond timings to a Broadlink IR packet. + + Positive values are pulse (high) durations; negative values are space + (low) durations. The Broadlink library's encoder expects absolute + durations. + """ + pulses = [abs(t) for t in timings] + return _bl_pulses_to_data(pulses) + + +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddConfigEntryEntitiesCallback, +) -> None: + """Set up Broadlink infrared entity.""" + # Uses legacy hass.data[DOMAIN] pattern + # pylint: disable-next=hass-use-runtime-data + device = hass.data[DOMAIN].devices[config_entry.entry_id] + async_add_entities([BroadlinkInfraredEntity(device)]) + + +class BroadlinkInfraredEntity(BroadlinkEntity, InfraredEntity): + """Broadlink infrared transmitter entity.""" + + _attr_has_entity_name = True + _attr_translation_key = "infrared_emitter" + + def __init__(self, device: BroadlinkDevice) -> None: + """Initialize the entity.""" + super().__init__(device) + self._attr_unique_id = f"{device.unique_id}-emitter" + + async def async_send_command(self, command: InfraredCommand) -> None: + """Send an IR command via the Broadlink device.""" + packet = _timings_to_broadlink_packet(command.get_raw_timings()) + try: + await self._device.async_request(self._device.api.send_data, packet) + except (BroadlinkException, OSError) as err: + raise HomeAssistantError( + translation_domain=DOMAIN, + translation_key="send_command_failed", + translation_placeholders={"error": str(err)}, + ) from err diff --git a/homeassistant/components/broadlink/strings.json b/homeassistant/components/broadlink/strings.json index d5d91be6b0e78..c291e7a77a0c6 100644 --- a/homeassistant/components/broadlink/strings.json +++ b/homeassistant/components/broadlink/strings.json @@ -49,6 +49,11 @@ } }, "entity": { + "infrared": { + "infrared_emitter": { + "name": "IR emitter" + } + }, "select": { "day_of_week": { "name": "Day of week", @@ -82,6 +87,9 @@ "frequency_not_supported": { "message": "Broadlink devices cannot transmit on {frequency} MHz" }, + "send_command_failed": { + "message": "Failed to send IR command: {error}" + }, "transmit_failed": { "message": "Failed to transmit RF command: {error}" } diff --git a/tests/components/broadlink/test_infrared.py b/tests/components/broadlink/test_infrared.py new file mode 100644 index 0000000000000..7d4959bcaaf3f --- /dev/null +++ b/tests/components/broadlink/test_infrared.py @@ -0,0 +1,108 @@ +"""Tests for Broadlink infrared platform.""" + +from unittest.mock import call + +from broadlink.exceptions import BroadlinkException +from broadlink.remote import pulses_to_data +from infrared_protocols import NECCommand +import pytest + +from homeassistant.components.broadlink.const import DOMAIN +from homeassistant.components.infrared import async_send_command +from homeassistant.const import Platform +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers import device_registry as dr, entity_registry as er + +from . import get_device + +IR_DEVICES = ["Entrance", "Living Room", "Office", "Garage"] +NON_IR_DEVICE = "Bedroom" + + +async def test_infrared_setup_works( + hass: HomeAssistant, + device_registry: dr.DeviceRegistry, + entity_registry: er.EntityRegistry, +) -> None: + """Test the infrared entity is created for all IR-capable devices.""" + for device in map(get_device, IR_DEVICES): + mock_setup = await device.setup_entry(hass) + + device_entry = device_registry.async_get_device( + identifiers={(DOMAIN, mock_setup.entry.unique_id)} + ) + entries = er.async_entries_for_device(entity_registry, device_entry.id) + infrared_entities = [ + entry for entry in entries if entry.domain == Platform.INFRARED + ] + assert len(infrared_entities) == 1 + assert infrared_entities[0].unique_id == f"{device.mac}-emitter" + + +async def test_infrared_not_created_for_non_ir_device( + hass: HomeAssistant, + entity_registry: er.EntityRegistry, +) -> None: + """Test no infrared entity is created for non-IR devices.""" + device = get_device(NON_IR_DEVICE) + mock_setup = await device.setup_entry(hass) + + entries = er.async_entries_for_config_entry( + entity_registry, mock_setup.entry.entry_id + ) + infrared_entities = [ + entry for entry in entries if entry.domain == Platform.INFRARED + ] + assert len(infrared_entities) == 0 + + +async def test_infrared_send_command( + hass: HomeAssistant, + entity_registry: er.EntityRegistry, +) -> None: + """Test sending an IR command dispatches to the Broadlink API.""" + device = get_device("Entrance") + mock_setup = await device.setup_entry(hass) + + entries = er.async_entries_for_config_entry( + entity_registry, mock_setup.entry.entry_id + ) + infrared_entity = next( + entry for entry in entries if entry.domain == Platform.INFRARED + ) + + command = NECCommand(address=0x20, command=0x10) + await async_send_command(hass, infrared_entity.entity_id, command) + + expected_pulses = [abs(t) for t in command.get_raw_timings()] + expected_packet = pulses_to_data(expected_pulses) + + assert mock_setup.api.send_data.call_count == 1 + assert mock_setup.api.send_data.call_args == call(expected_packet) + + +@pytest.mark.parametrize("error", [BroadlinkException("boom"), OSError("boom")]) +async def test_infrared_send_command_error_translates( + hass: HomeAssistant, + entity_registry: er.EntityRegistry, + error: Exception, +) -> None: + """Test that Broadlink API errors translate to HomeAssistantError.""" + device = get_device("Entrance") + mock_setup = await device.setup_entry(hass) + mock_setup.api.send_data.side_effect = error + + entries = er.async_entries_for_config_entry( + entity_registry, mock_setup.entry.entry_id + ) + infrared_entity = next( + entry for entry in entries if entry.domain == Platform.INFRARED + ) + + command = NECCommand(address=0x20, command=0x10) + with pytest.raises(HomeAssistantError) as exc_info: + await async_send_command(hass, infrared_entity.entity_id, command) + + assert exc_info.value.translation_key == "send_command_failed" + assert exc_info.value.translation_domain == DOMAIN