Handle out of range values on rmu data (#173)

We need to apply the same out of range checks for RMU data, since these
fields also send out range data on error values.

Swapping scale and offset and applying same range checks for RMU data.

nibe/connection/encoders.py

@@ -1,6 +1,6 @@
 from abc import abstractmethod
 from binascii import hexlify
-from typing import Generic, List, Optional, SupportsInt, TypeVar, Union
+from typing import Generic, List, Optional, SupportsInt, TypeVar
 
 from construct import (
     Construct,
@@ -45,6 +45,13 @@
 }
 
 
+def is_hitting_integer_limit(size: str, int_value: int):
+    limit = integer_limit[size]
+    if limit < 0:
+        return int_value <= limit
+    return int_value >= limit
+
+
 _RawDataT = TypeVar("_RawDataT")
 
 
@@ -82,7 +89,7 @@ def decode(self, coil: Coil, raw: _RawDataT) -> CoilData:
         :raises DecodeException: If decoding fails"""
         try:
             value = self.decode_raw_value(coil.size, raw)
-            if self._is_hitting_integer_limit(coil.size, value):
+            if is_hitting_integer_limit(coil.size, value):
                 value = None
 
             return CoilData.from_raw_value(coil, value)
@@ -92,12 +99,6 @@ def decode(self, coil: Coil, raw: _RawDataT) -> CoilData:
                 f"Failed to decode {coil.name} coil from raw: {hexlify(raw).decode('utf-8')}, exception: {e}"
             ) from e
 
-    def _is_hitting_integer_limit(self, size: str, int_value: int):
-        limit = integer_limit[size]
-        if limit < 0:
-            return int_value <= limit
-        return int_value >= limit
-
 
 class CoilDataEncoderNibeGw(CoilDataEncoder[bytes]):
     """Encode and decode coil data."""

nibe/connection/nibegw.py

@@ -72,6 +72,7 @@
 from nibe.heatpump import HeatPump, ProductInfo
 
 from . import verify_connectivity_read_write_alarm
+from .encoders import is_hitting_integer_limit
 
 logger = logging.getLogger("nibe").getChild(__name__)
 
@@ -530,40 +531,58 @@ class FixedPointStrange(Adapter):
     be fixed.
     """
 
-    def __init__(self, subcon, scale, offset, ndigits=1) -> None:
+    def __init__(self, subcon, scale, offset, size, ndigits=1) -> None:
         super().__init__(subcon)
         self._offset = offset
         self._scale = scale
         self._ndigits = ndigits
+        self._size = size
 
     def _decode(self, obj, context, path):
-        scaled = obj * self._scale
-        if scaled >= self._offset:
-            scaled += self._offset
+        value = obj
+        if value >= self._offset:
+            value += self._offset
         else:
-            scaled -= self._offset
-        return round(scaled, self._ndigits)
+            value -= self._offset
+
+        # For now skip limit checks, since we don't know
+        # how the pump handles these for this special case
+        # for negative offsets, we could never reach the
+        # integer limits after offset has been applied.
+
+        return round(value * self._scale, self._ndigits)
 
     def _encode(self, obj, context, path):
-        if obj >= 0:
-            val = obj - self._offset
+        val = obj / self._scale
+        if val >= 0:
+            val -= self._offset
         else:
-            val = obj + self._offset
-        return val / self._scale
+            val += self._offset
+        return val
 
 
 class FixedPoint(Adapter):
-    def __init__(self, subcon, scale, offset, ndigits=1) -> None:
+    def __init__(self, subcon, scale, offset, size, ndigits=1) -> None:
         super().__init__(subcon)
         self._offset = offset
         self._scale = scale
         self._ndigits = ndigits
+        self._size = size
 
     def _decode(self, obj, context, path):
-        return round(obj * self._scale + self._offset, self._ndigits)
+        value = obj + self._offset
+
+        # Limits seem to be applied after offset
+        # have been applied. This may possible depend
+        # on the sign of the offset, but that is unknown
+        # at the moment.
+        if is_hitting_integer_limit(self._size, value):
+            return None
+
+        return round(value * self._scale, self._ndigits)
 
     def _encode(self, obj, context, path):
-        return (obj - self._offset) / self._scale
+        return obj / self._scale - self._offset
 
 
 StringData = Struct(
@@ -595,33 +614,33 @@ def _encode(self, obj, context, path):
             "hw_production" / Flag,
         ),
     ),
-    "bt1_outdoor_temperature" / FixedPointStrange(Int16sl, 0.1, -0.5),
-    "bt7_hw_top" / FixedPoint(Int16sl, 0.1, -0.5),
+    "bt1_outdoor_temperature" / FixedPointStrange(Int16sl, 0.1, -5, "s16"),
+    "bt7_hw_top" / FixedPoint(Int16sl, 0.1, -5, "s16"),
     "setpoint_or_offset_s1"
     / IfThenElse(
         lambda this: this.flags.use_room_sensor_s1,
-        FixedPoint(Int8ub, 0.1, 5.0),
-        FixedPoint(Int8sb, 1.0, 0),
+        FixedPoint(Int8ub, 0.1, 50, "u8"),
+        FixedPoint(Int8sb, 1.0, 0, "s8"),
     ),
     "setpoint_or_offset_s2"
     / IfThenElse(
         lambda this: this.flags.use_room_sensor_s2,
-        FixedPoint(Int8ub, 0.1, 5.0),
-        FixedPoint(Int8sb, 1.0, 0),
+        FixedPoint(Int8ub, 0.1, 50, "u8"),
+        FixedPoint(Int8sb, 1.0, 0, "s8"),
     ),
     "setpoint_or_offset_s3"
     / IfThenElse(
         lambda this: this.flags.use_room_sensor_s3,
-        FixedPoint(Int8ub, 0.1, 5.0),
-        FixedPoint(Int8sb, 1.0, 0),
+        FixedPoint(Int8ub, 0.1, 50, "u8"),
+        FixedPoint(Int8sb, 1.0, 0, "s8"),
     ),
     "setpoint_or_offset_s4"
     / IfThenElse(
         lambda this: this.flags.use_room_sensor_s4,
-        FixedPoint(Int8ub, 0.1, 5.0),
-        FixedPoint(Int8sb, 1.0, 0),
+        FixedPoint(Int8ub, 0.1, 50, "u8"),
+        FixedPoint(Int8sb, 1.0, 0, "s8"),
     ),
-    "bt50_room_temp_sX" / FixedPoint(Int16sl, 0.1, -0.5),
+    "bt50_room_temp_sX" / FixedPoint(Int16sl, 0.1, -5, "s16"),
     "temporary_lux" / Int8ub,
     "hw_time_hour" / Int8ub,
     "hw_time_min" / Int8ub,
@@ -754,18 +773,18 @@ def _encode(self, obj, context, path):
                 lambda this: this.index,
                 {
                     "TEMPORARY_LUX": Int8ub,
-                    "TEMPERATURE": FixedPoint(Int16ul, 0.1, -0.7),
+                    "TEMPERATURE": FixedPoint(Int16ul, 0.1, -7.0, size="s16"),
                     "FUNCTIONS": FlagsEnum(
                         Int8ub,
                         allow_additive_heating=0x01,
                         allow_heating=0x02,
                         allow_cooling=0x04,
                     ),
                     "OPERATIONAL_MODE": Int8ub,
-                    "SETPOINT_S1": FixedPoint(Int16sl, 0.1, 0.0),
-                    "SETPOINT_S2": FixedPoint(Int16sl, 0.1, 0.0),
-                    "SETPOINT_S3": FixedPoint(Int16sl, 0.1, 0.0),
-                    "SETPOINT_S4": FixedPoint(Int16sl, 0.1, 0.0),
+                    "SETPOINT_S1": FixedPoint(Int16sl, 0.1, 0.0, size="s16"),
+                    "SETPOINT_S2": FixedPoint(Int16sl, 0.1, 0.0, size="s16"),
+                    "SETPOINT_S3": FixedPoint(Int16sl, 0.1, 0.0, size="s16"),
+                    "SETPOINT_S4": FixedPoint(Int16sl, 0.1, 0.0, size="s16"),
                 },
                 default=Select(
                     Int16ul,

tests/connection/test_nibegw_message_parsing.py

@@ -329,9 +329,54 @@ def test_zero_outdoor(self):
             unknown5=b"\x01\x00",
         )
 
+    def test_invalid_values(self):
+        data = self.parse(
+            "0500 0080 9b 9b 9b ff 0080 00 00 00 00 00 03 79 15 34 00 03 00 00 01 00"
+        )
+
+        assert data == Container(
+            bt1_outdoor_temperature=0.0,
+            bt7_hw_top=None,
+            setpoint_or_offset_s1=20.5,
+            setpoint_or_offset_s2=20.5,
+            setpoint_or_offset_s3=20.5,
+            setpoint_or_offset_s4=-1.0,
+            bt50_room_temp_sX=None,
+            temporary_lux=0,
+            hw_time_hour=0,
+            hw_time_min=0,
+            fan_mode=0,
+            operational_mode=0,
+            flags=Container(
+                unknown_8000=False,
+                unknown_4000=False,
+                unknown_2000=False,
+                unknown_1000=False,
+                unknown_0800=False,
+                allow_cooling=False,
+                allow_heating=True,
+                allow_additive_heating=True,
+                use_room_sensor_s4=False,
+                use_room_sensor_s3=True,
+                use_room_sensor_s2=True,
+                use_room_sensor_s1=True,
+                unknown_0008=True,
+                unknown_0004=False,
+                unknown_0002=False,
+                hw_production=True,
+            ),
+            clock_time_hour=21,
+            clock_time_min=52,
+            alarm=0,
+            unknown4=b"\x03",
+            fan_time_hour=0,
+            fan_time_min=0,
+            unknown5=b"\x01\x00",
+        )
+
     @staticmethod
     def parse(txt_raw):
-        raw = binascii.unhexlify(txt_raw)
+        raw = binascii.unhexlify(txt_raw.replace(" ", ""))
         return RmuData.parse(raw)