From b7a35b00513a28e613103789390002fb0c7bf23f Mon Sep 17 00:00:00 2001
From: sverre <sverre.hamre@gmail.com>
Date: Sun, 30 Jan 2022 19:26:07 +0100
Subject: [PATCH] Adds back end support for getting battery voltage by
 communicating with the command interface uuid. Tested and working with the
 airthings pluss only. #70

---
 custom_components/airthings_wave/airthings.py | 70 ++++++++++++++++++-
 1 file changed, 68 insertions(+), 2 deletions(-)

diff --git a/custom_components/airthings_wave/airthings.py b/custom_components/airthings_wave/airthings.py
index 98ca30f..b2588d4 100644
--- a/custom_components/airthings_wave/airthings.py
+++ b/custom_components/airthings_wave/airthings.py
@@ -12,6 +12,7 @@
 _LOGGER = logging.getLogger(__name__)
 
 # Use full UUID since we do not use UUID from bluepy.btle
+CHAR_UUID_CCCD   = btle.UUID('2902') # Client Characteristic Configuration Descriptor (CCCD)
 CHAR_UUID_MANUFACTURER_NAME = UUID('00002a29-0000-1000-8000-00805f9b34fb')
 CHAR_UUID_SERIAL_NUMBER_STRING = UUID('00002a25-0000-1000-8000-00805f9b34fb')
 CHAR_UUID_MODEL_NUMBER_STRING = UUID('00002a24-0000-1000-8000-00805f9b34fb')
@@ -25,6 +26,7 @@
 CHAR_UUID_WAVE_PLUS_DATA = UUID('b42e2a68-ade7-11e4-89d3-123b93f75cba')
 CHAR_UUID_WAVE_2_DATA = UUID('b42e4dcc-ade7-11e4-89d3-123b93f75cba')
 CHAR_UUID_WAVEMINI_DATA = UUID('b42e3b98-ade7-11e4-89d3-123b93f75cba')
+COMMAND_UUID = UUID('b42e2d06-ade7-11e4-89d3-123b93f75cba') # "Access Control Point" Characteristic
 
 Characteristic = namedtuple('Characteristic', ['uuid', 'name', 'format'])
 
@@ -48,7 +50,8 @@ def __str__(self):
 
 sensors_characteristics_uuid = [CHAR_UUID_DATETIME, CHAR_UUID_TEMPERATURE, CHAR_UUID_HUMIDITY, CHAR_UUID_RADON_1DAYAVG,
                                 CHAR_UUID_RADON_LONG_TERM_AVG, CHAR_UUID_ILLUMINANCE_ACCELEROMETER,
-                                CHAR_UUID_WAVE_PLUS_DATA,CHAR_UUID_WAVE_2_DATA,CHAR_UUID_WAVEMINI_DATA]
+                                CHAR_UUID_WAVE_PLUS_DATA,CHAR_UUID_WAVE_2_DATA,CHAR_UUID_WAVEMINI_DATA,
+                                COMMAND_UUID]
 
 sensors_characteristics_uuid_str = [str(x) for x in sensors_characteristics_uuid]
 
@@ -127,6 +130,46 @@ def decode_data(self, raw_data):
         return data
 
 
+class CommandDecode:
+    def __init__(self, name, format_type, cmd):
+        self.name = name
+        self.format_type = format_type
+        self.cmd = cmd
+
+    def decode_data(self, raw_data): 
+        cmd = raw_data[0:1]
+        if cmd != self.cmd:
+            _LOGGER.warning("Result for Wrong command received, expected {} got {}".format(self.cmd.hex(), cmd.hex()))
+            return {}
+                
+        val = struct.unpack(
+            self.format_type,
+            raw_data[2:])
+        res = {}
+        res['ambientlight'] = val[2]
+        res['measurement_periods'] =  val[5]
+        res['voltage'] = val[17] / 1000.0
+
+        V_MAX=3.2
+        V_MIN=2.2
+        res['battery']= max(0, min(100, round( (res['voltage']-V_MIN)/(V_MAX-V_MIN)*100)))
+
+        return res
+
+
+class MyDelegate(btle.DefaultDelegate):
+    def __init__(self):
+        btle.DefaultDelegate.__init__(self)
+        # ... initialise here
+        self.data = None
+
+    def handleNotification(self, cHandle, data):
+        if self.data is None:
+            self.data = data
+        else:
+            self.data = self.data + data
+
+
 sensor_decoders = {str(CHAR_UUID_WAVE_PLUS_DATA):WavePlussDecode(name="Pluss", format_type='BBBBHHHHHHHH', scale=0),
                    str(CHAR_UUID_DATETIME):WaveDecodeDate(name="date_time", format_type='HBBBBB', scale=0),
                    str(CHAR_UUID_HUMIDITY):BaseDecode(name="humidity", format_type='H', scale=1.0/100.0),
@@ -137,6 +180,8 @@ def decode_data(self, raw_data):
                    str(CHAR_UUID_WAVE_2_DATA):Wave2Decode(name="Wave2", format_type='<4B8H', scale=1.0),
                    str(CHAR_UUID_WAVEMINI_DATA):WaveMiniDecode(name="WaveMini", format_type='<HHHHHHLL', scale=1.0),}
 
+command_decoders = {str(COMMAND_UUID):CommandDecode(name="Battery", format_type='<L12B6H', cmd=struct.pack('<B', 0x6d))}
+
 
 class AirthingsWaveDetect:
     def __init__(self, scan_interval, mac=None):
@@ -158,7 +203,7 @@ def _parse_serial_number(self, manufacturer_data):
 
     def find_devices(self, scans=50, timeout=0.1):
         # Search for devices, scan for BLE devices scans times for timeout seconds
-        # Get manufacturer data and try to match match it to airthings ID.
+        # Get manufacturer data and try to match it to airthings ID.
         scanner = btle.Scanner()
         for _count in range(scans):
             advertisements = scanner.scan(timeout)
@@ -178,6 +223,8 @@ def connect(self, mac, retries=10):
             tries += 1
             try:
                 self._dev = btle.Peripheral(mac.lower())
+                self.delgate = MyDelegate()
+                self._dev.withDelegate( self.delgate )
                 break
             except Exception as e:
                 print(e)
@@ -238,15 +285,34 @@ def get_sensor_data(self):
                 if self._dev is not None:
                     try:
                         for characteristic in characteristics:
+                            sensor_data = None
                             if str(characteristic.uuid) in sensor_decoders:
                                 char = self._dev.getCharacteristics(uuid=characteristic.uuid)[0]
                                 data = char.read()
                                 sensor_data = sensor_decoders[str(characteristic.uuid)].decode_data(data)
                                 _LOGGER.debug("{} Got sensordata {}".format(mac, sensor_data))
+                                
+                            if str(characteristic.uuid) in command_decoders:
+                                self.delgate.data = None # Clear the delegate so it is ready for new data.
+                                char = self._dev.getCharacteristics(uuid=characteristic.uuid)[0]
+                                # Do these steps to get notification to work, I do not know how it works, this link should explain it
+                                # https://devzone.nordicsemi.com/guides/short-range-guides/b/bluetooth-low-energy/posts/ble-characteristics-a-beginners-tutorial
+                                desc, = char.getDescriptors(forUUID=CHAR_UUID_CCCD)
+                                desc.write(struct.pack('<H', 1), True)
+                                char.write(command_decoders[str(characteristic.uuid)].cmd)
+                                for i in range(3):
+                                    if self._dev.waitForNotifications(0.1):
+                                        _LOGGER.debug("Received notification, total data received len {}".format(len(self.delgate.data)))
+                                
+                                sensor_data = command_decoders[str(characteristic.uuid)].decode_data(self.delgate.data)
+                                _LOGGER.debug("{} Got cmddata {}".format(mac, sensor_data))
+
+                            if sensor_data is not None:
                                 if self.sensordata.get(mac) is None:
                                     self.sensordata[mac] = sensor_data
                                 else:
                                     self.sensordata[mac].update(sensor_data)
+
                     except btle.BTLEDisconnectError:
                         _LOGGER.exception("Disconnected")
                         self._dev = None