Added BLE Advertising data

Added packets for processing data returned from BLE Advertising and Scan Response, some helper functions for processing, and an example of gathering all advertising and scan response packets being broadcast by any Atmotube PROs in the area.

Author: aefarrell

atmotube/__init__.py

@@ -1,13 +1,18 @@
+from .ble import (get_ble_packet,
+                  ble_callback_wrapper)
 from .gatt import (InvalidAtmotubeService,
                    gatt_notify,
                    start_gatt_notifications,
                    get_available_characteristics)
 from .packets import (InvalidByteData,
-                      AtmotubePacket,
+                      AtmotubeGATTPacket,
+                      AtmotubeBLEPacket,
                       AtmotubeProStatus,
                       AtmotubeProSPS30,
                       AtmotubeProBME280,
-                      AtmotubeProSGPC3)
+                      AtmotubeProSGPC3,
+                      AtmotubeProBLEAdvertising,
+                      AtmotubeProBLEScanResponse)
 from .uuids import (AtmotubeProService_UUID,
                     AtmotubeProGATT_UUID,
                     AtmotubeProUART_UUID)

atmotube/ble.py

@@ -0,0 +1,48 @@
+from enum import IntEnum
+from bleak import BLEDevice
+from bleak.backends.scanner import AdvertisementData
+
+from .packets import (AtmotubeBLEPacket,
+                      AtmotubeProBLEAdvertising,
+                      AtmotubeProBLEScanResponse)
+
+import inspect
+import logging
+
+
+class AtmotubeProBLE_CONSTS(IntEnum):
+    MANUFACTURER_DATA_ID = int(0xFFFF)
+    ADVERTISING_BYTE_LENGTH = 12
+    SCAN_RESPONSE_BYTE_LENGTH = 9
+
+PACKET_MAP = {
+    AtmotubeProBLE_CONSTS.ADVERTISING_BYTE_LENGTH: AtmotubeProBLEAdvertising,
+    AtmotubeProBLE_CONSTS.SCAN_RESPONSE_BYTE_LENGTH: AtmotubeProBLEScanResponse
+}
+
+def get_ble_packet(b:bytearray) -> AtmotubeBLEPacket | None:
+    packet_cls = PACKET_MAP.get(len(b), None)
+    if packet_cls:
+        return packet_cls(b)
+    else:
+        return None
+
+def ble_callback_wrapper(callback):
+    if inspect.iscoroutinefunction(callback):
+        async def wrapped_callback(device:BLEDevice, adv:AdvertisementData) -> None:
+            mfr_data = adv.manufacturer_data.get(
+                        AtmotubeProBLE_CONSTS.MANUFACTURER_DATA_ID,
+                        bytearray(b''))
+            packet = get_ble_packet(mfr_data)
+            await callback(device,packet)
+    else:
+        def wrapped_callback(device:BLEDevice, adv:AdvertisementData) -> None:
+            mfr_data = adv.manufacturer_data.get(
+                        AtmotubeProBLE_CONSTS.MANUFACTURER_DATA_ID,
+                        bytearray(b''))
+            packet = get_ble_packet(mfr_data)
+            callback(device,packet)
+    
+    return wrapped_callback
+
+

atmotube/gatt.py

@@ -7,10 +7,10 @@
 
 from .uuids import AtmotubeProService_UUID, AtmotubeProGATT_UUID
 from .packets import (
-    AtmotubePacket,
+    AtmotubeGATTPacket,
     AtmotubeProStatus, AtmotubeProSPS30, AtmotubeProBME280, AtmotubeProSGPC3)
 
-PacketList: TypeAlias = list[tuple[AtmotubeProGATT_UUID, AtmotubePacket]]
+PacketList: TypeAlias = list[tuple[AtmotubeProGATT_UUID, AtmotubeGATTPacket]]
 
 ATMOTUBE_PRO_PACKETS = {AtmotubeProGATT_UUID.STATUS: AtmotubeProStatus,
                         AtmotubeProGATT_UUID.SPS30: AtmotubeProSPS30,
@@ -42,8 +42,8 @@ def get_available_characteristics(client: BleakClient) -> PacketList:
 
 
 def gatt_notify(client: BleakClient, uuid: str | AtmotubeProGATT_UUID,
-                packet_cls: AtmotubePacket,
-                callback: Callable[[AtmotubePacket], None]) -> Awaitable:
+                packet_cls: AtmotubeGATTPacket,
+                callback: Callable[[AtmotubeGATTPacket], None]) -> Awaitable:
     """
     Start GATT notifications for a specific characteristic UUID.
 
@@ -52,9 +52,9 @@ def gatt_notify(client: BleakClient, uuid: str | AtmotubeProGATT_UUID,
     :param uuid: The UUID of the characteristic to notify
     :type uuid: str | AtmotubeProGATT_UUID
     :param packet_cls: The packet class to instantiate from the received data
-    :type packet_cls: AtmotubePacket
+    :type packet_cls: AtmotubeGATTPacket
     :param callback: The callback function to call when a packet is received
-    :type callback: Callable[[AtmotubePacket], None]
+    :type callback: Callable[[AtmotubeGATTPacket], None]
     :return: An awaitable object representing the notification task
     :rtype: Awaitable
     """
@@ -74,15 +74,15 @@ def packet_callback(char: BleakGATTCharacteristic,
 
 async def start_gatt_notifications(
         client: BleakClient,
-        callback: Callable[[AtmotubePacket], None],
+        callback: Callable[[AtmotubeGATTPacket], None],
         packet_list: PacketList = list(ATMOTUBE_PRO_PACKETS.items())) -> None:
     """
     Start GATT notifications for all specified characteristics.
 
     :param client: The BleakClient instance of the connected Atmotube device
     :type client: BleakClient
     :param callback: The callback function to call when a packet is received
-    :type callback: Callable[[AtmoTubePacket], None]
+    :type callback: Callable[[AtmotubeGATTPacket], None]
     :param packet_list: The list of UUIDs and packet classes to notify
     :type packet_list: PacketList
     """

atmotube/packets.py

@@ -1,5 +1,12 @@
 from abc import abstractmethod
-from ctypes import LittleEndianStructure, c_ubyte, c_byte, c_short, c_int
+from ctypes import (BigEndianStructure,
+                    LittleEndianStructure,
+                    c_ubyte,
+                    c_byte,
+                    c_short,
+                    c_int,
+                    c_char,
+                    c_long)
 from datetime import datetime
 from typing import TypeAlias
 
@@ -14,7 +21,7 @@ class InvalidByteData(Exception):
 # LittleEndianStructure so be mindful that this class is intended to be
 # abstract. It is only exposed to the user to use as a type hint for
 # functions that accept any Atmotube packet.
-class AtmotubePacket(LittleEndianStructure):
+class AtmotubeGATTPacket(LittleEndianStructure):
     """
     Abstract base class for Atmotube data packets.
     """
@@ -49,7 +56,7 @@ def _process_bytes(self) -> None:
         ...
 
 
-class AtmotubeProStatus(AtmotubePacket):
+class AtmotubeProStatus(AtmotubeGATTPacket):
     """
     Represents the status packet from an Atmotube device.
     """
@@ -77,7 +84,7 @@ def _process_bytes(self) -> None:
         self.battery_level = self._battery
 
     def __str__(self) -> str:
-        return (f"StatusPacket(date_time={str(self.date_time)}, "
+        return (f"AtmotubeProStatus(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}, "
@@ -100,7 +107,7 @@ def __eq__(self, other: object) -> bool:
             return False
 
 
-class AtmotubeProSPS30(AtmotubePacket):
+class AtmotubeProSPS30(AtmotubeGATTPacket):
     """
     Represents the SPS30 particulate matter sensor data packet from an
     Atmotube device.
@@ -112,7 +119,7 @@ class AtmotubeProSPS30(AtmotubePacket):
         ('_pm4', c_byte*3),
     ]
 
-    _pack_: int = 1
+    _pack_: bool = True
     _layout_: str = "ms"
     _byte_size_: int = 12
 
@@ -127,7 +134,7 @@ def _process_bytes(self) -> None:
         self.pm4 = self.pm_from_bytes(self._pm4)
 
     def __str__(self) -> str:
-        return (f"SPS30Packet(date_time={str(self.date_time)}, "
+        return (f"AtmotubeProSPS30(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³)")
 
@@ -142,7 +149,7 @@ def __eq__(self, other: object) -> bool:
             return False
 
 
-class AtmotubeProBME280(AtmotubePacket):
+class AtmotubeProBME280(AtmotubeGATTPacket):
     """
     Represents the BME280 environmental sensor data packet from an
     Atmotube device.
@@ -154,7 +161,7 @@ class AtmotubeProBME280(AtmotubePacket):
         ('_T_dec', c_short),
         ]
 
-    _pack_: int = 1
+    _pack_: bool = True
     _layout_: str = "ms"
     _byte_size_: int = 8
 
@@ -164,7 +171,7 @@ def _process_bytes(self) -> None:
         self.pressure = self._P / 100.0 if self._P > 0 else None
 
     def __str__(self) -> str:
-        return (f"BME280Packet(date_time={str(self.date_time)}, "
+        return (f"AtmotubeProBME280(date_time={str(self.date_time)}, "
                 f"humidity={self.humidity}%, "
                 f"temperature={self.temperature}°C, "
                 f"pressure={self.pressure}mbar)")
@@ -179,7 +186,7 @@ def __eq__(self, other: object) -> bool:
             return False
 
 
-class AtmotubeProSGPC3(AtmotubePacket):
+class AtmotubeProSGPC3(AtmotubeGATTPacket):
     """
     Represents the SGPC3 air quality sensor data packet from an
     Atmotube device.
@@ -188,15 +195,15 @@ class AtmotubeProSGPC3(AtmotubePacket):
         ('_tvoc', c_short)
     ]
 
-    _pack_: int = 1
+    _pack_: bool = True
     _layout_: str = "ms"
     _byte_size_: int = 4
 
     def _process_bytes(self) -> None:
         self.tvoc = self._tvoc/1000.0 if self._tvoc > 0 else None
 
     def __str__(self) -> str:
-        return (f"SGPC3Packet(date_time={str(self.date_time)}, "
+        return (f"AtmotubeProSGPC3(date_time={str(self.date_time)}, "
                 f"tvoc={self.tvoc}ppb)")
 
     def __eq__(self, other: object) -> bool:
@@ -205,3 +212,153 @@ def __eq__(self, other: object) -> bool:
                         self.tvoc == other.tvoc))
         else:
             return False
+
+
+class AtmotubeBLEPacket(BigEndianStructure):
+    """
+    Abstract base class for Atmotube data packets.
+    """
+    _byte_size_: int = 0  # To be defined in subclasses
+
+    def __new__(cls, data: bytearray, date_time: datetime | None = 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: bytearray, date_time: datetime | None = None):
+
+        if date_time is None:
+            date_time = datetime.now()
+        self.date_time = date_time
+        self._process_bytes()
+
+    def __repr__(self) -> str:
+        return str(self)
+
+    @abstractmethod
+    def __eq__(self, other: object) -> bool:
+        ...
+
+    @abstractmethod
+    def __str__(self) -> str:
+        ...
+
+    @abstractmethod
+    def _process_bytes(self) -> None:
+        ...
+
+
+class AtmotubeProBLEAdvertising(AtmotubeBLEPacket):
+    """
+    Represents the BLE advertising packet from an Atmotube PRO device.
+    """
+    #>hhbblbb
+    _fields_: FieldList = [
+        ('_tvoc', c_short, 16),
+        ('_devid', c_short, 16),
+        ('_rh', c_byte, 8),
+        ('_T', c_byte, 8),
+        ('_P', c_int, 32),
+        ("_bit_8", c_ubyte, 1),
+        ("_pre_heating", c_ubyte, 1),
+        ("_bit_6",  c_ubyte, 1),
+        ("_charging_timer", c_ubyte, 1),
+        ("_charging", c_ubyte, 1),
+        ("_bonding", c_ubyte, 1),
+        ("_error", c_ubyte, 1),
+        ("_pm_sensor", c_ubyte, 1),
+        ("_battery", c_ubyte, 8),
+    ]
+
+    _pack_: bool = True
+    _layout_: str = "ms"
+    _byte_size_: int = 12
+
+    def _process_bytes(self) -> None:
+        self.tvoc = self._tvoc/1000.0 if self._tvoc > 0 else None
+        self.device_id = self._devid
+        self.humidity = self._rh if self._rh > 0 else None
+        self.temperature = self._T
+        self.pressure = self._P / 100.0 if self._P > 0 else None
+        self.pm_sensor_status = bool(self._pm_sensor)
+        self.error_flag = bool(self._error)
+        self.bonding_flag = bool(self._bonding)
+        self.charging = bool(self._charging)
+        self.charging_timer = bool(self._charging_timer)
+        self.pre_heating = bool(self._pre_heating)
+        self.battery_level = self._battery
+    
+    def __str__(self) -> str:
+        return (f"AtmotubeProBLEAdvertising(date_time={str(self.date_time)}, "
+                f"device_id={self.device_id}, "
+                f"tvoc={self.tvoc}ppb, "
+                f"humidity={self.humidity}%, "
+                f"temperature={self.temperature}°C, "
+                f"pressure={self.pressure}mbar, "
+                f"pm_sensor_status={self.pm_sensor_status}, "
+                f"error_flag={self.error_flag}, "
+                f"bonding_flag={self.bonding_flag}, "
+                f"charging={self.charging}, "
+                f"charging_timer={self.charging_timer}, "
+                f"pre_heating={self.pre_heating}, "
+                f"battery_level={self.battery_level}%)")
+    
+    def __eq__(self, other: object) -> bool:
+        if isinstance(other, AtmotubeProBLEAdvertising):
+            return all((self.date_time == other.date_time,
+                        self.device_id == other.device_id,
+                        self.tvoc == other.tvoc,
+                        self.humidity == other.humidity,
+                        self.temperature == other.temperature,
+                        self.pressure == other.pressure,
+                        self.pm_sensor_status == other.pm_sensor_status,
+                        self.error_flag == other.error_flag,
+                        self.bonding_flag == other.bonding_flag,
+                        self.charging == other.charging,
+                        self.charging_timer == other.charging_timer,
+                        self.pre_heating == other.pre_heating,
+                        self.battery_level == other.battery_level))
+        else:
+            return False
+
+
+class AtmotubeProBLEScanResponse(AtmotubeBLEPacket):
+    """
+    Represents the BLE scan response packet from an Atmotube PRO device.
+    """
+    _fields_: FieldList = [
+        ('_pm1', c_short),
+        ('_pm2_5', c_short),
+        ('_pm10', c_short),
+        ('_fw_maj', c_ubyte),
+        ('_fw_min', c_ubyte),
+        ('_fw_bld', c_ubyte),
+    ]
+
+    _pack_: bool = True
+    _layout_: str = "ms"
+    _byte_size_: int = 9
+
+    def _process_bytes(self) -> None:
+        self.pm1 = self._pm1 if self._pm1 > 0 else None
+        self.pm2_5 = self._pm2_5 if self._pm2_5 > 0 else None
+        self.pm10 = self._pm10 if self._pm10 > 0 else None
+        self.firmware_version = f"{self._fw_maj}.{self._fw_min}.{self._fw_bld}"
+    
+    def __str__(self) -> str:
+        return (f"AtmotubeProBLEScanResponse(date_time={str(self.date_time)}, "
+                f"pm1={self.pm1}µg/m³, "
+                f"pm2_5={self.pm2_5}µg/m³, "
+                f"pm10={self.pm10}µg/m³, "
+                f"firmware_version={self.firmware_version})")
+    
+    def __eq__(self, other: object) -> bool:
+        if isinstance(other, AtmotubeProBLEScanResponse):
+            return all((self.date_time == other.date_time,
+                        self.pm1 == other.pm1,
+                        self.pm2_5 == other.pm2_5,
+                        self.pm10 == other.pm10,
+                        self.firmware_version == other.firmware_version))
+        else:
+            return False