feat: Add Somfy RTS integration

homeassistant/components/somfy_rts/__init__.py

@@ -0,0 +1,25 @@
+"""The Somfy RTS integration."""
+
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.storage import Store
+
+from .const import DOMAIN, STORAGE_VERSION
+from .entity import SomfyRTSConfigEntry, SomfyRTSData
+
+PLATFORMS: list[Platform] = [Platform.BUTTON, Platform.COVER]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: SomfyRTSConfigEntry) -> bool:
+    """Set up Somfy RTS from a config entry."""
+    store = Store(hass, STORAGE_VERSION, f"{DOMAIN}/{entry.entry_id}")
+    stored = await store.async_load()
+    rolling_code = stored["rolling_code"] if stored is not None else 1
+    entry.runtime_data = SomfyRTSData(store=store, rolling_code=rolling_code)
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: SomfyRTSConfigEntry) -> bool:
+    """Unload a config entry."""
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

homeassistant/components/somfy_rts/button.py

@@ -0,0 +1,42 @@
+"""Button platform for Somfy RTS."""
+
+from rf_protocols import SomfyRTSButton
+
+from homeassistant.components.button import ButtonEntity
+from homeassistant.const import EntityCategory
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+
+from .entity import SomfyRTSConfigEntry, SomfyRTSEntity
+
+PARALLEL_UPDATES = 1
+
+# Somfy RTS PROG command requires 4 retransmit frames to reliably enter pairing mode.
+_PROG_FRAME_REPEATS = 4
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: SomfyRTSConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up the Somfy RTS button platform."""
+    async_add_entities([SomfyRTSProgButton(config_entry)])
+
+
+class SomfyRTSProgButton(SomfyRTSEntity, ButtonEntity):
+    """Button that sends the Somfy RTS PROG command to enter pairing mode."""
+
+    _attr_entity_category = EntityCategory.CONFIG
+    _attr_translation_key = "prog"
+
+    def __init__(self, entry: SomfyRTSConfigEntry) -> None:
+        """Initialize the PROG button."""
+        super().__init__(entry)
+        self._attr_unique_id = f"{entry.entry_id}_prog"
+
+    async def async_press(self) -> None:
+        """Send the PROG command."""
+        await self._async_send_command(
+            SomfyRTSButton.PROG, frame_repeats=_PROG_FRAME_REPEATS
+        )

homeassistant/components/somfy_rts/config_flow.py

@@ -0,0 +1,86 @@
+"""Config flow for the Somfy RTS integration."""
+
+from typing import Any
+
+from rf_protocols import SomfyRTSButton, SomfyRTSCommand
+import voluptuous as vol
+
+from homeassistant.components.radio_frequency import async_get_transmitters
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers import entity_registry as er, selector
+
+from .const import CONF_ADDRESS, CONF_TRANSMITTER, DOMAIN
+
+
+def _parse_address(value: str) -> int | None:
+    """Parse a hex string into a 24-bit Somfy RTS remote address.
+
+    Returns None if the value is not a valid hex string or out of range.
+    """
+    try:
+        address = int(value.strip(), 16)
+    except ValueError:
+        return None
+    if not (0x1 <= address <= 0xFFFFFF):
+        return None
+    return address
+
+
+class SomfyRTSConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Somfy RTS."""
+
+    VERSION = 1
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the initial step."""
+        sample_command = SomfyRTSCommand(
+            address=0x1, rolling_code=0, button=SomfyRTSButton.MY
+        )
+        try:
+            transmitters = async_get_transmitters(
+                self.hass, sample_command.frequency, sample_command.modulation
+            )
+        except HomeAssistantError:
+            return self.async_abort(reason="no_transmitters")
+
+        if not transmitters:
+            return self.async_abort(reason="no_compatible_transmitters")
+
+        errors: dict[str, str] = {}
+
+        if user_input is not None:
+            address = _parse_address(user_input[CONF_ADDRESS])
+            if address is None:
+                errors[CONF_ADDRESS] = "invalid_address"
+            else:
+                address_hex = format(address, "06X")
+                await self.async_set_unique_id(address_hex)
+                self._abort_if_unique_id_configured()
+
+                registry = er.async_get(self.hass)
+                entity_entry = registry.async_get(user_input[CONF_TRANSMITTER])
+                assert entity_entry is not None
+
+                return self.async_create_entry(
+                    title=f"Somfy RTS {address_hex}",
+                    data={
+                        CONF_ADDRESS: address,
+                        CONF_TRANSMITTER: entity_entry.id,
+                    },
+                )
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_ADDRESS): selector.TextSelector(),
+                    vol.Required(CONF_TRANSMITTER): selector.EntitySelector(
+                        selector.EntitySelectorConfig(include_entities=transmitters),
+                    ),
+                }
+            ),
+            errors=errors,
+        )

homeassistant/components/somfy_rts/const.py

@@ -0,0 +1,10 @@
+"""Constants for the Somfy RTS integration."""
+
+from typing import Final
+
+DOMAIN: Final = "somfy_rts"
+
+CONF_ADDRESS: Final = "address"
+CONF_TRANSMITTER: Final = "transmitter"
+
+STORAGE_VERSION: Final = 1

homeassistant/components/somfy_rts/cover.py

@@ -0,0 +1,70 @@
+"""Cover platform for Somfy RTS."""
+
+from typing import Any
+
+from rf_protocols import SomfyRTSButton
+
+from homeassistant.components.cover import (
+    CoverDeviceClass,
+    CoverEntity,
+    CoverEntityFeature,
+)
+from homeassistant.const import STATE_CLOSED, STATE_OPEN
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+from homeassistant.helpers.restore_state import RestoreEntity
+
+from .entity import SomfyRTSConfigEntry, SomfyRTSEntity
+
+PARALLEL_UPDATES = 1
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: SomfyRTSConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up the Somfy RTS cover platform."""
+    async_add_entities([SomfyRTSCover(config_entry)])
+
+
+class SomfyRTSCover(SomfyRTSEntity, CoverEntity, RestoreEntity):
+    """A Somfy RTS cover controlled via RF."""
+
+    _attr_assumed_state = True
+    _attr_device_class = CoverDeviceClass.SHUTTER
+    _attr_is_closed: bool | None = None
+    _attr_name = None
+    _attr_supported_features = (
+        CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE | CoverEntityFeature.STOP
+    )
+
+    def __init__(self, entry: SomfyRTSConfigEntry) -> None:
+        """Initialize the cover."""
+        super().__init__(entry)
+        self._attr_unique_id = entry.entry_id
+
+    async def async_added_to_hass(self) -> None:
+        """Restore last known cover state."""
+        await super().async_added_to_hass()
+        if (last_state := await self.async_get_last_state()) is not None:
+            if last_state.state == STATE_OPEN:
+                self._attr_is_closed = False
+            elif last_state.state == STATE_CLOSED:
+                self._attr_is_closed = True
+
+    async def async_open_cover(self, **kwargs: Any) -> None:
+        """Open the cover."""
+        await self._async_send_command(SomfyRTSButton.UP)
+        self._attr_is_closed = False
+        self.async_write_ha_state()
+
+    async def async_close_cover(self, **kwargs: Any) -> None:
+        """Close the cover."""
+        await self._async_send_command(SomfyRTSButton.DOWN)
+        self._attr_is_closed = True
+        self.async_write_ha_state()
+
+    async def async_stop_cover(self, **kwargs: Any) -> None:
+        """Stop the cover."""
+        await self._async_send_command(SomfyRTSButton.MY)

homeassistant/components/somfy_rts/entity.py

@@ -0,0 +1,109 @@
+"""Common entity for the Somfy RTS integration."""
+
+import logging
+
+from rf_protocols import SomfyRTSButton, SomfyRTSCommand
+
+from homeassistant.components.radio_frequency import async_send_command
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import STATE_UNAVAILABLE
+from homeassistant.core import Event, EventStateChangedData, callback
+from homeassistant.helpers import entity_registry as er
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.entity import Entity
+from homeassistant.helpers.event import async_track_state_change_event
+from homeassistant.helpers.storage import Store
+
+from .const import CONF_ADDRESS, CONF_TRANSMITTER, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class SomfyRTSData:
+    """Shared runtime data for a Somfy RTS config entry."""
+
+    def __init__(self, *, store: Store, rolling_code: int) -> None:
+        """Initialize runtime data."""
+        self.store = store
+        self.rolling_code = rolling_code
+
+
+type SomfyRTSConfigEntry = ConfigEntry[SomfyRTSData]
+
+
+class SomfyRTSEntity(Entity):
+    """Somfy RTS base entity."""
+
+    _attr_has_entity_name = True
+    _attr_should_poll = False
+
+    def __init__(self, entry: SomfyRTSConfigEntry) -> None:
+        """Initialize the entity."""
+        self._entry = entry
+        self._transmitter: str = entry.data[CONF_TRANSMITTER]
+        self._address: int = entry.data[CONF_ADDRESS]
+        address_hex = format(self._address, "06X")
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, entry.entry_id)},
+            manufacturer="Somfy",
+            model="RTS Remote",
+            name=f"Somfy RTS {address_hex}",
+        )
+
+    async def async_added_to_hass(self) -> None:
+        """Subscribe to transmitter entity state changes."""
+        await super().async_added_to_hass()
+
+        transmitter_entity_id = er.async_validate_entity_id(
+            er.async_get(self.hass), self._transmitter
+        )
+
+        @callback
+        def _async_transmitter_state_changed(
+            event: Event[EventStateChangedData],
+        ) -> None:
+            """Handle transmitter entity state changes."""
+            new_state = event.data["new_state"]
+            transmitter_available = (
+                new_state is not None and new_state.state != STATE_UNAVAILABLE
+            )
+            if transmitter_available != self.available:
+                _LOGGER.info(
+                    "Transmitter %s used by %s is %s",
+                    transmitter_entity_id,
+                    self.entity_id,
+                    "available" if transmitter_available else "unavailable",
+                )
+                self._attr_available = transmitter_available
+                self.async_write_ha_state()
+
+        self.async_on_remove(
+            async_track_state_change_event(
+                self.hass,
+                [transmitter_entity_id],
+                _async_transmitter_state_changed,
+            )
+        )
+
+        transmitter_state = self.hass.states.get(transmitter_entity_id)
+        self._attr_available = (
+            transmitter_state is not None
+            and transmitter_state.state != STATE_UNAVAILABLE
+        )
+
+    async def _async_send_command(
+        self, button: SomfyRTSButton, *, frame_repeats: int = 3
+    ) -> None:
+        """Increment the rolling code, persist it, and transmit the command."""
+        data = self._entry.runtime_data
+        data.rolling_code += 1
+        await data.store.async_save({"rolling_code": data.rolling_code})
+        command = SomfyRTSCommand(
+            address=self._address,
+            rolling_code=data.rolling_code,
+            button=button,
+            frame_repeats=frame_repeats,
+        )
+        await async_send_command(
+            self.hass, self._transmitter, command, context=self._context
+        )

homeassistant/components/somfy_rts/manifest.json

@@ -0,0 +1,12 @@
+{
+  "domain": "somfy_rts",
+  "name": "Somfy RTS",
+  "codeowners": [],
+  "config_flow": true,
+  "dependencies": ["radio_frequency"],
+  "documentation": "https://www.home-assistant.io/integrations/somfy_rts",
+  "integration_type": "device",
+  "iot_class": "assumed_state",
+  "quality_scale": "bronze",
+  "requirements": ["rf-protocols==2.2.0"]
+}