Add myuplink number platform (#111154)

* Add number platform

* Use constant for SERVICE_SET_VALUE

homeassistant/components/myuplink/__init__.py

@@ -22,6 +22,7 @@ from .coordinator import MyUplinkDataCoordinator
 
 PLATFORMS: list[Platform] = [
     Platform.BINARY_SENSOR,
+    Platform.NUMBER,
     Platform.SENSOR,
     Platform.SWITCH,
     Platform.UPDATE,

homeassistant/components/myuplink/helpers.py

@@ -2,10 +2,15 @@
 
 from myuplink import DevicePoint
 
+from homeassistant.components.number import NumberEntityDescription
+from homeassistant.components.sensor import SensorEntityDescription
 from homeassistant.const import Platform
 
 
-def find_matching_platform(device_point: DevicePoint) -> Platform:
+def find_matching_platform(
+    device_point: DevicePoint,
+    description: SensorEntityDescription | NumberEntityDescription | None = None,
+) -> Platform:
     """Find entity platform for a DevicePoint."""
     if (
         len(device_point.enum_values) == 2
@@ -16,4 +21,13 @@ def find_matching_platform(device_point: DevicePoint) -> Platform:
             return Platform.SWITCH
         return Platform.BINARY_SENSOR
 
+    if (
+        description
+        and description.native_unit_of_measurement == "DM"
+        or (device_point.raw["maxValue"] and device_point.raw["minValue"])
+    ):
+        if device_point.writable:
+            return Platform.NUMBER
+        return Platform.SENSOR
+
     return Platform.SENSOR

homeassistant/components/myuplink/number.py

@@ -0,0 +1,132 @@
+"""Number entity for myUplink."""
+
+
+from aiohttp import ClientError
+from myuplink import DevicePoint
+
+from homeassistant.components.number import NumberEntity, NumberEntityDescription
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from . import MyUplinkDataCoordinator
+from .const import DOMAIN
+from .entity import MyUplinkEntity
+from .helpers import find_matching_platform
+
+DEVICE_POINT_UNIT_DESCRIPTIONS: dict[str, NumberEntityDescription] = {
+    "DM": NumberEntityDescription(
+        key="degree_minutes",
+        icon="mdi:thermometer-lines",
+        native_unit_of_measurement="DM",
+    ),
+}
+
+CATEGORY_BASED_DESCRIPTIONS: dict[str, dict[str, NumberEntityDescription]] = {
+    "NIBEF": {
+        "40940": NumberEntityDescription(
+            key="degree_minutes",
+            icon="mdi:thermometer-lines",
+            native_unit_of_measurement="DM",
+        ),
+    },
+}
+
+
+def get_description(device_point: DevicePoint) -> NumberEntityDescription | None:
+    """Get description for a device point.
+
+    Priorities:
+    1. Category specific prefix e.g "NIBEF"
+    2. Global parameter_unit e.g. "DM"
+    3. Default to None
+    """
+    prefix, _, _ = device_point.category.partition(" ")
+    description = CATEGORY_BASED_DESCRIPTIONS.get(prefix, {}).get(
+        device_point.parameter_id
+    )
+
+    if description is None:
+        description = DEVICE_POINT_UNIT_DESCRIPTIONS.get(device_point.parameter_unit)
+
+    return description
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up myUplink number."""
+    entities: list[NumberEntity] = []
+    coordinator: MyUplinkDataCoordinator = hass.data[DOMAIN][config_entry.entry_id]
+
+    # Setup device point number entities
+    for device_id, point_data in coordinator.data.points.items():
+        for point_id, device_point in point_data.items():
+            description = get_description(device_point)
+            if find_matching_platform(device_point, description) == Platform.NUMBER:
+                entities.append(
+                    MyUplinkNumber(
+                        coordinator=coordinator,
+                        device_id=device_id,
+                        device_point=device_point,
+                        entity_description=description,
+                        unique_id_suffix=point_id,
+                    )
+                )
+
+    async_add_entities(entities)
+
+
+class MyUplinkNumber(MyUplinkEntity, NumberEntity):
+    """Representation of a myUplink number entity."""
+
+    def __init__(
+        self,
+        coordinator: MyUplinkDataCoordinator,
+        device_id: str,
+        device_point: DevicePoint,
+        entity_description: NumberEntityDescription | None,
+        unique_id_suffix: str,
+    ) -> None:
+        """Initialize the number."""
+        super().__init__(
+            coordinator=coordinator,
+            device_id=device_id,
+            unique_id_suffix=unique_id_suffix,
+        )
+
+        # Internal properties
+        self.point_id = device_point.parameter_id
+        self._attr_name = device_point.parameter_name
+        self._attr_native_min_value = (
+            device_point.raw["minValue"] if device_point.raw["minValue"] else -30000
+        ) * float(device_point.raw.get("scaleValue", 1))
+        self._attr_native_max_value = (
+            device_point.raw["maxValue"] if device_point.raw["maxValue"] else 30000
+        ) * float(device_point.raw.get("scaleValue", 1))
+        self._attr_step_value = device_point.raw.get("stepValue", 20)
+        if entity_description is not None:
+            self.entity_description = entity_description
+
+    @property
+    def native_value(self) -> float:
+        """Number state value."""
+        device_point = self.coordinator.data.points[self.device_id][self.point_id]
+        return float(device_point.value)
+
+    async def async_set_native_value(self, value: float) -> None:
+        """Update the current value."""
+        try:
+            await self.coordinator.api.async_set_device_points(
+                self.device_id, data={self.point_id: str(value)}
+            )
+        except ClientError as err:
+            raise HomeAssistantError(
+                f"Failed to set new value {value} for {self.point_id}/{self.entity_id}"
+            ) from err
+
+        await self.coordinator.async_request_refresh()

tests/components/myuplink/test_number.py

@@ -0,0 +1,87 @@
+"""Tests for myuplink switch module."""
+
+from unittest.mock import MagicMock
+
+from aiohttp import ClientError
+import pytest
+
+from homeassistant.components.number import SERVICE_SET_VALUE
+from homeassistant.const import ATTR_ENTITY_ID, Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers import entity_registry as er
+
+TEST_PLATFORM = Platform.NUMBER
+pytestmark = pytest.mark.parametrize("platforms", [(TEST_PLATFORM,)])
+
+ENTITY_ID = "number.f730_cu_3x400v_degree_minutes"
+ENTITY_FRIENDLY_NAME = "F730 CU 3x400V Degree minutes"
+ENTITY_UID = "batman-r-1234-20240201-123456-aa-bb-cc-dd-ee-ff-40940"
+
+
+async def test_entity_registry(
+    hass: HomeAssistant,
+    entity_registry: er.EntityRegistry,
+    mock_myuplink_client: MagicMock,
+    setup_platform: None,
+) -> None:
+    """Test that the entities are registered in the entity registry."""
+
+    entry = entity_registry.async_get(ENTITY_ID)
+    assert entry.unique_id == ENTITY_UID
+
+
+async def test_attributes(
+    hass: HomeAssistant,
+    mock_myuplink_client: MagicMock,
+    setup_platform: None,
+) -> None:
+    """Test the switch attributes are correct."""
+
+    state = hass.states.get(ENTITY_ID)
+    assert state.state == "-875.0"
+    assert state.attributes == {
+        "friendly_name": ENTITY_FRIENDLY_NAME,
+        "icon": "mdi:thermometer-lines",
+        "min": -3000,
+        "max": 3000,
+        "mode": "auto",
+        "step": 1.0,
+        "unit_of_measurement": "DM",
+    }
+
+
+async def test_set_value(
+    hass: HomeAssistant,
+    mock_myuplink_client: MagicMock,
+    setup_platform: None,
+) -> None:
+    """Test the value of the number entity can be set."""
+
+    await hass.services.async_call(
+        TEST_PLATFORM,
+        SERVICE_SET_VALUE,
+        {ATTR_ENTITY_ID: ENTITY_ID, "value": -125},
+        blocking=True,
+    )
+    await hass.async_block_till_done()
+    mock_myuplink_client.async_set_device_points.assert_called_once()
+
+
+async def test_api_failure(
+    hass: HomeAssistant,
+    mock_myuplink_client: MagicMock,
+    setup_platform: None,
+) -> None:
+    """Test handling of exception from API."""
+
+    with pytest.raises(HomeAssistantError):
+        mock_myuplink_client.async_set_device_points.side_effect = ClientError
+        await hass.services.async_call(
+            TEST_PLATFORM,
+            SERVICE_SET_VALUE,
+            {ATTR_ENTITY_ID: ENTITY_ID, "value": -125},
+            blocking=True,
+        )
+        await hass.async_block_till_done()
+        mock_myuplink_client.async_set_device_points.assert_called_once()