Add support for Shelly text virtual component (#121735)

* Add support for text component

* Add tests

* Improve const names

* Remove unnecessary code

---------

Co-authored-by: Maciej Bieniek <478555+bieniu@users.noreply.github.com>

homeassistant/components/shelly/__init__.py

@@ -63,6 +63,7 @@ PLATFORMS: Final = [
     Platform.LIGHT,
     Platform.SENSOR,
     Platform.SWITCH,
+    Platform.TEXT,
     Platform.UPDATE,
     Platform.VALVE,
 ]

homeassistant/components/shelly/const.py

@@ -241,5 +241,7 @@ SHELLY_PLUS_RGBW_CHANNELS = 4
 
 VIRTUAL_COMPONENTS_MAP = {
     "binary_sensor": {"type": "boolean", "mode": "label"},
+    "sensor": {"type": "text", "mode": "label"},
     "switch": {"type": "boolean", "mode": "toggle"},
+    "text": {"type": "text", "mode": "field"},
 }

homeassistant/components/shelly/sensor.py

@@ -9,6 +9,7 @@ from aioshelly.block_device import Block
 from aioshelly.const import RPC_GENERATIONS
 
 from homeassistant.components.sensor import (
+    DOMAIN as SENSOR_PLATFORM,
     RestoreSensor,
     SensorDeviceClass,
     SensorEntity,
@@ -52,8 +53,10 @@ from .entity import (
     async_setup_entry_rpc,
 )
 from .utils import (
+    async_remove_orphaned_virtual_entities,
     get_device_entry_gen,
     get_device_uptime,
+    get_virtual_component_ids,
     is_rpc_wifi_stations_disabled,
 )
 
@@ -1016,6 +1019,11 @@ RPC_SENSORS: Final = {
             or status[key].get("xfreq") is None
         ),
     ),
+    "text": RpcSensorDescription(
+        key="text",
+        sub_key="value",
+        has_entity_name=True,
+    ),
 }
 
 
@@ -1035,9 +1043,26 @@ async def async_setup_entry(
                 RpcSleepingSensor,
             )
         else:
+            coordinator = config_entry.runtime_data.rpc
+            assert coordinator
+
             async_setup_entry_rpc(
                 hass, config_entry, async_add_entities, RPC_SENSORS, RpcSensor
             )
+
+            # the user can remove virtual components from the device configuration, so
+            # we need to remove orphaned entities
+            virtual_sensor_ids = get_virtual_component_ids(
+                coordinator.device.config, SENSOR_PLATFORM
+            )
+            async_remove_orphaned_virtual_entities(
+                hass,
+                config_entry.entry_id,
+                coordinator.mac,
+                SENSOR_PLATFORM,
+                "text",
+                virtual_sensor_ids,
+            )
         return
 
     if config_entry.data[CONF_SLEEP_PERIOD]:

homeassistant/components/shelly/text.py

@@ -0,0 +1,89 @@
+"""Text for Shelly."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Final
+
+from aioshelly.const import RPC_GENERATIONS
+
+from homeassistant.components.text import (
+    DOMAIN as TEXT_PLATFORM,
+    TextEntity,
+    TextEntityDescription,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from .coordinator import ShellyConfigEntry
+from .entity import (
+    RpcEntityDescription,
+    ShellyRpcAttributeEntity,
+    async_setup_entry_rpc,
+)
+from .utils import (
+    async_remove_orphaned_virtual_entities,
+    get_device_entry_gen,
+    get_virtual_component_ids,
+)
+
+
+@dataclass(frozen=True, kw_only=True)
+class RpcTextDescription(RpcEntityDescription, TextEntityDescription):
+    """Class to describe a RPC text entity."""
+
+
+RPC_TEXT_ENTITIES: Final = {
+    "text": RpcTextDescription(
+        key="text",
+        sub_key="value",
+        has_entity_name=True,
+    ),
+}
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ShellyConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up sensors for device."""
+    if get_device_entry_gen(config_entry) in RPC_GENERATIONS:
+        coordinator = config_entry.runtime_data.rpc
+        assert coordinator
+
+        async_setup_entry_rpc(
+            hass, config_entry, async_add_entities, RPC_TEXT_ENTITIES, RpcText
+        )
+
+        # the user can remove virtual components from the device configuration, so
+        # we need to remove orphaned entities
+        virtual_text_ids = get_virtual_component_ids(
+            coordinator.device.config, TEXT_PLATFORM
+        )
+        async_remove_orphaned_virtual_entities(
+            hass,
+            config_entry.entry_id,
+            coordinator.mac,
+            TEXT_PLATFORM,
+            "text",
+            virtual_text_ids,
+        )
+
+
+class RpcText(ShellyRpcAttributeEntity, TextEntity):
+    """Represent a RPC text entity."""
+
+    entity_description: RpcTextDescription
+
+    @property
+    def native_value(self) -> str | None:
+        """Return value of sensor."""
+        if TYPE_CHECKING:
+            assert isinstance(self.attribute_value, str | None)
+
+        return self.attribute_value
+
+    async def async_set_value(self, value: str) -> None:
+        """Change the value."""
+        await self.call_rpc("Text.Set", {"id": self._id, "value": value})

homeassistant/components/shelly/utils.py

@@ -323,7 +323,7 @@ def get_rpc_channel_name(device: RpcDevice, key: str) -> str:
             return f"{device_name} {key.replace(':', '_')}"
         if key.startswith("em1"):
             return f"{device_name} EM{key.split(':')[-1]}"
-        if key.startswith("boolean:"):
+        if key.startswith(("boolean:", "text:")):
             return key.replace(":", " ").title()
         return device_name
 

tests/components/shelly/test_sensor.py

@@ -854,3 +854,104 @@ async def test_rpc_disabled_xfreq(
 
     entry = entity_registry.async_get(entity_id)
     assert not entry
+
+
+@pytest.mark.parametrize(
+    ("name", "entity_id"),
+    [
+        ("Virtual sensor", "sensor.test_name_virtual_sensor"),
+        (None, "sensor.test_name_text_203"),
+    ],
+)
+async def test_rpc_device_virtual_sensor(
+    hass: HomeAssistant,
+    entity_registry: EntityRegistry,
+    mock_rpc_device: Mock,
+    monkeypatch: pytest.MonkeyPatch,
+    name: str | None,
+    entity_id: str,
+) -> None:
+    """Test a virtual sensor for RPC device."""
+    config = deepcopy(mock_rpc_device.config)
+    config["text:203"] = {
+        "name": name,
+        "meta": {"ui": {"view": "label"}},
+    }
+    monkeypatch.setattr(mock_rpc_device, "config", config)
+
+    status = deepcopy(mock_rpc_device.status)
+    status["text:203"] = {"value": "lorem ipsum"}
+    monkeypatch.setattr(mock_rpc_device, "status", status)
+
+    await init_integration(hass, 3)
+
+    state = hass.states.get(entity_id)
+    assert state
+    assert state.state == "lorem ipsum"
+
+    entry = entity_registry.async_get(entity_id)
+    assert entry
+    assert entry.unique_id == "123456789ABC-text:203-text"
+
+    monkeypatch.setitem(mock_rpc_device.status["text:203"], "value", "dolor sit amet")
+    mock_rpc_device.mock_update()
+    assert hass.states.get(entity_id).state == "dolor sit amet"
+
+
+async def test_rpc_remove_virtual_sensor_when_mode_field(
+    hass: HomeAssistant,
+    entity_registry: EntityRegistry,
+    device_registry: DeviceRegistry,
+    mock_rpc_device: Mock,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Test if the virtual sensor will be removed if the mode has been changed to a field."""
+    config = deepcopy(mock_rpc_device.config)
+    config["text:200"] = {"name": None, "meta": {"ui": {"view": "field"}}}
+    monkeypatch.setattr(mock_rpc_device, "config", config)
+
+    status = deepcopy(mock_rpc_device.status)
+    status["text:200"] = {"value": "lorem ipsum"}
+    monkeypatch.setattr(mock_rpc_device, "status", status)
+
+    config_entry = await init_integration(hass, 3, skip_setup=True)
+    device_entry = register_device(device_registry, config_entry)
+    entity_id = register_entity(
+        hass,
+        SENSOR_DOMAIN,
+        "test_name_text_200",
+        "text:200-text",
+        config_entry,
+        device_id=device_entry.id,
+    )
+
+    await hass.config_entries.async_setup(config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    entry = entity_registry.async_get(entity_id)
+    assert not entry
+
+
+async def test_rpc_remove_virtual_sensor_when_orphaned(
+    hass: HomeAssistant,
+    entity_registry: EntityRegistry,
+    device_registry: DeviceRegistry,
+    mock_rpc_device: Mock,
+) -> None:
+    """Check whether the virtual sensor will be removed if it has been removed from the device configuration."""
+    config_entry = await init_integration(hass, 3, skip_setup=True)
+    device_entry = register_device(device_registry, config_entry)
+    entity_id = register_entity(
+        hass,
+        SENSOR_DOMAIN,
+        "test_name_text_200",
+        "text:200-text",
+        config_entry,
+        device_id=device_entry.id,
+    )
+
+    await hass.config_entries.async_setup(config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    entry = entity_registry.async_get(entity_id)
+    assert not entry

tests/components/shelly/test_text.py

@@ -0,0 +1,129 @@
+"""Tests for Shelly text platform."""
+
+from copy import deepcopy
+from unittest.mock import Mock
+
+import pytest
+
+from homeassistant.components.text import (
+    ATTR_VALUE,
+    DOMAIN as TEXT_PLATFORM,
+    SERVICE_SET_VALUE,
+)
+from homeassistant.const import ATTR_ENTITY_ID
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.device_registry import DeviceRegistry
+from homeassistant.helpers.entity_registry import EntityRegistry
+
+from . import init_integration, register_device, register_entity
+
+
+@pytest.mark.parametrize(
+    ("name", "entity_id"),
+    [
+        ("Virtual text", "text.test_name_virtual_text"),
+        (None, "text.test_name_text_203"),
+    ],
+)
+async def test_rpc_device_virtual_text(
+    hass: HomeAssistant,
+    entity_registry: EntityRegistry,
+    mock_rpc_device: Mock,
+    monkeypatch: pytest.MonkeyPatch,
+    name: str | None,
+    entity_id: str,
+) -> None:
+    """Test a virtual text for RPC device."""
+    config = deepcopy(mock_rpc_device.config)
+    config["text:203"] = {
+        "name": name,
+        "meta": {"ui": {"view": "field"}},
+    }
+    monkeypatch.setattr(mock_rpc_device, "config", config)
+
+    status = deepcopy(mock_rpc_device.status)
+    status["text:203"] = {"value": "lorem ipsum"}
+    monkeypatch.setattr(mock_rpc_device, "status", status)
+
+    await init_integration(hass, 3)
+
+    state = hass.states.get(entity_id)
+    assert state
+    assert state.state == "lorem ipsum"
+
+    entry = entity_registry.async_get(entity_id)
+    assert entry
+    assert entry.unique_id == "123456789ABC-text:203-text"
+
+    monkeypatch.setitem(mock_rpc_device.status["text:203"], "value", "dolor sit amet")
+    mock_rpc_device.mock_update()
+    assert hass.states.get(entity_id).state == "dolor sit amet"
+
+    monkeypatch.setitem(mock_rpc_device.status["text:203"], "value", "sed do eiusmod")
+    await hass.services.async_call(
+        TEXT_PLATFORM,
+        SERVICE_SET_VALUE,
+        {ATTR_ENTITY_ID: entity_id, ATTR_VALUE: "sed do eiusmod"},
+        blocking=True,
+    )
+    mock_rpc_device.mock_update()
+    assert hass.states.get(entity_id).state == "sed do eiusmod"
+
+
+async def test_rpc_remove_virtual_text_when_mode_label(
+    hass: HomeAssistant,
+    entity_registry: EntityRegistry,
+    device_registry: DeviceRegistry,
+    mock_rpc_device: Mock,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Test if the virtual text will be removed if the mode has been changed to a label."""
+    config = deepcopy(mock_rpc_device.config)
+    config["text:200"] = {"name": None, "meta": {"ui": {"view": "label"}}}
+    monkeypatch.setattr(mock_rpc_device, "config", config)
+
+    status = deepcopy(mock_rpc_device.status)
+    status["text:200"] = {"value": "lorem ipsum"}
+    monkeypatch.setattr(mock_rpc_device, "status", status)
+
+    config_entry = await init_integration(hass, 3, skip_setup=True)
+    device_entry = register_device(device_registry, config_entry)
+    entity_id = register_entity(
+        hass,
+        TEXT_PLATFORM,
+        "test_name_text_200",
+        "text:200-text",
+        config_entry,
+        device_id=device_entry.id,
+    )
+
+    await hass.config_entries.async_setup(config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    entry = entity_registry.async_get(entity_id)
+    assert not entry
+
+
+async def test_rpc_remove_virtual_text_when_orphaned(
+    hass: HomeAssistant,
+    entity_registry: EntityRegistry,
+    device_registry: DeviceRegistry,
+    mock_rpc_device: Mock,
+) -> None:
+    """Check whether the virtual text will be removed if it has been removed from the device configuration."""
+    config_entry = await init_integration(hass, 3, skip_setup=True)
+    device_entry = register_device(device_registry, config_entry)
+    entity_id = register_entity(
+        hass,
+        TEXT_PLATFORM,
+        "test_name_text_200",
+        "text:200-text",
+        config_entry,
+        device_id=device_entry.id,
+    )
+
+    await hass.config_entries.async_setup(config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    entry = entity_registry.async_get(entity_id)
+    assert not entry