Add Broadlink infrared emitter support

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.REMOTE: {"RM4MINI", "RM4PRO", "RMMINI", "RMMINIB", "RMPRO"},
     Platform.SELECT: {"HYS"},

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"
+
+    def __init__(self, device: BroadlinkDevice) -> None:
+        """Initialize the entity."""
+        super().__init__(device)
+        self._attr_unique_id = device.unique_id
+
+    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

homeassistant/components/broadlink/strings.json

@@ -49,6 +49,11 @@
     }
   },
   "entity": {
+    "infrared": {
+      "infrared": {
+        "name": "IR transmitter"
+      }
+    },
     "select": {
       "day_of_week": {
         "name": "Day of week",
@@ -77,5 +82,10 @@
         "name": "Total consumption"
       }
     }
+  },
+  "exceptions": {
+    "send_command_failed": {
+      "message": "Failed to send IR command: {error}"
+    }
   }
 }

tests/components/broadlink/test_infrared.py

@@ -0,0 +1,109 @@
+"""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 == device.mac
+        assert mock_setup.api.auth.call_count == 1
+
+
+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