json

homeassistant/components/esphome/entry_data.py

@@ -29,6 +29,7 @@ from aioesphomeapi import (
     Event,
     EventInfo,
     FanInfo,
+    InfraredProxyInfo,
     LightInfo,
     LockInfo,
     MediaPlayerInfo,
@@ -84,6 +85,7 @@ INFO_TYPE_TO_PLATFORM: dict[type[EntityInfo], Platform] = {
     DateTimeInfo: Platform.DATETIME,
     EventInfo: Platform.EVENT,
     FanInfo: Platform.FAN,
+    InfraredProxyInfo: Platform.REMOTE,
     LightInfo: Platform.LIGHT,
     LockInfo: Platform.LOCK,
     MediaPlayerInfo: Platform.MEDIA_PLAYER,
@@ -187,6 +189,7 @@ class RuntimeEntryData:
     entity_removal_callbacks: dict[EntityInfoKey, list[CALLBACK_TYPE]] = field(
         default_factory=dict
     )
+    infrared_proxy_receive_callbacks: list[CALLBACK_TYPE] = field(default_factory=list)
 
     @property
     def name(self) -> str:
@@ -518,6 +521,27 @@ class RuntimeEntryData:
             ),
         )
 
+    @callback
+    def async_on_infrared_proxy_receive(
+        self, hass: HomeAssistant, receive_event: Any
+    ) -> None:
+        """Handle an infrared proxy receive event."""
+        # Fire a Home Assistant event with the infrared data
+        device_info = self.device_info
+        if not device_info:
+            return
+
+        hass.bus.async_fire(
+            f"{DOMAIN}_infrared_proxy_received",
+            {
+                "device_name": device_info.name,
+                "device_mac": device_info.mac_address,
+                "entry_id": self.entry_id,
+                "key": receive_event.key,
+                "timings": receive_event.timings,
+            },
+        )
+
     @callback
     def async_register_assist_satellite_config_updated_callback(
         self,

homeassistant/components/esphome/manager.py

@@ -692,6 +692,11 @@ class ESPHomeManager:
                 cli.subscribe_zwave_proxy_request(self._async_zwave_proxy_request)
             )
 
+        if device_info.infrared_proxy_feature_flags:
+            entry_data.disconnect_callbacks.add(
+                cli.subscribe_infrared_proxy_receive(self._async_infrared_proxy_receive)
+            )
+
         cli.subscribe_home_assistant_states_and_services(
             on_state=entry_data.async_update_state,
             on_service_call=self.async_on_service_call,
@@ -722,6 +727,10 @@ class ESPHomeManager:
             self.hass, self.entry_data.device_info, zwave_home_id
         )
 
+    def _async_infrared_proxy_receive(self, receive_event: Any) -> None:
+        """Handle an infrared proxy receive event."""
+        self.entry_data.async_on_infrared_proxy_receive(self.hass, receive_event)
+
     async def on_disconnect(self, expected_disconnect: bool) -> None:
         """Run disconnect callbacks on API disconnect."""
         entry_data = self.entry_data

homeassistant/components/esphome/remote.py

@@ -0,0 +1,172 @@
+"""Support for ESPHome infrared proxy remote components."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+from functools import partial
+import json
+import logging
+from typing import Any
+
+from aioesphomeapi import (
+    EntityInfo,
+    EntityState,
+    InfraredProxyCapability,
+    InfraredProxyInfo,
+    InfraredProxyTimingParams,
+)
+
+from homeassistant.components.remote import RemoteEntity, RemoteEntityFeature
+from homeassistant.core import callback
+from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
+
+from .const import DOMAIN
+from .entity import EsphomeEntity, platform_async_setup_entry
+
+_LOGGER = logging.getLogger(__name__)
+
+PARALLEL_UPDATES = 0
+
+
+class EsphomeInfraredProxy(EsphomeEntity[InfraredProxyInfo, EntityState], RemoteEntity):
+    """An infrared proxy remote implementation for ESPHome."""
+
+    @callback
+    def _on_static_info_update(self, static_info: EntityInfo) -> None:
+        """Set attrs from static info."""
+        super()._on_static_info_update(static_info)
+        static_info = self._static_info
+        capabilities = static_info.capabilities
+
+        # Set supported features based on capabilities
+        features = RemoteEntityFeature(0)
+        if capabilities & InfraredProxyCapability.RECEIVER:
+            features |= RemoteEntityFeature.LEARN_COMMAND
+        self._attr_supported_features = features
+
+    @callback
+    def _on_device_update(self) -> None:
+        """Call when device updates or entry data changes."""
+        super()._on_device_update()
+        if self._entry_data.available:
+            # Infrared proxy entities should go available directly
+            # when the device comes online.
+            self.async_write_ha_state()
+
+    @property
+    def is_on(self) -> bool:
+        """Return true if remote is on."""
+        # ESPHome infrared proxies are always on when available
+        return self.available
+
+    async def async_turn_on(self, **kwargs: Any) -> None:
+        """Turn the remote on."""
+        # ESPHome infrared proxies are always on, nothing to do
+        _LOGGER.debug("Turn on called for %s (no-op)", self.name)
+
+    async def async_turn_off(self, **kwargs: Any) -> None:
+        """Turn the remote off."""
+        # ESPHome infrared proxies cannot be turned off
+        _LOGGER.debug("Turn off called for %s (no-op)", self.name)
+
+    async def async_send_command(self, command: Iterable[str], **kwargs: Any) -> None:
+        """Send commands to a device.
+
+        Commands should be JSON strings containing either:
+        1. Protocol-based format: {"protocol": "NEC", "address": 0x04, "command": 0x08}
+        2. Pulse-width format: {
+            "timing": {
+                "frequency": 38000,
+                "length_in_bits": 32,
+                "header_high_us": 9000,
+                "header_low_us": 4500,
+                ...
+            },
+            "data": [0x01, 0x02, 0x03, 0x04]
+        }
+        """
+        self._check_capabilities()
+
+        for cmd in command:
+            try:
+                cmd_data = json.loads(cmd)
+            except json.JSONDecodeError as err:
+                raise ServiceValidationError(
+                    f"Command must be valid JSON: {err}"
+                ) from err
+
+            # Check if this is a protocol-based command
+            if "protocol" in cmd_data:
+                self._client.infrared_proxy_transmit_protocol(
+                    self._static_info.key,
+                    cmd,  # Pass the original JSON string
+                )
+            # Check if this is a pulse-width command
+            elif "timing" in cmd_data and "data" in cmd_data:
+                timing_data = cmd_data["timing"]
+                data_array = cmd_data["data"]
+
+                # Convert array of integers to bytes
+                if not isinstance(data_array, list):
+                    raise ServiceValidationError(
+                        "Data must be an array of integers (0-255)"
+                    )
+
+                try:
+                    data_bytes = bytes(data_array)
+                except (ValueError, TypeError) as err:
+                    raise ServiceValidationError(
+                        f"Invalid data array: {err}. Each element must be an integer between 0 and 255."
+                    ) from err
+
+                timing = InfraredProxyTimingParams(
+                    frequency=timing_data.get("frequency", 38000),
+                    length_in_bits=timing_data.get("length_in_bits", 32),
+                    header_high_us=timing_data.get("header_high_us", 0),
+                    header_low_us=timing_data.get("header_low_us", 0),
+                    one_high_us=timing_data.get("one_high_us", 0),
+                    one_low_us=timing_data.get("one_low_us", 0),
+                    zero_high_us=timing_data.get("zero_high_us", 0),
+                    zero_low_us=timing_data.get("zero_low_us", 0),
+                    footer_high_us=timing_data.get("footer_high_us", 0),
+                    footer_low_us=timing_data.get("footer_low_us", 0),
+                    repeat_high_us=timing_data.get("repeat_high_us", 0),
+                    repeat_low_us=timing_data.get("repeat_low_us", 0),
+                    minimum_idle_time_us=timing_data.get("minimum_idle_time_us", 0),
+                    msb_first=timing_data.get("msb_first", True),
+                    repeat_count=timing_data.get("repeat_count", 1),
+                )
+                self._client.infrared_proxy_transmit(
+                    self._static_info.key,
+                    timing,
+                    data_bytes,
+                )
+            else:
+                raise ServiceValidationError(
+                    "Command must contain either 'protocol' or both 'timing' and 'data' fields"
+                )
+
+    def _check_capabilities(self) -> None:
+        """Check if the device supports transmission."""
+        if not self._static_info.capabilities & InfraredProxyCapability.TRANSMITTER:
+            raise HomeAssistantError(
+                translation_domain=DOMAIN,
+                translation_key="infrared_proxy_transmitter_not_supported",
+            )
+
+    async def async_learn_command(self, **kwargs: Any) -> None:
+        """Learn a command from a device."""
+        # Learning is handled through the receive event subscription
+        # which is managed at the entry_data level
+        raise HomeAssistantError(
+            "Learning commands is handled automatically through receive events. "
+            "Listen for esphome_infrared_proxy_received events instead."
+        )
+
+
+async_setup_entry = partial(
+    platform_async_setup_entry,
+    info_type=InfraredProxyInfo,
+    entity_type=EsphomeInfraredProxy,
+    state_type=EntityState,
+)

homeassistant/components/esphome/strings.json

@@ -140,6 +140,9 @@
     "error_uploading": {
       "message": "Error during OTA (Over-The-Air) update of {configuration}. Try again in ESPHome dashboard for more information."
     },
+    "infrared_proxy_transmitter_not_supported": {
+      "message": "Device does not support infrared transmission"
+    },
     "ota_in_progress": {
       "message": "An OTA (Over-The-Air) update is already in progress for {configuration}."
     }

tests/components/esphome/test_remote.py

@@ -0,0 +1,431 @@
+"""Test ESPHome infrared proxy remotes."""
+
+from unittest.mock import patch
+
+from aioesphomeapi import (
+    APIClient,
+    InfraredProxyCapability,
+    InfraredProxyInfo,
+    InfraredProxyReceiveEvent,
+)
+import pytest
+
+from homeassistant.components.remote import DOMAIN as REMOTE_DOMAIN, RemoteEntityFeature
+from homeassistant.const import STATE_ON, STATE_UNAVAILABLE
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
+
+
+async def test_infrared_proxy_transmitter_only(
+    hass: HomeAssistant,
+    mock_client: APIClient,
+    mock_esphome_device,
+) -> None:
+    """Test an infrared proxy remote with transmitter capability only."""
+    entity_info = [
+        InfraredProxyInfo(
+            object_id="myremote",
+            key=1,
+            name="my remote",
+            capabilities=InfraredProxyCapability.TRANSMITTER,
+        )
+    ]
+    states = []
+    user_service = []
+    await mock_esphome_device(
+        mock_client=mock_client,
+        entity_info=entity_info,
+        user_service=user_service,
+        states=states,
+    )
+    await hass.async_block_till_done()
+
+    # Test initial state
+    state = hass.states.get("remote.test_my_remote")
+    assert state is not None
+    assert state.state == STATE_ON
+    # Transmitter-only should not support learn
+    assert state.attributes["supported_features"] == 0
+
+
+async def test_infrared_proxy_receiver_capability(
+    hass: HomeAssistant,
+    mock_client: APIClient,
+    mock_esphome_device,
+) -> None:
+    """Test an infrared proxy remote with receiver capability."""
+    entity_info = [
+        InfraredProxyInfo(
+            object_id="myremote",
+            key=1,
+            name="my remote",
+            capabilities=InfraredProxyCapability.TRANSMITTER
+            | InfraredProxyCapability.RECEIVER,
+        )
+    ]
+    states = []
+    user_service = []
+    await mock_esphome_device(
+        mock_client=mock_client,
+        entity_info=entity_info,
+        user_service=user_service,
+        states=states,
+    )
+    await hass.async_block_till_done()
+
+    # Test initial state
+    state = hass.states.get("remote.test_my_remote")
+    assert state is not None
+    assert state.state == STATE_ON
+    # Should support learn command
+    assert state.attributes["supported_features"] == RemoteEntityFeature.LEARN_COMMAND
+
+
+async def test_infrared_proxy_unavailability(
+    hass: HomeAssistant,
+    mock_client: APIClient,
+    mock_esphome_device,
+) -> None:
+    """Test infrared proxy remote availability."""
+    entity_info = [
+        InfraredProxyInfo(
+            object_id="myremote",
+            key=1,
+            name="my remote",
+            capabilities=InfraredProxyCapability.TRANSMITTER,
+        )
+    ]
+    states = []
+    user_service = []
+    device = await mock_esphome_device(
+        mock_client=mock_client,
+        entity_info=entity_info,
+        user_service=user_service,
+        states=states,
+    )
+    await hass.async_block_till_done()
+
+    # Test initial state
+    state = hass.states.get("remote.test_my_remote")
+    assert state is not None
+    assert state.state == STATE_ON
+
+    # Test device becomes unavailable
+    await device.mock_disconnect(True)
+    await hass.async_block_till_done()
+    state = hass.states.get("remote.test_my_remote")
+    assert state.state == STATE_UNAVAILABLE
+
+    # Test device becomes available again
+    await device.mock_connect()
+    await hass.async_block_till_done()
+    state = hass.states.get("remote.test_my_remote")
+    assert state.state == STATE_ON
+
+
+async def test_infrared_proxy_receive_event(
+    hass: HomeAssistant,
+    mock_client: APIClient,
+    mock_esphome_device,
+) -> None:
+    """Test infrared proxy receive event firing."""
+    entity_info = [
+        InfraredProxyInfo(
+            object_id="myremote",
+            key=1,
+            name="my remote",
+            capabilities=InfraredProxyCapability.RECEIVER,
+        )
+    ]
+    states = []
+    user_service = []
+    device = await mock_esphome_device(
+        mock_client=mock_client,
+        entity_info=entity_info,
+        user_service=user_service,
+        states=states,
+    )
+    await hass.async_block_till_done()
+
+    events = []
+
+    def event_listener(event):
+        events.append(event)
+
+    hass.bus.async_listen("esphome_infrared_proxy_received", event_listener)
+
+    # Simulate receiving an infrared signal
+    receive_event = InfraredProxyReceiveEvent(
+        key=1,
+        timings=[1000, 500, 1000, 500, 500, 1000],
+    )
+    # Get entry_data from the config entry
+    entry_data = device.entry.runtime_data
+    entry_data.async_on_infrared_proxy_receive(hass, receive_event)
+    await hass.async_block_till_done()
+
+    # Verify event was fired
+    assert len(events) == 1
+    event_data = events[0].data
+    assert event_data["key"] == 1
+    assert event_data["timings"] == [1000, 500, 1000, 500, 500, 1000]
+    assert event_data["device_name"] == "test"
+    assert "entry_id" in event_data
+
+
+async def test_infrared_proxy_send_command_protocol(
+    hass: HomeAssistant,
+    mock_client: APIClient,
+    mock_esphome_device,
+) -> None:
+    """Test sending protocol-based commands."""
+    entity_info = [
+        InfraredProxyInfo(
+            object_id="myremote",
+            key=1,
+            name="my remote",
+            capabilities=InfraredProxyCapability.TRANSMITTER,
+        )
+    ]
+    states = []
+    user_service = []
+    await mock_esphome_device(
+        mock_client=mock_client,
+        entity_info=entity_info,
+        user_service=user_service,
+        states=states,
+    )
+    await hass.async_block_till_done()
+
+    # Test protocol-based command
+    with patch.object(
+        mock_client, "infrared_proxy_transmit_protocol"
+    ) as mock_transmit_protocol:
+        await hass.services.async_call(
+            REMOTE_DOMAIN,
+            "send_command",
+            {
+                "entity_id": "remote.test_my_remote",
+                "command": ['{"protocol": "NEC", "address": 4, "command": 8}'],
+            },
+            blocking=True,
+        )
+        await hass.async_block_till_done()
+
+        mock_transmit_protocol.assert_called_once_with(
+            1, '{"protocol": "NEC", "address": 4, "command": 8}'
+        )
+
+
+async def test_infrared_proxy_send_command_pulse_width(
+    hass: HomeAssistant,
+    mock_client: APIClient,
+    mock_esphome_device,
+) -> None:
+    """Test sending pulse-width based commands."""
+    entity_info = [
+        InfraredProxyInfo(
+            object_id="myremote",
+            key=1,
+            name="my remote",
+            capabilities=InfraredProxyCapability.TRANSMITTER,
+        )
+    ]
+    states = []
+    user_service = []
+    await mock_esphome_device(
+        mock_client=mock_client,
+        entity_info=entity_info,
+        user_service=user_service,
+        states=states,
+    )
+    await hass.async_block_till_done()
+
+    # Test pulse-width command
+    with patch.object(mock_client, "infrared_proxy_transmit") as mock_transmit:
+        await hass.services.async_call(
+            REMOTE_DOMAIN,
+            "send_command",
+            {
+                "entity_id": "remote.test_my_remote",
+                "command": [
+                    '{"timing": {"frequency": 38000, "length_in_bits": 32}, "data": [1, 2, 3, 4]}'
+                ],
+            },
+            blocking=True,
+        )
+        await hass.async_block_till_done()
+
+        assert mock_transmit.call_count == 1
+        call_args = mock_transmit.call_args
+        assert call_args[0][0] == 1  # key
+        assert call_args[0][2] == b"\x01\x02\x03\x04"  # decoded data
+
+
+async def test_infrared_proxy_send_command_invalid_json(
+    hass: HomeAssistant,
+    mock_client: APIClient,
+    mock_esphome_device,
+) -> None:
+    """Test sending invalid JSON command."""
+    entity_info = [
+        InfraredProxyInfo(
+            object_id="myremote",
+            key=1,
+            name="my remote",
+            capabilities=InfraredProxyCapability.TRANSMITTER,
+        )
+    ]
+    states = []
+    user_service = []
+    await mock_esphome_device(
+        mock_client=mock_client,
+        entity_info=entity_info,
+        user_service=user_service,
+        states=states,
+    )
+    await hass.async_block_till_done()
+
+    # Test invalid JSON
+    with pytest.raises(
+        ServiceValidationError,
+        match="Command must be valid JSON",
+    ):
+        await hass.services.async_call(
+            REMOTE_DOMAIN,
+            "send_command",
+            {"entity_id": "remote.test_my_remote", "command": ["not valid json"]},
+            blocking=True,
+        )
+
+
+async def test_infrared_proxy_send_command_invalid_data_array(
+    hass: HomeAssistant,
+    mock_client: APIClient,
+    mock_esphome_device,
+) -> None:
+    """Test sending command with invalid data array."""
+    entity_info = [
+        InfraredProxyInfo(
+            object_id="myremote",
+            key=1,
+            name="my remote",
+            capabilities=InfraredProxyCapability.TRANSMITTER,
+        )
+    ]
+    states = []
+    user_service = []
+    await mock_esphome_device(
+        mock_client=mock_client,
+        entity_info=entity_info,
+        user_service=user_service,
+        states=states,
+    )
+    await hass.async_block_till_done()
+
+    # Test invalid data type (not an array)
+    with pytest.raises(
+        ServiceValidationError,
+        match="Data must be an array of integers",
+    ):
+        await hass.services.async_call(
+            REMOTE_DOMAIN,
+            "send_command",
+            {
+                "entity_id": "remote.test_my_remote",
+                "command": ['{"timing": {"frequency": 38000}, "data": "not_an_array"}'],
+            },
+            blocking=True,
+        )
+
+    # Test invalid array values (out of range)
+    with pytest.raises(
+        ServiceValidationError,
+        match="Invalid data array",
+    ):
+        await hass.services.async_call(
+            REMOTE_DOMAIN,
+            "send_command",
+            {
+                "entity_id": "remote.test_my_remote",
+                "command": ['{"timing": {"frequency": 38000}, "data": [1, 2, 300, 4]}'],
+            },
+            blocking=True,
+        )
+
+
+async def test_infrared_proxy_send_command_no_transmitter(
+    hass: HomeAssistant,
+    mock_client: APIClient,
+    mock_esphome_device,
+) -> None:
+    """Test sending command to receiver-only device."""
+    entity_info = [
+        InfraredProxyInfo(
+            object_id="myremote",
+            key=1,
+            name="my remote",
+            capabilities=InfraredProxyCapability.RECEIVER,  # No transmitter
+        )
+    ]
+    states = []
+    user_service = []
+    await mock_esphome_device(
+        mock_client=mock_client,
+        entity_info=entity_info,
+        user_service=user_service,
+        states=states,
+    )
+    await hass.async_block_till_done()
+
+    # Test send_command raises error
+    with pytest.raises(
+        HomeAssistantError,
+        match="does not support infrared transmission",
+    ):
+        await hass.services.async_call(
+            REMOTE_DOMAIN,
+            "send_command",
+            {
+                "entity_id": "remote.test_my_remote",
+                "command": ['{"protocol": "NEC", "address": 4, "command": 8}'],
+            },
+            blocking=True,
+        )
+
+
+async def test_infrared_proxy_learn_command_not_implemented(
+    hass: HomeAssistant,
+    mock_client: APIClient,
+    mock_esphome_device,
+) -> None:
+    """Test that learn_command raises appropriate error."""
+    entity_info = [
+        InfraredProxyInfo(
+            object_id="myremote",
+            key=1,
+            name="my remote",
+            capabilities=InfraredProxyCapability.RECEIVER,
+        )
+    ]
+    states = []
+    user_service = []
+    await mock_esphome_device(
+        mock_client=mock_client,
+        entity_info=entity_info,
+        user_service=user_service,
+        states=states,
+    )
+    await hass.async_block_till_done()
+
+    # Test learn_command raises error
+    with pytest.raises(
+        HomeAssistantError,
+        match="Learning commands is handled automatically",
+    ):
+        await hass.services.async_call(
+            REMOTE_DOMAIN,
+            "learn_command",
+            {"entity_id": "remote.test_my_remote"},
+            blocking=True,
+        )