Add radio_frequency entity integration

.core_files.yaml

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

homeassistant/components/radio_frequency/__init__.py

@@ -0,0 +1,190 @@
+"""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 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 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
+    @abstractmethod
+    def supported_frequency_ranges(self) -> list[tuple[int, int]]:
+        """Return list of (min_hz, max_hz) tuples."""
+
+    @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.
+        """

homeassistant/components/radio_frequency/const.py

@@ -0,0 +1,5 @@
+"""Constants for the Radio Frequency integration."""
+
+from typing import Final
+
+DOMAIN: Final = "radio_frequency"

homeassistant/components/radio_frequency/icons.json

@@ -0,0 +1,7 @@
+{
+  "entity_component": {
+    "_": {
+      "default": "mdi:radio-tower"
+    }
+  }
+}

homeassistant/components/radio_frequency/manifest.json

@@ -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"]
+}

homeassistant/components/radio_frequency/strings.json

@@ -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"
+    }
+  }
+}

homeassistant/components/radio_frequency/translations/en.json

@@ -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"
+    }
+  }
+}

tests/components/radio_frequency/__init__.py

@@ -0,0 +1 @@
+"""Tests for the Radio Frequency integration."""

tests/components/radio_frequency/conftest.py

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