Change timestamp to date_time object

I think this makes this more generally useful for people. You can put anything in the date_time field, even a timestamp if you want, but the default is datetime.now()

Author: aefarrell

README.md

@@ -9,56 +9,64 @@ Currently it is just a set of helper classes for taking bytearrays returned by t
 
 ### Status Packet
 
-The `StatusPacket` class takes a bytearray returned from the Status GATT characteristic and an optional timestamp and creates a data structure with fields corresponding to the different status flags:
+The `StatusPacket` class takes a bytearray returned from the Status GATT characteristic and an optional datetime and creates a data structure with fields corresponding to the different status flags:
 
 ```python
 from atmotube import StatusPacket
+from datetime import datetime
 
+example_date = datetime(2024, 1, 1, 12, 0, 0)
 example_status = bytearray(b'Ad')
-packet = StatusPacket(example_status)
+packet = StatusPacket(example_status, date_time=example_date)
 
 print(packet)
-# StatusPacket(pm_sensor_status=True, error_flag=False, bonding_flag=False, charging=False, charging_timer=False, pre_heating=True, battery_level=100%)
+# StatusPacket(date_time=2024-01-01 12:00:00, pm_sensor_status=True, error_flag=False, bonding_flag=False, charging=False, charging_timer=False, pre_heating=True, battery_level=100%)
 ```
 
 ### SPS30 Packet
 
-The `SPS30Packet` takes a bytearray returned from the SPS30 GATT characteristic with an optional timestamp and creates a data structure with fields corresponding to the PM measurements in ug/m^3
+The `SPS30Packet` takes a bytearray returned from the SPS30 GATT characteristic with an optional datetime and creates a data structure with fields corresponding to the PM measurements in ug/m^3
 
 ```python
 from atmotube import SPS30Packet
+from datetime import datetime
 
+example_date = datetime(2024, 1, 1, 12, 0, 0)
 example_sps30 = bytearray(b'd\x00\x00\xb9\x00\x00J\x01\x00o\x00\x00')
-packet = SPS30Packet(example_sps30)
+packet = SPS30Packet(example_sps30, date_time=example_date)
 
 print(packet)
-# SPS30Packet(pm1=1.0µg/m³, pm2_5=1.85µg/m³, pm10=3.3µg/m³, pm4=1.11µg/m³)
+# SPS30Packet(date_time=2024-01-01 12:00:00, pm1=1.0µg/m³, pm2_5=1.85µg/m³, pm10=3.3µg/m³, pm4=1.11µg/m³)
 ```
 
 ### BME280 Packet
 
-The `BME280Packet` takes a bytearray returned from the BME280 GATT characteristic with an optional timestamp and creates a datastructure with fields corresponding the temperature, pressure, and humidity.
+The `BME280Packet` takes a bytearray returned from the BME280 GATT characteristic with an optional datetime and creates a datastructure with fields corresponding the temperature, pressure, and humidity.
 
 ```python
 from atmotube import BME280Packet
+from datetime import datetime
 
+example_date = datetime(2024, 1, 1, 12, 0, 0)
 example_bme280 = bytearray(b'\x0e\x17\x8ao\x01\x00\x1a\t')
-packet = BME280Packet(example_bme280)
+packet = BME280Packet(example_bme280, date_time=example_date)
 
 print(packet)
-# BME280Packet(humidity=14%, temperature=23.3°C, pressure=940.9mbar)
+# BME280Packet(date_time=2024-01-01 12:00:00, humidity=14%, temperature=23.3°C, pressure=940.9mbar)
 ```
 
 ### SGPC3 Packet
 
-The `SGPC3Packet` takes a bytearray returned from the SGPC3 GATT characteristic with an optional timestamp and creates a datastructure with a single field for the VOC.
+The `SGPC3Packet` takes a bytearray returned from the SGPC3 GATT characteristic with an optional datetime and creates a datastructure with a single field for the VOC.
 
 ```python
 from atmotube import SGPC3Packet
+from datetime import datetime
 
+example_date = datetime(2024, 1, 1, 12, 0, 0)
 example_sgpc3 = bytearray(b'\x02\x00\x00\x00')
-packet = SGPC3Packet(example_sgpc3)
+packet = SGPC3Packet(example_sgpc3, date_time=example_date)
 
 print(packet)
-# SGPC3Packet(tvoc=0.002ppb)
+# SGPC3Packet(date_time=2024-01-01 12:00:00, tvoc=0.002ppb)
 ```

atmotube/packets.py

@@ -1,5 +1,5 @@
 from ctypes import LittleEndianStructure, c_ubyte, c_byte, c_short, c_int
-import time
+from datetime import datetime
 
 
 class InvalidByteData(Exception):
@@ -9,16 +9,16 @@ class InvalidByteData(Exception):
 class AtmoTubePacket(LittleEndianStructure):
     _byte_size_ = 0  # To be defined in subclasses
 
-    def __new__(cls, data, ts=None):
+    def __new__(cls, data, date_time=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):
-        if ts is None:
-            ts = time.time()
-        self.timestamp = ts
+    def __init__(self, data, date_time=None):
+        if date_time is None:
+            date_time = datetime.now()
+        self.date_time = date_time
         self._process_bytes()
 
     def __repr__(self):
@@ -53,7 +53,7 @@ def _process_bytes(self):
         self.battery_level = self._battery
 
     def __str__(self):
-        return (f"StatusPacket(timestamp={time.ctime(self.timestamp)}, "
+        return (f"StatusPacket(date_time={str(self.date_time)}, "
                 f"pm_sensor_status={self.pm_sensor_status}, "
                 f"error_flag={self.error_flag}, "
                 f"bonding_flag={self.bonding_flag}, "
@@ -86,7 +86,7 @@ def _process_bytes(self):
         self.pm4 = self.pm_from_bytes(self._pm4)
 
     def __str__(self):
-        return (f"SPS30Packet(timestamp={time.ctime(self.timestamp)}, "
+        return (f"SPS30Packet(date_time={str(self.date_time)}, "
                 f"pm1={self.pm1}µg/m³, pm2_5={self.pm2_5}µg/m³, "
                 f"pm10={self.pm10}µg/m³, pm4={self.pm4}µg/m³)")
 
@@ -109,7 +109,7 @@ def _process_bytes(self):
         self.pressure = self._P / 100.0 if self._P > 0 else None
 
     def __str__(self):
-        return (f"BME280Packet(timestamp={time.ctime(self.timestamp)}, "
+        return (f"BME280Packet(date_time={str(self.date_time)}, "
                 f"humidity={self.humidity}%, "
                 f"temperature={self.temperature}°C, "
                 f"pressure={self.pressure}mbar)")
@@ -128,5 +128,5 @@ def _process_bytes(self):
         self.tvoc = self._tvoc/1000.0 if self._tvoc > 0 else None
 
     def __str__(self):
-        return (f"SGPC3Packet(timestamp={time.ctime(self.timestamp)}, "
+        return (f"SGPC3Packet(date_time={str(self.date_time)}, "
                 f"tvoc={self.tvoc}ppb)")

examples/data_logging_example.py

@@ -10,7 +10,6 @@
 from atmotube import BME280Packet, SGPC3Packet
 import asyncio
 import logging
-import time
 
 ATMOTUBE = "C2:2B:42:15:30:89"  # the mac address of my Atmotube
 
@@ -48,23 +47,23 @@ async def sgp30_cb(char, data):
 def log_packet(packet):
     match packet:
         case StatusPacket():
-            logging.info(f"{time.ctime(packet.timestamp)} - Status Packet - "
+            logging.info(f"{str(packet.date_time)} - Status Packet - "
                          f"Battery: {packet.battery_level}%, "
                          f"Charging: {packet.charging}, "
                          f"Error: {packet.error_flag}")
         case SPS30Packet():
-            logging.info(f"{time.ctime(packet.timestamp)} - SPS30 Packet - "
+            logging.info(f"{str(packet.date_time)} - SPS30 Packet - "
                          f"PM1: {packet.pm1} µg/m³, "
                          f"PM2.5: {packet.pm2_5} µg/m³, "
                          f"PM4: {packet.pm4} µg/m³, "
                          f"PM10: {packet.pm10} µg/m³")
         case BME280Packet():
-            logging.info(f"{time.ctime(packet.timestamp)} - BME280 Packet - "
+            logging.info(f"{str(packet.date_time)} - BME280 Packet - "
                          f"Humidity: {packet.humidity}%, "
                          f"Temperature: {packet.temperature}°C, "
                          f"Pressure: {packet.pressure} mbar")
         case SGPC3Packet():
-            logging.info(f"{time.ctime(packet.timestamp)} - SGPC3 Packet - "
+            logging.info(f"{str(packet.date_time)} - SGPC3 Packet - "
                          f"TVOC: {packet.tvoc} ppb")
         case _:
             logging.info("Unknown packet type")

tests/test_packets.py

@@ -1,22 +1,22 @@
 from atmotube import InvalidByteData
 from atmotube import StatusPacket, SPS30Packet, BME280Packet, SGPC3Packet
+from datetime import datetime
 
 import pytest
-import time
 
 # Status Packet tests
-timestamp = 1766865019.792235
+datetime_obj = datetime(2024, 1, 1, 12, 0, 0)
 status_test_byte = bytearray(b'Ad')
 status_test_invalid_byte = bytearray(b'A')
-status_packet_str = (f"StatusPacket(timestamp={time.ctime(timestamp)}, "
+status_packet_str = (f"StatusPacket(date_time={str(datetime_obj)}, "
                      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, ts=timestamp)
+    packet = StatusPacket(status_test_byte, date_time=datetime_obj)
     assert packet.pm_sensor_status is True
     assert packet.error_flag is False
     assert packet.bonding_flag is False
@@ -27,59 +27,59 @@ def test_status_packet():
     assert str(packet) == status_packet_str
     assert repr(packet) == status_packet_str
     with pytest.raises(InvalidByteData):
-        StatusPacket(status_test_invalid_byte, ts=timestamp)
+        StatusPacket(status_test_invalid_byte, date_time=datetime_obj)
 
 
 # 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)}, "
+sps30_test_str = (f"SPS30Packet(date_time={str(datetime_obj)}, "
                   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, ts=timestamp)
+    packet = SPS30Packet(sps30_test_byte, date_time=datetime_obj)
     assert packet.pm1 == 1.0
     assert packet.pm2_5 == 1.85
     assert packet.pm10 == 3.3
     assert packet.pm4 == 1.11
     assert str(packet) == sps30_test_str
     assert repr(packet) == sps30_test_str
     with pytest.raises(InvalidByteData):
-        SPS30Packet(sps30_test_invalid_byte, ts=timestamp)
+        SPS30Packet(sps30_test_invalid_byte, date_time=datetime_obj)
 
 
 # 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)}, "
+bme280_test_str = (f"BME280Packet(date_time={str(datetime_obj)}, "
                    f"humidity=14%, temperature=23.3°C, "
                    f"pressure=940.9mbar)")
 
 
 def test_bme280_packet():
-    packet = BME280Packet(bme280_test_byte, ts=timestamp)
+    packet = BME280Packet(bme280_test_byte, date_time=datetime_obj)
     assert packet.humidity == 14
     assert packet.temperature == 23.3
     assert packet.pressure == 940.9
     assert str(packet) == bme280_test_str
     assert repr(packet) == bme280_test_str
     with pytest.raises(InvalidByteData):
-        BME280Packet(bme280_test_invalid_byte, ts=timestamp)
+        BME280Packet(bme280_test_invalid_byte, date_time=datetime_obj)
 
 
 # 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)}, "
+sgpc3_test_str = (f"SGPC3Packet(date_time={str(datetime_obj)}, "
                   f"tvoc=0.002ppb)")
 
 
 def test_sgpc3_packet():
-    packet = SGPC3Packet(sgpc3_test_byte, ts=timestamp)
+    packet = SGPC3Packet(sgpc3_test_byte, date_time=datetime_obj)
     assert packet.tvoc == 0.002
     assert str(packet) == sgpc3_test_str
     assert repr(packet) == sgpc3_test_str
     with pytest.raises(InvalidByteData):
-        SGPC3Packet(sgpc3_test_invalid_byte, ts=timestamp)
+        SGPC3Packet(sgpc3_test_invalid_byte, date_time=datetime_obj)