Compare commits

...

2 Commits

Author SHA1 Message Date
Paulus Schoutsen ec82342115 remove name 2026-06-05 22:11:52 -04:00
Claude adc9c0835f Add websocket API to list radio frequency transmitters
Add a radio_frequency/list websocket command that returns the available RF
transmitters. Each transmitter is described by its entity id, the device and
config entry it belongs to, the frequency ranges it can operate on and the
modulation types it supports.

To avoid a circular import between the package __init__ and the new
websocket_api module, DATA_COMPONENT is moved to const.py (and re-exported
from the package for backwards compatibility), mirroring how the camera
integration structures its data key.
2026-06-03 21:06:50 +02:00
5 changed files with 170 additions and 5 deletions
@@ -11,15 +11,16 @@ from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_validation as cv, entity_registry as er
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.typing import ConfigType
from homeassistant.util.hass_dict import HassKey
from .const import DOMAIN
from . import websocket_api
from .const import DATA_COMPONENT, DOMAIN
from .entity import (
RadioFrequencyTransmitterEntity,
RadioFrequencyTransmitterEntityDescription,
)
__all__ = [
"DATA_COMPONENT",
"DOMAIN",
"ModulationType",
"RadioFrequencyTransmitterEntity",
@@ -30,9 +31,6 @@ __all__ = [
_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
@@ -46,6 +44,8 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
](_LOGGER, DOMAIN, hass, SCAN_INTERVAL)
await component.async_setup(config)
websocket_api.async_setup(hass)
return True
@@ -2,4 +2,13 @@
from typing import Final
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.util.hass_dict import HassKey
from .entity import RadioFrequencyTransmitterEntity
DOMAIN: Final = "radio_frequency"
DATA_COMPONENT: HassKey[EntityComponent[RadioFrequencyTransmitterEntity]] = HassKey(
DOMAIN
)
@@ -2,6 +2,7 @@
"domain": "radio_frequency",
"name": "Radio Frequency",
"codeowners": ["@home-assistant/core"],
"dependencies": ["websocket_api"],
"documentation": "https://www.home-assistant.io/integrations/radio_frequency",
"integration_type": "entity",
"quality_scale": "internal",
@@ -0,0 +1,57 @@
"""The Radio Frequency websocket API."""
from typing import Any
from rf_protocols import ModulationType
import voluptuous as vol
from homeassistant.components import websocket_api
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import entity_registry as er
from .const import DATA_COMPONENT
@callback
def async_setup(hass: HomeAssistant) -> None:
"""Set up the radio frequency websocket API."""
websocket_api.async_register_command(hass, ws_list_transmitters)
@websocket_api.require_admin
@websocket_api.websocket_command({vol.Required("type"): "radio_frequency/list"})
@callback
def ws_list_transmitters(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict[str, Any],
) -> None:
"""Return the available radio frequency transmitters.
Each transmitter is described by its entity id, the device and config
entry it belongs to (when registered), the frequency ranges it can
operate on and the modulation types it supports.
"""
component = hass.data[DATA_COMPONENT]
ent_reg = er.async_get(hass)
transmitters: list[dict[str, Any]] = []
for entity in component.entities:
entry = ent_reg.async_get(entity.entity_id)
transmitters.append(
{
"entity_id": entity.entity_id,
"device_id": entry.device_id if entry else None,
"config_entry_id": entry.config_entry_id if entry else None,
"supported_frequency_ranges": [
[low, high] for low, high in entity.supported_frequency_ranges
],
"supported_modulations": [
modulation.value
for modulation in ModulationType
if entity.supports_modulation(modulation)
],
}
)
connection.send_result(msg["id"], {"transmitters": transmitters})
@@ -0,0 +1,98 @@
"""Tests for the Radio Frequency websocket API."""
import pytest
from homeassistant.components.radio_frequency import DATA_COMPONENT
from homeassistant.core import HomeAssistant
from . import ENTITY_ID
from .common import MockRadioFrequencyEntity
from tests.typing import WebSocketGenerator
@pytest.mark.usefixtures("mock_rf_entity")
async def test_list_transmitters(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
"""Test listing radio frequency transmitters."""
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "radio_frequency/list"})
response = await client.receive_json()
assert response["success"]
assert response["result"] == {
"transmitters": [
{
"entity_id": ENTITY_ID,
"device_id": None,
"config_entry_id": None,
"supported_frequency_ranges": [[433_000_000, 434_000_000]],
"supported_modulations": ["OOK"],
}
]
}
@pytest.mark.usefixtures("init_radio_frequency")
async def test_list_transmitters_empty(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
"""Test listing transmitters when none are registered."""
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "radio_frequency/list"})
response = await client.receive_json()
assert response["success"]
assert response["result"] == {"transmitters": []}
@pytest.mark.usefixtures("init_radio_frequency")
async def test_list_multiple_transmitters_with_ranges(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
"""Test listing transmitters reports each one's supported frequency ranges."""
component = hass.data[DATA_COMPONENT]
await component.async_add_entities(
[
MockRadioFrequencyEntity(
"transmitter_one",
frequency_ranges=[(433_000_000, 434_000_000)],
),
MockRadioFrequencyEntity(
"transmitter_two",
frequency_ranges=[
(868_000_000, 868_500_000),
(915_000_000, 928_000_000),
],
),
]
)
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "radio_frequency/list"})
response = await client.receive_json()
assert response["success"]
transmitters = response["result"]["transmitters"]
assert len(transmitters) == 2
all_ranges = [
transmitter["supported_frequency_ranges"] for transmitter in transmitters
]
assert [[433_000_000, 434_000_000]] in all_ranges
assert [[868_000_000, 868_500_000], [915_000_000, 928_000_000]] in all_ranges
@pytest.mark.usefixtures("mock_rf_entity")
async def test_list_transmitters_requires_admin(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
hass_read_only_access_token: str,
) -> None:
"""Test listing transmitters is only allowed for admins."""
client = await hass_ws_client(hass, hass_read_only_access_token)
await client.send_json_auto_id({"type": "radio_frequency/list"})
response = await client.receive_json()
assert not response["success"]
assert response["error"]["code"] == "unauthorized"