SPS30 short messages, #25

README.md

@@ -102,6 +102,7 @@ The [project site] contain some help for those brave enough to use its internals
 
 - WIP 0.6.1
   - fix `pms.sensors.sensirion` module name and docs
+  - reliably recognize SPS30 sensor [#25](https://github.com/avaldebe/PyPMS/issues/25)
 - 0.6.0
   - [project site]
   - reorganize internal modules

src/pms/core/sensor.py

@@ -16,7 +16,7 @@
 else:  # pragma: no cover
     import importlib_metadata as metadata
 
-from pms import WrongMessageChecksum, WrongMessageFormat
+from pms import WrongMessageChecksum, WrongMessageFormat, logger
 
 from .types import Cmd, Commands, Message, ObsData
 
@@ -79,7 +79,8 @@ def check(self, buffer: bytes, command: str) -> bool:
         """Validate buffer contents"""
         try:
             self.Message.decode(buffer, self.command(command))
-        except (WrongMessageFormat, WrongMessageChecksum):
+        except (WrongMessageFormat, WrongMessageChecksum) as e:
+            logger.debug(f"decode error {e}")
             return False
         else:
             return True

src/pms/sensors/sensirion/sps30.py

@@ -10,7 +10,7 @@
 from dataclasses import dataclass, field
 from typing import Tuple
 
-from pms import SensorWarmingUp, WrongMessageChecksum, WrongMessageFormat
+from pms import SensorWarmingUp, WrongMessageChecksum, WrongMessageFormat, logger
 
 from .. import base
 
@@ -34,23 +34,33 @@ class Message(base.Message):
 
     data_records = slice(10)
 
+    @classmethod
+    def unpack(cls, message: bytes, header: bytes, length: int) -> Tuple[float, ...]:
+        # error messages: recoverable errors (throw away observation)
+        if message.endswith(b"\x7E\x00\x00\x43\x00\xBC\x7E"):
+            raise SensorWarmingUp("short message: command not allowed in current state")
+        if message.endswith(b"\x7E\x00\x03\x00\x00\xFC\x7E"):
+            raise SensorWarmingUp("short message: no data")
+
+        # byte de-stuffing
+        for k, v in {
+            b"\x7D\x5E": b"\x7E",
+            b"\x7D\x5D": b"\x7D",
+            b"\x7D\x31": b"\x11",
+            b"\x7D\x33": b"\x13",
+        }.items():
+            if k in message:  # pragma: no cover
+                message = message.replace(k, v)
+        print(message)
+        return super().unpack(message, header, length)
+
     @property
     def header(self) -> bytes:
         return self.message[:5]
 
     @property
     def payload(self) -> bytes:
-        """byte de-stuffing"""
-        p = self.message[5:-2]
-        for k, v in {
-            b"\x7E": b"\x7D\x5E",
-            b"\x7D": b"\x7D\x5D",
-            b"\x11": b"\x7D\x31",
-            b"\x13": b"\x7D\x33",
-        }.items():
-            if v in p:  # pragma: no cover
-                p = p.replace(v, k)
-        return p
+        return self.message[5:-2]
 
     @property
     def checksum(self) -> int:
@@ -71,6 +81,7 @@ def _validate(cls, message: bytes, header: bytes, length: int) -> base.Message:
         assert length == len_payload + 7, f"wrong payload length {length} != {len_payload+7}"
 
         # validate message: recoverable errors (throw away observation)
+        print(message)
         msg = cls(message)
         if msg.header != header:
             raise WrongMessageFormat(f"message header: {msg.header!r}")

tests/sensor/test_message.py

@@ -100,8 +100,8 @@
         ),
         pytest.param(
             sps30,
-            "7E00030000FC7E",
-            r"message header: b'~\x00\x03\x00\x00'",
+            "7E00000000FF7E",
+            r"message header: b'~\x00\x00\x00\x00'",
             id="SPS30 wrong header",
         ),
         pytest.param(
@@ -122,6 +122,18 @@
             "message empty: warming up sensor",
             id="SPS30 empty message",
         ),
+        pytest.param(
+            sps30,
+            "7E00000000FF7E7E00004300BC7E",
+            "short message: command not allowed in current state",
+            id="SPS30 command not allowed",
+        ),
+        pytest.param(
+            sps30,
+            "7E00000000FF7E7E00030000FC7E",
+            "short message: no data",
+            id="SPS30 no data",
+        ),
         pytest.param(
             mcu680, "5A5A3F0F0835198A0188", "message length: 10", id="MCU680 short message"
         ),