feat: Add Somfy RTS integration

homeassistant/components/somfy_rts/__init__.py

@@ -0,0 +1,26 @@
+"""The Somfy RTS integration."""
+
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.storage import Store
+
+from .const import CONF_ROLLING_CODE, DOMAIN, STORAGE_VERSION
+from .entity import SomfyRTSConfigEntry, SomfyRTSData
+
+PLATFORMS: list[Platform] = [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()
+    entry_default = entry.data.get(CONF_ROLLING_CODE, 1)
+    rolling_code = stored.get("rolling_code", entry_default) if isinstance(stored, dict) else entry_default
+    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/config_flow.py

@@ -0,0 +1,137 @@
+"""Config flow for the Somfy RTS integration."""
+
+from typing import Any
+
+from rf_protocols.codes.somfy.rts import SomfyRTSButton
+from rf_protocols.commands.somfy_rts import SomfyRTSCommand
+import voluptuous as vol
+
+from homeassistant.components.radio_frequency import async_get_transmitters, async_send_command
+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_ROLLING_CODE, 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
+
+    def __init__(self) -> None:
+        """Initialize config flow."""
+        self._address: int | None = None
+        self._transmitter_id: str | None = None
+        self._rolling_code: int = 0
+
+    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])
+                if entity_entry is None:
+                    errors[CONF_TRANSMITTER] = "invalid_entity"
+                else:
+                    self._address = address
+                    self._transmitter_id = entity_entry.id
+                    return await self.async_step_prog()
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=self.add_suggested_values_to_schema(
+                vol.Schema(
+                    {
+                        vol.Required(CONF_ADDRESS): selector.TextSelector(),
+                        vol.Required(CONF_TRANSMITTER): selector.EntitySelector(
+                            selector.EntitySelectorConfig(include_entities=transmitters),
+                        ),
+                    }
+                ),
+                user_input or {},
+            ),
+            errors=errors,
+      last_step = False,
+        )
+
+    async def async_step_prog(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the PROG pairing step."""
+        assert self._address is not None
+        assert self._transmitter_id is not None
+        errors: dict[str, str] = {}
+
+        if user_input is not None:
+            if user_input.get("send_prog"):
+                self._rolling_code += 1
+                command = SomfyRTSCommand(
+                    address=self._address,
+                    rolling_code=self._rolling_code,
+                    button=SomfyRTSButton.PROG,
+                    frame_repeats=4,
+                )
+                try:
+                    await async_send_command(self.hass, self._transmitter_id, command)
+                except HomeAssistantError:
+                    errors["base"] = "prog_failed"
+                    self._rolling_code -= 1
+            else:
+                address_hex = format(self._address, "06X")
+                return self.async_create_entry(
+                    title=f"Somfy RTS {address_hex}",
+                    data={
+                        CONF_ADDRESS: self._address,
+                        CONF_TRANSMITTER: self._transmitter_id,
+                        CONF_ROLLING_CODE: self._rolling_code,
+                    },
+                )
+
+        return self.async_show_form(
+            step_id="prog",
+            data_schema=vol.Schema(
+                {
+                    vol.Optional("send_prog", default=False): selector.BooleanSelector(),
+                }
+            ),
+            errors=errors,
+        )

homeassistant/components/somfy_rts/const.py

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

homeassistant/components/somfy_rts/cover.py

@@ -0,0 +1,147 @@
+"""Cover platform for Somfy RTS."""
+
+import logging
+from typing import Any
+
+from rf_protocols.codes.somfy.rts import SomfyRTSButton
+from rf_protocols.commands.somfy_rts import SomfyRTSCommand
+
+from homeassistant.components.cover import (
+    CoverDeviceClass,
+    CoverEntity,
+    CoverEntityFeature,
+)
+from homeassistant.components.radio_frequency import async_send_command
+from homeassistant.const import STATE_CLOSED, STATE_OPEN, STATE_UNAVAILABLE
+from homeassistant.core import Event, EventStateChangedData, HomeAssistant, callback
+from homeassistant.helpers import entity_registry as er
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+from homeassistant.helpers.event import async_track_state_change_event
+from homeassistant.helpers.restore_state import RestoreEntity
+
+from .const import CONF_ADDRESS, CONF_TRANSMITTER, DOMAIN
+from .entity import SomfyRTSConfigEntry
+
+_LOGGER = logging.getLogger(__name__)
+
+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(CoverEntity, RestoreEntity):
+    """A Somfy RTS cover controlled via RF."""
+
+    _attr_assumed_state = True
+    _attr_available = False
+    _attr_device_class = CoverDeviceClass.SHUTTER
+    _attr_has_entity_name = True
+    _attr_is_closed: bool | None = None
+    _attr_name = None
+    _attr_should_poll = False
+    _attr_supported_features = (
+        CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE | CoverEntityFeature.STOP
+    )
+
+    def __init__(self, entry: SomfyRTSConfigEntry) -> None:
+        """Initialize the cover."""
+        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_unique_id = entry.entry_id
+        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:
+        """Restore last known state and subscribe to transmitter availability."""
+        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
+
+        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:
+            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:
+        """Transmit the command and persist the rolling code after success."""
+        data = self._entry.runtime_data
+        async with data.lock:
+            rolling_code = data.rolling_code + 1
+            command = SomfyRTSCommand(
+                address=self._address,
+                rolling_code=rolling_code,
+                button=button,
+                frame_repeats=frame_repeats,
+            )
+            await async_send_command(
+                self.hass, self._transmitter, command, context=self._context
+            )
+            data.rolling_code = rolling_code
+            await data.store.async_save({"rolling_code": data.rolling_code})
+
+    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,19 @@
+"""Shared data types for the Somfy RTS integration."""
+
+import asyncio
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.helpers.storage import Store
+
+
+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
+        self.lock = asyncio.Lock()
+
+
+type SomfyRTSConfigEntry = ConfigEntry[SomfyRTSData]

homeassistant/components/somfy_rts/manifest.json

@@ -0,0 +1,12 @@
+{
+  "domain": "somfy_rts",
+  "name": "Somfy RTS",
+  "codeowners": ["@l-henke"],
+  "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==3.0.0"]
+}