mirror of
https://github.com/home-assistant/core.git
synced 2026-04-15 22:26:12 +02:00
Compare commits
4 Commits
168007
...
radio-freq
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
258dfdda8f | ||
|
|
dcc0745fd2 | ||
|
|
174c86fd36 | ||
|
|
e0d7e3702c |
@@ -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/**
|
||||
|
||||
@@ -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,
|
||||
|
||||
79
homeassistant/components/esphome/radio_frequency.py
Normal file
79
homeassistant/components/esphome/radio_frequency.py
Normal file
@@ -0,0 +1,79 @@
|
||||
"""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 = [
|
||||
interval
|
||||
for timing in command.get_raw_timings()
|
||||
for interval in (timing.high_us, -timing.low_us)
|
||||
]
|
||||
_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
|
||||
),
|
||||
)
|
||||
189
homeassistant/components/radio_frequency/__init__.py
Normal file
189
homeassistant/components/radio_frequency/__init__.py
Normal file
@@ -0,0 +1,189 @@
|
||||
"""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.
|
||||
|
||||
An empty list means no compatible transmitters.
|
||||
|
||||
Raises:
|
||||
HomeAssistantError: 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 any(
|
||||
low <= frequency <= high for low, high in entity.supported_frequency_ranges
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
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:
|
||||
HomeAssistantError: If the radio_frequency 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 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
|
||||
def supported_frequency_ranges(self) -> list[tuple[int, int]]:
|
||||
"""Return list of (min_hz, max_hz) tuples."""
|
||||
raise NotImplementedError
|
||||
|
||||
@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.
|
||||
"""
|
||||
5
homeassistant/components/radio_frequency/const.py
Normal file
5
homeassistant/components/radio_frequency/const.py
Normal file
@@ -0,0 +1,5 @@
|
||||
"""Constants for the Radio Frequency integration."""
|
||||
|
||||
from typing import Final
|
||||
|
||||
DOMAIN: Final = "radio_frequency"
|
||||
7
homeassistant/components/radio_frequency/icons.json
Normal file
7
homeassistant/components/radio_frequency/icons.json
Normal file
@@ -0,0 +1,7 @@
|
||||
{
|
||||
"entity_component": {
|
||||
"_": {
|
||||
"default": "mdi:radio-tower"
|
||||
}
|
||||
}
|
||||
}
|
||||
9
homeassistant/components/radio_frequency/manifest.json
Normal file
9
homeassistant/components/radio_frequency/manifest.json
Normal file
@@ -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==0.0.1"]
|
||||
}
|
||||
13
homeassistant/components/radio_frequency/strings.json
Normal file
13
homeassistant/components/radio_frequency/strings.json
Normal file
@@ -0,0 +1,13 @@
|
||||
{
|
||||
"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"
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
{
|
||||
"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"
|
||||
}
|
||||
}
|
||||
}
|
||||
1
homeassistant/generated/entity_platforms.py
generated
1
homeassistant/generated/entity_platforms.py
generated
@@ -31,6 +31,7 @@ class EntityPlatforms(StrEnum):
|
||||
IMAGE_PROCESSING = "image_processing"
|
||||
INFRARED = "infrared"
|
||||
LAWN_MOWER = "lawn_mower"
|
||||
RADIO_FREQUENCY = "radio_frequency"
|
||||
LIGHT = "light"
|
||||
LOCK = "lock"
|
||||
MEDIA_PLAYER = "media_player"
|
||||
|
||||
3
requirements_all.txt
generated
3
requirements_all.txt
generated
@@ -2831,6 +2831,9 @@ renson-endura-delta==1.7.2
|
||||
# homeassistant.components.reolink
|
||||
reolink-aio==0.19.1
|
||||
|
||||
# homeassistant.components.radio_frequency
|
||||
rf-protocols==0.0.1
|
||||
|
||||
# homeassistant.components.idteck_prox
|
||||
rfk101py==0.0.1
|
||||
|
||||
|
||||
3
requirements_test_all.txt
generated
3
requirements_test_all.txt
generated
@@ -2409,6 +2409,9 @@ renson-endura-delta==1.7.2
|
||||
# homeassistant.components.reolink
|
||||
reolink-aio==0.19.1
|
||||
|
||||
# homeassistant.components.radio_frequency
|
||||
rf-protocols==0.0.1
|
||||
|
||||
# homeassistant.components.rflink
|
||||
rflink==0.0.67
|
||||
|
||||
|
||||
211
tests/components/esphome/test_radio_frequency.py
Normal file
211
tests/components/esphome/test_radio_frequency.py
Normal file
@@ -0,0 +1,211 @@
|
||||
"""Test ESPHome radio frequency platform."""
|
||||
|
||||
from aioesphomeapi import (
|
||||
APIClient,
|
||||
APIConnectionError,
|
||||
RadioFrequencyCapability,
|
||||
RadioFrequencyInfo,
|
||||
RadioFrequencyModulation,
|
||||
)
|
||||
import pytest
|
||||
from rf_protocols import ModulationType, OOKCommand, Timing
|
||||
|
||||
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=[
|
||||
Timing(high_us=350, low_us=1050),
|
||||
Timing(high_us=350, low_us=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=[Timing(high_us=350, low_us=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
|
||||
1
tests/components/radio_frequency/__init__.py
Normal file
1
tests/components/radio_frequency/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""Tests for the Radio Frequency integration."""
|
||||
71
tests/components/radio_frequency/conftest.py
Normal file
71
tests/components/radio_frequency/conftest.py
Normal file
@@ -0,0 +1,71 @@
|
||||
"""Common fixtures for the Radio Frequency tests."""
|
||||
|
||||
from typing import override
|
||||
|
||||
import pytest
|
||||
from rf_protocols import ModulationType, RadioFrequencyCommand, Timing
|
||||
|
||||
from homeassistant.components.radio_frequency import 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[Timing]:
|
||||
"""Return mock timings."""
|
||||
return [Timing(high_us=350, low_us=1050), Timing(high_us=350, low_us=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 = frequency_ranges or [(433_000_000, 434_000_000)]
|
||||
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
|
||||
def mock_rf_entity() -> MockRadioFrequencyEntity:
|
||||
"""Return a mock radio frequency entity."""
|
||||
return MockRadioFrequencyEntity("test_rf_transmitter")
|
||||
171
tests/components/radio_frequency/test_init.py
Normal file
171
tests/components/radio_frequency/test_init.py
Normal file
@@ -0,0 +1,171 @@
|
||||
"""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 .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("init_integration")
|
||||
async def test_get_transmitters_with_frequency_ranges(
|
||||
hass: HomeAssistant,
|
||||
mock_rf_entity: MockRadioFrequencyEntity,
|
||||
) -> None:
|
||||
"""Test transmitter with frequency ranges filters correctly."""
|
||||
component = hass.data[DATA_COMPONENT]
|
||||
await component.async_add_entities([mock_rf_entity])
|
||||
|
||||
# 433.92 MHz is within 433-434 MHz range
|
||||
result = async_get_transmitters(hass, 433_920_000, ModulationType.OOK)
|
||||
assert result == [mock_rf_entity.entity_id]
|
||||
|
||||
# 868 MHz is outside the range
|
||||
result = async_get_transmitters(hass, 868_000_000, ModulationType.OOK)
|
||||
assert result == []
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("init_integration")
|
||||
async def test_rf_entity_initial_state(
|
||||
hass: HomeAssistant,
|
||||
mock_rf_entity: MockRadioFrequencyEntity,
|
||||
) -> None:
|
||||
"""Test radio frequency entity has no state before any command is sent."""
|
||||
component = hass.data[DATA_COMPONENT]
|
||||
await component.async_add_entities([mock_rf_entity])
|
||||
|
||||
state = hass.states.get("radio_frequency.test_rf_transmitter")
|
||||
assert state is not None
|
||||
assert state.state == STATE_UNKNOWN
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("init_integration")
|
||||
async def test_async_send_command_success(
|
||||
hass: HomeAssistant,
|
||||
mock_rf_entity: MockRadioFrequencyEntity,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""Test sending command via async_send_command helper."""
|
||||
component = hass.data[DATA_COMPONENT]
|
||||
await component.async_add_entities([mock_rf_entity])
|
||||
|
||||
now = dt_util.utcnow()
|
||||
freezer.move_to(now)
|
||||
|
||||
command = MockRadioFrequencyCommand(frequency=433_920_000)
|
||||
await async_send_command(hass, mock_rf_entity.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("radio_frequency.test_rf_transmitter")
|
||||
assert state is not None
|
||||
assert state.state == now.isoformat(timespec="milliseconds")
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("init_integration")
|
||||
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."""
|
||||
component = hass.data[DATA_COMPONENT]
|
||||
await component.async_add_entities([mock_rf_entity])
|
||||
|
||||
state = hass.states.get("radio_frequency.test_rf_transmitter")
|
||||
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, mock_rf_entity.entity_id, command)
|
||||
|
||||
state = hass.states.get("radio_frequency.test_rf_transmitter")
|
||||
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_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,
|
||||
mock_rf_entity: MockRadioFrequencyEntity,
|
||||
restored_value: str,
|
||||
expected_state: str,
|
||||
) -> None:
|
||||
"""Test radio frequency entity state restore."""
|
||||
mock_restore_cache(
|
||||
hass, [State("radio_frequency.test_rf_transmitter", 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([mock_rf_entity])
|
||||
|
||||
state = hass.states.get("radio_frequency.test_rf_transmitter")
|
||||
assert state is not None
|
||||
assert state.state == expected_state
|
||||
Reference in New Issue
Block a user