feat: add support for raw data parsing (#30)

Author: bdraco

src/sensorpro_ble/parser.py

@@ -11,9 +11,9 @@
 import logging
 from struct import unpack
 
-from bluetooth_data_tools import short_address
+from bluetooth_data_tools import parse_advertisement_data_bytes, short_address
 from bluetooth_sensor_state_data import BluetoothData
-from home_assistant_bluetooth import BluetoothServiceInfo
+from habluetooth import BluetoothServiceInfoBleak
 from sensor_state_data import SensorLibrary
 
 _LOGGER = logging.getLogger(__name__)
@@ -30,12 +30,19 @@
 class SensorProBluetoothDeviceData(BluetoothData):
     """Date update for SensorPro Bluetooth devices."""
 
-    def _start_update(self, service_info: BluetoothServiceInfo) -> None:
+    def _start_update(self, service_info: BluetoothServiceInfoBleak) -> None:
         """Update from BLE advertisement data."""
         _LOGGER.debug("Parsing sensorpro BLE advertisement data: %s", service_info)
-        if 43605 not in service_info.manufacturer_data:
+        if service_info.raw:
+            # If we have the raw data we don't need to work out
+            # which one is the newest.
+            _, _, _, changed_manufacturer_data, _ = parse_advertisement_data_bytes(
+                service_info.raw
+            )
+        else:
+            changed_manufacturer_data = self.changed_manufacturer_data(service_info)
+        if 43605 not in changed_manufacturer_data:
             return
-        changed_manufacturer_data = self.changed_manufacturer_data(service_info)
         if not changed_manufacturer_data or len(changed_manufacturer_data) > 1:
             # If len(changed_manufacturer_data) > 1 it means we switched
             # ble adapters so we do not know which data is the latest

tests/test_parser.py

@@ -1,4 +1,9 @@
-from bluetooth_sensor_state_data import BluetoothServiceInfo, SensorUpdate
+from uuid import UUID
+
+from bleak.backends.device import BLEDevice
+from bluetooth_data_tools import monotonic_time_coarse
+from bluetooth_sensor_state_data import SensorUpdate
+from habluetooth import BluetoothServiceInfoBleak
 from sensor_state_data import (
     DeviceKey,
     SensorDescription,
@@ -11,11 +16,44 @@
 from sensorpro_ble.parser import SensorProBluetoothDeviceData
 
 
+def make_bluetooth_service_info(  # noqa: PLR0913
+    name: str,
+    manufacturer_data: dict[int, bytes],
+    service_uuids: list[str],
+    address: str,
+    rssi: int,
+    service_data: dict[UUID, bytes],
+    source: str,
+    tx_power: int = 0,
+    raw: bytes | None = None,
+) -> BluetoothServiceInfoBleak:
+    return BluetoothServiceInfoBleak(
+        name=name,
+        manufacturer_data=manufacturer_data,
+        service_uuids=service_uuids,
+        address=address,
+        rssi=rssi,
+        service_data=service_data,
+        source=source,
+        device=BLEDevice(
+            name=name,
+            address=address,
+            details={},
+            rssi=rssi,
+        ),
+        time=monotonic_time_coarse(),
+        advertisement=None,
+        connectable=True,
+        tx_power=tx_power,
+        raw=raw,
+    )
+
+
 def test_can_create():
     SensorProBluetoothDeviceData()
 
 
-MFR_T201 = BluetoothServiceInfo(
+MFR_T201 = make_bluetooth_service_info(
     name="T201",
     address="aa:bb:cc:dd:ee:ff",
     rssi=-60,
@@ -27,7 +65,7 @@ def test_can_create():
     source="local",
 )
 
-MFR_T301 = BluetoothServiceInfo(
+MFR_T301 = make_bluetooth_service_info(
     name="T301",
     address="aa:bb:cc:dd:ee:ff",
     rssi=-60,
@@ -37,8 +75,19 @@ def test_can_create():
     source="local",
 )
 
-MFR_T301_NO_NAME = BluetoothServiceInfo(
-    name=None,
+MFR_T301_RAW = make_bluetooth_service_info(
+    name="T301",
+    address="aa:bb:cc:dd:ee:ff",
+    rssi=-60,
+    service_data={},
+    manufacturer_data={44: b""},
+    service_uuids=[],
+    source="local",
+    raw=b"\x14\xffU\xaa\x01\x05\xa4\xc18\x1aWv\x01\x07\tV\x17\x0ca\x00\x01",
+)
+
+MFR_T301_NO_NAME = make_bluetooth_service_info(
+    name="",
     address="aa:bb:cc:dd:ee:ff",
     rssi=-60,
     service_data={},
@@ -194,6 +243,79 @@ def test_t301():
     )
 
 
+def test_t301_raw():
+    parser = SensorProBluetoothDeviceData()
+    update = parser.update(MFR_T301_RAW)
+    assert update == SensorUpdate(
+        title="T301 EEFF",
+        devices={
+            None: SensorDeviceInfo(
+                name="T301 EEFF",
+                model=5,
+                manufacturer="SensorPro",
+                sw_version=None,
+                hw_version=None,
+            )
+        },
+        entity_descriptions={
+            DeviceKey(key="voltage", device_id=None): SensorDescription(
+                device_key=DeviceKey(key="voltage", device_id=None),
+                device_class=SensorDeviceClass.VOLTAGE,
+                native_unit_of_measurement=Units.ELECTRIC_POTENTIAL_VOLT,
+            ),
+            DeviceKey(key="humidity", device_id=None): SensorDescription(
+                device_key=DeviceKey(key="humidity", device_id=None),
+                device_class=SensorDeviceClass.HUMIDITY,
+                native_unit_of_measurement=Units.PERCENTAGE,
+            ),
+            DeviceKey(key="temperature", device_id=None): SensorDescription(
+                device_key=DeviceKey(key="temperature", device_id=None),
+                device_class=SensorDeviceClass.TEMPERATURE,
+                native_unit_of_measurement=Units.TEMP_CELSIUS,
+            ),
+            DeviceKey(key="battery", device_id=None): SensorDescription(
+                device_key=DeviceKey(key="battery", device_id=None),
+                device_class=SensorDeviceClass.BATTERY,
+                native_unit_of_measurement=Units.PERCENTAGE,
+            ),
+            DeviceKey(key="signal_strength", device_id=None): SensorDescription(
+                device_key=DeviceKey(key="signal_strength", device_id=None),
+                device_class=SensorDeviceClass.SIGNAL_STRENGTH,
+                native_unit_of_measurement=Units.SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
+            ),
+        },
+        entity_values={
+            DeviceKey(key="voltage", device_id=None): SensorValue(
+                device_key=DeviceKey(key="voltage", device_id=None),
+                name="Voltage",
+                native_value=2.63,
+            ),
+            DeviceKey(key="humidity", device_id=None): SensorValue(
+                device_key=DeviceKey(key="humidity", device_id=None),
+                name="Humidity",
+                native_value=59.0,
+            ),
+            DeviceKey(key="temperature", device_id=None): SensorValue(
+                device_key=DeviceKey(key="temperature", device_id=None),
+                name="Temperature",
+                native_value=23.9,
+            ),
+            DeviceKey(key="battery", device_id=None): SensorValue(
+                device_key=DeviceKey(key="battery", device_id=None),
+                name="Battery",
+                native_value=97,
+            ),
+            DeviceKey(key="signal_strength", device_id=None): SensorValue(
+                device_key=DeviceKey(key="signal_strength", device_id=None),
+                name="Signal " "Strength",
+                native_value=-60,
+            ),
+        },
+        binary_entity_descriptions={},
+        binary_entity_values={},
+    )
+
+
 def test_t301_passive():
     parser = SensorProBluetoothDeviceData()
     update = parser.update(MFR_T301_NO_NAME)