First echo implementation

SerialPort.py

@@ -1,50 +1,137 @@
-import errno
-import fcntl
-import os
-import select
-import struct
-import sys
-import termios
+import serial
+import re
+from threading import Thread, RLock
+import time
+import random
+
+class CommunicationReadTimeout(serial.SerialException):
+    pass
+
+class CommunicationReadNoMatch(Exception):
+    pass
 
 class SerialPort:
-    """SerialPort class with basic application-level protocol functions to write strings and read strings"""
+    """SerialPort class with basic application-level protocol 
+    functions to write strings and read strings"""
     port = None
-    bsdPath = None
-    vendorId = None
-    productId = None
-    serialNumber = None
-
-    def __init__(self, bsdPath = None, vendorId = None, productId = None, serialNumber = None):
+    def __init__(self, bsdPath=None, vendorId=None,
+                 productId=None, serialNumber=None):
         self.bsdPath = bsdPath
+        self.vendorId = vendorId
+        self.productId = productId
+        self.serialNumber = serialNumber
+        self.portLock = RLock()
+        self.transactionLock = RLock()
 
-    def flush(self):
-        return None
-
-    def drain(self):
-        return None
 
     def open(self):
-        try:
-            self.port = open(self.bsdPath, "r+")
-        except OSError as msg:
-            self.port = None
-            raise SerialException(msg.errno, "could not open port {}: {}".format(self.port, msg))
+        self.port = serial.Serial(self.bsdPath, 19200, timeout=0.3)
 
     def close(self):
         self.port.close()
 
-    def readString(self, matching=".*"):
-        string = self.port.readline()
+    def bytesAvailable(self):
+        return self.port.in_waiting
+
+    def readData(self, length):
+        with self.portLock:
+            data = self.port.read(length)
+            if len(data) != length:
+                raise CommunicationReadTimeout()
+
+        return data
+
+    def writeData(self, data):
+        with self.portLock:
+            nBytesWritten = self.port.write(data)
+            if nBytesWritten != len(data):
+                raise IOError("Not all bytes written to port")
+            self.port.flush()
+
+        return nBytesWritten
+
+    def readString(self):      
+        with self.portLock:
+            byte = None
+            data = bytearray(0)
+            while (byte != b''):
+                byte = self.readData(1)
+                data += byte
+                if byte == b'\n':
+                    break
+
+            string = data.decode(encoding='utf-8')
+
         return string
 
     def writeString(self, string):
-        return self.port.write(string)
-
+        nBytes = 0
+        with self.portLock:
+            data = bytearray(string, "utf-8")
+            nBytes = self.writeData(data)
+
+        return nBytes
+
+    def writeStringExpectMatchingString(self, string, replyPattern):
+        with self.transactionLock:
+            self.writeString(string)
+            reply = self.readString()
+            match = re.search(replyPattern, string)
+            if match is None:
+                raise CommunicationReadNoMatch("No match")
+
+        return reply
+
+    def writeStringReadFirstMatchingGroup(self, string, replyPattern):
+        with self.transactionLock:
+            groups = self.writeStringReadMatchingGroups(string, replyPattern)
+            if len(groups) >= 1:
+                return groups[0]
+            else:
+                raise CommunicationReadNoMatch("Unable to find first group with pattern:'{0}'".format(replyPattern))
+
+    def writeStringReadMatchingGroups(self, string, replyPattern):
+        with self.transactionLock:
+            self.writeString(string)
+            reply = self.readString()
+            match = re.search(replyPattern, string)
+
+            if match is not None:
+                return match.groups()
+            else:
+                raise CommunicationReadNoMatch("Unable to match pattern:'{0}' in reply:'{1}'".format(replyPattern, reply))
+
+
+class DebugEchoSerialPort(SerialPort):
+    def __init__(self):
+        self.buffer = bytearray()
+        self.delay = 0
+        super(DebugEchoSerialPort, self).__init__()
+
+    def open(self):
+        return
+
+    def close(self):
+        return
+
+    def bytesAvailable(self):
+        return len(self.buffer)
+
+    def readData(self, length):
+        with self.portLock:
+            time.sleep(self.delay*random.random())
+            data = bytearray()
+            for i in range(0, length):
+                if len(self.buffer) > 0:
+                    byte = self.buffer.pop(0)
+                    data.append(byte)
+                else:
+                    raise CommunicationReadTimeout("Unable to read data")
+
+        return data
 
+    def writeData(self, data):
+        with self.portLock:
+            self.buffer.extend(data)
 
-if __name__ == "__main__":
-    port = SerialPort("/dev/null")
-    port.open()
-    port.writeString("abcd")
-    string = port.readString()
-    port.close()
+        return len(data)

testEcho.py

@@ -0,0 +1,213 @@
+import unittest
+from serial import *
+from SerialPort import *
+import time
+from threading import Thread, Lock
+
+payloadData = b'1234'
+payloadString = '1234\n'
+globalLock = Lock()
+threadFailed = -1
+
+class BaseTestCases:
+
+    class TestEchoPort(unittest.TestCase):
+        port = None
+
+        def testCreate(self):
+            self.assertIsNotNone(self.port)
+
+        def testWriteData(self):
+            nBytes = self.port.writeData(payloadData)
+            self.assertTrue(nBytes == len(payloadData))
+
+        # def testWriteDataBytesAvailable(self):
+        #     nBytes = self.port.writeData(payloadData)
+        #     self.assertEqual(nBytes, self.port.bytesAvailable())
+
+        def testWriteDataReadEcho(self):
+            nBytes = self.port.writeData(payloadData)
+            self.assertTrue(nBytes == len(payloadData))
+
+            data = self.port.readData(length=len(payloadData))
+            self.assertTrue(data == payloadData)
+
+        def testWriteDataReadEchoSequence(self):
+            nBytes = self.port.writeData(payloadData)
+            nBytes = self.port.writeData(payloadData)
+
+            data = self.port.readData(length=len(payloadData))
+            self.assertTrue(data == payloadData)
+            data = self.port.readData(length=len(payloadData))
+            self.assertTrue(data == payloadData)
+
+        def testWriteDataReadEchoLarge(self):
+            for i in range(100):
+                nBytes = self.port.writeData(payloadData)
+
+            for i in range(100):
+                data = self.port.readData(length=len(payloadData))
+                self.assertTrue(data == payloadData)
+
+        def testWriteString(self):
+            nBytes = self.port.writeString(payloadString)
+            self.assertTrue(nBytes == len(payloadString))
+
+        def testWriteStringReadEcho(self):
+            nBytes = self.port.writeString(payloadString)
+            self.assertTrue(nBytes == len(payloadString))
+
+            string = self.port.readString()
+            self.assertTrue(string == payloadString, "String {0}, payload:{1}".format(string, payloadString))
+
+        def testWriteStringReadEchoSequence(self):
+            nBytes = self.port.writeString(payloadString)
+            nBytes = self.port.writeString(payloadString)
+
+            string = self.port.readString()
+            self.assertTrue(string == payloadString)
+            string = self.port.readString()
+            self.assertTrue(string == payloadString)
+
+        def testWriteStringReadEchoLarge(self):
+            for i in range(100):
+                nBytes = self.port.writeString(payloadString)
+
+            for i in range(100):
+                string = self.port.readString()
+                self.assertTrue(string == payloadString)
+
+        def testTimeoutReadData(self):
+            with self.assertRaises(CommunicationReadTimeout) as context:
+                self.port.readData(1)
+
+            self.port.writeData(payloadData)
+            with self.assertRaises(CommunicationReadTimeout) as context:
+                self.port.readData(len(payloadData)+1)
+
+        def testWriteStringReadMatchingEcho(self):
+            reply = self.port.writeStringExpectMatchingString(
+                        payloadString,
+                        replyPattern=payloadString)
+            self.assertEqual(reply, payloadString)
+
+        def testWriteStringReadMatchingPattern(self):
+            reply = self.port.writeStringExpectMatchingString(
+                        "abcd1234\n",
+                        replyPattern="abc.\\d{4}")
+            self.assertEqual(reply, "abcd1234\n")
+
+        def testWriteStringReadMatchingPatternFirstCaptureGroup(self):
+            firstGroup = self.port.writeStringReadFirstMatchingGroup(
+                        "abcd1234\n",
+                        replyPattern="abc.(\\d{4})")
+            self.assertEqual(firstGroup, "1234")
+
+        def testWriteStringReadMatchingPatternAllCaptureGroups(self):
+            (firstGroup, secondGroup) = self.port.writeStringReadMatchingGroups(
+                        "abcd1234\n",
+                        replyPattern="(abc.)(\\d{4})")
+            self.assertEqual(firstGroup, "abcd")
+            self.assertEqual(secondGroup, "1234")
+
+        def testFailedWriteStringReadMatchingPattern(self):
+            with self.assertRaises(CommunicationReadNoMatch) as context:
+                reply = self.port.writeStringExpectMatchingString(
+                            "abcd1234\n",
+                            replyPattern="abc.\\d{5}")
+
+        def testFailedWriteStringReadMatchingPatternFirstCaptureGroup(self):
+            with self.assertRaises(CommunicationReadNoMatch) as context:
+                firstGroup = self.port.writeStringReadFirstMatchingGroup(
+                            "abcd1234\n",
+                            replyPattern="abc.(\\d{5})")
+
+        def testFailedWriteStringReadMatchingPatternWithoutGroupFirstCaptureGroup(self):
+            with self.assertRaises(CommunicationReadNoMatch) as context:
+                firstGroup = self.port.writeStringReadFirstMatchingGroup(
+                            "abcd1234\n",
+                            replyPattern="abc.\\d{4}")
+
+        def testFailedWriteStringReadMatchingPatternAllCaptureGroups(self):
+            with self.assertRaises(CommunicationReadNoMatch) as context:
+                (firstGroup, secondGroup) = self.port.writeStringReadMatchingGroups(
+                            "abcd1234\n",
+                            replyPattern="(abc.)(\\d{5})")
+
+        def testThreadSafety(self):
+            global threadFailed, globalLock
+            threadFailed = -1
+            threadPool = []
+            for i in range(100):
+                process = Thread(target=threadReadWrite, kwargs=dict(port=self.port, index=i))
+                threadPool.append(process)
+
+            for process in threadPool:
+                process.start()
+
+            for i,p in enumerate(threadPool):
+                p.join(timeout=1)
+
+                try:
+                    globalLock.acquire()
+                    if threadFailed != -1:
+                        self.fail("Thread {0}?".format(threadFailed))
+                finally:
+                    globalLock.release()
+
+            for i,p in enumerate(threadPool):
+                p.join(timeout=1)
+
+
+def threadReadWrite(port, index):
+    global threadFailed, globalLock
+    payload = "abcd{0}\n".format(index)
+    with globalLock:
+        try:
+            port.writeStringExpectMatchingString(payload, replyPattern=payload)
+        except:
+            if threadFailed != -1:
+                threadFailed = index        
+
+
+class TestDebugEchoPort(BaseTestCases.TestEchoPort):
+
+    def setUp(self):
+        self.port = DebugEchoSerialPort()
+        self.assertIsNotNone(self.port)
+        self.port.open()
+
+    def tearDown(self):
+        self.port.close()
+
+
+
+class TestSlowDebugEchoPort(BaseTestCases.TestEchoPort):
+
+    def setUp(self):
+        self.port = DebugEchoSerialPort()
+        self.assertIsNotNone(self.port)
+        self.port.delay = 0.01
+        self.port.open()
+
+    def tearDown(self):
+        self.port.close()
+
+
+class TestRealEchoPort(BaseTestCases.TestEchoPort):
+
+    def setUp(self):
+        try:
+            self.port = SerialPort("/dev/cu.usbserial-ftDXIKC4")
+            self.assertIsNotNone(self.port)
+            self.port.open()
+        except:
+            self.fail("Unable to setUp serial port")
+
+
+    def tearDown(self):
+        self.port.close()
+
+
+if __name__ == '__main__':
+    unittest.main()