1
0
mirror of https://github.com/home-assistant/core.git synced 2026-05-08 17:49:37 +01:00

Add radio_frequency entity integration (#168447)

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: balloob <1444314+balloob@users.noreply.github.com>
This commit is contained in:
Paulus Schoutsen
2026-04-24 06:37:28 -04:00
committed by GitHub
parent eb42804871
commit dd2a90a31f
18 changed files with 688 additions and 0 deletions
+1
View File
@@ -36,6 +36,7 @@ base_platforms: &base_platforms
- homeassistant/components/image_processing/**
- homeassistant/components/infrared/**
- homeassistant/components/lawn_mower/**
- homeassistant/components/radio_frequency/**
- homeassistant/components/light/**
- homeassistant/components/lock/**
- homeassistant/components/media_player/**
Generated
+2
View File
@@ -1415,6 +1415,8 @@ CLAUDE.md @home-assistant/core
/tests/components/radarr/ @tkdrob
/homeassistant/components/radio_browser/ @frenck
/tests/components/radio_browser/ @frenck
/homeassistant/components/radio_frequency/ @home-assistant/core
/tests/components/radio_frequency/ @home-assistant/core
/homeassistant/components/radiotherm/ @vinnyfuria
/tests/components/radiotherm/ @vinnyfuria
/homeassistant/components/rainbird/ @konikvranik @allenporter
@@ -62,6 +62,7 @@ COMPONENTS_WITH_DEMO_PLATFORM = [
Platform.LAWN_MOWER,
Platform.LOCK,
Platform.NOTIFY,
Platform.RADIO_FREQUENCY,
Platform.SENSOR,
Platform.SWITCH,
Platform.WEATHER,
@@ -0,0 +1,67 @@
"""Demo platform that offers a fake radio frequency entity."""
from __future__ import annotations
from rf_protocols import RadioFrequencyCommand
from homeassistant.components import persistent_notification
from homeassistant.components.radio_frequency import RadioFrequencyTransmitterEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import DOMAIN
PARALLEL_UPDATES = 0
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the demo radio frequency platform."""
async_add_entities(
[
DemoRadioFrequency(
unique_id="rf_transmitter",
device_name="RF Blaster",
entity_name="Radio Frequency Transmitter",
),
]
)
class DemoRadioFrequency(RadioFrequencyTransmitterEntity):
"""Representation of a demo radio frequency entity."""
_attr_has_entity_name = True
_attr_should_poll = False
def __init__(
self,
unique_id: str,
device_name: str,
entity_name: str,
) -> None:
"""Initialize the demo radio frequency entity."""
self._attr_unique_id = unique_id
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, unique_id)},
name=device_name,
)
self._attr_name = entity_name
@property
def supported_frequency_ranges(self) -> list[tuple[int, int]]:
"""Return supported frequency ranges."""
return [(300_000_000, 928_000_000)]
async def async_send_command(self, command: RadioFrequencyCommand) -> None:
"""Send an RF command."""
persistent_notification.async_create(
self.hass,
str(command.get_raw_timings()),
title="Radio Frequency Command",
)
@@ -0,0 +1,228 @@
"""Provides functionality to interact with radio frequency devices."""
from __future__ import annotations
from abc import abstractmethod
from datetime import timedelta
import logging
from typing import final
from rf_protocols import ModulationType, RadioFrequencyCommand
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.core import Context, HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_validation as cv, entity_registry as er
from homeassistant.helpers.entity import EntityDescription
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.typing import ConfigType
from homeassistant.util import dt as dt_util
from homeassistant.util.hass_dict import HassKey
from .const import DOMAIN
__all__ = [
"DOMAIN",
"ModulationType",
"RadioFrequencyTransmitterEntity",
"RadioFrequencyTransmitterEntityDescription",
"async_get_transmitters",
"async_send_command",
]
_LOGGER = logging.getLogger(__name__)
DATA_COMPONENT: HassKey[EntityComponent[RadioFrequencyTransmitterEntity]] = HassKey(
DOMAIN
)
ENTITY_ID_FORMAT = DOMAIN + ".{}"
PLATFORM_SCHEMA = cv.PLATFORM_SCHEMA
PLATFORM_SCHEMA_BASE = cv.PLATFORM_SCHEMA_BASE
SCAN_INTERVAL = timedelta(seconds=30)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the radio_frequency domain."""
component = hass.data[DATA_COMPONENT] = EntityComponent[
RadioFrequencyTransmitterEntity
](_LOGGER, DOMAIN, hass, SCAN_INTERVAL)
await component.async_setup(config)
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up a config entry."""
return await hass.data[DATA_COMPONENT].async_setup_entry(entry)
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.data[DATA_COMPONENT].async_unload_entry(entry)
@callback
def async_get_transmitters(
hass: HomeAssistant,
frequency: int,
modulation: ModulationType,
) -> list[str]:
"""Get entity IDs of all RF transmitters supporting the given frequency.
Transmitters are filtered by both their supported frequency ranges and
their supported modulation types. An empty list means no compatible
transmitters.
Raises:
HomeAssistantError: If the component is not loaded or if no
transmitters exist.
"""
component = hass.data.get(DATA_COMPONENT)
if component is None:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="component_not_loaded",
)
entities = list(component.entities)
if not entities:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="no_transmitters",
)
return [
entity.entity_id
for entity in entities
if entity.supports_modulation(modulation)
and entity.supports_frequency(frequency)
]
async def async_send_command(
hass: HomeAssistant,
entity_id_or_uuid: str,
command: RadioFrequencyCommand,
context: Context | None = None,
) -> None:
"""Send an RF command to the specified radio_frequency entity.
Raises:
vol.Invalid: If `entity_id_or_uuid` is not a valid entity ID or known entity
registry UUID.
HomeAssistantError: If the radio_frequency component is not loaded or the
resolved entity is not found.
"""
component = hass.data.get(DATA_COMPONENT)
if component is None:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="component_not_loaded",
)
ent_reg = er.async_get(hass)
entity_id = er.async_validate_entity_id(ent_reg, entity_id_or_uuid)
entity = component.get_entity(entity_id)
if entity is None:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="entity_not_found",
translation_placeholders={"entity_id": entity_id},
)
if not entity.supports_frequency(command.frequency):
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="unsupported_frequency",
translation_placeholders={
"entity_id": entity_id,
"frequency": str(command.frequency),
},
)
if not entity.supports_modulation(command.modulation):
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="unsupported_modulation",
translation_placeholders={
"entity_id": entity_id,
"modulation": command.modulation,
},
)
if context is not None:
entity.async_set_context(context)
await entity.async_send_command_internal(command)
class RadioFrequencyTransmitterEntityDescription(
EntityDescription, frozen_or_thawed=True
):
"""Describes radio frequency transmitter entities."""
class RadioFrequencyTransmitterEntity(RestoreEntity):
"""Base class for radio frequency transmitter entities."""
entity_description: RadioFrequencyTransmitterEntityDescription
_attr_should_poll = False
_attr_state: None = None
__last_command_sent: str | None = None
@property
@abstractmethod
def supported_frequency_ranges(self) -> list[tuple[int, int]]:
"""Return list of (min_hz, max_hz) tuples."""
@callback
@final
def supports_frequency(self, frequency: int) -> bool:
"""Return whether the transmitter supports the given frequency."""
return any(
low <= frequency <= high for low, high in self.supported_frequency_ranges
)
@callback
@final
def supports_modulation(self, modulation: ModulationType) -> bool:
"""Return whether the transmitter supports the given modulation."""
return modulation == ModulationType.OOK
@property
@final
def state(self) -> str | None:
"""Return the entity state."""
return self.__last_command_sent
@final
async def async_send_command_internal(self, command: RadioFrequencyCommand) -> None:
"""Send an RF command and update state.
Should not be overridden, handles setting last sent timestamp.
"""
await self.async_send_command(command)
self.__last_command_sent = dt_util.utcnow().isoformat(timespec="milliseconds")
self.async_write_ha_state()
@final
async def async_internal_added_to_hass(self) -> None:
"""Call when the radio frequency entity is added to hass."""
await super().async_internal_added_to_hass()
state = await self.async_get_last_state()
if state is not None and state.state not in (STATE_UNAVAILABLE, None):
self.__last_command_sent = state.state
@abstractmethod
async def async_send_command(self, command: RadioFrequencyCommand) -> None:
"""Send an RF command.
Args:
command: The RF command to send.
Raises:
HomeAssistantError: If transmission fails.
"""
@@ -0,0 +1,5 @@
"""Constants for the Radio Frequency integration."""
from typing import Final
DOMAIN: Final = "radio_frequency"
@@ -0,0 +1,7 @@
{
"entity_component": {
"_": {
"default": "mdi:radio-tower"
}
}
}
@@ -0,0 +1,9 @@
{
"domain": "radio_frequency",
"name": "Radio Frequency",
"codeowners": ["@home-assistant/core"],
"documentation": "https://www.home-assistant.io/integrations/radio_frequency",
"integration_type": "entity",
"quality_scale": "internal",
"requirements": ["rf-protocols==2.0.0"]
}
@@ -0,0 +1,19 @@
{
"exceptions": {
"component_not_loaded": {
"message": "Radio Frequency component not loaded"
},
"entity_not_found": {
"message": "Radio Frequency entity `{entity_id}` not found"
},
"no_transmitters": {
"message": "No Radio Frequency transmitters available"
},
"unsupported_frequency": {
"message": "Radio Frequency entity `{entity_id}` does not support frequency {frequency} Hz"
},
"unsupported_modulation": {
"message": "Radio Frequency entity `{entity_id}` does not support modulation {modulation}"
}
}
}
+1
View File
@@ -36,6 +36,7 @@ class EntityPlatforms(StrEnum):
MEDIA_PLAYER = "media_player"
NOTIFY = "notify"
NUMBER = "number"
RADIO_FREQUENCY = "radio_frequency"
REMOTE = "remote"
SCENE = "scene"
SELECT = "select"
+1
View File
@@ -47,6 +47,7 @@ python-slugify==8.0.4
PyTurboJPEG==1.8.3
PyYAML==6.0.3
requests==2.33.1
rf-protocols==2.0.0
securetar==2026.4.1
SQLAlchemy==2.0.49
standard-aifc==3.13.0
+3
View File
@@ -2840,6 +2840,9 @@ renson-endura-delta==1.7.2
# homeassistant.components.reolink
reolink-aio==0.19.1
# homeassistant.components.radio_frequency
rf-protocols==2.0.0
# homeassistant.components.idteck_prox
rfk101py==0.0.1
+3
View File
@@ -2424,6 +2424,9 @@ renson-endura-delta==1.7.2
# homeassistant.components.reolink
reolink-aio==0.19.1
# homeassistant.components.radio_frequency
rf-protocols==2.0.0
# homeassistant.components.rflink
rflink==0.0.67
@@ -0,0 +1,53 @@
"""The tests for the kitchen_sink radio frequency platform."""
from unittest.mock import patch
from freezegun.api import FrozenDateTimeFactory
import pytest
from rf_protocols import OOKCommand
from homeassistant.components.kitchen_sink import DOMAIN
from homeassistant.components.radio_frequency import async_send_command
from homeassistant.const import STATE_UNKNOWN, Platform
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
ENTITY_RF_TRANSMITTER = "radio_frequency.rf_blaster_radio_frequency_transmitter"
@pytest.fixture
async def radio_frequency_only() -> None:
"""Enable only the radio_frequency platform."""
with patch(
"homeassistant.components.kitchen_sink.COMPONENTS_WITH_DEMO_PLATFORM",
[Platform.RADIO_FREQUENCY],
):
yield
@pytest.fixture(autouse=True)
async def setup_comp(hass: HomeAssistant, radio_frequency_only: None) -> None:
"""Set up demo component."""
assert await async_setup_component(hass, DOMAIN, {DOMAIN: {}})
await hass.async_block_till_done()
async def test_send_command(
hass: HomeAssistant, freezer: FrozenDateTimeFactory
) -> None:
"""Test sending a radio frequency command."""
state = hass.states.get(ENTITY_RF_TRANSMITTER)
assert state
assert state.state == STATE_UNKNOWN
now = dt_util.parse_datetime("2021-01-09 12:00:00+00:00")
assert now is not None
freezer.move_to(now)
command = OOKCommand(frequency=433_920_000, timings=[350, -1050, 350, -350])
await async_send_command(hass, ENTITY_RF_TRANSMITTER, command)
state = hass.states.get(ENTITY_RF_TRANSMITTER)
assert state
assert state.state == now.isoformat(timespec="milliseconds")
@@ -0,0 +1,3 @@
"""Tests for the Radio Frequency integration."""
ENTITY_ID = "radio_frequency.test_rf_transmitter"
@@ -0,0 +1,83 @@
"""Common fixtures for the Radio Frequency tests."""
from typing import override
import pytest
from rf_protocols import ModulationType, RadioFrequencyCommand
from homeassistant.components.radio_frequency import (
DATA_COMPONENT,
RadioFrequencyTransmitterEntity,
)
from homeassistant.components.radio_frequency.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
@pytest.fixture
async def init_integration(hass: HomeAssistant) -> None:
"""Set up the Radio Frequency integration for testing."""
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
class MockRadioFrequencyCommand(RadioFrequencyCommand):
"""Mock RF command for testing."""
def __init__(
self,
*,
frequency: int = 433_920_000,
modulation: ModulationType = ModulationType.OOK,
repeat_count: int = 0,
) -> None:
"""Initialize mock command."""
super().__init__(
frequency=frequency, modulation=modulation, repeat_count=repeat_count
)
@override
def get_raw_timings(self) -> list[int]:
"""Return mock timings."""
return [350, -1050, 350, -350]
class MockRadioFrequencyEntity(RadioFrequencyTransmitterEntity):
"""Mock radio frequency entity for testing."""
_attr_has_entity_name = True
_attr_name = "Test RF transmitter"
def __init__(
self,
unique_id: str,
frequency_ranges: list[tuple[int, int]] | None = None,
) -> None:
"""Initialize mock entity."""
self._attr_unique_id = unique_id
self._frequency_ranges = (
[(433_000_000, 434_000_000)]
if frequency_ranges is None
else frequency_ranges
)
self.send_command_calls: list[RadioFrequencyCommand] = []
@property
def supported_frequency_ranges(self) -> list[tuple[int, int]]:
"""Return supported frequency ranges."""
return self._frequency_ranges
async def async_send_command(self, command: RadioFrequencyCommand) -> None:
"""Mock send command."""
self.send_command_calls.append(command)
@pytest.fixture
async def mock_rf_entity(
hass: HomeAssistant, init_integration: None
) -> MockRadioFrequencyEntity:
"""Return a mock radio frequency entity."""
entity = MockRadioFrequencyEntity("test_rf_transmitter")
component = hass.data[DATA_COMPONENT]
await component.async_add_entities([entity])
return entity
@@ -0,0 +1,200 @@
"""Tests for the Radio Frequency integration setup."""
from unittest.mock import AsyncMock
from freezegun.api import FrozenDateTimeFactory
import pytest
from rf_protocols import ModulationType
from homeassistant.components.radio_frequency import (
DATA_COMPONENT,
DOMAIN,
async_get_transmitters,
async_send_command,
)
from homeassistant.const import STATE_UNAVAILABLE, STATE_UNKNOWN
from homeassistant.core import HomeAssistant, State
from homeassistant.exceptions import HomeAssistantError
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from . import ENTITY_ID
from .conftest import MockRadioFrequencyCommand, MockRadioFrequencyEntity
from tests.common import mock_restore_cache
async def test_get_transmitters_component_not_loaded(hass: HomeAssistant) -> None:
"""Test getting transmitters raises when the component is not loaded."""
with pytest.raises(HomeAssistantError, match="component_not_loaded"):
async_get_transmitters(hass, 433_920_000, ModulationType.OOK)
@pytest.mark.usefixtures("init_integration")
async def test_get_transmitters_no_entities(hass: HomeAssistant) -> None:
"""Test getting transmitters raises when none are registered."""
with pytest.raises(
HomeAssistantError,
match="No Radio Frequency transmitters available",
):
async_get_transmitters(hass, 433_920_000, ModulationType.OOK)
@pytest.mark.usefixtures("mock_rf_entity")
async def test_get_transmitters_with_frequency_ranges(hass: HomeAssistant) -> None:
"""Test transmitter with frequency ranges filters correctly."""
# 433.92 MHz is within 433-434 MHz range
result = async_get_transmitters(hass, 433_920_000, ModulationType.OOK)
assert result == [ENTITY_ID]
# 868 MHz is outside the range
result = async_get_transmitters(hass, 868_000_000, ModulationType.OOK)
assert result == []
@pytest.mark.usefixtures("mock_rf_entity")
async def test_get_transmitters_filters_by_modulation(hass: HomeAssistant) -> None:
"""Test transmitters are filtered by supported modulation."""
result = async_get_transmitters(hass, 433_920_000, "no_matching_modulation") # type: ignore[arg-type]
assert result == []
@pytest.mark.usefixtures("mock_rf_entity")
async def test_rf_entity_initial_state(hass: HomeAssistant) -> None:
"""Test radio frequency entity has no state before any command is sent."""
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.state == STATE_UNKNOWN
async def test_async_send_command_success(
hass: HomeAssistant,
mock_rf_entity: MockRadioFrequencyEntity,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test sending command via async_send_command helper."""
now = dt_util.utcnow()
freezer.move_to(now)
command = MockRadioFrequencyCommand(frequency=433_920_000)
await async_send_command(hass, ENTITY_ID, command)
assert len(mock_rf_entity.send_command_calls) == 1
assert mock_rf_entity.send_command_calls[0] is command
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.state == now.isoformat(timespec="milliseconds")
async def test_async_send_command_error_does_not_update_state(
hass: HomeAssistant,
mock_rf_entity: MockRadioFrequencyEntity,
) -> None:
"""Test that state is not updated when async_send_command raises an error."""
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.state == STATE_UNKNOWN
command = MockRadioFrequencyCommand(frequency=433_920_000)
mock_rf_entity.async_send_command = AsyncMock(
side_effect=HomeAssistantError("Transmission failed")
)
with pytest.raises(HomeAssistantError, match="Transmission failed"):
await async_send_command(hass, ENTITY_ID, command)
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.state == STATE_UNKNOWN
@pytest.mark.usefixtures("init_integration")
async def test_async_send_command_entity_not_found(hass: HomeAssistant) -> None:
"""Test async_send_command raises error when entity not found."""
command = MockRadioFrequencyCommand(frequency=433_920_000)
with pytest.raises(
HomeAssistantError,
match="Radio Frequency entity `radio_frequency.nonexistent_entity` not found",
):
await async_send_command(hass, "radio_frequency.nonexistent_entity", command)
async def test_async_send_command_unsupported_frequency(
hass: HomeAssistant,
mock_rf_entity: MockRadioFrequencyEntity,
) -> None:
"""Test async_send_command raises when the frequency is not supported."""
command = MockRadioFrequencyCommand(frequency=868_000_000)
with pytest.raises(
HomeAssistantError,
match=(
f"Radio Frequency entity `{ENTITY_ID}` "
"does not support frequency 868000000 Hz"
),
):
await async_send_command(hass, ENTITY_ID, command)
assert mock_rf_entity.send_command_calls == []
@pytest.mark.usefixtures("mock_rf_entity")
async def test_async_send_command_unsupported_modulation(
hass: HomeAssistant,
mock_rf_entity: MockRadioFrequencyEntity,
) -> None:
"""Test async_send_command raises when the modulation is not supported."""
command = MockRadioFrequencyCommand(
frequency=433_920_000,
modulation="incorrect_modulation", # type: ignore[arg-type]
)
with pytest.raises(
HomeAssistantError,
match=(
f"Radio Frequency entity `{ENTITY_ID}` "
"does not support modulation incorrect_modulation"
),
):
await async_send_command(hass, ENTITY_ID, command)
assert mock_rf_entity.send_command_calls == []
async def test_async_send_command_component_not_loaded(hass: HomeAssistant) -> None:
"""Test async_send_command raises error when component not loaded."""
command = MockRadioFrequencyCommand(frequency=433_920_000)
with pytest.raises(HomeAssistantError, match="component_not_loaded"):
await async_send_command(hass, "radio_frequency.some_entity", command)
@pytest.mark.parametrize(
("restored_value", "expected_state"),
[
("2026-01-01T12:00:00.000+00:00", "2026-01-01T12:00:00.000+00:00"),
(STATE_UNAVAILABLE, STATE_UNKNOWN),
],
)
async def test_rf_entity_state_restore(
hass: HomeAssistant,
restored_value: str,
expected_state: str,
) -> None:
"""Test radio frequency entity state restore."""
mock_restore_cache(hass, [State(ENTITY_ID, restored_value)])
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
component = hass.data[DATA_COMPONENT]
await component.async_add_entities(
[MockRadioFrequencyEntity("test_rf_transmitter")]
)
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.state == expected_state
+2
View File
@@ -75,6 +75,7 @@
'onboarding',
'person',
'power',
'radio_frequency',
'remote',
'repairs',
'scene',
@@ -182,6 +183,7 @@
'onboarding',
'person',
'power',
'radio_frequency',
'remote',
'repairs',
'scene',