Use configured speed ranges for HomeSeer FC200+ fan controllers in zwave_js

homeassistant/components/zwave_js/discovery.py

@@ -44,6 +44,7 @@
 from .const import LOGGER
 from .discovery_data_template import (
     BaseDiscoverySchemaDataTemplate,
+    ConfigurableFanSpeedDataTemplate,
     CoverTiltDataTemplate,
     DynamicCurrentTempClimateDataTemplate,
     NumericSensorDataTemplate,
@@ -259,6 +260,21 @@ def get_config_parameter_discovery_schema(
             type={"number"},
         ),
     ),
+    # HomeSeer HS-FC200+
+    ZWaveDiscoverySchema(
+        platform="fan",
+        hint="configured_fan_speed",
+        manufacturer_id={0x000C},
+        product_id={0x0001},
+        product_type={0x0203},
+        primary_value=SWITCH_MULTILEVEL_CURRENT_VALUE_SCHEMA,
+        data_template=ConfigurableFanSpeedDataTemplate(
+            configuration_option=ZwaveValueID(
+                5, CommandClass.CONFIGURATION, endpoint=0
+            ),
+            configuration_value_to_speeds={0: [33, 66, 99], 1: [24, 49, 74, 99]},
+        ),
+    ),
     # Fibaro Shutter Fibaro FGR222
     ZWaveDiscoverySchema(
         platform="cover",

homeassistant/components/zwave_js/discovery_data_template.py

@@ -3,6 +3,7 @@
 
 from collections.abc import Iterable
 from dataclasses import dataclass, field
+import logging
 from typing import Any
 
 from zwave_js_server.const import CommandClass
@@ -76,7 +77,11 @@
     MultilevelSensorType,
 )
 from zwave_js_server.model.node import Node as ZwaveNode
-from zwave_js_server.model.value import Value as ZwaveValue, get_value_id
+from zwave_js_server.model.value import (
+    ConfigurationValue as ZwaveConfigurationValue,
+    Value as ZwaveValue,
+    get_value_id,
+)
 from zwave_js_server.util.command_class.meter import get_meter_scale_type
 from zwave_js_server.util.command_class.multilevel_sensor import (
     get_multilevel_sensor_scale_type,
@@ -218,6 +223,8 @@
     IRRADIATION_WATTS_PER_SQUARE_METER: UNIT_WATT_PER_SQUARE_METER,
 }
 
+_LOGGER = logging.getLogger(__name__)
+
 
 @dataclass
 class ZwaveValueID:
@@ -422,3 +429,98 @@ def values_to_watch(self, resolved_data: dict[str, Any]) -> Iterable[ZwaveValue]
     def current_tilt_value(resolved_data: dict[str, Any]) -> ZwaveValue | None:
         """Get current tilt ZwaveValue from resolved data."""
         return resolved_data["tilt_value"]
+
+
+@dataclass
+class FanSpeedDataTemplate:
+    """Mixin to define get_speed_config."""
+
+    def get_speed_config(self, resolved_data: dict[str, Any]) -> list[int] | None:
+        """
+        Get the fan speed configuration for this device.
+
+        Values should indicate the highest allowed device setting for each
+        actual speed, and should be sorted in ascending order.
+
+        Empty lists are not permissible.
+        """
+        # pylint: disable=no-self-use
+        raise NotImplementedError
+
+
+@dataclass
+class ConfigurableFanSpeedValueMix:
+    """Mixin data class for defining configurable fan speeds."""
+
+    configuration_option: ZwaveValueID
+    configuration_value_to_speeds: dict[int, list[int]]
+
+    def __post_init__(self) -> None:
+        """
+        Validate inputs.
+
+        These inputs are hardcoded in `discovery.py`, so these checks should
+        only fail due to developer error.
+        """
+        for speeds in self.configuration_value_to_speeds.values():
+            assert len(speeds) > 0
+            assert sorted(speeds) == speeds
+
+
+@dataclass
+class ConfigurableFanSpeedDataTemplate(
+    BaseDiscoverySchemaDataTemplate, FanSpeedDataTemplate, ConfigurableFanSpeedValueMix
+):
+    """
+    Gets fan speeds based on a configuration value.
+
+    Example:
+      ZWaveDiscoverySchema(
+          platform="fan",
+          hint="configured_fan_speed",
+          ...
+          data_template=ConfigurableFanSpeedDataTemplate(
+            configuration_option=ZwaveValueID(
+                5, CommandClass.CONFIGURATION, endpoint=0
+            ),
+            configuration_value_to_speeds={0: [32, 65, 99], 1: [24, 49, 74, 99]},
+          ),
+      ),
+
+    `configuration_option` is a reference to the setting that determines how
+    many speeds are supported.
+
+    `configuration_value_to_speeds` maps the values from `configuration_option`
+    to a list of speeds.  The specified speeds indicate the maximum setting on
+    the underlying switch for each actual speed.
+    """
+
+    def resolve_data(self, value: ZwaveValue) -> dict[str, ZwaveConfigurationValue]:
+        """Resolve helper class data for a discovered value."""
+        zwave_value: ZwaveValue = self._get_value_from_id(
+            value.node, self.configuration_option
+        )
+        return {"configuration_value": zwave_value}
+
+    def values_to_watch(self, resolved_data: dict[str, Any]) -> Iterable[ZwaveValue]:
+        """Return list of all ZwaveValues that should be watched."""
+        return [
+            resolved_data["configuration_value"],
+        ]
+
+    def get_speed_config(
+        self, resolved_data: dict[str, ZwaveConfigurationValue]
+    ) -> list[int] | None:
+        """Get current speed configuration from resolved data."""
+        zwave_value: ZwaveValue = resolved_data["configuration_value"]
+
+        if zwave_value.value is None:
+            _LOGGER.warning("Unable to read fan speed configuration value")
+            return None
+
+        speed_config = self.configuration_value_to_speeds.get(zwave_value.value)
+        if speed_config is None:
+            _LOGGER.warning("Unrecognized speed configuration value")
+            return None
+
+        return speed_config

homeassistant/components/zwave_js/fan.py

@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 import math
-from typing import Any
+from typing import Any, cast
 
 from zwave_js_server.client import Client as ZwaveClient
 from zwave_js_server.const import TARGET_VALUE_PROPERTY
@@ -24,11 +24,12 @@
 
 from .const import DATA_CLIENT, DOMAIN
 from .discovery import ZwaveDiscoveryInfo
+from .discovery_data_template import FanSpeedDataTemplate
 from .entity import ZWaveBaseEntity
 
 SUPPORTED_FEATURES = SUPPORT_SET_SPEED
 
-SPEED_RANGE = (1, 99)  # off is not included
+DEFAULT_SPEED_RANGE = (1, 99)  # off is not included
 
 
 async def async_setup_entry(
@@ -43,7 +44,11 @@ async def async_setup_entry(
     def async_add_fan(info: ZwaveDiscoveryInfo) -> None:
         """Add Z-Wave fan."""
         entities: list[ZWaveBaseEntity] = []
-        entities.append(ZwaveFan(config_entry, client, info))
+        if info.platform_hint == "configured_fan_speed":
+            entities.append(ConfiguredSpeedRangeZwaveFan(config_entry, client, info))
+        else:
+            entities.append(ZwaveFan(config_entry, client, info))
+
         async_add_entities(entities)
 
     config_entry.async_on_unload(
@@ -58,19 +63,23 @@ def async_add_fan(info: ZwaveDiscoveryInfo) -> None:
 class ZwaveFan(ZWaveBaseEntity, FanEntity):
     """Representation of a Z-Wave fan."""
 
-    async def async_set_percentage(self, percentage: int | None) -> None:
-        """Set the speed percentage of the fan."""
-        target_value = self.get_zwave_value(TARGET_VALUE_PROPERTY)
+    def __init__(
+        self, config_entry: ConfigEntry, client: ZwaveClient, info: ZwaveDiscoveryInfo
+    ) -> None:
+        """Initialize the fan."""
+        super().__init__(config_entry, client, info)
+        self._target_value = self.get_zwave_value(TARGET_VALUE_PROPERTY)
 
-        if percentage is None:
-            # Value 255 tells device to return to previous value
-            zwave_speed = 255
-        elif percentage == 0:
+    async def async_set_percentage(self, percentage: int) -> None:
+        """Set the speed percentage of the fan."""
+        if percentage == 0:
             zwave_speed = 0
         else:
-            zwave_speed = math.ceil(percentage_to_ranged_value(SPEED_RANGE, percentage))
+            zwave_speed = math.ceil(
+                percentage_to_ranged_value(DEFAULT_SPEED_RANGE, percentage)
+            )
 
-        await self.info.node.async_set_value(target_value, zwave_speed)
+        await self.info.node.async_set_value(self._target_value, zwave_speed)
 
     async def async_turn_on(
         self,
@@ -80,12 +89,15 @@ async def async_turn_on(
         **kwargs: Any,
     ) -> None:
         """Turn the device on."""
-        await self.async_set_percentage(percentage)
+        if percentage is None:
+            # Value 255 tells device to return to previous value
+            await self.info.node.async_set_value(self._target_value, 255)
+        else:
+            await self.async_set_percentage(percentage)
 
     async def async_turn_off(self, **kwargs: Any) -> None:
         """Turn the device off."""
-        target_value = self.get_zwave_value(TARGET_VALUE_PROPERTY)
-        await self.info.node.async_set_value(target_value, 0)
+        await self.info.node.async_set_value(self._target_value, 0)
 
     @property
     def is_on(self) -> bool | None:  # type: ignore
@@ -101,7 +113,9 @@ def percentage(self) -> int | None:
         if self.info.primary_value.value is None:
             # guard missing value
             return None
-        return ranged_value_to_percentage(SPEED_RANGE, self.info.primary_value.value)
+        return ranged_value_to_percentage(
+            DEFAULT_SPEED_RANGE, self.info.primary_value.value
+        )
 
     @property
     def percentage_step(self) -> float:
@@ -111,9 +125,103 @@ def percentage_step(self) -> float:
     @property
     def speed_count(self) -> int:
         """Return the number of speeds the fan supports."""
-        return int_states_in_range(SPEED_RANGE)
+        return int_states_in_range(DEFAULT_SPEED_RANGE)
 
     @property
     def supported_features(self) -> int:
         """Flag supported features."""
         return SUPPORTED_FEATURES
+
+
+class ConfiguredSpeedRangeZwaveFan(ZwaveFan):
+    """A Zwave fan with a configured speed range (e.g., 1-24 is low)."""
+
+    def __init__(
+        self, config_entry: ConfigEntry, client: ZwaveClient, info: ZwaveDiscoveryInfo
+    ) -> None:
+        """Initialize the fan."""
+        super().__init__(config_entry, client, info)
+        self.data_template = cast(
+            FanSpeedDataTemplate, self.info.platform_data_template
+        )
+
+    async def async_set_percentage(self, percentage: int) -> None:
+        """Set the speed percentage of the fan."""
+        zwave_speed = self.percentage_to_zwave_speed(percentage)
+        await self.info.node.async_set_value(self._target_value, zwave_speed)
+
+    @property
+    def available(self) -> bool:
+        """Return whether the entity is available."""
+        return super().available and self.has_speed_configuration
+
+    @property
+    def percentage(self) -> int | None:
+        """Return the current speed percentage."""
+        if self.info.primary_value.value is None:
+            # guard missing value
+            return None
+
+        return self.zwave_speed_to_percentage(self.info.primary_value.value)
+
+    @property
+    def percentage_step(self) -> float:
+        """Return the step size for percentage."""
+        # This is the same implementation as the base fan type, but
+        # it needs to be overridden here because the ZwaveFan does
+        # something different for fans with unknown speeds.
+        return 100 / self.speed_count
+
+    @property
+    def has_speed_configuration(self) -> bool:
+        """Check if the speed configuration is valid."""
+        return self.data_template.get_speed_config(self.info.platform_data) is not None
+
+    @property
+    def speed_configuration(self) -> list[int]:
+        """Return the speed configuration for this fan."""
+        speed_configuration = self.data_template.get_speed_config(
+            self.info.platform_data
+        )
+
+        # Entity should be unavailable if this isn't set
+        assert speed_configuration is not None
+
+        return speed_configuration
+
+    @property
+    def speed_count(self) -> int:
+        """Return the number of speeds the fan supports."""
+        return len(self.speed_configuration)
+
+    def percentage_to_zwave_speed(self, percentage: int) -> int:
+        """Map a percentage to a ZWave speed."""
+        if percentage == 0:
+            return 0
+
+        # Since the percentage steps are computed with rounding, we have to
+        # search to find the appropriate speed.
+        for speed_limit in self.speed_configuration:
+            step_percentage = self.zwave_speed_to_percentage(speed_limit)
+            if percentage <= step_percentage:
+                return speed_limit
+
+        # This shouldn't actually happen; the last entry in
+        # `self.speed_configuration` should map to 100%.
+        return self.speed_configuration[-1]
+
+    def zwave_speed_to_percentage(self, zwave_speed: int) -> int:
+        """Convert a Zwave speed to a percentage."""
+        if zwave_speed == 0:
+            return 0
+
+        percentage = 0.0
+        for speed_limit in self.speed_configuration:
+            percentage += self.percentage_step
+            if zwave_speed <= speed_limit:
+                break
+
+        # This choice of rounding function is to provide consistency with how
+        # the UI handles steps e.g., for a 3-speed fan, you get steps at 33,
+        # 67, and 100.
+        return round(percentage)

tests/components/zwave_js/conftest.py

@@ -332,6 +332,12 @@ def in_wall_smart_fan_control_state_fixture():
     return json.loads(load_fixture("zwave_js/in_wall_smart_fan_control_state.json"))
 
 
+@pytest.fixture(name="hs_fc200_state", scope="session")
+def hs_fc200_state_fixture():
+    """Load the HS FC200+ node state fixture data."""
+    return json.loads(load_fixture("zwave_js/fan_hs_fc200_state.json"))
+
+
 @pytest.fixture(name="gdc_zw062_state", scope="session")
 def motorized_barrier_cover_state_fixture():
     """Load the motorized barrier cover node state fixture data."""
@@ -697,6 +703,14 @@ def in_wall_smart_fan_control_fixture(client, in_wall_smart_fan_control_state):
     return node
 
 
+@pytest.fixture(name="hs_fc200")
+def hs_fc200_fixture(client, hs_fc200_state):
+    """Mock a fan node."""
+    node = Node(client, copy.deepcopy(hs_fc200_state))
+    client.driver.controller.nodes[node.node_id] = node
+    return node
+
+
 @pytest.fixture(name="null_name_check")
 def null_name_check_fixture(client, null_name_check_state):
     """Mock a node with no name."""