Merge pull request #2 from rccoleman/async

Convert to async

Author: rccoleman

README.md

@@ -31,10 +31,6 @@ Now, run `python test.py` and you should get a prompt that looks like this:
 
 `1 = ON, 2 = OFF, 3 = Status, Other = quit:`
 
-You'll also see an `INFO` message when the machine at first when the current machine state is detected and whenever it changes after that.  It looks like this:
-
-`INFO:lmdirect:Device is OFF`.
-
 You can hit `1` to turn the machine on, `2` to turn it off, `3` to dump a dict of all the config and status items that it's received from your machine, and any other key to quit.  The app requests all status & config information every 5 seconds, so you should see the values change when the state of the machine changes.
 
 Note that the machine only accepts a single connection at a time, so you cannot run this app and the mobile app at the same time.  The second one will block until you close the first instance.  This means that you can't experiment by running this app and manipulating settings using the mobile app simultaneously, but you can change settings on the machine itself and see the values update here.

lmdirect/__init__.py

@@ -2,13 +2,10 @@
 from . import cmds as CMD
 from .const import *
 
-import socket, select
-from threading import Thread
-from time import sleep
-
+import asyncio
+from functools import partial
 import logging
 
-
 _LOGGER = logging.getLogger(__name__)
 _LOGGER.setLevel(logging.INFO)
 
@@ -18,53 +15,49 @@ def __init__(self, key):
         """Init LMDirect"""
         self.run = True
         self.cipher = AESCipher(key)
-        self.response_thread = None
-        self.status_thread = None
+        self.response_task = None
+        self.status_task = None
         self.current_status = {}
-        self.is_on = None
 
-    def connect(self, addr):
+    async def connect(self, addr):
         """Conmnect to espresso machine"""
         TCP_PORT = 1774
 
-        self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        self.s.connect((addr, TCP_PORT))
+        """Connect to the machine"""
+        self.reader, self.writer = await asyncio.open_connection(addr, TCP_PORT)
 
-        """Start listening for responses"""
-        self.response_thread = Thread(target=self.response, name="Response")
-        self.response_thread.start()
+        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_thread = Thread(target=self.status, name="Status")
-        self.status_thread.start()
+        self.status_task = loop.create_task(self.status())
 
-    def close(self):
+    async def close(self):
         """Stop listening for responses and close the socket"""
         self.run = False
-        threads = [self.response_thread, self.status_thread]
 
-        [t.join() for t in threads if t is not None]
+        await asyncio.gather(*[self.response_task, self.status_task])
 
-        self.s.shutdown(socket.SHUT_WR)
-        self.s.close()
+        """Close the connection"""
+        self.writer.close()
 
-    def response(self):
+    async def response(self):
         """Start thread to receive responses"""
         BUFFER_SIZE = 1000
 
         while self.run:
-            reading = [self.s]
-            writing = []
-            exceptions = []
-            select.select(reading, writing, exceptions, 1)
-            if self.s in reading:
-                encoded_data = self.s.recv(BUFFER_SIZE)
-                _LOGGER.debug(encoded_data)
-                if encoded_data is not None:
-                    plaintext = self.cipher.decrypt(encoded_data[1:-1])
-                    self.process_data(plaintext)
-
-    def process_data(self, plaintext):
+            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])
+                plaintext = await loop.run_in_executor(None, fn)
+                await self.process_data(plaintext)
+
+    async def process_data(self, plaintext):
         """Process incoming packet"""
 
         """Separate the preamble from the data"""
@@ -82,39 +75,18 @@ def process_data(self, plaintext):
             )
             return
 
-        if preamble == CMD.D8_PREAMBLE:
-            _LOGGER.debug("D8 status")
-            self.populate_items(data, CMD.D8_MAP)
-
-            new_state = self.current_status["POWER"] == 1
-            if new_state is not self.is_on:
-                self.is_on = new_state
-
-                _LOGGER.info("Device is {}".format("ON" if self.is_on else "OFF"))
-        elif preamble == CMD.E9_PREAMBLE:
-            _LOGGER.debug("E9 status")
-            self.populate_items(data, CMD.E9_MAP)
-        elif preamble == CMD.SHORT_PREAMBLE:
-            _LOGGER.debug("Short status")
-            self.populate_items(data, CMD.SHORT_MAP)
-        elif preamble == CMD.EB_PREAMBLE:
-            _LOGGER.debug("EB auto schedule")
-            self.populate_items(data, CMD.EB_MAP)
-
-        _LOGGER.debug(self.current_status)
+        if any(preamble in x for x in CMD.PREAMBLES):
+            await self.populate_items(data, CMD.RESP_MAP[preamble])
+            _LOGGER.debug(self.current_status)
 
-    def populate_items(self, data, map):
+    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)
+            value = int(data[index: index + size], 16)
             if any(x in map[elem] for x in ["TEMP", "PREBREWING_K"]):
                 value = value / 10
-            elif "UNITS" in map[elem]:
-                print(data)
-                print("{}, {}".format(value, chr(value)))
-                value = u"\xb0F" if chr(value) == "f" else u"\xb0C"
             elif "AUTO_BITFIELD" in map[elem]:
                 for i in range(0, 7):
                     setting = ENABLED if value & 0x01 else DISABLED
@@ -123,19 +95,21 @@ def populate_items(self, data, map):
                 continue
             self.current_status[map[elem]] = value
 
-    def status(self):
+    async def status(self):
         """Send periodic status requests"""
         while self.run:
-            self.send_cmd(CMD.STATUS)
-            self.send_cmd(CMD.CONFIG)
-            self.send_cmd(CMD.AUTO_SCHED)
-            sleep(5)
+            await self.send_cmd(CMD.STATUS)
+            await self.send_cmd(CMD.CONFIG)
+            await self.send_cmd(CMD.AUTO_SCHED)
+            await asyncio.sleep(5)
 
-    def send_cmd(self, cmd):
+    async def send_cmd(self, cmd):
         """Send command to espresso machine"""
-        ciphertext = "@" + self.cipher.encrypt(cmd).decode("utf-8") + "%"
+        loop = asyncio.get_running_loop()
+        fn = partial(self.cipher.encrypt, cmd)
+        ciphertext = "@" + (await loop.run_in_executor(None, fn)).decode("utf-8") + "%"
         _LOGGER.debug(ciphertext)
 
         _LOGGER.debug("Before sending: {}".format(ciphertext))
-        self.s.send(bytes(ciphertext, "utf-8"))
-        _LOGGER.debug("After sending")
+        self.writer.write(bytes(ciphertext, "utf-8"))
+        await self.writer.drain()

lmdirect/aescipher.py

@@ -5,7 +5,7 @@
 import logging
 
 _LOGGER = logging.getLogger(__name__)
-_LOGGER.setLevel(logging.DEBUG)
+_LOGGER.setLevel(logging.INFO)
 
 
 class AESCipher:

lmdirect/cmds.py

@@ -20,6 +20,13 @@
 E9_PREAMBLE = "0000001F"
 EB_PREAMBLE = "0310001D"
 
+PREAMBLES = [
+    SHORT_PREAMBLE,
+    D8_PREAMBLE,
+    E9_PREAMBLE,
+    EB_PREAMBLE,
+]
+
 RESPONSE_GOOD = "OK"
 
 # Response Maps
@@ -153,4 +160,11 @@ def size(self):
     5: "THU_AUTO",
     6: "FRI_AUTO",
     7: "SAT_AUTO",
+}
+
+RESP_MAP = {
+    D8_PREAMBLE: D8_MAP,
+    E9_PREAMBLE: E9_MAP,
+    SHORT_PREAMBLE: SHORT_MAP,
+    EB_PREAMBLE: EB_MAP,
 }

lmdirect/const.py

@@ -1,3 +1,3 @@
 """Constants for lmdirect"""
 ENABLED = "ENABLED"
-DISABLED = "DISABLED"
+DISABLED = "DISABLED"

test.py

@@ -1,38 +1,54 @@
+import asyncio
 from lmdirect import LMDirect
-from lmdirect.cmds import ON, OFF
 import json
+import sys
 import logging
+from lmdirect.cmds import ON, OFF
 
 logging.basicConfig(level=logging.DEBUG)
 _LOGGER = logging.getLogger(__name__)
 _LOGGER.setLevel(logging.DEBUG)
 
-try:
-    with open("config.json") as config_file:
-        data = json.load(config_file)
 
-        key = data["key"]
-        ip_addr = data["ip_addr"]
-except Exception as err:
-    print(err)
-    exit(1)
+def read_config():
+    """Read key and machine IP from config file"""
+    try:
+        with open("config.json") as config_file:
+            data = json.load(config_file)
 
-lmdirect = LMDirect(key)
-lmdirect.connect(ip_addr)
+            key = data["key"]
+            ip_addr = data["ip_addr"]
+    except Exception as err:
+        print(err)
+        exit(1)
 
-while True:
-    try:
-        response = input("1 = ON, 2 = OFF, 3 = Status, Other = quit: ")
-        if response == "1":
-            lmdirect.send_cmd(ON)
-        elif response == "2":
-            lmdirect.send_cmd(OFF)
-        elif response == "3":
-            print(lmdirect.current_status)
-        else:
+    return key, ip_addr
+
+
+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)
+    await lmdirect.connect(ip_addr)
+
+    while True:
+        try:
+            print("\n1 = ON, 2 = OFF, 3 = Status, Other = quit: ")
+            response = (await loop.run_in_executor(None, sys.stdin.readline)).rstrip()
+
+            if response == "1":
+                await lmdirect.send_cmd(ON)
+            elif response == "2":
+                await lmdirect.send_cmd(OFF)
+            elif response == "3":
+                print(lmdirect.current_status)
+            else:
+                break
+        except KeyboardInterrupt:
             break
-    except KeyboardInterrupt:
-        break
 
-lmdirect.close()
-exit(0)
+    await lmdirect.close()
+
+asyncio.run(main())