Add radio_frequency platform to ESPHome

homeassistant/components/esphome/entry_data.py

@@ -35,6 +35,7 @@
     MediaPlayerInfo,
     MediaPlayerSupportedFormat,
     NumberInfo,
+    RadioFrequencyInfo,
     SelectInfo,
     SensorInfo,
     SensorState,
@@ -88,6 +89,7 @@
     FanInfo: Platform.FAN,
     InfraredInfo: Platform.INFRARED,
     LightInfo: Platform.LIGHT,
+    RadioFrequencyInfo: Platform.RADIO_FREQUENCY,
     LockInfo: Platform.LOCK,
     MediaPlayerInfo: Platform.MEDIA_PLAYER,
     NumberInfo: Platform.NUMBER,

homeassistant/components/esphome/radio_frequency.py

@@ -0,0 +1,75 @@
+"""Radio Frequency platform for ESPHome."""
+
+from __future__ import annotations
+
+from functools import partial
+import logging
+
+from aioesphomeapi import (
+    EntityState,
+    RadioFrequencyCapability,
+    RadioFrequencyInfo,
+    RadioFrequencyModulation,
+)
+from rf_protocols import ModulationType, RadioFrequencyCommand
+
+from homeassistant.components.radio_frequency import RadioFrequencyTransmitterEntity
+from homeassistant.core import callback
+
+from .entity import (
+    EsphomeEntity,
+    convert_api_error_ha_error,
+    platform_async_setup_entry,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+PARALLEL_UPDATES = 0
+
+MODULATION_TYPE_TO_ESPHOME: dict[ModulationType, RadioFrequencyModulation] = {
+    ModulationType.OOK: RadioFrequencyModulation.OOK,
+}
+
+
+class EsphomeRadioFrequencyEntity(
+    EsphomeEntity[RadioFrequencyInfo, EntityState], RadioFrequencyTransmitterEntity
+):
+    """ESPHome radio frequency entity using native API."""
+
+    @property
+    def supported_frequency_ranges(self) -> list[tuple[int, int]]:
+        """Return supported frequency ranges from device info."""
+        return [(self._static_info.frequency_min, self._static_info.frequency_max)]
+
+    @callback
+    def _on_device_update(self) -> None:
+        """Call when device updates or entry data changes."""
+        super()._on_device_update()
+        if self._entry_data.available:
+            self.async_write_ha_state()
+
+    @convert_api_error_ha_error
+    async def async_send_command(self, command: RadioFrequencyCommand) -> None:
+        """Send an RF command."""
+        timings = command.get_raw_timings()
+        _LOGGER.debug("Sending RF command: %s", timings)
+
+        self._client.radio_frequency_transmit_raw_timings(
+            self._static_info.key,
+            frequency=command.frequency,
+            timings=timings,
+            modulation=MODULATION_TYPE_TO_ESPHOME[command.modulation],
+            repeat_count=command.repeat_count + 1,
+            device_id=self._static_info.device_id,
+        )
+
+
+async_setup_entry = partial(
+    platform_async_setup_entry,
+    info_type=RadioFrequencyInfo,
+    entity_type=EsphomeRadioFrequencyEntity,
+    state_type=EntityState,
+    info_filter=lambda info: bool(
+        info.capabilities & RadioFrequencyCapability.TRANSMITTER
+    ),
+)

tests/components/esphome/test_radio_frequency.py

@@ -0,0 +1,208 @@
+"""Test ESPHome radio frequency platform."""
+
+from aioesphomeapi import (
+    APIClient,
+    APIConnectionError,
+    RadioFrequencyCapability,
+    RadioFrequencyInfo,
+    RadioFrequencyModulation,
+)
+import pytest
+from rf_protocols import ModulationType, OOKCommand
+
+from homeassistant.components import radio_frequency
+from homeassistant.const import STATE_UNAVAILABLE
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+
+from .conftest import MockESPHomeDevice, MockESPHomeDeviceType
+
+ENTITY_ID = "radio_frequency.test_rf"
+
+
+async def _mock_rf_device(
+    mock_esphome_device: MockESPHomeDeviceType,
+    mock_client: APIClient,
+    capabilities: RadioFrequencyCapability = RadioFrequencyCapability.TRANSMITTER,
+    frequency_min: int = 433_000_000,
+    frequency_max: int = 434_000_000,
+    supported_modulations: int = 1,
+) -> MockESPHomeDevice:
+    entity_info = [
+        RadioFrequencyInfo(
+            object_id="rf",
+            key=1,
+            name="RF",
+            capabilities=capabilities,
+            frequency_min=frequency_min,
+            frequency_max=frequency_max,
+            supported_modulations=supported_modulations,
+        )
+    ]
+    return await mock_esphome_device(
+        mock_client=mock_client, entity_info=entity_info, states=[]
+    )
+
+
+@pytest.mark.parametrize(
+    ("capabilities", "entity_created"),
+    [
+        (RadioFrequencyCapability.TRANSMITTER, True),
+        (RadioFrequencyCapability.RECEIVER, False),
+        (
+            RadioFrequencyCapability.TRANSMITTER | RadioFrequencyCapability.RECEIVER,
+            True,
+        ),
+        (RadioFrequencyCapability(0), False),
+    ],
+)
+async def test_radio_frequency_entity_transmitter(
+    hass: HomeAssistant,
+    mock_client: APIClient,
+    mock_esphome_device: MockESPHomeDeviceType,
+    capabilities: RadioFrequencyCapability,
+    entity_created: bool,
+) -> None:
+    """Test radio frequency entity with transmitter capability is created."""
+    await _mock_rf_device(mock_esphome_device, mock_client, capabilities)
+
+    state = hass.states.get(ENTITY_ID)
+    assert (state is not None) == entity_created
+
+
+async def test_radio_frequency_multiple_entities_mixed_capabilities(
+    hass: HomeAssistant,
+    mock_client: APIClient,
+    mock_esphome_device: MockESPHomeDeviceType,
+) -> None:
+    """Test multiple radio frequency entities with mixed capabilities."""
+    entity_info = [
+        RadioFrequencyInfo(
+            object_id="rf_transmitter",
+            key=1,
+            name="RF Transmitter",
+            capabilities=RadioFrequencyCapability.TRANSMITTER,
+        ),
+        RadioFrequencyInfo(
+            object_id="rf_receiver",
+            key=2,
+            name="RF Receiver",
+            capabilities=RadioFrequencyCapability.RECEIVER,
+        ),
+        RadioFrequencyInfo(
+            object_id="rf_transceiver",
+            key=3,
+            name="RF Transceiver",
+            capabilities=(
+                RadioFrequencyCapability.TRANSMITTER | RadioFrequencyCapability.RECEIVER
+            ),
+        ),
+    ]
+    await mock_esphome_device(
+        mock_client=mock_client,
+        entity_info=entity_info,
+        states=[],
+    )
+
+    # Only transmitter and transceiver should be created
+    assert hass.states.get("radio_frequency.test_rf_transmitter") is not None
+    assert hass.states.get("radio_frequency.test_rf_receiver") is None
+    assert hass.states.get("radio_frequency.test_rf_transceiver") is not None
+
+
+async def test_radio_frequency_send_command_success(
+    hass: HomeAssistant,
+    mock_client: APIClient,
+    mock_esphome_device: MockESPHomeDeviceType,
+) -> None:
+    """Test sending RF command successfully."""
+    await _mock_rf_device(mock_esphome_device, mock_client)
+
+    command = OOKCommand(
+        frequency=433_920_000,
+        timings=[350, -1050, 350, -350],
+    )
+    await radio_frequency.async_send_command(hass, ENTITY_ID, command)
+
+    mock_client.radio_frequency_transmit_raw_timings.assert_called_once()
+    call_args = mock_client.radio_frequency_transmit_raw_timings.call_args
+    assert call_args[0][0] == 1  # key
+    assert call_args[1]["frequency"] == 433_920_000
+    assert call_args[1]["modulation"] == RadioFrequencyModulation.OOK
+    assert call_args[1]["repeat_count"] == 1
+    assert call_args[1]["device_id"] == 0
+    assert call_args[1]["timings"] == [350, -1050, 350, -350]
+
+
+async def test_radio_frequency_send_command_failure(
+    hass: HomeAssistant,
+    mock_client: APIClient,
+    mock_esphome_device: MockESPHomeDeviceType,
+) -> None:
+    """Test sending RF command with APIConnectionError raises HomeAssistantError."""
+    await _mock_rf_device(mock_esphome_device, mock_client)
+
+    mock_client.radio_frequency_transmit_raw_timings.side_effect = APIConnectionError(
+        "Connection lost"
+    )
+
+    command = OOKCommand(
+        frequency=433_920_000,
+        timings=[350, -1050],
+    )
+
+    with pytest.raises(HomeAssistantError) as exc_info:
+        await radio_frequency.async_send_command(hass, ENTITY_ID, command)
+    assert exc_info.value.translation_domain == "esphome"
+    assert exc_info.value.translation_key == "error_communicating_with_device"
+
+
+async def test_radio_frequency_entity_availability(
+    hass: HomeAssistant,
+    mock_client: APIClient,
+    mock_esphome_device: MockESPHomeDeviceType,
+) -> None:
+    """Test radio frequency entity becomes available after device reconnects."""
+    mock_device = await _mock_rf_device(mock_esphome_device, mock_client)
+
+    state = hass.states.get(ENTITY_ID)
+    assert state is not None
+    assert state.state != STATE_UNAVAILABLE
+
+    await mock_device.mock_disconnect(False)
+    await hass.async_block_till_done()
+
+    state = hass.states.get(ENTITY_ID)
+    assert state is not None
+    assert state.state == STATE_UNAVAILABLE
+
+    await mock_device.mock_connect()
+    await hass.async_block_till_done()
+
+    state = hass.states.get(ENTITY_ID)
+    assert state is not None
+    assert state.state != STATE_UNAVAILABLE
+
+
+async def test_radio_frequency_supported_frequency_ranges(
+    hass: HomeAssistant,
+    mock_client: APIClient,
+    mock_esphome_device: MockESPHomeDeviceType,
+) -> None:
+    """Test supported frequency ranges are exposed from device info."""
+    await _mock_rf_device(
+        mock_esphome_device,
+        mock_client,
+        frequency_min=433_000_000,
+        frequency_max=434_000_000,
+    )
+
+    transmitters = radio_frequency.async_get_transmitters(
+        hass, 433_920_000, ModulationType.OOK
+    )
+    assert len(transmitters) == 1
+
+    transmitters = radio_frequency.async_get_transmitters(
+        hass, 868_000_000, ModulationType.OOK
+    )
+    assert len(transmitters) == 0