Adding device serial number and firmware (#31)

* Implemented some essential tests.

* Added device serial and firware version.

* Removed duplicate property.

* Changed CLI command name; only fetch on request.

* Removed assumptions on unknown bytes; added some essential tests.

* Updated readme.

* Hound fixes.

Author: merll

README.md

@@ -10,6 +10,7 @@ with support for more features and better device handling.
 * Reading device status: locked, low battery, valve state, window open, target temperature, active mode
 * Writing settings: target temperature, auto mode presets, temperature offset
 * Setting the active mode: auto, manual, boost, away
+* Reading the device serial number and firmware version
 
 ## Not (yet) supported)
 

eq3bt/eq3btsmart.py

@@ -84,6 +84,9 @@ def __init__(self, _mac, connection_cls=BTLEConnection):
         self._away_duration = timedelta(days=30)
         self._away_end = None
 
+        self._firmware_version = None
+        self._device_serial = None
+
         self._conn = connection_cls(_mac)
         self._conn.set_callback(PROP_NTFY_HANDLE, self.handle_notification)
 
@@ -149,9 +152,21 @@ def handle_notification(self, data):
             parsed = self.parse_schedule(data)
             self._schedule[parsed.day] = parsed
 
+        elif data[0] == PROP_ID_RETURN:
+            parsed = DeviceId.parse(data)
+            _LOGGER.debug("Parsed device data: %s", parsed)
+            self._firmware_version = parsed.version
+            self._device_serial = parsed.serial
+
         else:
             _LOGGER.debug("Unknown notification %s (%s)", data[0], codecs.encode(data, 'hex'))
 
+    def query_id(self):
+        """Query device identification information, e.g. the serial number."""
+        _LOGGER.debug("Querying id..")
+        value = struct.pack('B', PROP_ID_QUERY)
+        self._conn.make_request(PROP_WRITE_HANDLE, value)
+
     def update(self):
         """Update the data from the thermostat. Always sets the current time."""
         _LOGGER.debug("Querying the device..")
@@ -388,5 +403,11 @@ def max_temp(self):
         return EQ3BT_MAX_TEMP
 
     @property
-    def away_end(self):
-        return self._away_end
+    def firmware_version(self):
+        """Return the firmware version."""
+        return self._firmware_version
+
+    @property
+    def device_serial(self):
+        """Return the device serial number."""
+        return self._device_serial

eq3bt/eq3cli.py

@@ -152,6 +152,16 @@ def away(dev, away_end, temperature):
         click.echo("Disabling away mode")
     dev.set_away(away_end, temperature)
 
+
+@cli.command()
+@pass_dev
+def device(dev):
+    """ Displays basic device information. """
+    dev.query_id()
+    click.echo("Firmware version: %s" % dev.firmware_version)
+    click.echo("Device serial:    %s" % dev.device_serial)
+
+
 @cli.command()
 @click.pass_context
 def state(ctx):

eq3bt/structures.py

@@ -1,9 +1,10 @@
 """ Contains construct adapters and structures. """
-from construct import Struct, Adapter, ExprAdapter, Int8ub, Enum, FlagsEnum, Const, Pass, GreedyRange, GreedyBytes, IfThenElse, If, Bytes, Byte
-import struct
+from construct import (Struct, Adapter, Int8ub, Enum, FlagsEnum, Const,
+                       GreedyRange, GreedyBytes, IfThenElse, Bytes, Byte)
 from datetime import datetime, time
-from math import floor
 
+
+PROP_ID_RETURN = 1
 PROP_INFO_RETURN = 2
 PROP_SCHEDULE_SET = 0x10
 PROP_SCHEDULE_RETURN = 0x21
@@ -71,6 +72,13 @@ def _encode(self, obj, ctx, path):
         return (obj.day, year, hour, obj.month)
 
 
+class DeviceSerialAdapter(Adapter):
+    """ Adapter to decode the device serial number. """
+    def _decode(self, obj, context, path):
+        return bytearray(n - 0x30
+                         for n in obj).decode()
+
+
 Status = "Status" / Struct(
     "cmd" / Const(PROP_INFO_RETURN, Int8ub),
     Const(0x01, Int8ub),
@@ -93,3 +101,12 @@ def _encode(self, obj, ctx, path):
         "next_change_at" / TimeAdapter(Int8ub),
     )),
 )
+
+DeviceId = "DeviceId" / Struct(
+    "cmd" / Const(PROP_ID_RETURN, Int8ub),
+    "version" / Int8ub,
+    Int8ub,
+    Int8ub,
+    "serial" / DeviceSerialAdapter(Bytes(10)),
+    Int8ub,
+)

eq3bt/tests/test_thermostat.py

@@ -1,12 +1,51 @@
 from unittest import TestCase
+
+import codecs
+from datetime import datetime
+
 from eq3bt import Thermostat, TemperatureException
+from eq3bt.eq3btsmart import (PROP_NTFY_HANDLE, PROP_ID_QUERY,
+                              PROP_INFO_QUERY, Mode)
+
+
+ID_RESPONSE = b'01780000807581626163606067659e'
+STATUS_RESPONSES = {
+    'auto': b'020100000428',
+    'manual': b'020101000428',
+    'window': b'020110000428',
+    'away': b'0201020004231d132e03',
+    'boost': b'020104000428',
+    'low_batt': b'020180000428',
+    'valve_at_22': b'020100160428',
+}
+
 
 class FakeConnection:
     def __init__(self, mac):
-        pass
+        self._callbacks = {}
+        self._res = 'auto'
 
     def set_callback(self, handle, cb):
-        pass
+        self._callbacks[handle] = cb
+
+    def set_status(self, key):
+        if key in STATUS_RESPONSES:
+            self._res = key
+        else:
+            raise ValueError("Invalid key for status test response.")
+
+    def make_request(self, handle, value, timeout=1, with_response=True):
+        """Write a GATT Command without callback - not utf-8."""
+        if with_response:
+            cb = self._callbacks.get(PROP_NTFY_HANDLE)
+
+            if value[0] == PROP_ID_QUERY:
+                data = ID_RESPONSE
+            elif value[0] == PROP_INFO_QUERY:
+                data = STATUS_RESPONSES[self._res]
+            else:
+                return
+            cb(codecs.decode(data, 'hex'))
 
 
 class TestThermostat(TestCase):
@@ -28,8 +67,38 @@ def test_parse_schedule(self):
     def test_handle_notification(self):
         self.fail()
 
+    def test_query_id(self):
+        self.thermostat.query_id()
+        self.assertEqual(self.thermostat.firmware_version, 120)
+        self.assertEqual(self.thermostat.device_serial, "PEQ2130075")
+
     def test_update(self):
-        self.fail()
+        th = self.thermostat
+
+        th._conn.set_status('auto')
+        th.update()
+        self.assertEqual(th.valve_state, 0)
+        self.assertEqual(th.mode, Mode.Auto)
+        self.assertEqual(th.target_temperature, 20.0)
+        self.assertFalse(th.locked)
+        self.assertFalse(th.low_battery)
+        self.assertFalse(th.boost)
+        self.assertFalse(th.window_open)
+
+        th._conn.set_status('manual')
+        th.update()
+        self.assertTrue(th.mode, Mode.Manual)
+
+        th._conn.set_status('away')
+        th.update()
+        self.assertEqual(th.mode, Mode.Away)
+        self.assertEqual(th.target_temperature, 17.5)
+        self.assertEqual(th.away_end, datetime(2019, 3, 29, 23, 00))
+
+        th._conn.set_status('boost')
+        th.update()
+        self.assertTrue(th.boost)
+        self.assertEqual(th.mode, Mode.Boost)
 
     def test_query_schedule(self):
         self.fail()
@@ -53,10 +122,16 @@ def test_boost(self):
         self.fail()
 
     def test_valve_state(self):
-        self.fail()
+        th = self.thermostat
+        th._conn.set_status('valve_at_22')
+        th.update()
+        self.assertEqual(th.valve_state, 22)
 
     def test_window_open(self):
-        self.fail()
+        th = self.thermostat
+        th._conn.set_status('window')
+        th.update()
+        self.assertTrue(th.window_open)
 
     def test_window_open_config(self):
         self.fail()
@@ -65,7 +140,10 @@ def test_locked(self):
         self.fail()
 
     def test_low_battery(self):
-        self.fail()
+        th = self.thermostat
+        th._conn.set_status('low_batt')
+        th.update()
+        self.assertTrue(th.low_battery)
 
     def test_temperature_offset(self):
         self.fail()