Added some error handling

Minimal error checking, simply that the bytearray is the correct size. Otherwise it is kind of up to the user to decide what a valid looking result is.

Author: aefarrell

atmotube/__init__.py

@@ -1,3 +1,4 @@
+from .packets import InvalidByteData
 from .packets import StatusPacket, SPS30Packet, BME280Packet, SGPC3Packet
 from .uuids import AtmoTube_GATT_UUIDs
 

atmotube/packets.py

@@ -2,8 +2,17 @@
 import time
 
 
+class InvalidByteData(Exception):
+    pass
+
+
 class AtmoTubePacket(LittleEndianStructure):
+    _byte_size_ = 0  # To be defined in subclasses
+
     def __new__(cls, data, ts=None):
+        if len(data) != cls._byte_size_:
+            raise InvalidByteData(f"Expected {cls._byte_size_} bytes, "
+                                  f"got {len(data)} bytes")
         return cls.from_buffer_copy(data)
 
     def __init__(self, data, ts=None):
@@ -12,6 +21,9 @@ def __init__(self, data, ts=None):
         self.timestamp = ts
         self._process_bytes()
 
+    def __repr__(self):
+        return str(self)
+
     def _process_bytes(self):
         ...
 
@@ -29,6 +41,8 @@ class StatusPacket(AtmoTubePacket):
                 ("_battery",            c_ubyte, 8),
     ]
 
+    _byte_size_ = 2
+
     def _process_bytes(self):
         self.pm_sensor_status = bool(self._pm_sensor)
         self.error_flag = bool(self._error)
@@ -39,7 +53,7 @@ def _process_bytes(self):
         self.battery_level = self._battery
 
     def __str__(self):
-        return (f"StatusPacket("
+        return (f"StatusPacket(timestamp={time.ctime(self.timestamp)}, "
                 f"pm_sensor_status={self.pm_sensor_status}, "
                 f"error_flag={self.error_flag}, "
                 f"bonding_flag={self.bonding_flag}, "
@@ -56,7 +70,10 @@ class SPS30Packet(AtmoTubePacket):
         ('_pm10',  c_byte*3),
         ('_pm4',   c_byte*3),
     ]
+
     _pack_ = 1
+    _layout_ = "ms"
+    _byte_size_ = 12
 
     def pm_from_bytes(self, byte_array):
         res = int.from_bytes(byte_array, byteorder='little', signed=True)
@@ -69,7 +86,8 @@ def _process_bytes(self):
         self.pm4 = self.pm_from_bytes(self._pm4)
 
     def __str__(self):
-        return (f"SPS30Packet(pm1={self.pm1}µg/m³, pm2_5={self.pm2_5}µg/m³, "
+        return (f"SPS30Packet(timestamp={time.ctime(self.timestamp)}, "
+                f"pm1={self.pm1}µg/m³, pm2_5={self.pm2_5}µg/m³, "
                 f"pm10={self.pm10}µg/m³, pm4={self.pm4}µg/m³)")
 
 
@@ -80,15 +98,19 @@ class BME280Packet(AtmoTubePacket):
         ('_P',     c_int),
         ('_T_dec', c_short),
         ]
+
     _pack_ = 1
+    _layout_ = "ms"
+    _byte_size_ = 8
 
     def _process_bytes(self):
         self.humidity = self._rh if self._rh > 0 else None
         self.temperature = self._T_dec / 100.0
         self.pressure = self._P / 100.0 if self._P > 0 else None
 
     def __str__(self):
-        return (f"BME280Packet(humidity={self.humidity}%, "
+        return (f"BME280Packet(timestamp={time.ctime(self.timestamp)}, "
+                f"humidity={self.humidity}%, "
                 f"temperature={self.temperature}°C, "
                 f"pressure={self.pressure}mbar)")
 
@@ -97,10 +119,14 @@ class SGPC3Packet(AtmoTubePacket):
     _fields_ = [
         ('_tvoc', c_short)
     ]
+
     _pack_ = 1
+    _layout_ = "ms"
+    _byte_size_ = 4
 
     def _process_bytes(self):
         self.tvoc = self._tvoc/1000.0 if self._tvoc > 0 else None
 
     def __str__(self):
-        return f"SGPC3Packet(tvoc={self.tvoc}ppb)"
+        return (f"SGPC3Packet(timestamp={time.ctime(self.timestamp)}, "
+                f"tvoc={self.tvoc}ppb)")

examples/data_logging_example.py

@@ -90,6 +90,5 @@ async def runner():
 
 
 if __name__ == "__main__":
-    logger = logging.getLogger()
-    logger.setLevel(logging.INFO)
+    logging.basicConfig(level=logging.INFO)
     main()

tests/test_packets.py

@@ -1,46 +1,85 @@
+from atmotube import InvalidByteData
 from atmotube import StatusPacket, SPS30Packet, BME280Packet, SGPC3Packet
 
+import pytest
+import time
+
+# Status Packet tests
+timestamp = 1766865019.792235
 status_test_byte = bytearray(b'Ad')
-sps30_test_byte = bytearray(b'd\x00\x00\xb9\x00\x00J\x01\x00o\x00\x00')
-bme280_test_byte = bytearray(b'\x0e\x17\x8ao\x01\x00\x1a\t')
-sgpc3_test_byte = bytearray(b'\x02\x00\x00\x00')
+status_test_invalid_byte = bytearray(b'A')
+status_packet_str = (f"StatusPacket(timestamp={time.ctime(timestamp)}, "
+                     f"pm_sensor_status=True, "
+                     f"error_flag=False, bonding_flag=False, "
+                     f"charging=False, charging_timer=False, "
+                     f"pre_heating=True, battery_level=100%)")
 
 
 def test_status_packet():
-    packet = StatusPacket(status_test_byte)
+    packet = StatusPacket(status_test_byte, ts=timestamp)
     assert packet.pm_sensor_status is True
     assert packet.error_flag is False
     assert packet.bonding_flag is False
     assert packet.charging is False
     assert packet.charging_timer is False
     assert packet.pre_heating is True
     assert packet.battery_level == 100
-    assert str(packet) == ("StatusPacket(pm_sensor_status=True, "
-                           "error_flag=False, bonding_flag=False, "
-                           "charging=False, charging_timer=False, "
-                           "pre_heating=True, battery_level=100%)")
+    assert str(packet) == status_packet_str
+    assert repr(packet) == status_packet_str
+    with pytest.raises(InvalidByteData):
+        StatusPacket(status_test_invalid_byte, ts=timestamp)
+
+
+# SPS30 Packet tests
+sps30_test_byte = bytearray(b'd\x00\x00\xb9\x00\x00J\x01\x00o\x00\x00')
+sps30_test_invalid_byte = bytearray(b'd\x00\x00\xb9\x00\x00J\x01')
+sps30_test_str = (f"SPS30Packet(timestamp={time.ctime(timestamp)}, "
+                  f"pm1=1.0µg/m³, pm2_5=1.85µg/m³, "
+                  f"pm10=3.3µg/m³, pm4=1.11µg/m³)")
 
 
 def test_sps30_packet():
-    packet = SPS30Packet(sps30_test_byte)
+    packet = SPS30Packet(sps30_test_byte, ts=timestamp)
     assert packet.pm1 == 1.0
     assert packet.pm2_5 == 1.85
     assert packet.pm10 == 3.3
     assert packet.pm4 == 1.11
-    assert str(packet) == ("SPS30Packet(pm1=1.0µg/m³, pm2_5=1.85µg/m³, "
-                           "pm10=3.3µg/m³, pm4=1.11µg/m³)")
+    assert str(packet) == sps30_test_str
+    assert repr(packet) == sps30_test_str
+    with pytest.raises(InvalidByteData):
+        SPS30Packet(sps30_test_invalid_byte, ts=timestamp)
+
+
+# BME280 Packet tests
+bme280_test_byte = bytearray(b'\x0e\x17\x8ao\x01\x00\x1a\t')
+bme280_test_invalid_byte = bytearray(b'\x0e\x17\x8ao\x01')
+bme280_test_str = (f"BME280Packet(timestamp={time.ctime(timestamp)}, "
+                   f"humidity=14%, temperature=23.3°C, "
+                   f"pressure=940.9mbar)")
 
 
 def test_bme280_packet():
-    packet = BME280Packet(bme280_test_byte)
+    packet = BME280Packet(bme280_test_byte, ts=timestamp)
     assert packet.humidity == 14
     assert packet.temperature == 23.3
     assert packet.pressure == 940.9
-    assert str(packet) == ("BME280Packet(humidity=14%, temperature=23.3°C, "
-                           "pressure=940.9mbar)")
+    assert str(packet) == bme280_test_str
+    assert repr(packet) == bme280_test_str
+    with pytest.raises(InvalidByteData):
+        BME280Packet(bme280_test_invalid_byte, ts=timestamp)
+
+
+# SGPC3 Packet tests
+sgpc3_test_byte = bytearray(b'\x02\x00\x00\x00')
+sgpc3_test_invalid_byte = bytearray(b'\x02\x00\x00')
+sgpc3_test_str = (f"SGPC3Packet(timestamp={time.ctime(timestamp)}, "
+                  f"tvoc=0.002ppb)")
 
 
 def test_sgpc3_packet():
-    packet = SGPC3Packet(sgpc3_test_byte)
+    packet = SGPC3Packet(sgpc3_test_byte, ts=timestamp)
     assert packet.tvoc == 0.002
-    assert str(packet) == "SGPC3Packet(tvoc=0.002ppb)"
+    assert str(packet) == sgpc3_test_str
+    assert repr(packet) == sgpc3_test_str
+    with pytest.raises(InvalidByteData):
+        SGPC3Packet(sgpc3_test_invalid_byte, ts=timestamp)