Merge pull request #3 from rccoleman/async

Improve compatibility with the Home Assistant integration

lmdirect/__init__.py

@@ -13,11 +13,17 @@
 class LMDirect:
     def __init__(self, key):
         """Init LMDirect"""
-        self.run = True
-        self.cipher = AESCipher(key)
-        self.response_task = None
-        self.status_task = None
-        self.current_status = {}
+        self._run = True
+        self._cipher = AESCipher(key)
+        self._read_response_task = None
+        self._poll_status_task = None
+        self._current_status = {}
+        self._callback = None
+
+    def register_callback(self, callback):
+        """Register callback for updates"""
+        if callable(callback):
+            self._callback = callback
 
     async def connect(self, addr):
         """Conmnect to espresso machine"""
@@ -29,34 +35,40 @@ async def connect(self, addr):
         loop = asyncio.get_event_loop()
 
         """Start listening for responses & sending status requests"""
-        self.response_task = loop.create_task(self.response())
-
-        """Start sending status requests"""
-        self.status_task = loop.create_task(self.status())
+        self._read_response_task = loop.create_task(
+            self.read_response_task(), name="Response Task"
+        )
 
     async def close(self):
         """Stop listening for responses and close the socket"""
-        self.run = False
+        self._run = False
+        tasks = [self._read_response_task]
+
+        if self._poll_status_task:
+            tasks.append(self._poll_status_task)
 
-        await asyncio.gather(*[self.response_task, self.status_task])
+        await asyncio.gather(*tasks)
 
         """Close the connection"""
         self.writer.close()
 
-    async def response(self):
+    async def read_response_task(self):
         """Start thread to receive responses"""
         BUFFER_SIZE = 1000
 
-        while self.run:
+        while self._run:
             encoded_data = await self.reader.read(BUFFER_SIZE)
 
             _LOGGER.debug(encoded_data)
             if encoded_data is not None:
-                loop = asyncio.get_running_loop()
-                fn = partial(self.cipher.decrypt, encoded_data[1:-1])
+                loop = asyncio.get_event_loop()
+                fn = partial(self._cipher.decrypt, encoded_data[1:-1])
                 plaintext = await loop.run_in_executor(None, fn)
                 await self.process_data(plaintext)
 
+                if self._callback is not None:
+                    self._callback(self._current_status)
+
     async def process_data(self, plaintext):
         """Process incoming packet"""
 
@@ -77,36 +89,50 @@ async def process_data(self, plaintext):
 
         if any(preamble in x for x in CMD.PREAMBLES):
             await self.populate_items(data, CMD.RESP_MAP[preamble])
-            _LOGGER.debug(self.current_status)
+            _LOGGER.debug(self._current_status)
 
     async def populate_items(self, data, map):
         for elem in map:
             index = elem.index * 2
             size = elem.size * 2
 
-            value = int(data[index: index + size], 16)
-            if any(x in map[elem] for x in ["TEMP", "PREBREWING_K"]):
+            value = int(data[index : index + size], 16)
+            if any(x in map[elem] for x in ["TSET", "TEMP", "PREBREWING_K"]):
                 value = value / 10
             elif "AUTO_BITFIELD" in map[elem]:
                 for i in range(0, 7):
                     setting = ENABLED if value & 0x01 else DISABLED
-                    self.current_status[CMD.AUTO_BITFIELD_MAP[i]] = setting
+                    self._current_status[CMD.AUTO_BITFIELD_MAP[i]] = setting
                     value = value >> 1
                 continue
-            self.current_status[map[elem]] = value
+            self._current_status[map[elem]] = value
+
+    @property
+    def current_status(self):
+        """Return a dict of all the properties that have been received"""
+        return self._current_status
 
-    async def status(self):
+    async def create_polling_task(self):
+        """Start a polling task"""
+        self._poll_status_task = asyncio.get_event_loop().create_task(
+            self.poll_status_task(), name="Request Status Task"
+        )
+
+    async def poll_status_task(self):
         """Send periodic status requests"""
-        while self.run:
-            await self.send_cmd(CMD.STATUS)
-            await self.send_cmd(CMD.CONFIG)
-            await self.send_cmd(CMD.AUTO_SCHED)
+        while self._run:
+            await self.request_status()
             await asyncio.sleep(5)
 
+    async def request_status(self):
+        await self.send_cmd(CMD.STATUS)
+        await self.send_cmd(CMD.CONFIG)
+        await self.send_cmd(CMD.AUTO_SCHED)
+
     async def send_cmd(self, cmd):
         """Send command to espresso machine"""
-        loop = asyncio.get_running_loop()
-        fn = partial(self.cipher.encrypt, cmd)
+        loop = asyncio.get_event_loop()
+        fn = partial(self._cipher.encrypt, cmd)
         ciphertext = "@" + (await loop.run_in_executor(None, fn)).decode("utf-8") + "%"
         _LOGGER.debug(ciphertext)
 

lmdirect/cmds.py

@@ -50,7 +50,7 @@ def size(self):
 # 04 D8: Steam Temp (124.0C)
 # B6
 
-SHORT_MAP = {Element(4, 2): "COFFEE_TEMP", Element(6, 2): "STEAM_TEMP"}
+SHORT_MAP = {Element(4, 2): "TEMP_COFFEE", Element(6, 2): "TEMP_STEAM"}
 
 # R
 # 40 00 00 20: Preamble
@@ -63,7 +63,7 @@ def size(self):
 # 70
 
 D8_MAP = {
-    Element(27, 1): "POWER",
+    Element(27, 1): "MACHINE_STATUS",
     # Element(32, 2): "COFFEE_TEMP",
     # Element(34, 2): "STEAM_TEMP",
 }
@@ -91,9 +91,8 @@ def size(self):
 # 35: 66: Degrees f/c
 
 E9_MAP = {
-    Element(0): "STBY_TIMER",
-    Element(11, 2): "COFFEE_TEMP_SET",
-    Element(13, 2): "STEAM_TEMP_SET",
+    Element(11, 2): "TSET_COFFEE",
+    Element(13, 2): "TSET_STEAM",
     Element(15): "ENABLE_PREBREWING",
     Element(16): "TON_PREBREWING_K1",
     Element(17): "TON_PREBREWING_K2",
@@ -109,7 +108,6 @@ def size(self):
     Element(30, 2): "DOSE_K4",
     Element(32, 2): "DOSE_K5",
     Element(34): "DOSE_TEA",
-    # Element(35): "T_UNITS",
 }
 
 # Response to R 03 10 00 1D EB

test.py

@@ -25,13 +25,19 @@ def read_config():
     return key, ip_addr
 
 
+def update(data):
+    print("Updated: {}".format(data))
+
+
 async def main():
     """Main execution loop"""
     loop = asyncio.get_event_loop()
     key, ip_addr = await loop.run_in_executor(None, read_config)
 
     lmdirect = LMDirect(key)
+    lmdirect.register_callback(update)
     await lmdirect.connect(ip_addr)
+    # await lmdirect.create_polling_task()
 
     while True:
         try:
@@ -51,4 +57,5 @@ async def main():
 
     await lmdirect.close()
 
+
 asyncio.run(main())