Add locked port

Author: ccrisan

qtoggleserver/eq3bt/eq3btthermostat.py

@@ -18,20 +18,23 @@ class EQ3BTThermostat(ble.BLEPeripheral):
 
     STATUS_MANUAL_MASK = 0x01
     STATUS_BOOST_MASK = 0x04
+    STATUS_LOCKED_MASK = 0x20
 
     STATUS_BITS_INDEX = 2
     STATUS_TEMP_INDEX = 5
 
     WRITE_TEMP_HEADER = 0x41
     WRITE_MANUAL_HEADER = 0x40
     WRITE_BOOST_HEADER = 0x45
+    WRITE_LOCKED_HEADER = 0x80
 
     def __init__(self, **kwargs) -> None:
         super().__init__(**kwargs)
 
         self._temp: Optional[float] = None
         self._manual: Optional[bool] = False
         self._boost: Optional[bool] = False
+        self._locked: Optional[bool] = False
 
     async def set_temp(self, temp: float) -> None:
         self.debug('setting temperature to %.1f degrees', temp)
@@ -63,13 +66,24 @@ async def set_boost(self, boost: bool) -> None:
     def get_boost(self) -> Optional[bool]:
         return self._boost
 
+    async def set_locked(self, locked: bool) -> None:
+        self.debug(['unlocked', 'locked'][locked])
+
+        await self.write(self.WRITE_HANDLE, bytes([self.WRITE_LOCKED_HEADER, int(locked)]))
+        self.debug('successfully set locked')
+        self._locked = locked
+
+    def get_locked(self) -> Optional[bool]:
+        return self._locked
+
     async def make_port_args(self) -> List[Type[core_ports.BasePort]]:
-        from .ports import Temperature, Manual, Boost
+        from .ports import Temperature, Manual, Boost, Locked
 
         return [
             Temperature,
             Manual,
-            Boost
+            Boost,
+            Locked
         ]
 
     async def poll(self) -> None:
@@ -96,13 +110,15 @@ async def _read_config(self) -> None:
         if data[0] != self.STATUS_RECV_HEADER:
             raise EQ3Exception(f'Unexpected notification data header: {data[0]:02X}')
 
+        self._temp = data[self.STATUS_TEMP_INDEX] / 2.0
         self._manual = bool(data[self.STATUS_BITS_INDEX] & self.STATUS_MANUAL_MASK)
         self._boost = bool(data[self.STATUS_BITS_INDEX] & self.STATUS_BOOST_MASK)
-        self._temp = data[self.STATUS_TEMP_INDEX] / 2.0
+        self._locked = bool(data[self.STATUS_BITS_INDEX] & self.STATUS_LOCKED_MASK)
 
         self.debug('temperature is %.1f degrees', self._temp)
         self.debug('manual mode is %s', ['disabled', 'enabled'][self._manual])
         self.debug('boost mode is %s', ['disabled', 'enabled'][self._boost])
+        self.debug('thermostat is %s', ['unlocked', 'locked'][self._locked])
 
     @staticmethod
     def _make_status_value() -> List[int]:

qtoggleserver/eq3bt/ports.py

@@ -44,6 +44,20 @@ async def write_value(self, value: bool) -> None:
         await self.get_peripheral().set_boost(value)
 
 
+class Locked(EQ3BTPort):
+    TYPE = ports.TYPE_BOOLEAN
+    WRITABLE = True
+
+    ID = 'locked'
+
+    async def read_value(self) -> Optional[bool]:
+        return self.get_peripheral().get_locked()
+
+    @ble.port_exceptions
+    async def write_value(self, value: bool) -> None:
+        await self.get_peripheral().set_locked(value)
+
+
 class Temperature(EQ3BTPort):
     TYPE = 'number'
     WRITABLE = True