Tweak sensor value setter to validate dtype

src/aiokatcp/sensor.py

@@ -30,6 +30,7 @@
 import enum
 import functools
 import inspect
+import numbers
 import time
 import warnings
 import weakref
@@ -168,6 +169,7 @@ def __init__(
         self.stype = sensor_type
         type_info = core.get_type(sensor_type)
         self.type_name = type_info.name
+        self._core_type = type_info.type_
         self._classic_observers: Set[ClassicObserver[_T]] = set()
         self._delta_observers: Set[DeltaObserver[_T]] = set()
         self.name = name
@@ -177,7 +179,7 @@ def __init__(
         if default is None:
             value: _T = type_info.default(sensor_type)
         else:
-            value = default
+            value = self._check_value_type(default)
         self._reading = Reading(time.time(), initial_status, value)
         if auto_strategy is None:
             self.auto_strategy = SensorSampler.Strategy.AUTO
@@ -186,6 +188,37 @@ def __init__(
         self.auto_strategy_parameters = tuple(auto_strategy_parameters)
         # TODO: should validate the parameters against the strategy.
 
+    def _check_value_type(self, value: _T) -> _T:
+        """Verify incoming value is compatible with sensor type.
+
+        This also handles the special case of :class:`core.Timestamp` where
+        it is used with `float` type interchangeably.
+
+        Raises
+        ------
+        TypeError
+            If the incoming `value` type is not compatible with the sensor's
+            core type.
+        """
+
+        if isinstance(value, self._core_type) and not issubclass(self._core_type, enum.Enum):
+            # The more general case of e.g. numpy types not being directly
+            # comparable with python types.
+            return value
+        elif self.stype is core.Timestamp and isinstance(value, numbers.Real):
+            # core.Timestamp can also be a float
+            return value  # type: ignore
+        elif type(value) in [bytes, str, core.Address] and self.stype is core.Address:
+            return core.Address(value)  # type: ignore
+        elif isinstance(value, self.stype):
+            # To ensure specific types of enum.Enums are handled correctly
+            return value
+        else:
+            raise TypeError(
+                f"Value type {type(value)} is not compatible with Sensor type "
+                f"{self.stype} with core type {self._core_type}"
+            )
+
     def notify(self, reading: Reading[_T], old_reading: Reading[_T]) -> None:
         """Notify all observers of changes to this sensor.
 
@@ -202,6 +235,9 @@ def set_value(
     ) -> None:
         """Set the current value of the sensor.
 
+        Also validate that the incoming value type is compatible with the core
+        type of this sensor.
+
         Parameters
         ----------
         value
@@ -214,12 +250,19 @@ def set_value(
         timestamp
             The time at which the sensor value was determined (seconds).
             If not given, it defaults to :func:`time.time`.
+
+        Raises
+        ------
+        TypeError
+            If the incoming `value` type is not compatible with the sensor's
+            core type.
         """
+        checked_value = self._check_value_type(value)
         if timestamp is None:
             timestamp = time.time()
         if status is None:
-            status = self.status_func(value)
-        reading = Reading(timestamp, status, value)
+            status = self.status_func(checked_value)
+        reading = Reading(timestamp, status, checked_value)
         old_reading = self._reading
         self._reading = reading
         self.notify(reading, old_reading)

tests/test_sensor.py

@@ -30,6 +30,7 @@
 """
 
 import asyncio
+import enum
 import gc
 import unittest
 import weakref
@@ -39,6 +40,7 @@
 
 import pytest
 
+from aiokatcp.core import Address, Timestamp
 from aiokatcp.sensor import (
     AggregateSensor,
     Reading,
@@ -80,6 +82,49 @@ def status_func(value):
     assert sensor.status == Sensor.Status.NOMINAL
 
 
+def test_sensor_value_setter(diverse_sensors):
+    """Check whether the value type verification works."""
+    diverse_sensors["test-bool-sensor"].value = True
+    assert diverse_sensors["test-bool-sensor"]
+    with pytest.raises(TypeError):
+        diverse_sensors["test-bool-sensor"].value = "True"
+
+    diverse_sensors["test-int-sensor"].value = 1234
+    assert diverse_sensors["test-int-sensor"].value == 1234
+    with pytest.raises(TypeError):
+        diverse_sensors["test-int-sensor"].value = "1234"
+
+    diverse_sensors["test-float-sensor"].value = 1234.5
+    assert diverse_sensors["test-float-sensor"].value == 1234.5
+    with pytest.raises(TypeError):
+        diverse_sensors["test-float-sensor"].value = "1234"
+
+    diverse_sensors["test-str-sensor"].value = "foo"
+    assert diverse_sensors["test-str-sensor"].value == "foo"
+    with pytest.raises(TypeError):
+        diverse_sensors["test-str-sensor"].value = 1234
+
+    diverse_sensors["test-bytes-sensor"].value = b"bar"
+    assert diverse_sensors["test-bytes-sensor"].value == b"bar"
+    with pytest.raises(TypeError):
+        diverse_sensors["test-bytes-sensor"].value = "foo"
+
+    diverse_sensors["test-address-sensor"].value = "1.2.3.4"
+    assert diverse_sensors["test-address-sensor"].value == Address("1.2.3.4")
+    with pytest.raises(TypeError):
+        diverse_sensors["test-address-sensor"].value = 5678
+
+    diverse_sensors["test-timestamp-sensor"].value = 12345678
+    assert diverse_sensors["test-timestamp-sensor"].value == Timestamp(12345678)
+    with pytest.raises(TypeError):
+        diverse_sensors["test-timestamp-sensor"].value = "12345678"
+
+    diverse_sensors["test-enum-sensor"].value = MyEnum.ONE
+    assert diverse_sensors["test-enum-sensor"].value == MyEnum.ONE
+    with pytest.raises(TypeError):
+        diverse_sensors["test-enum-sensor"].value = OtherEnum.ABC
+
+
 @pytest.fixture
 def classic_observer():
     def classic(sensor, reading):
@@ -160,6 +205,80 @@ def alt_sensors():
     return [Sensor(float, f"name{i}") for i in range(5)]
 
 
+class MyEnum(enum.Enum):
+    ZERO = 0
+    ONE = 1
+
+
+class OtherEnum(enum.Enum):
+    ABC = 0
+    DEF = 1
+
+
+@pytest.fixture
+def diverse_sensors():
+    # Another set of sensors with different data types
+    diverse_ss = SensorSet()
+    diverse_ss.add(
+        Sensor(
+            bool,
+            "test-bool-sensor",
+            default=False,
+        )
+    )
+    diverse_ss.add(
+        Sensor(
+            int,
+            "test-int-sensor",
+            default=0,
+        )
+    )
+    diverse_ss.add(
+        Sensor(
+            float,
+            "test-float-sensor",
+            default=0.0,
+        )
+    )
+    diverse_ss.add(
+        Sensor(
+            str,
+            "test-str-sensor",
+            default="zero",
+        )
+    )
+    diverse_ss.add(
+        Sensor(
+            bytes,
+            "test-bytes-sensor",
+            default=b"zero",
+        )
+    )
+    diverse_ss.add(
+        Sensor(
+            Address,
+            "test-address-sensor",
+            default=Address("0.0.0.0"),
+        )
+    )
+    diverse_ss.add(
+        Sensor(
+            Timestamp,
+            "test-timestamp-sensor",
+            default=Timestamp(0.0),
+        )
+    )
+    diverse_ss.add(
+        Sensor(
+            MyEnum,
+            "test-enum-sensor",
+            default=MyEnum.ZERO,
+        )
+    )
+
+    return diverse_ss
+
+
 @pytest.fixture
 def ss(add_callback, remove_callback, sensors):
     ss = SensorSet()