Validate Modbus write register ranges

Author: rosenrot00

custom_components/solax_modbus/__init__.py

@@ -6,6 +6,7 @@
 import importlib
 import json
 import logging
+import struct
 import time as _mtime
 from dataclasses import dataclass, replace
 from datetime import timedelta
@@ -76,9 +77,11 @@
     REG_HOLDING,
     REG_INPUT,
     REGISTER_F32,
+    REGISTER_INT_RANGES,
     REGISTER_S16,
     REGISTER_S32,
     REGISTER_STR,
+    REGISTER_TYPE_WORDS,
     REGISTER_U8H,
     REGISTER_U8L,
     REGISTER_U16,
@@ -480,6 +483,10 @@ class PendingWrite:
     register_data_type: str | None = None
 
 
+class RegisterEncodingError(HomeAssistantError):
+    """Raised when a value cannot be represented by its Modbus register type."""
+
+
 class SolaXModbusHub:
     """Thread safe wrapper class for pymodbus."""
 
@@ -1366,11 +1373,17 @@ def _validate_write_response(self, response: Any, *, unit: int, address: int, op
             raise HomeAssistantError(f"{self._name}: {operation} was rejected by device {unit} at register 0x{address:x}: {response}")
         return response
 
-    def _encode_multi_write_payload(self, payload: list[tuple[Any, Any]]) -> list[int]:
-        """Encode a complete multi-register payload before any data is sent."""
-        if not isinstance(payload, list) or not payload:
-            raise HomeAssistantError(f"{self._name}: multi-register write requires a non-empty payload")
-
+    def _encode_write_value(
+        self,
+        payload: int | float,
+        register_data_type: str | None,
+        *,
+        single_register: bool,
+    ) -> list[int]:
+        """Validate and encode one value before it reaches the transport."""
+        effective_type = register_data_type or (REGISTER_S16 if single_register else None)
+        if effective_type is None:
+            raise RegisterEncodingError(f"{self._name}: unsupported register data type {register_data_type}")
         data_type_enum = cast(Any, DataType)
         data_types: dict[str, Any] = {
             REGISTER_U16: data_type_enum.UINT16,
@@ -1379,13 +1392,44 @@ def _encode_multi_write_payload(self, payload: list[tuple[Any, Any]]) -> list[in
             REGISTER_F32: data_type_enum.FLOAT32,
             REGISTER_S32: data_type_enum.INT32,
         }
+        data_type = data_types.get(effective_type)
+        word_count = REGISTER_TYPE_WORDS.get(effective_type)
+        if data_type is None or word_count is None:
+            raise RegisterEncodingError(f"{self._name}: unsupported register data type {register_data_type}")
+        if single_register and word_count != 1:
+            raise RegisterEncodingError(
+                f"{self._name}: register data type {effective_type} requires {word_count} registers and cannot be written as a single register"
+            )
+
+        try:
+            if effective_type == REGISTER_F32:
+                value: int | float = float(payload)
+            else:
+                value = int(payload)
+                minimum, maximum = REGISTER_INT_RANGES[effective_type]
+                if value < minimum or value > maximum:
+                    raise RegisterEncodingError(f"{self._name}: value {value} is outside the {effective_type} register range {minimum}..{maximum}")
+            registers = cast(list[int], convert_to_registers(value, data_type, self.plugin.order32))
+        except RegisterEncodingError:
+            raise
+        except (OverflowError, TypeError, ValueError, struct.error) as ex:
+            raise RegisterEncodingError(f"{self._name}: cannot encode value {payload!r} as {effective_type}: {ex}") from ex
+
+        if len(registers) != word_count:
+            raise RegisterEncodingError(f"{self._name}: encoding {effective_type} produced {len(registers)} registers instead of {word_count}")
+        return registers
+
+    def _encode_multi_write_payload(self, payload: list[tuple[Any, Any]]) -> list[int]:
+        """Encode a complete multi-register payload before any data is sent."""
+        if not isinstance(payload, list) or not payload:
+            raise HomeAssistantError(f"{self._name}: multi-register write requires a non-empty payload")
+
         registers: list[int] = []
         for item in payload:
             try:
                 key, value = item
                 if key.startswith("_"):
                     register_data_type = key
-                    value = int(value)
                 else:
                     descr = self.writeLocals[key]
                     reverse_options = getattr(descr, "reverse_option_dict", None)
@@ -1399,13 +1443,9 @@ def _encode_multi_write_payload(self, payload: list[tuple[Any, Any]]) -> list[in
                         value = descr.scale(value, descr, self.data)
                     else:
                         value = value * descr.scale
-                    value = int(value)
                     register_data_type = descr.register_data_type
 
-                data_type = data_types.get(register_data_type)
-                if data_type is None:
-                    raise ValueError(f"unsupported register data type {register_data_type}")
-                registers.extend(convert_to_registers(value, data_type, self.plugin.order32))
+                registers.extend(self._encode_write_value(value, register_data_type, single_register=False))
             except Exception as ex:
                 raise HomeAssistantError(f"{self._name}: cannot encode multi-register write item {item!r}: {ex}") from ex
 
@@ -1438,11 +1478,8 @@ async def _async_transport_write(
         )
 
     async def async_lowlevel_write_register(self, unit: int, address: int, payload: int, register_data_type: str | None = None) -> Any:
-        if register_data_type == REGISTER_U16:
-            regs = convert_to_registers(int(payload), DataType.UINT16, self.plugin.order32)  # type: ignore[attr-defined]
-        else:
-            regs = convert_to_registers(int(payload), DataType.INT16, self.plugin.order32)  # type: ignore[attr-defined]
         try:
+            regs = self._encode_write_value(payload, register_data_type, single_register=True)
             response = await self._async_transport_write(
                 unit=unit,
                 address=address,
@@ -1479,6 +1516,8 @@ async def async_write_register(self, unit: int, address: int, payload: int, regi
                 payload,
                 register_data_type=register_data_type,
             )
+        except RegisterEncodingError:
+            raise
         except HomeAssistantError as ex:
             self.writequeue[(unit, address)] = request
             if self.wakeupButton:
@@ -1515,10 +1554,7 @@ async def async_write_registers_single(
         self, unit: int, address: int, payload: int, register_data_type: str | None = None
     ) -> Any:  # Needs adapting for register queue
         """Write registers multi, but write only one register of type 16bit"""
-        if register_data_type == REGISTER_U16:
-            regs = convert_to_registers(int(payload), DataType.UINT16, self.plugin.order32)  # type: ignore[attr-defined]
-        else:
-            regs = convert_to_registers(int(payload), DataType.INT16, self.plugin.order32)  # type: ignore[attr-defined]
+        regs = self._encode_write_value(payload, register_data_type, single_register=True)
         return await self._async_transport_write(
             unit=unit,
             address=address,

custom_components/solax_modbus/const.py

@@ -108,6 +108,20 @@ class UnitOfReactivePower(StrEnum):  # type: ignore[no-redef]
 REGISTER_WORDS = "_words"  # nr or words must be specified in wordcount
 REGISTER_U8L = "_int8L"
 REGISTER_U8H = "_int8H"
+REGISTER_U16_MAX = 0xFFFF
+REGISTER_INT_RANGES: dict[str, tuple[int, int]] = {
+    REGISTER_U16: (0, REGISTER_U16_MAX),
+    REGISTER_S16: (-(1 << 15), (1 << 15) - 1),
+    REGISTER_U32: (0, (1 << 32) - 1),
+    REGISTER_S32: (-(1 << 31), (1 << 31) - 1),
+}
+REGISTER_TYPE_WORDS: dict[str, int] = {
+    REGISTER_U16: 1,
+    REGISTER_S16: 1,
+    REGISTER_U32: 2,
+    REGISTER_S32: 2,
+    REGISTER_F32: 2,
+}
 WRITE_SINGLE_MODBUS = 1  # use write_single_modbus command
 WRITE_MULTISINGLE_MODBUS = 2  # use write_mutiple modbus command for single register
 WRITE_DATA_LOCAL = 3  # write only to local data storage (not persistent)

custom_components/solax_modbus/plugin_solis.py

@@ -29,6 +29,7 @@
     REGISTER_S16,
     REGISTER_S32,
     REGISTER_U16,
+    REGISTER_U16_MAX,
     REGISTER_U32,
     REGISTER_WORDS,
     SLEEPMODE_LASTAWAKE,
@@ -350,18 +351,7 @@ def value_function_pv4_power(initval: int, descr: Any, datadict: dict[str, Any])
 
 # ================================= Number Declarations ============================================================
 
-MAX_CURRENTS: list[tuple[str, int | float]] = [
-    ("0602", 62.5),  # 3kW 48v
-    ("0102", 62.5),  # 3kW 48v AC Only?
-    ("110F", 62.5),  # 3.6kW 48v
-    ("160F3", 100),  # 5kW 48v
-    ("160F4", 60),  # 3.6kW 48v
-    ("160F5", 62.5),  # 3.6kW 48v
-    ("1031", 100),  # 5kW 48v
-    ("134F", 100),  # 5kW 48v
-    ("6031", 100),  # 6kW 48v
-    ("110C", 25),  # 10kW HV
-]
+U16_CURRENT_MAX = REGISTER_U16_MAX * 0.1
 
 NUMBER_TYPES = [
     ###
@@ -862,13 +852,13 @@ def value_function_pv4_power(initval: int, descr: Any, datadict: dict[str, Any])
         register=43116,
         fmt="f",
         native_min_value=0,
-        native_max_value=20,
+        native_max_value=U16_CURRENT_MAX,
         native_step=1,
         scale=0.1,
         native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
         device_class=NumberDeviceClass.CURRENT,
         allowedtypes=HYBRID,
-        max_exceptions=MAX_CURRENTS,
+        register_data_type=REGISTER_U16,
         entity_category=EntityCategory.CONFIG,
     ),
     SolisModbusNumberEntityDescription(
@@ -877,13 +867,13 @@ def value_function_pv4_power(initval: int, descr: Any, datadict: dict[str, Any])
         register=43117,
         fmt="f",
         native_min_value=0,
-        native_max_value=20,
+        native_max_value=U16_CURRENT_MAX,
         native_step=1,
         scale=0.1,
         native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
         device_class=NumberDeviceClass.CURRENT,
         allowedtypes=HYBRID,
-        max_exceptions=MAX_CURRENTS,
+        register_data_type=REGISTER_U16,
         entity_category=EntityCategory.CONFIG,
     ),
     SolisModbusNumberEntityDescription(
@@ -892,13 +882,13 @@ def value_function_pv4_power(initval: int, descr: Any, datadict: dict[str, Any])
         register=43118,
         fmt="f",
         native_min_value=0,
-        native_max_value=20,
+        native_max_value=U16_CURRENT_MAX,
         native_step=1,
         scale=0.1,
         native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
         device_class=NumberDeviceClass.CURRENT,
         allowedtypes=HYBRID,
-        max_exceptions=MAX_CURRENTS,
+        register_data_type=REGISTER_U16,
         entity_category=EntityCategory.CONFIG,
     ),
     SolisModbusNumberEntityDescription(
@@ -907,13 +897,13 @@ def value_function_pv4_power(initval: int, descr: Any, datadict: dict[str, Any])
         register=43141,
         fmt="f",
         native_min_value=0,
-        native_max_value=20,
+        native_max_value=U16_CURRENT_MAX,
         native_step=1,
         scale=0.1,
         native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
         device_class=NumberDeviceClass.CURRENT,
         allowedtypes=HYBRID,
-        max_exceptions=MAX_CURRENTS,
+        register_data_type=REGISTER_U16,
         entity_category=EntityCategory.CONFIG,
     ),
     SolisModbusNumberEntityDescription(
@@ -922,13 +912,13 @@ def value_function_pv4_power(initval: int, descr: Any, datadict: dict[str, Any])
         register=43142,
         fmt="f",
         native_min_value=0,
-        native_max_value=20,
+        native_max_value=U16_CURRENT_MAX,
         native_step=1,
         scale=0.1,
         native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
         device_class=NumberDeviceClass.CURRENT,
         allowedtypes=HYBRID,
-        max_exceptions=MAX_CURRENTS,
+        register_data_type=REGISTER_U16,
         entity_category=EntityCategory.CONFIG,
     ),
     SolisModbusNumberEntityDescription(