When using an analog sensor, wait for valid value

Problem reported by Max on nxt-python mailing list: when the color
sensor is reset, because of a program start or stop for example, it
takes the NXT firmware some time to return a valid value.

Add a new get_valid_input_values to retry reading when value is not
valid. Raise an nxt.sensor.analog.InvalidSensorReading on timeout.

Use this new method for all "analog" sensors.

Unrelated change: use dataclass for nxt.sensor.analog.RawReading class
and rename mode to sensor_mode for coherency. This should not have any
impact on user code as this is used mostly internally, and standard
usage stay unchanged.

Author: schodet

nxt/sensor/analog.py

@@ -13,39 +13,40 @@
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 
+import time
+from dataclasses import dataclass
+
 import nxt.sensor
 
 
+class InvalidReading(Exception):
+    """Exception raised on timeout trying to get valid readings."""
+
+    pass
+
+
+@dataclass
 class RawReading:
     """A object holding the raw sensor values for a sensor."""
 
-    def __init__(self, values):
-        (
-            self.port,
-            self.valid,
-            self.calibrated,
-            self.sensor_type,
-            self.mode,
-            self.raw_value,
-            self.normalized_value,
-            self.scaled_value,
-            self.calibrated_value,
-        ) = values
-
-    def __repr__(self):
-        return str(
-            (
-                self.port,
-                self.valid,
-                self.calibrated,
-                self.sensor_type,
-                self.mode,
-                self.raw_value,
-                self.normalized_value,
-                self.scaled_value,
-                self.calibrated_value,
-            )
-        )
+    #: Input port identifier.
+    port: nxt.sensor.Port
+    #: ``True`` if the value is valid, else ``False``.
+    valid: bool
+    #: Always ``False``, there is no calibration in NXT firmware.
+    calibrated: bool
+    #: Sensor type.
+    sensor_type: nxt.sensor.Type
+    #: Sensor mode.
+    sensor_mode: nxt.sensor.Mode
+    #: Raw analog to digital converter value.
+    raw_value: int
+    #: Normalized value.
+    normalized_value: int
+    #: Scaled value.
+    scaled_value: int
+    #: Always normalized value, there is no calibration in NXT firmware.
+    calibrated_value: int
 
 
 class BaseAnalogSensor(nxt.sensor.Sensor):
@@ -57,7 +58,25 @@ def get_input_values(self):
         :return: An object with the read values.
         :rtype: RawReading
         """
-        return RawReading(self._brick.get_input_values(self._port))
+        return RawReading(*self._brick.get_input_values(self._port))
+
+    def get_valid_input_values(self):
+        """Wait until input is valid, then get raw sensor readings.
+
+        :return: An object with the read values.
+        :rtype: RawReading
+        :raises InvalidReading: On timeout trying to get valid readings.
+        """
+        tries = 10
+        tries_delay_s = 0.1
+        readings = self.get_input_values()
+        while not readings.valid:
+            tries -= 1
+            if tries == 0:
+                raise InvalidReading()
+            time.sleep(tries_delay_s)
+            readings = self.get_input_values()
+        return readings
 
     def reset_input_scaled_value(self):
         """Reset sensor scaled value.

nxt/sensor/generic.py

@@ -34,7 +34,7 @@ def is_pressed(self):
         :return: ``True`` if the sensor is pressed, else ``False``.
         :rtype: bool
         """
-        return bool(self.get_input_values().scaled_value)
+        return bool(self.get_valid_input_values().scaled_value)
 
     get_sample = is_pressed
 
@@ -70,7 +70,7 @@ def get_lightness(self):
         :return: Detected light level between 0 and 1023.
         :rtype: int
         """
-        return self.get_input_values().scaled_value
+        return self.get_valid_input_values().scaled_value
 
     get_sample = get_lightness
 
@@ -103,7 +103,7 @@ def get_loudness(self):
         :return: Detected sound level between 0 and 1023.
         :rtype: int
         """
-        return self.get_input_values().scaled_value
+        return self.get_valid_input_values().scaled_value
 
     get_sample = get_loudness
 
@@ -266,7 +266,7 @@ def get_light_color(self):
         :return: Light color, one of the Type.COLOR_* values.
         :rtype: Type
         """
-        return self.get_input_values().sensor_type
+        return self.get_valid_input_values().sensor_type
 
     def get_reflected_light(self, color):
         """Get detected light level.
@@ -276,7 +276,7 @@ def get_reflected_light(self, color):
         :rtype: int
         """
         self.set_light_color(color)
-        return self.get_input_values().scaled_value
+        return self.get_valid_input_values().scaled_value
 
     def get_color(self):
         """Get detected color.
@@ -288,7 +288,7 @@ def get_color(self):
         determine color).
         """
         self.get_reflected_light(Type.COLOR_FULL)
-        return self.DetectedColor(self.get_input_values().scaled_value)
+        return self.DetectedColor(self.get_valid_input_values().scaled_value)
 
     get_sample = get_color
 

nxt/sensor/hitechnic.py

@@ -352,7 +352,7 @@ def set_range_short(self):
     def get_raw_value(self):
         """Unscaled value read from sensor."""
 
-        return self._MAX_DISTANCE - self.get_input_values().raw_value
+        return self._MAX_DISTANCE - self.get_valid_input_values().raw_value
 
     def get_processed_value(self):
         """Derived from the square root of the raw value."""
@@ -496,7 +496,7 @@ def __init__(self, brick, port):
         self.offset = 0
 
     def get_rotation_speed(self):
-        return self.get_input_values().scaled_value - self.offset
+        return self.get_valid_input_values().scaled_value - self.offset
 
     def set_zero(self, value):
         self.offset = value

nxt/sensor/mindsensors.py

@@ -62,7 +62,7 @@ def set_long_range(self, val):
 
     def get_sample(self):
         """Returns the processed meaningful values of the sensor"""
-        return self.Reading(self.get_input_values())
+        return self.Reading(self.get_valid_input_values())
 
 
 class Compassv2(BaseDigitalSensor):

tests/conftest.py

@@ -60,9 +60,13 @@ def sleepf(delay):
     mtime.time.side_effect = timef
     mtime.sleep.side_effect = sleepf
 
-    with patch("nxt.brick.time", new=mtime), patch(
-        "nxt.motcont.time", new=mtime
-    ), patch("nxt.motor.time", new=mtime), patch("nxt.sensor.digital.time", new=mtime):
+    with (
+        patch("nxt.brick.time", new=mtime),
+        patch("nxt.motcont.time", new=mtime),
+        patch("nxt.motor.time", new=mtime),
+        patch("nxt.sensor.analog.time", new=mtime),
+        patch("nxt.sensor.digital.time", new=mtime),
+    ):
         yield mtime
 
 

tests/test_sensors.py

@@ -50,19 +50,64 @@ def test_analog(self, mbrick):
         assert v.valid is True
         assert v.calibrated is False
         assert v.sensor_type == Type.SWITCH
-        assert v.mode == Mode.BOOL
+        assert v.sensor_mode == Mode.BOOL
         assert v.raw_value == 1
         assert v.normalized_value == 2
         assert v.scaled_value == 3
         assert v.calibrated_value == 4
-        assert str(v).startswith("(")
+        assert str(v).startswith("RawReading(")
         s.reset_input_scaled_value()
         assert mbrick.mock_calls == [
             call.set_input_mode(Port.S1, Type.SWITCH, Mode.BOOL),
             call.get_input_values(Port.S1),
             call.reset_input_scaled_value(Port.S1),
         ]
 
+    def test_analog_retry(self, mbrick, mtime):
+        s = mbrick.get_sensor(Port.S1, nxt.sensor.analog.BaseAnalogSensor)
+        mbrick.get_input_values.side_effect = [
+            (Port.S1, False, False, Type.SWITCH, Mode.BOOL, 1, 2, 3, 4),
+            (Port.S1, False, False, Type.SWITCH, Mode.BOOL, 1, 2, 3, 4),
+            (Port.S1, True, False, Type.SWITCH, Mode.BOOL, 1, 2, 3, 4),
+        ]
+        s.set_input_mode(Type.SWITCH, Mode.BOOL)
+        v = s.get_valid_input_values()
+        assert v.port == Port.S1
+        assert v.valid is True
+        assert v.calibrated is False
+        assert v.sensor_type == Type.SWITCH
+        assert v.sensor_mode == Mode.BOOL
+        assert v.raw_value == 1
+        assert v.normalized_value == 2
+        assert v.scaled_value == 3
+        assert v.calibrated_value == 4
+        assert mbrick.mock_calls == [
+            call.set_input_mode(Port.S1, Type.SWITCH, Mode.BOOL),
+            call.get_input_values(Port.S1),
+            call.get_input_values(Port.S1),
+            call.get_input_values(Port.S1),
+        ]
+        assert mtime.sleep.mock_calls == [
+            call(0.1),
+            call(0.1),
+        ]
+
+    def test_analog_retry_fail(self, mbrick, mtime):
+        retries = 10
+        s = mbrick.get_sensor(Port.S1, nxt.sensor.analog.BaseAnalogSensor)
+        mbrick.get_input_values.side_effect = retries * [
+            (Port.S1, False, False, Type.SWITCH, Mode.BOOL, 1, 2, 3, 4),
+        ]
+        s.set_input_mode(Type.SWITCH, Mode.BOOL)
+        with pytest.raises(nxt.sensor.analog.InvalidReading):
+            s.get_valid_input_values()
+        assert mbrick.mock_calls == [
+            call.set_input_mode(Port.S1, Type.SWITCH, Mode.BOOL),
+        ] + retries * [
+            call.get_input_values(Port.S1),
+        ]
+        assert mtime.sleep.mock_calls == (retries - 1) * [call(0.1)]
+
     def test_touch(self, mbrick):
         assert (
             nxt.sensor.generic.Touch.get_sample is nxt.sensor.generic.Touch.is_pressed