Add node.async_set_raw_config_parameter_value command

zwave_js_server/const/__init__.py

@@ -11,9 +11,9 @@
 __version__ = metadata.version(PACKAGE_NAME)
 
 # minimal server schema version we can handle
-MIN_SERVER_SCHEMA_VERSION = 32
+MIN_SERVER_SCHEMA_VERSION = 33
 # max server schema version we can handle (and our code is compatible with)
-MAX_SERVER_SCHEMA_VERSION = 32
+MAX_SERVER_SCHEMA_VERSION = 33
 
 VALUE_UNKNOWN = "unknown"
 
@@ -494,3 +494,13 @@ class ControllerStatus(IntEnum):
     UNRESPONSIVE = 1
     # The controller is unable to transmit
     JAMMED = 2
+
+
+class SupervisionStatus(IntEnum):
+    """Enum for all known supervision statuses."""
+
+    # https://github.com/zwave-js/node-zwave-js/blob/cc_api_options/packages/core/src/consts/Transmission.ts#L304
+    NO_SUPPORT = 0
+    WORKING = 1
+    FAIL = 2
+    SUCCESS = 255

zwave_js_server/model/node/__init__.py

@@ -5,7 +5,7 @@
 import copy
 import logging
 from datetime import datetime
-from typing import TYPE_CHECKING, Any, cast
+from typing import TYPE_CHECKING, Any, Literal, cast
 
 from ...const import (
     INTERVIEW_FAILED,
@@ -39,8 +39,10 @@
 )
 from ..value import (
     ConfigurationValue,
+    ConfigurationValueFormat,
     MetaDataType,
     SetValueResult,
+    SupervisionResult,
     Value,
     ValueDataType,
     ValueMetadata,
@@ -941,6 +943,80 @@ async def async_has_device_config_changed(self) -> bool | None:
             return cast(bool, changed)
         return None
 
+    async def async_set_raw_config_parameter_value(
+        self,
+        new_value: int | str,
+        property_: int | str,
+        property_key: int | str | None = None,
+        value_size: Literal[1, 2, 4] | None = None,
+        value_format: ConfigurationValueFormat | None = None,
+    ) -> SupervisionResult | None:
+        """Send setRawConfigParameterValue."""
+        if (property_is_name := isinstance(property_, str)) or (
+            property_key_is_name := isinstance(property_key, str)
+        ):
+            attr_to_value = {}
+            key = "property_name" if property_is_name else "property_"
+            attr_to_value[key] = property_
+            key = "property_key_name" if property_key_is_name else "property_key"
+            attr_to_value[key] = property_key
+            try:
+                zwave_value = next(
+                    config_value
+                    for config_value in self.get_configuration_values().values()
+                    if all(
+                        getattr(config_value, attr_name) == value
+                        for attr_name, value in attr_to_value.items()
+                    )
+                )
+            except StopIteration:
+                raise NotFoundError(
+                    f"Configuration parameter with parameter {property_} and bitmask "
+                    f"{property_key} on node {self} could not be found"
+                ) from None
+
+        if not isinstance(new_value, str):
+            value = new_value
+        else:
+            try:
+                value = int(next(
+                    k for k, v in zwave_value.metadata.states.items() if v == new_value
+                ))
+            except StopIteration:
+                raise NotFoundError(
+                    f"Configuration parameter {zwave_value.value_id} does not have "
+                    f"{new_value} as a valid state. If this is a valid call, you must "
+                    "use the state key instead of the string."
+                ) from None
+
+        if (value_size is not None and value_format is None) or (
+            value_size is None and value_format is not None
+        ):
+            raise ValueError(
+                "value_size and value_format must either both be included or not "
+                "included"
+            )
+
+        options = {
+            "value": value,
+            "parameter": zwave_value.property_,
+            "bitMask": zwave_value.property_key,
+            "valueSize": value_size,
+            "valueFormat": value_format,
+        }
+
+        data = await self.async_send_command(
+            "setRawConfigParameterValue",
+            options={k: v for k, v in options.items() if v is not None},
+            require_schema=33,
+        )
+
+        result: int | None = data.get("result")
+
+        if result is not None:
+            return SupervisionResult(result)
+        return None
+
     def handle_test_powerlevel_progress(self, event: Event) -> None:
         """Process a test power level progress event."""
         event.data["test_power_level_progress"] = TestPowerLevelProgress(

zwave_js_server/model/value.py

@@ -2,10 +2,16 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from enum import StrEnum
+from enum import IntEnum, StrEnum
 from typing import TYPE_CHECKING, Any, TypedDict
 
-from ..const import VALUE_UNKNOWN, CommandClass, ConfigurationValueType, SetValueStatus
+from ..const import (
+    VALUE_UNKNOWN,
+    CommandClass,
+    ConfigurationValueType,
+    SetValueStatus,
+    SupervisionStatus,
+)
 from ..event import Event
 from ..util.helpers import parse_buffer
 from .duration import Duration, DurationDataType
@@ -301,6 +307,47 @@ class ValueNotification(Value):
     # format is the same as a Value message, subclassed for easier identifying and future use
 
 
+class ConfigurationValueFormat(IntEnum):
+    """Enum of all known configuration value formats."""
+
+    # https://github.com/zwave-js/node-zwave-js/blob/cc_api_options/packages/core/src/values/Metadata.ts#L157
+    SIGNED_INTEGER = 0
+    UNSIGNED_INTEGER = 1
+    ENUMERATED = 2
+    BIT_FIELD = 3
+
+
+class SupervisionResultDataType(TypedDict, total=False):
+    """Represent a Supervision result data dict type."""
+
+    # https://github.com/zwave-js/node-zwave-js/blob/cc_api_options/packages/core/src/consts/Transmission.ts#L311
+    status: int
+    remainingDuration: DurationDataType  # optional unless status is 1 (working)
+
+
+@dataclass
+class SupervisionResult:
+    """Represent a Supervision result type."""
+
+    data: SupervisionResultDataType
+    status: SupervisionStatus = field(init=False)
+    remaining_duration: Duration | None = field(init=False, default=None)
+
+    def __post_init__(self) -> None:
+        """Post initialization."""
+        self.status = SupervisionStatus(self.data["status"])
+        if remaining_duration := self.data.get("remainingDuration"):
+            self.remaining_duration = Duration(remaining_duration)
+
+        if self.status == SupervisionStatus.WORKING ^ bool(
+            self.remaining_duration is None
+        ):
+            raise ValueError(
+                "SupervisionStatus of WORKING requires a remaining duration, all "
+                "other statuses don't include it"
+            )
+
+
 class ConfigurationValue(Value):
     """Model for a Configuration Value."""