Compare commits

...

3 Commits

Author SHA1 Message Date
Paulus Schoutsen
4fe57129c1 Adopt rf-protocols 1.0.0 get_raw_timings signature
In rf-protocols 1.0.0, RadioFrequencyCommand.get_raw_timings() returns
a flat list[int] of signed alternating microseconds directly, so the
ESPHome platform can pass it straight through without unpacking Timing
objects.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-17 23:29:29 -04:00
Paulus Schoutsen
1136a9eb7d Remove None support from ESPHome supported_frequency_ranges
Always return frequency ranges from device info, matching the
updated RadioFrequencyTransmitterEntity contract.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-17 15:11:18 -04:00
Paulus Schoutsen
0a01eb60be Add radio_frequency platform to ESPHome
Implement the new radio_frequency platform on ESPHome, mirroring
the infrared platform pattern.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-17 15:11:15 -04:00
3 changed files with 285 additions and 0 deletions

View File

@@ -35,6 +35,7 @@ from aioesphomeapi import (
MediaPlayerInfo,
MediaPlayerSupportedFormat,
NumberInfo,
RadioFrequencyInfo,
SelectInfo,
SensorInfo,
SensorState,
@@ -88,6 +89,7 @@ INFO_TYPE_TO_PLATFORM: dict[type[EntityInfo], Platform] = {
FanInfo: Platform.FAN,
InfraredInfo: Platform.INFRARED,
LightInfo: Platform.LIGHT,
RadioFrequencyInfo: Platform.RADIO_FREQUENCY,
LockInfo: Platform.LOCK,
MediaPlayerInfo: Platform.MEDIA_PLAYER,
NumberInfo: Platform.NUMBER,

View File

@@ -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
),
)

View File

@@ -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