Add additional entities for Shelly BLU TRV

homeassistant/components/shelly/binary_sensor.py

@@ -15,11 +15,12 @@
 )
 from homeassistant.const import STATE_ON, EntityCategory
 from homeassistant.core import HomeAssistant
+from homeassistant.helpers.device_registry import CONNECTION_BLUETOOTH, DeviceInfo
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.restore_state import RestoreEntity
 
 from .const import CONF_SLEEP_PERIOD
-from .coordinator import ShellyConfigEntry
+from .coordinator import ShellyConfigEntry, ShellyRpcCoordinator
 from .entity import (
     BlockEntityDescription,
     RestEntityDescription,
@@ -59,6 +60,36 @@ class RestBinarySensorDescription(RestEntityDescription, BinarySensorEntityDescr
     """Class to describe a REST binary sensor."""
 
 
+class RpcBinarySensor(ShellyRpcAttributeEntity, BinarySensorEntity):
+    """Represent a RPC binary sensor entity."""
+
+    entity_description: RpcBinarySensorDescription
+
+    @property
+    def is_on(self) -> bool:
+        """Return true if RPC sensor state is on."""
+        return bool(self.attribute_value)
+
+
+class RpcBluTrvBinarySensor(RpcBinarySensor):
+    """Represent a RPC BluTrv binary sensor."""
+
+    def __init__(
+        self,
+        coordinator: ShellyRpcCoordinator,
+        key: str,
+        attribute: str,
+        description: RpcBinarySensorDescription,
+    ) -> None:
+        """Initialize."""
+
+        super().__init__(coordinator, key, attribute, description)
+        ble_addr: str = coordinator.device.config[key]["addr"]
+        self._attr_device_info = DeviceInfo(
+            connections={(CONNECTION_BLUETOOTH, ble_addr)}
+        )
+
+
 SENSORS: dict[tuple[str, str], BlockBinarySensorDescription] = {
     ("device", "overtemp"): BlockBinarySensorDescription(
         key="device|overtemp",
@@ -232,6 +263,15 @@ class RestBinarySensorDescription(RestEntityDescription, BinarySensorEntityDescr
         sub_key="value",
         has_entity_name=True,
     ),
+    "calibration": RpcBinarySensorDescription(
+        key="blutrv",
+        sub_key="errors",
+        name="Calibration",
+        device_class=BinarySensorDeviceClass.PROBLEM,
+        value=lambda status, _: False if status is None else "not_calibrated" in status,
+        entity_category=EntityCategory.DIAGNOSTIC,
+        entity_class=RpcBluTrvBinarySensor,
+    ),
 }
 
 
@@ -320,17 +360,6 @@ def is_on(self) -> bool:
         return bool(self.attribute_value)
 
 
-class RpcBinarySensor(ShellyRpcAttributeEntity, BinarySensorEntity):
-    """Represent a RPC binary sensor entity."""
-
-    entity_description: RpcBinarySensorDescription
-
-    @property
-    def is_on(self) -> bool:
-        """Return true if RPC sensor state is on."""
-        return bool(self.attribute_value)
-
-
 class BlockSleepingBinarySensor(
     ShellySleepingBlockAttributeEntity, BinarySensorEntity, RestoreEntity
 ):

homeassistant/components/shelly/entity.py

@@ -196,10 +196,16 @@ def async_setup_rpc_attribute_entities(
             elif description.use_polling_coordinator:
                 if not sleep_period:
                     entities.append(
-                        sensor_class(polling_coordinator, key, sensor_id, description)
+                        get_entity_class(sensor_class, description)(
+                            polling_coordinator, key, sensor_id, description
+                        )
                     )
             else:
-                entities.append(sensor_class(coordinator, key, sensor_id, description))
+                entities.append(
+                    get_entity_class(sensor_class, description)(
+                        coordinator, key, sensor_id, description
+                    )
+                )
     if not entities:
         return
 
@@ -232,7 +238,9 @@ def async_restore_rpc_attribute_entities(
 
         if description := sensors.get(attribute):
             entities.append(
-                sensor_class(coordinator, key, attribute, description, entry)
+                get_entity_class(sensor_class, description)(
+                    coordinator, key, attribute, description, entry
+                )
             )
 
     if not entities:
@@ -293,6 +301,7 @@ class RpcEntityDescription(EntityDescription):
     supported: Callable = lambda _: False
     unit: Callable[[dict], str | None] | None = None
     options_fn: Callable[[dict], list[str]] | None = None
+    entity_class: Callable | None = None
 
 
 @dataclass(frozen=True)
@@ -673,3 +682,13 @@ async def async_update(self) -> None:
             "Entity %s comes from a sleeping device, update is not possible",
             self.entity_id,
         )
+
+
+def get_entity_class(
+    sensor_class: Callable, description: RpcEntityDescription
+) -> Callable:
+    """Return entity class."""
+    if description.entity_class is not None:
+        return description.entity_class
+
+    return sensor_class

homeassistant/components/shelly/icons.json

@@ -12,6 +12,9 @@
       }
     },
     "number": {
+      "external_temperature": {
+        "default": "mdi:thermometer-check"
+      },
       "valve_position": {
         "default": "mdi:pipe-valve"
       }
@@ -29,6 +32,9 @@
       "tilt": {
         "default": "mdi:angle-acute"
       },
+      "valve_position": {
+        "default": "mdi:pipe-valve"
+      },
       "valve_status": {
         "default": "mdi:valve"
       }

homeassistant/components/shelly/number.py

@@ -18,9 +18,10 @@
     NumberMode,
     RestoreNumber,
 )
-from homeassistant.const import PERCENTAGE, EntityCategory
+from homeassistant.const import PERCENTAGE, EntityCategory, UnitOfTemperature
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.device_registry import CONNECTION_BLUETOOTH, DeviceInfo
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.entity_registry import RegistryEntry
 
@@ -57,6 +58,74 @@ class RpcNumberDescription(RpcEntityDescription, NumberEntityDescription):
     min_fn: Callable[[dict], float] | None = None
     step_fn: Callable[[dict], float] | None = None
     mode_fn: Callable[[dict], NumberMode] | None = None
+    method: str
+    method_params_fn: Callable[[int, float], dict]
+
+
+class RpcNumber(ShellyRpcAttributeEntity, NumberEntity):
+    """Represent a RPC number entity."""
+
+    entity_description: RpcNumberDescription
+
+    def __init__(
+        self,
+        coordinator: ShellyRpcCoordinator,
+        key: str,
+        attribute: str,
+        description: RpcNumberDescription,
+    ) -> None:
+        """Initialize sensor."""
+        super().__init__(coordinator, key, attribute, description)
+
+        if description.max_fn is not None:
+            self._attr_native_max_value = description.max_fn(
+                coordinator.device.config[key]
+            )
+        if description.min_fn is not None:
+            self._attr_native_min_value = description.min_fn(
+                coordinator.device.config[key]
+            )
+        if description.step_fn is not None:
+            self._attr_native_step = description.step_fn(coordinator.device.config[key])
+        if description.mode_fn is not None:
+            self._attr_mode = description.mode_fn(coordinator.device.config[key])
+
+    @property
+    def native_value(self) -> float | None:
+        """Return value of number."""
+        if TYPE_CHECKING:
+            assert isinstance(self.attribute_value, float | None)
+
+        return self.attribute_value
+
+    async def async_set_native_value(self, value: float) -> None:
+        """Change the value."""
+        if TYPE_CHECKING:
+            assert isinstance(self._id, int)
+
+        await self.call_rpc(
+            self.entity_description.method,
+            self.entity_description.method_params_fn(self._id, value),
+        )
+
+
+class RpcBluTrvNumber(RpcNumber):
+    """Represent a RPC BluTrv number."""
+
+    def __init__(
+        self,
+        coordinator: ShellyRpcCoordinator,
+        key: str,
+        attribute: str,
+        description: RpcNumberDescription,
+    ) -> None:
+        """Initialize."""
+
+        super().__init__(coordinator, key, attribute, description)
+        ble_addr: str = coordinator.device.config[key]["addr"]
+        self._attr_device_info = DeviceInfo(
+            connections={(CONNECTION_BLUETOOTH, ble_addr)}
+        )
 
 
 NUMBERS: dict[tuple[str, str], BlockNumberDescription] = {
@@ -78,6 +147,25 @@ class RpcNumberDescription(RpcEntityDescription, NumberEntityDescription):
 
 
 RPC_NUMBERS: Final = {
+    "external_temperature": RpcNumberDescription(
+        key="blutrv",
+        sub_key="current_C",
+        translation_key="external_temperature",
+        name="External temperature",
+        native_min_value=-50,
+        native_max_value=50,
+        native_step=0.1,
+        mode=NumberMode.BOX,
+        entity_category=EntityCategory.CONFIG,
+        native_unit_of_measurement=UnitOfTemperature.CELSIUS,
+        method="BluTRV.Call",
+        method_params_fn=lambda idx, value: {
+            "id": idx,
+            "method": "Trv.SetExternalTemperature",
+            "params": {"id": 0, "t_C": value},
+        },
+        entity_class=RpcBluTrvNumber,
+    ),
     "number": RpcNumberDescription(
         key="number",
         sub_key="value",
@@ -92,6 +180,28 @@ class RpcNumberDescription(RpcEntityDescription, NumberEntityDescription):
         unit=lambda config: config["meta"]["ui"]["unit"]
         if config["meta"]["ui"]["unit"]
         else None,
+        method="Number.Set",
+        method_params_fn=lambda idx, value: {"id": idx, "value": value},
+    ),
+    "valve_position": RpcNumberDescription(
+        key="blutrv",
+        sub_key="pos",
+        translation_key="valve_position",
+        name="Valve position",
+        native_min_value=0,
+        native_max_value=100,
+        native_step=1,
+        mode=NumberMode.SLIDER,
+        native_unit_of_measurement=PERCENTAGE,
+        method="BluTRV.Call",
+        method_params_fn=lambda idx, value: {
+            "id": idx,
+            "method": "Trv.SetPosition",
+            "params": {"id": 0, "pos": value},
+        },
+        removal_condition=lambda config, _status, key: config[key].get("enable", True)
+        is True,
+        entity_class=RpcBluTrvNumber,
     ),
 }
 
@@ -190,44 +300,3 @@ async def _set_state_full_path(self, path: str, params: Any) -> Any:
             ) from err
         except InvalidAuthError:
             await self.coordinator.async_shutdown_device_and_start_reauth()
-
-
-class RpcNumber(ShellyRpcAttributeEntity, NumberEntity):
-    """Represent a RPC number entity."""
-
-    entity_description: RpcNumberDescription
-
-    def __init__(
-        self,
-        coordinator: ShellyRpcCoordinator,
-        key: str,
-        attribute: str,
-        description: RpcNumberDescription,
-    ) -> None:
-        """Initialize sensor."""
-        super().__init__(coordinator, key, attribute, description)
-
-        if description.max_fn is not None:
-            self._attr_native_max_value = description.max_fn(
-                coordinator.device.config[key]
-            )
-        if description.min_fn is not None:
-            self._attr_native_min_value = description.min_fn(
-                coordinator.device.config[key]
-            )
-        if description.step_fn is not None:
-            self._attr_native_step = description.step_fn(coordinator.device.config[key])
-        if description.mode_fn is not None:
-            self._attr_mode = description.mode_fn(coordinator.device.config[key])
-
-    @property
-    def native_value(self) -> float | None:
-        """Return value of number."""
-        if TYPE_CHECKING:
-            assert isinstance(self.attribute_value, float | None)
-
-        return self.attribute_value
-
-    async def async_set_native_value(self, value: float) -> None:
-        """Change the value."""
-        await self.call_rpc("Number.Set", {"id": self._id, "value": value})