bosch_shc: add number platform (Micromodule Relay pulse length) (#183136)

Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>

homeassistant/components/bosch_shc/__init__.py

@@ -19,6 +19,7 @@ PLATFORMS = [
     Platform.BINARY_SENSOR,
     Platform.COVER,
     Platform.EVENT,
+    Platform.NUMBER,
     Platform.SENSOR,
     Platform.SWITCH,
     Platform.VALVE,

homeassistant/components/bosch_shc/number.py

@@ -0,0 +1,137 @@
+"""Platform for number integration."""
+
+from collections.abc import Callable, Coroutine
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any, override
+
+from boschshcpy import SHCMicromoduleRelay
+from boschshcpy.device import SHCDevice
+
+from homeassistant.components.number import (
+    NumberDeviceClass,
+    NumberEntity,
+    NumberEntityDescription,
+    NumberMode,
+)
+from homeassistant.const import EntityCategory, UnitOfTime
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+
+from . import BoschConfigEntry
+from .entity import SHCEntity
+
+PARALLEL_UPDATES = 1
+
+
+@dataclass(frozen=True, kw_only=True)
+class SHCNumberEntityDescription[_DeviceT: SHCDevice](NumberEntityDescription):
+    """Describes a SHC number entity."""
+
+    value_fn: Callable[[_DeviceT], float | None]
+    set_value_fn: Callable[[_DeviceT, float], Coroutine[Any, Any, None]]
+
+
+def _impulse_length_value_fn(device: SHCMicromoduleRelay) -> float | None:
+    # ImpulseSwitchService.impulse_length indexes the raw state dict
+    # directly, so a partial poll that omits the field raises KeyError,
+    # not just AttributeError.
+    try:
+        raw = device.impulse_length
+    except AttributeError, KeyError:
+        return None
+    if raw is None:
+        return None
+    return float(raw) / 10.0
+
+
+async def _impulse_length_set_value_fn(
+    device: SHCMicromoduleRelay, value: float
+) -> None:
+    await device.async_set_impulse_length(round(value * 10))
+
+
+IMPULSE_LENGTH = "impulse_length"
+
+NUMBER_TYPES: dict[str, SHCNumberEntityDescription] = {
+    IMPULSE_LENGTH: SHCNumberEntityDescription[SHCMicromoduleRelay](
+        key=IMPULSE_LENGTH,
+        translation_key=IMPULSE_LENGTH,
+        entity_category=EntityCategory.CONFIG,
+        device_class=NumberDeviceClass.DURATION,
+        native_unit_of_measurement=UnitOfTime.SECONDS,
+        native_min_value=0.1,
+        native_max_value=60.0,
+        native_step=0.1,
+        mode=NumberMode.BOX,
+        value_fn=_impulse_length_value_fn,
+        set_value_fn=_impulse_length_set_value_fn,
+    ),
+}
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: BoschConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up the SHC number platform."""
+    session = config_entry.runtime_data
+    shc_info = session.information
+    if TYPE_CHECKING:
+        assert shc_info is not None and shc_info.unique_id is not None
+
+    entities: list[SHCNumber] = []
+    for device in session.device_helper.micromodule_impulse_relays:
+        # KeyError means a partial poll temporarily omits the field, not that
+        # the device is unsupported, so the entity is still created with an
+        # unknown state until a later poll or callback fills it in.
+        try:
+            supported = device.impulse_length is not None
+        except KeyError:
+            supported = True
+        if not supported:
+            continue
+        entities.append(
+            SHCNumber(
+                hass=hass,
+                device=device,
+                parent_id=shc_info.unique_id,
+                entry_id=config_entry.entry_id,
+                description=NUMBER_TYPES[IMPULSE_LENGTH],
+            )
+        )
+
+    async_add_entities(entities)
+
+
+class SHCNumber[_DeviceT: SHCDevice](SHCEntity, NumberEntity):
+    """Generic SHC number entity, driven by a SHCNumberEntityDescription."""
+
+    entity_description: SHCNumberEntityDescription[_DeviceT]
+    _device: _DeviceT
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        device: _DeviceT,
+        parent_id: str,
+        entry_id: str,
+        description: SHCNumberEntityDescription[_DeviceT],
+    ) -> None:
+        """Initialize the number entity."""
+        self.entity_description = description
+        super().__init__(
+            hass=hass, device=device, parent_id=parent_id, entry_id=entry_id
+        )
+        self._attr_unique_id = f"{device.serial}_{description.key}"
+
+    @property
+    @override
+    def native_value(self) -> float | None:
+        """Return the current value."""
+        return self.entity_description.value_fn(self._device)
+
+    @override
+    async def async_set_native_value(self, value: float) -> None:
+        """Set a new value, writing it to the device."""
+        await self.entity_description.set_value_fn(self._device, value)

homeassistant/components/bosch_shc/strings.json

@@ -59,6 +59,11 @@
         }
       }
     },
+    "number": {
+      "impulse_length": {
+        "name": "Pulse length"
+      }
+    },
     "sensor": {
       "air_quality": {
         "name": "Air quality"

tests/components/bosch_shc/conftest.py

@@ -3,7 +3,7 @@
 from collections.abc import Generator
 from types import SimpleNamespace
 from typing import Any
-from unittest.mock import MagicMock, create_autospec, patch
+from unittest.mock import MagicMock, PropertyMock, create_autospec, patch
 
 from boschshcpy import (
     AlarmService,
@@ -360,6 +360,8 @@ def micromodule_relay_device(
     supports_switch_configuration: bool = False,
     swap_inputs: bool = False,
     swap_outputs: bool = False,
+    impulse_length: float | None = None,
+    impulse_length_raises_key_error: bool = False,
 ) -> SHCMicromoduleRelay:
     """Build a minimal device double for the micromodule_relays bucket."""
     device = create_autospec(SHCMicromoduleRelay, instance=True, spec_set=True)
@@ -376,6 +378,12 @@ def micromodule_relay_device(
     device.supports_switch_configuration = supports_switch_configuration
     device.swap_inputs = swap_inputs
     device.swap_outputs = swap_outputs
+    if impulse_length_raises_key_error:
+        type(device).impulse_length = PropertyMock(
+            side_effect=KeyError("impulseLength")
+        )
+    else:
+        device.impulse_length = impulse_length
     return device
 
 

tests/components/bosch_shc/test_number.py

@@ -0,0 +1,101 @@
+"""Tests for the Bosch SHC number platform."""
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from homeassistant.components.number import (
+    ATTR_VALUE,
+    DOMAIN as NUMBER_DOMAIN,
+    SERVICE_SET_VALUE,
+)
+from homeassistant.const import ATTR_ENTITY_ID
+from homeassistant.core import HomeAssistant
+
+from .conftest import micromodule_relay_device, setup_integration
+
+from tests.common import MockConfigEntry
+
+IMPULSE_LENGTH_ENTITY_ID = "number.relay_pulse_length"
+
+
+@pytest.mark.parametrize(
+    "device_buckets",
+    [{"micromodule_impulse_relays": [micromodule_relay_device(impulse_length=50)]}],
+    indirect=True,
+)
+@pytest.mark.usefixtures("mock_session")
+async def test_micromodule_impulse_relay_impulse_length_value(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """The impulse length is reported, converted from tenths of a second."""
+    await setup_integration(hass, mock_config_entry)
+
+    state = hass.states.get(IMPULSE_LENGTH_ENTITY_ID)
+    assert state is not None
+    assert state.state == "5.0"
+
+
+@pytest.mark.parametrize(
+    "device_buckets",
+    [{"micromodule_impulse_relays": [micromodule_relay_device(impulse_length=50)]}],
+    indirect=True,
+)
+@pytest.mark.usefixtures("mock_session")
+async def test_micromodule_impulse_relay_impulse_length_set_value(
+    hass: HomeAssistant,
+    mock_session: MagicMock,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Setting a value writes it to the device, converted to tenths of a second."""
+    await setup_integration(hass, mock_config_entry)
+    device = mock_session.device_helper.micromodule_impulse_relays[0]
+
+    await hass.services.async_call(
+        NUMBER_DOMAIN,
+        SERVICE_SET_VALUE,
+        {ATTR_ENTITY_ID: IMPULSE_LENGTH_ENTITY_ID, ATTR_VALUE: 2.5},
+        blocking=True,
+    )
+    device.async_set_impulse_length.assert_awaited_once_with(25)
+
+
+@pytest.mark.parametrize(
+    "device_buckets",
+    [{"micromodule_impulse_relays": [micromodule_relay_device(impulse_length=None)]}],
+    indirect=True,
+)
+@pytest.mark.usefixtures("mock_session")
+async def test_micromodule_impulse_relay_no_impulse_length_support(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """No number entity is created for a relay without impulse-length support."""
+    await setup_integration(hass, mock_config_entry)
+
+    assert hass.states.get(IMPULSE_LENGTH_ENTITY_ID) is None
+
+
+@pytest.mark.parametrize(
+    "device_buckets",
+    [
+        {
+            "micromodule_impulse_relays": [
+                micromodule_relay_device(impulse_length_raises_key_error=True)
+            ]
+        }
+    ],
+    indirect=True,
+)
+@pytest.mark.usefixtures("mock_session")
+async def test_micromodule_impulse_relay_impulse_length_partial_poll(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """The entity is still created if the initial poll omits the field."""
+    await setup_integration(hass, mock_config_entry)
+
+    state = hass.states.get(IMPULSE_LENGTH_ENTITY_ID)
+    assert state is not None
+    assert state.state == "unknown"