Tidy up code and working quite well

Author: dala318

lib/device.py

@@ -1,111 +1,129 @@
+import threading
 import socket
+import logging
+from time import sleep
 
 PH803W_DEFAULT_TCP_PORT = 12416
-PH803W_PING_INTERVAL = 4000
-RECONNECT_DELAY = 10000
-RESPONSE_TIMEOUT = 5000
+PH803W_PING_INTERVAL = 4
+RECONNECT_DELAY = 10
+# RESPONSE_TIMEOUT = 5000
+ABORT_AFTER_CONSECUTIVE_EMPTY = 10
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class DeviceError(ConnectionError):
+    pass
 
 
 class Device(object):
     def __init__(self, host):
-        self.result = {}
         self.host = host
+        self.passcode = ""
+        self._measurements = []
+        self._measurements_counter = 0
         self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
         self._loop = True
+        self._empty_counter = 0
+        self._pong_thread = None
 
-    async def run_async(self):
-        self.run()
+    async def run_async(self, once: bool = True) -> bool:
+        return self.run(once)
 
-    def run(self, once: bool = False) -> bool:
+    def run(self, once: bool = True) -> bool:
+        if once:
+            self._connect()
+            return self._run(once)
+        else:
+            measurements = -1
+            while self._loop:
+                if measurements == self._measurements_counter:
+                    _LOGGER.error("Aborting reconnects, no new measurements")
+                    break
+                measurements = self._measurements_counter
+                self._connect()
+                try:
+                    self._run(once)
+                except Exception as e:
+                    _LOGGER.warning("Exception in run loop", e)
+                    sleep(RECONNECT_DELAY)                
+            return not self._loop and self._measurements_counter > 0
+
+    def _connect(self) -> bool:
         self._loop = True
         self._socket.connect((self.host, PH803W_DEFAULT_TCP_PORT))
 
+        # Send request for connection
         data = bytes.fromhex("0000000303000006")
         self._socket.sendall(data)
+
+        # Receive response and passcode
         response = self._socket.recv(1024)
         passcode_lenth = response[9]
         passcode_raw = response[10 : 10 + passcode_lenth]
-        passcode = passcode_raw.decode("utf-8")
-        # print(passcode)
+        self.passcode = passcode_raw.decode("utf-8")
+        _LOGGER.debug(self.passcode)
 
+        # Send passcode confirmation
         data = (
             bytes.fromhex("000000030f00000800")
             + passcode_lenth.to_bytes(1, "little")
             + passcode_raw
         )
         self._socket.sendall(data)
+
+        # Receive confirmation
         response = self._socket.recv(1024)
         if response[8] != 0:
-            #     print("Error connecting")
-            return False
+            raise DeviceError("Error connecting")
 
-        # Connection established, from now on some cyclig bahavior
+    def _run(self, once: bool = True) -> bool:
+        # Connection established, start requesting data,
+        # from now on some cyclig bahavior
         data = bytes.fromhex("000000030400009002")
         self._socket.sendall(data)
-        empty_counter = 0
-        data = bytes.fromhex("0000000303000015")
-        while self._loop and empty_counter < 10:
+
+        # If continous reading ping/pong needs to be run cyclic
+        if not once:
+            self._pong_thread = threading.Thread(target=self._ping_loop)
+            self._pong_thread.daemon = True
+            self._pong_thread.start()
+        else:
+            self._send_ping()
+
+        while self._loop:
             response = self._socket.recv(1024)
+            if self._empty_counter > ABORT_AFTER_CONSECUTIVE_EMPTY:
+                _LOGGER.error("Too many empty consecutive packages")
+                raise DeviceError("Too many empty consecutive packages")
             if len(response) == 0:
-                empty_counter += 1
+                self._empty_counter += 1
                 continue
-            empty_counter = 0
-            # print(response)
-            if len(response) == 18:
-                flag1 = response[8]
-                if flag1 & 0b0000_0100:
-                    print("In water")
-                flag2 = response[9]
-                if flag2 & 0b0000_0010:
-                    print("ORP on")
-                if flag2 & 0b0000_0001:
-                    print("PH on")
-                # state_raw = response[8 : 9]
-                ph_raw = response[10:12]
-                ph = int.from_bytes(ph_raw, "big") * 0.01
-                redox_raw = response[12:14]
-                redox = int.from_bytes(redox_raw, "big") - 2000
-                unknown1_raw = response[14:16]
-                unknown1 = int.from_bytes(unknown1_raw, "big")
-                unknown2_raw = response[15:18]
-                unknown2 = int.from_bytes(unknown2_raw, "big")
-                print(
-                    "pH: %s, Redox: %s, U1: %s, U2: %s"
-                    % (ph, redox, unknown1, unknown2),
-                    flush=True,
-                )
-            self._socket.sendall(data)
-            response = self._socket.recv(1024)
-            if once:
-                break
+            self._empty_counter = 0
+
+            self._handle_response(response)
+
+            if once and len(self._measurements) > 0:
+                self._loop = False
+        
+        return (once and len(self._measurements) > 0) or not once
 
     def _handle_response(self, data):
         if data[0] != 0 and data[1] != 0 and data[2] != 0 and data[2] != 3:
-            # print("Ignore data package because invalid prefix: %s" % data[0:3])
-            self.result["status"] = [
-                "Error",
-                "Ignore data package because invalid prefix",
-            ]
+            _LOGGER.warning("Ignore data package because invalid prefix: %s" % data[0:3])
             return
         data_length = data[4]
         if len(data) != data_length + 5:
             if len(data) > data_length:
                 additional_data = data[data_length : len(data)]
                 data = data[0:data_length]
-                # print(
-                #     "Split into two data packages because additional data detected. First %s - Second %s}"
-                #     % (data.toString("hex"), additional_data.toString("hex"))
-                # )
+                _LOGGER.debug("Split into two data packages because additional data detected.")
                 self._handle_response(additional_data)
             else:
-                # print(
-                #     "Ignore data package because invalid length(%s): %s"
-                #     % (data_length, data)
-                # )
-                self.result["status"] = [
-                    "Error",
-                    "Ignore data package because invalid length",
-                ]
+                _LOGGER.warning(
+                    "Ignore data package because invalid length(%s): %s"
+                    % (data_length, data)
+                )
                 return
 
         message_type = data[7]
@@ -120,46 +138,54 @@ def _handle_response(self, data):
         elif message_type == 0x94:
             self._handle_data_extended_response(data)
         else:
-            # print(
-            #     "Ignore data package because invalid message type %s: %s"
-            #     % (message_type, data)
-            # )
-            self.result["status"] = [
-                "Ignore",
-                "Ignore data package because invalid length",
-                message_type,
-                data,
-            ]
+            pass
+            _LOGGER.warning(
+                "Ignore data package because invalid message type %s: %s"
+                % (message_type, data)
+            )
 
     def _handle_passcode_response(self, data):
         pass
+        _LOGGER.warning("Passcode resonse ignored")
 
-    def _handle_login_response(self, data):
-        pass
-
-    def _handle_ping_pong_response(self):
+    def _handle_login_response(self, data):        
         pass
+        _LOGGER.warning("Login resonse ignored")
 
+    def _handle_data_response(self, data):
+        if len(data) == 18:
+            meas = Measurement(data)
+            self._measurements.append(meas)
+            self._measurements_counter += 1
+        else:
+            pass
+        _LOGGER.debug(meas)
+        
     def _handle_data_extended_response(self, data):
         pass
+        _LOGGER.warning("Extended data ignored")
+
+    def _handle_ping_pong_response(self):
+        # if self._pong_thread is None or self._pong_thread.done:
+        #     self._pong_thread = asyncio.create_task(self._async_queue_ping())
+        #     pass
+        # else:
+        #     _LOGGER.debug("Pong thread alredy running")
+        _LOGGER.debug("Pong message received")
 
     def _send_ping(self):
-        pass
+        pong_data = bytes.fromhex("0000000303000015")
+        self._socket.sendall(pong_data)
+        _LOGGER.debug("Ping sent")
 
-    #     if (this.pingWaitTimeout) {
-    #         clearTimeout(this.pingWaitTimeout);
-    #         this.pingWaitTimeout = null;
-    #     }
-    #     if (this.pingTimeout) {
-    #         clearTimeout(this.pingTimeout);
-    #         this.pingTimeout = null;
-    #     }
-    #     debug('received pong');
-    #     this.pingTimeout = setTimeout(() => {
-    #         this.pingTimeout = null;
-    #         this._sendPing();
-    #     }, this.options.pingInterval || PH803W_PING_INTERVAL);
-    # }    def _handle_data_response(self, data):
+    def _ping_loop(self):
+        while self._loop:
+            sleep(PH803W_PING_INTERVAL)
+            self._send_ping()
+
+    # async def _async_queue_ping(self):
+    #     await asyncio.sleep(PH803W_PING_INTERVAL)
+    #     self._send_ping()
 
     def abort(self):
         self._loop = False
@@ -176,3 +202,25 @@ def __enter__(self):
 
     def __exit__(self, type, value, traceback):
         self._socket.close()
+
+class Measurement:
+    def __init__(self, data) -> None:
+        flag1 = data[8]
+        self.in_water = flag1 & 0b0000_0100 != 0
+        flag2 = data[9]
+        self.orp_on = flag2 & 0b0000_0010 != 0
+        self.ph_on = flag2 & 0b0000_0001 != 0
+        ph_raw = data[10:12]
+        self.ph = int.from_bytes(ph_raw, "big") * 0.01
+        redox_raw = data[12:14]
+        self.redox = int.from_bytes(redox_raw, "big") - 2000
+        unknown1_raw = data[14:16]
+        self.unknown1 = int.from_bytes(unknown1_raw, "big")
+        unknown2_raw = data[15:18]
+        self.unknown2 = int.from_bytes(unknown2_raw, "big")
+
+    def __str__(self) -> str:
+        return (
+            "pH: %s, Redox: %s, In-water: %s, pH-on: %s, Orp-on: %s"
+            % (self.ph, self.redox, self.in_water, self.ph_on, self.orp_on)
+        )

lib/discovery.py

@@ -1,8 +1,11 @@
 import socket
 import asyncio
+import logging
 
 PH803W_UDP_PORT = 12414
 
+_LOGGER = logging.getLogger(__name__)
+
 
 class DiscoveryError(ConnectionError):
     pass
@@ -27,28 +30,30 @@ def run(self):
         # Send dicovery request broadcast
         data = bytes.fromhex("0000000303000003")
         self._socket.sendto(data, ("<broadcast>", PH803W_UDP_PORT))
-        # print("Sent request message!")
+        self._socket.settimeout(2)
+        _LOGGER.debug("Sent request message!")
 
         # Receive device reaponse to discovery
         data, remote = self._socket.recvfrom(1024)
         if data[0] != 0 and data[1] != 0 and data[2] != 0 and data[2] != 3:
-            # print('Ignore data package because invalid prefix: %s' % data[0:3])
+            _LOGGER.error('Ignore data package because invalid prefix: %s' % data[0:3])
             raise DiscoveryError("Ignore data package because invalid prefix")
         data_length = data[4]
         if len(data) != data_length + 5:
-            # print('Ignore data package because invalid length(%s): %s' % (data_length, data))
+            _LOGGER.error('Ignore data package because invalid length(%s): %s' % (data_length, data))
             raise DiscoveryError("Ignore data package because invalid length")
         if data[7] == 3:
+            _LOGGER.error("Unknown response message type")
             raise DiscoveryError("Unknown response message type")
         if data[7] != 4:
-            # print("Ignore data package because invalid message type ${data[7]}")
+            _LOGGER.error("Ignore data package because invalid message type ${data[7]}")
             raise DiscoveryError("Ignore data package because invalid message type")
 
         # Parsing result of correct type
-        # print(
-        #     "Parsing discovered device: %s: %s - %s" % (remote[0], remote[1], data[7:])
-        # )
-        self.device = Device(remote[0], data)
+        _LOGGER.info(
+            "Parsing discovered device: %s: %s - %s" % (remote[0], remote[1], data[7:])
+        )
+        self.device = DeviceDiscovery(remote[0], data)
 
     def close(self):
         self._socket.close()
@@ -64,10 +69,18 @@ def __exit__(self, type, value, traceback):
         self.close()
 
 
-class Device:
+class DeviceDiscovery:
     def __init__(self, ip: str, data):
         self.ip = ip
 
+        if data is None:
+            _LOGGER.info("Initializing empty device with ip:%s" % ip)
+            self.id1 = ""
+            self.id2 = ""
+            self.api_server = ""
+            self.version_server = ""
+            return
+
         id1_length = data[9]
         id1_raw = data[10 : 10 + id1_length]
         self.id1 = id1_raw.decode("utf-8")

lib/main.py

@@ -0,0 +1,25 @@
+if __name__ == '__main__':
+
+    import asyncio
+    import discovery, device
+    import logging
+
+    logging.basicConfig(level=logging.DEBUG)
+
+    loop = asyncio.get_event_loop()
+
+    try:
+        disc = discovery.Discovery()
+        loop.run_until_complete(disc.run_async())
+        result = disc.get_result()
+        print(result)
+    except:
+        result = discovery.DeviceDiscovery('192.168.1.89', None)
+
+    dev = device.Device(result.ip)
+    # listener = loop.create_task(dev.run_async())
+    # asyncio.wait([listener])
+    loop.run_until_complete(dev.run_async(once=False))
+    #print(dev.get_result())
+
+    loop.close()