From b2dee94fe5fc53cc00119fe694e2bd574c658b07 Mon Sep 17 00:00:00 2001 From: Raman Gupta <7243222+raman325@users.noreply.github.com> Date: Sun, 15 Oct 2023 19:17:52 -0400 Subject: [PATCH] Add node.async_set_raw_config_parameter_value command --- zwave_js_server/const/__init__.py | 14 ++++- zwave_js_server/model/node/__init__.py | 78 +++++++++++++++++++++++++- zwave_js_server/model/value.py | 51 ++++++++++++++++- 3 files changed, 138 insertions(+), 5 deletions(-) diff --git a/zwave_js_server/const/__init__.py b/zwave_js_server/const/__init__.py index fa4afcdb3..8d84eb0df 100644 --- a/zwave_js_server/const/__init__.py +++ b/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 diff --git a/zwave_js_server/model/node/__init__.py b/zwave_js_server/model/node/__init__.py index 0980e3002..b8f2628ec 100644 --- a/zwave_js_server/model/node/__init__.py +++ b/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( diff --git a/zwave_js_server/model/value.py b/zwave_js_server/model/value.py index bec3c9ee8..489d8621d 100644 --- a/zwave_js_server/model/value.py +++ b/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."""