mirror of
https://github.com/home-assistant/core.git
synced 2026-04-20 18:16:18 +00:00
Compare commits
3 Commits
dev
...
esphome-ra
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4fe57129c1 | ||
|
|
1136a9eb7d | ||
|
|
0a01eb60be |
@@ -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,
|
||||
|
||||
75
homeassistant/components/esphome/radio_frequency.py
Normal file
75
homeassistant/components/esphome/radio_frequency.py
Normal 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
|
||||
),
|
||||
)
|
||||
208
tests/components/esphome/test_radio_frequency.py
Normal file
208
tests/components/esphome/test_radio_frequency.py
Normal 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
|
||||
Reference in New Issue
Block a user