Standardize NaN check.

Use `helers.data_types.SupportsNanCheck.isnan()` method to check if value is a valid number.
denpamusic · Oct 21, 2024 · c0da10c · c0da10c
1 parent 8338e96
commit c0da10c
Show file tree

Hide file tree

Showing 8 changed files with 62 additions and 27 deletions.
diff --git a/pyplumio/helpers/data_types.py b/pyplumio/helpers/data_types.py
@@ -3,10 +3,13 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
+import math
 import socket
 import struct
 from typing import ClassVar, Final, Generic, TypeVar
 
+from pyplumio.const import BYTE_UNDEFINED
+
 T = TypeVar("T")
 DataTypeT = TypeVar("DataTypeT", bound="DataType")
 
@@ -278,6 +281,28 @@ def size(self) -> int:
         return self._size
 
 
+class SupportsNanCheck(ABC):
+    """Represents a data type, that supports NaN (not a number).
+
+    Plum devices usually reserve uppermost boundary of data type as
+    an invalid value or NaN.
+    """
+
+    __slots__ = ()
+
+    def isnan(self) -> bool:
+        """Return True if value is not a number, False otherwise."""
+        if (
+            hasattr(self, "value")
+            and hasattr(self, "size")
+            and self.value != (BYTE_UNDEFINED * self.size)
+            and not math.isnan(self.value)
+        ):
+            return False
+
+        return True
+
+
 class SignedChar(BuiltInDataType[int]):
     """Represents a signed char."""
 
@@ -286,7 +311,7 @@ class SignedChar(BuiltInDataType[int]):
     _struct = struct.Struct("<b")
 
 
-class UnsignedChar(BuiltInDataType[int]):
+class UnsignedChar(BuiltInDataType[int], SupportsNanCheck):
     """Represents an unsigned char."""
 
     __slots__ = ()
@@ -326,15 +351,15 @@ class UnsignedInt(BuiltInDataType[int]):
     _struct = struct.Struct("<I")
 
 
-class Float(BuiltInDataType[int]):
+class Float(BuiltInDataType[float], SupportsNanCheck):
     """Represents a float."""
 
     __slots__ = ()
 
     _struct = struct.Struct("<f")
 
 
-class Double(BuiltInDataType[int]):
+class Double(BuiltInDataType[float]):
     """Represents a double."""
 
     __slots__ = ()

diff --git a/pyplumio/structures/boiler_load.py b/pyplumio/structures/boiler_load.py
@@ -4,7 +4,7 @@
 
 from typing import Any, Final
 
-from pyplumio.const import BYTE_UNDEFINED
+from pyplumio.helpers.data_types import UnsignedChar
 from pyplumio.structures import StructureDecoder
 from pyplumio.utils import ensure_dict
 
@@ -20,10 +20,10 @@ def decode(
         self, message: bytearray, offset: int = 0, data: dict[str, Any] | None = None
     ) -> tuple[dict[str, Any], int]:
         """Decode bytes and return message data and offset."""
-        boiler_load = message[offset]
-        offset += 1
+        boiler_load = UnsignedChar.from_bytes(message, offset)
+        offset += boiler_load.size
 
-        if boiler_load == BYTE_UNDEFINED:
+        if boiler_load.isnan():
             return ensure_dict(data), offset
 
         return (ensure_dict(data, {ATTR_BOILER_LOAD: boiler_load}), offset)
diff --git a/pyplumio/structures/boiler_power.py b/pyplumio/structures/boiler_power.py
@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import math
 from typing import Any, Final
 
 from pyplumio.helpers.data_types import Float
@@ -24,7 +23,7 @@ def decode(
         boiler_power = Float.from_bytes(message, offset)
         offset += boiler_power.size
 
-        if math.isnan(boiler_power.value):
+        if boiler_power.isnan():
             return ensure_dict(data), offset
 
         return ensure_dict(data, {ATTR_BOILER_POWER: boiler_power.value}), offset
diff --git a/pyplumio/structures/fan_power.py b/pyplumio/structures/fan_power.py
@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import math
 from typing import Any, Final
 
 from pyplumio.helpers.data_types import Float
@@ -24,7 +23,7 @@ def decode(
         fan_power = Float.from_bytes(message, offset)
         offset += fan_power.size
 
-        if math.isnan(fan_power.value):
+        if fan_power.isnan():
             return ensure_dict(data), offset
 
         return ensure_dict(data, {ATTR_FAN_POWER: fan_power.value}), offset
diff --git a/pyplumio/structures/fuel_level.py b/pyplumio/structures/fuel_level.py
@@ -4,7 +4,7 @@
 
 from typing import Any, Final
 
-from pyplumio.const import BYTE_UNDEFINED
+from pyplumio.helpers.data_types import UnsignedChar
 from pyplumio.structures import StructureDecoder
 from pyplumio.utils import ensure_dict
 
@@ -22,15 +22,16 @@ def decode(
         self, message: bytearray, offset: int = 0, data: dict[str, Any] | None = None
     ) -> tuple[dict[str, Any], int]:
         """Decode bytes and return message data and offset."""
-        fuel_level = message[offset]
-        offset += 1
+        fuel_level = UnsignedChar.from_bytes(message, offset)
+        offset += fuel_level.size
 
-        if fuel_level == BYTE_UNDEFINED:
+        if fuel_level.isnan():
             return ensure_dict(data), offset
 
-        if fuel_level >= FUEL_LEVEL_OFFSET:
+        fuel_level_value = fuel_level.value
+        if fuel_level_value >= FUEL_LEVEL_OFFSET:
             # Observed on at least ecoMAX 860P6-O.
             # See: https://github.com/denpamusic/PyPlumIO/issues/19
-            fuel_level -= FUEL_LEVEL_OFFSET
+            fuel_level_value -= FUEL_LEVEL_OFFSET
 
-        return (ensure_dict(data, {ATTR_FUEL_LEVEL: fuel_level}), offset)
+        return (ensure_dict(data, {ATTR_FUEL_LEVEL: fuel_level_value}), offset)
diff --git a/pyplumio/structures/lambda_sensor.py b/pyplumio/structures/lambda_sensor.py
@@ -6,8 +6,8 @@
 import math
 from typing import Any, Final
 
-from pyplumio.const import BYTE_UNDEFINED, LambdaState
-from pyplumio.helpers.data_types import UnsignedShort
+from pyplumio.const import LambdaState
+from pyplumio.helpers.data_types import UnsignedChar, UnsignedShort
 from pyplumio.structures import StructureDecoder
 from pyplumio.utils import ensure_dict
 
@@ -25,23 +25,24 @@ def decode(
         self, message: bytearray, offset: int = 0, data: dict[str, Any] | None = None
     ) -> tuple[dict[str, Any], int]:
         """Decode bytes and return message data and offset."""
-        lambda_state = message[offset]
-        offset += 1
-        if lambda_state == BYTE_UNDEFINED:
+        lambda_state = UnsignedChar.from_bytes(message, offset)
+        offset += lambda_state.size
+        if lambda_state.isnan():
             return ensure_dict(data), offset
 
         lambda_target = message[offset]
         offset += 1
         level = UnsignedShort.from_bytes(message, offset)
         offset += level.size
+        lambda_state_value = lambda_state.value
         with suppress(ValueError):
-            lambda_state = LambdaState(lambda_state)
+            lambda_state_value = LambdaState(lambda_state_value)
 
         return (
             ensure_dict(
                 data,
                 {
-                    ATTR_LAMBDA_STATE: lambda_state,
+                    ATTR_LAMBDA_STATE: lambda_state_value,
                     ATTR_LAMBDA_TARGET: lambda_target,
                     ATTR_LAMBDA_LEVEL: (
                         None if math.isnan(level.value) else (level.value / 10)

diff --git a/pyplumio/structures/temperatures.py b/pyplumio/structures/temperatures.py
@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import math
 from typing import Any, Final
 
 from pyplumio.helpers.data_types import Float
@@ -65,7 +64,7 @@ def decode(
             offset += 1
             temp = Float.from_bytes(message, offset)
             offset += temp.size
-            if (not math.isnan(temp.value)) and 0 <= index < len(TEMPERATURES):
+            if not temp.isnan() and 0 <= index < len(TEMPERATURES):
                 # Temperature exists and index is in the correct range.
                 data[TEMPERATURES[index]] = temp.value
 

diff --git a/tests/helpers/test_data_types.py b/tests/helpers/test_data_types.py
@@ -2,6 +2,7 @@
 
 import pytest
 
+from pyplumio.const import BYTE_UNDEFINED
 from pyplumio.helpers import data_types
 
 
@@ -87,15 +88,20 @@ def test_int() -> None:
 
 def test_unsigned_char() -> None:
     """Test an signed char data type."""
+    assert data_types.UnsignedChar().isnan()
     buffer = bytearray([0x3])
     data_type = data_types.UnsignedChar.from_bytes(buffer)
+    assert not data_type.isnan()
     assert data_type.value == 3
     assert data_type.size == 1
     assert data_type.to_bytes() == buffer
     assert repr(data_type) == "UnsignedChar(value=3)"
     assert repr(data_types.UnsignedChar()) == "UnsignedChar()"
     assert data_type == data_types.UnsignedChar.from_bytes(buffer)
     assert data_type == 3
+    buffer = bytearray([BYTE_UNDEFINED])
+    data_type = data_types.UnsignedChar.from_bytes(buffer)
+    assert data_type.isnan()
 
 
 def test_ushort() -> None:
@@ -126,15 +132,20 @@ def test_uint() -> None:
 
 def test_float() -> None:
     """Test a float data type."""
+    assert data_types.Float().isnan()
     buffer = bytearray([0x0, 0x0, 0x40, 0x41])
     data_type = data_types.Float.from_bytes(buffer)
+    assert not data_type.isnan()
     assert data_type.value == 12.0
     assert data_type.size == 4
     assert data_type.to_bytes() == buffer
     assert repr(data_type) == "Float(value=12.0)"
     assert repr(data_types.Float()) == "Float()"
     assert data_type == data_types.Float.from_bytes(buffer)
     assert data_type == 12.0
+    buffer = bytearray([BYTE_UNDEFINED, BYTE_UNDEFINED, BYTE_UNDEFINED, BYTE_UNDEFINED])
+    data_type = data_types.Float.from_bytes(buffer)
+    assert data_type.isnan()
 
 
 def test_double() -> None: