decode HPMA115xx ACK messages, #26

Author: avaldebe

src/pms/sensors/honeywell/hpma115s0.py

@@ -41,12 +41,16 @@ def checksum(self) -> int:
         return self.message[-1]
 
     @classmethod
-    def _validate(cls, message: bytes, header: bytes, length: int) -> base.Message:
-        # validate ACK message
-        if header == b"\xA5\xA5" and length == 2:
-            assert message.endswith(header)
-            return cls(message)
+    def decode(cls, message: bytes, command: base.Cmd) -> Tuple[float, ...]:
+        # decode ACK message
+        if command.answer_header == b"\xA5\xA5" and command.answer_length == 2:
+            if not message.endswith(command.answer_header):
+                raise WrongMessageFormat("message does not end in ACK")
+            return tuple()
+        return super().decode(message, command)
 
+    @classmethod
+    def _validate(cls, message: bytes, header: bytes, length: int) -> base.Message:
         # consistency check: bug in message singnature
         assert len(header) == 3, f"wrong header length {len(header)}"
         assert header[:1] == b"\x40", f"wrong header start {header!r}"

tests/sensor/test_sensor.py

@@ -3,7 +3,7 @@
 
 import pytest
 
-from pms import SensorWarning, sensors
+from pms import SensorWarning, WrongMessageFormat
 from pms.core import Sensor, Supported
 
 
@@ -33,9 +33,17 @@ def test_pre_heat(sensor):
 
 @pytest.mark.parametrize("sensor", ["HPMA115S0", "HPMA115C0"])
 @pytest.mark.parametrize("command", ["passive_mode", "wake"])
-@pytest.mark.parametrize("buffer", [b"\xA5\xA5", b"\xA5\xA5\xA5\xA5"])
-def test_HPMA115xx_ACK_message(sensor, command, buffer):
-    assert Sensor[sensor].check(buffer, command)
+@pytest.mark.parametrize(
+    "buffer,check",
+    [
+        pytest.param(b"\xA5\xA5", True, id="ACK"),
+        pytest.param(b"\xA5\xA5\xA5\xA5", True, id="ACK ACK"),
+        pytest.param(b"\x40\x0d\x04", False, id="no ACK"),
+        pytest.param(b"\xA5\xA5\x40\x0d\x04", False, id="no ACK at end"),
+    ],
+)
+def test_HPMA115xx_ACK_message(sensor, command, buffer, check):
+    assert Sensor[sensor].check(buffer, command) == check
 
 
 class RawData(NamedTuple):