diff --git a/1-simple/cobolt.py b/1-simple/cobolt.py new file mode 100644 index 00000000..c213856f --- /dev/null +++ b/1-simple/cobolt.py @@ -0,0 +1,32 @@ +try: + import serial as SP +except: + exit("pyserial must be installed with: pip install pyserial") + +""" Simplest, most trivial strategy to talk to the Cobolt +laser, change its power. See manual in ../manuals/ page 27""" + +port = SP.Serial("COM1") # Use '/dev/xxx on macOS' + +""" Read laser power. Note: +1) b'' means binary +2) ends with \r meaning ASCII 13 dec +3) readline() will read up to next \n +""" + +port.write(b'pa?\r') +reply = port.readline() +power = int(reply) +print("Power is {0}.".format(power)) + +# Set power +port.write(b'p 0.01\r') +# There is no reply according to protocol defined p.27 + +# Read power again +port.write(b'pa?\r') +reply = port.readline() +power = int(reply) +print("Power is now {0}.".format(power)) + +port.close() \ No newline at end of file diff --git a/2-class/cobolt.py b/2-class/cobolt.py new file mode 100644 index 00000000..2a428279 --- /dev/null +++ b/2-class/cobolt.py @@ -0,0 +1,41 @@ +try: + import serial as SP +except: + exit("pyserial must be installed with: pip install pyserial") + +""" Class-oriented strategy to talk to the Cobolt +laser, change its power. See manual in ../manuals/ page 27""" + +class CoboltLaser: + + def __init__(self, portName): + """ portName : "COM1", "/dev/xxx", etc... + opens automatically on creation """ + self.port = None + self.port = SP.Serial(portName) + + def __del__(self): + if self.port is not None: + self.port.close() + + def setPower(self, powerInWatts): + command = "p {0:0.3f}\r".format(powerInWatts) + data = bytearray(command, "utf-8") + self.port.write(data) + + def power(self): + self.port.write(b'pa?\r') + reply = self.port.readline() + print(reply) + power = float(reply.decode()) + return power + + +if __name__ == "__main__": + try: + laser = CoboltLaser("COM1") + except: + exit("No laser found on COM1") + + laser.setPower(powerInWatts=0.1) + print("Power is {0:0.3f} W".format(laser.power())) diff --git a/3-class+debugPort/CoboltDebugSerial.py b/3-class+debugPort/CoboltDebugSerial.py new file mode 100644 index 00000000..db31d4c2 --- /dev/null +++ b/3-class+debugPort/CoboltDebugSerial.py @@ -0,0 +1,69 @@ +class CoboltDebugSerial: + def __init__(self): + self.outputBuffer = bytearray() + self.lineEnding = b'\r' + self.power = 0.1 + self.requestedPower = 0 + self.isOpen = False + + def open(self): + if self.isOpen: + raise IOError() + else: + self.isOpen = True + + return + + def close(self): + if not self.isOpen: + raise IOError() + else: + self.isOpen = False + + return + + def read(self, length) -> bytearray: + data = bytearray() + for i in range(0, length): + if len(self.outputBuffer) > 0: + byte = self.outputBuffer.pop(0) + data.append(byte) + else: + raise CommunicationReadTimeout("Unable to read data") + + return data + + + def write(self, data:bytearray) -> int : + string = data.decode('utf-8') + + match = re.search("pa\\?", string) + if match is not None: + replyData = bytearray("{0}\r".format(self.power), encoding='utf-8') + self.outputBuffer.extend(replyData) + return len(data) + + match = re.search("p\\?", string) + if match is not None: + replyData = bytearray("{0}\r".format(self.requestedPower), encoding='utf-8') + self.outputBuffer.extend(replyData) + return len(data) + + match = re.search("p (\\d+\\.?\\d+)\r", string) + if match is not None: + self.power = match.groups()[0] + return len(data) + + raise ValueError("Unknown command: {0}".format(data)) + + def readline(self) -> bytearray: + data = bytearray() + byte = b'' + while byte != self.lineEnding: + if len(self.outputBuffer) > 0: + byte = self.outputBuffer.pop(0) + data.append(byte) + else: + break + + return data diff --git a/3-class+debugPort/cobolt.py b/3-class+debugPort/cobolt.py new file mode 100644 index 00000000..9e92ce6b --- /dev/null +++ b/3-class+debugPort/cobolt.py @@ -0,0 +1,121 @@ +try: + import serial as SP +except: + exit("pyserial must be installed with: pip install pyserial") +import re + +class CoboltDebugSerial: + def __init__(self): + self.outputBuffer = bytearray() + self.lineEnding = b'\r' + self.power = 0.1 + self.requestedPower = 0 + self.isOpen = True + + def open(self): + if self.isOpen: + raise IOError() + else: + self.isOpen = True + + return + + def close(self): + if not self.isOpen: + raise IOError() + else: + self.isOpen = False + + return + + def read(self, length) -> bytearray: + data = bytearray() + for i in range(0, length): + if len(self.outputBuffer) > 0: + byte = self.outputBuffer.pop(0) + data.append(byte) + else: + raise CommunicationReadTimeout("Unable to read data") + + return data + + + def write(self, data:bytearray) -> int : + string = data.decode('utf-8') + + match = re.search("pa\\?", string) + if match is not None: + replyData = bytearray("{0}\r".format(self.power), encoding='utf-8') + self.outputBuffer.extend(replyData) + return len(data) + + match = re.search("p\\?", string) + if match is not None: + replyData = bytearray("{0}\r".format(self.requestedPower), encoding='utf-8') + self.outputBuffer.extend(replyData) + return len(data) + + match = re.search("p (\\d+\\.?\\d+)\r", string) + if match is not None: + self.power = match.groups()[0] + return len(data) + + raise ValueError("Unknown command: {0}".format(data)) + + def readline(self) -> bytearray: + data = bytearray() + byte = b'' + while byte != self.lineEnding: + if len(self.outputBuffer) > 0: + byte = self.outputBuffer.pop(0) + data.append(byte) + else: + break + + return data + +""" Class-oriented strategy to talk to the Cobolt +laser, change its power. See manual in ../manuals/ page 27""" + +class CoboltLaser: + + def __init__(self, portName): + """ portName : "COM1", "/dev/xxx", etc... + opens automatically on creation """ + self.port = None + if portName == "debug": + self.port = CoboltDebugSerial() + else: + self.port = SP.Serial(portName) + + def __del__(self): + if self.port is not None: + self.port.close() + + def setPower(self, powerInWatts): + command = "p {0:0.3f}\r".format(powerInWatts) + data = bytearray(command, "utf-8") + self.port.write(data) + + def power(self): + self.port.write(b'pa?\r') + reply = self.port.readline() + power = float(reply.decode()) + return power + + +if __name__ == "__main__": + try: + laser = CoboltLaser("COM1") + except: + laser = CoboltLaser("debug") + + laser.setPower(powerInWatts=0.1) + print("Power is {0:0.3f} W".format(laser.power())) + + laser.setPower(powerInWatts=0.01) + print("Power is {0:0.3f} W".format(laser.power())) + + laser.setPower(powerInWatts=0.001) + print("Power is {0:0.3f} W".format(laser.power())) + diff --git a/CoboltDebugSerial.py b/CoboltDebugSerial.py new file mode 100644 index 00000000..2534e25c --- /dev/null +++ b/CoboltDebugSerial.py @@ -0,0 +1,108 @@ +import re +from CommunicationPort import * + +class CoboltDebugSerial: + def __init__(self): + self.outputBuffer = bytearray() + self.lineEnding = b'\r' + self.power = 0.1 + self.isOn = True + self.requestedPower = 0 + self._isOpen = True + + @property + def is_open(self): + return self._isOpen + + def open(self): + if self._isOpen: + raise IOError("port is already open") + else: + self._isOpen = True + + return + + def close(self): + self._isOpen = False + + return + + def flush(self): + return + + def read(self, length) -> bytearray: + data = bytearray() + for i in range(0, length): + if len(self.outputBuffer) > 0: + byte = self.outputBuffer.pop(0) + data.append(byte) + else: + raise CommunicationReadTimeout("Unable to read data") + + return data + + + def write(self, data:bytearray) -> int : + string = data.decode('utf-8') + + match = re.search("pa\\?", string) + if match is not None: + replyData = bytearray("{0:0.4f}\r\n".format(self.power), encoding='utf-8') + self.outputBuffer.extend(replyData) + return len(data) + + match = re.search("p\\?", string) + if match is not None: + replyData = bytearray("{0:0.4f}\r\n".format(self.requestedPower), encoding='utf-8') + self.outputBuffer.extend(replyData) + return len(data) + + match = re.search("p (\\d+\\.?\\d+)\r", string) + if match is not None: + self.power = float(match.groups()[0]) + replyData = bytearray("OK\r\n", encoding='utf-8') + self.outputBuffer.extend(replyData) + return len(data) + + match = re.search("l1\r", string) + if match is not None: + self.isOn = True + replyData = bytearray("OK\r\n", encoding='utf-8') + self.outputBuffer.extend(replyData) + return len(data) + + match = re.search("l0\r", string) + if match is not None: + self.isOn = False + replyData = bytearray("OK\r\n", encoding='utf-8') + self.outputBuffer.extend(replyData) + return len(data) + + match = re.search("sn\\?\r", string) + if match is not None: + replyData = bytearray("123456\r\n", encoding='utf-8') + self.outputBuffer.extend(replyData) + return len(data) + + match = re.search("ilk\\?\r", string) + if match is not None: + replyData = bytearray("1\r\n", encoding='utf-8') + self.outputBuffer.extend(replyData) + + return len(data) + + # Error: + replyData = bytearray("Syntax error: {0}", encoding='utf-8') + self.outputBuffer.extend(replyData) + + # def readline(self) -> bytearray: + # data = bytearray() + # byte = b'' + # while byte != self.lineEnding: + # if len(self.outputBuffer) > 0: + # byte = self.outputBuffer.pop(0) + # data.append(byte) + # else: + # break + + # return data diff --git a/CoboltDevice.py b/CoboltDevice.py index 166675a6..7347922b 100644 --- a/CoboltDevice.py +++ b/CoboltDevice.py @@ -1,79 +1,96 @@ from PhysicalDevice import * -from CommunicationPort import * -import numpy as np - - -class LaserSourceDevice: - def __init__(self): - self.requestedPower = 0 - self.power = 0 +from LaserSourceDevice import * - def turnOn(self): - self.doTurnOn() - - def turnOff(self): - self.doTurnOff() - - def setPower(self, power:float): - self.doSetPower(power) - - def power(self) -> float: - return self.doGetPower() +from CommunicationPort import * +from CoboltDebugSerial import * +import numpy as np +import re class CoboltDevice(PhysicalDevice, LaserSourceDevice): + def __init__(self, bsdPath=None, serialNumber: str = None, productId: np.uint32 = None, vendorId: np.uint32 = None): + + self.laserPower = 0 + self.requestedPower = 0 + self.interlockState = True self.laserSerialNumber = None - self.port = CommunicationPort(bsdPath=bsdPath) + + self.bsdPath = bsdPath PhysicalDevice.__init__(self, serialNumber, vendorId, productId) LaserSourceDevice.__init__(self) - self.port.open() + self.port = None def __del__(self): - if self.port is not None: + try: self.port.close() + except: + # ignore if already closed + return - def doInitializeDevice(self): # Doesn't work well, seems to know that the port is already open + def doInitializeDevice(self): try: + if self.bsdPath == "debug": + self.port = CommunicationPort(port=CoboltDebugSerial()) + else: + self.port = CommunicationPort(bsdPath=self.bsdPath) + + if self.port is None: + raise PhysicalDeviceUnableInitialize() self.port.open() self.doGetLaserSerialNumber() + except Exception as error: - self.port.close() - raise error + if self.port is not None: + self.port.close() + raise PhysicalDeviceUnableInitialize() - def doShutdownDevice(self): # Doesn't seem to do anything + def doShutdownDevice(self): self.port.close() return - def doGetInterlockState(self): - self.port.writeStringExpectMatchingString('ilk?\r', replyPattern='[0|1]') + def doGetInterlockState(self) -> bool: + value = self.port.writeStringExpectMatchingString('ilk?\r', replyPattern='([0|1])') + self.interlockState = bool(value) + return self.interlockState - def doGetLaserSerialNumber(self): - self.laserSerialNumber = self.port.writeStringExpectMatchingString('sn?\r', replyPattern='(\\d+)') + def doGetLaserSerialNumber(self) -> str: + self.laserSerialNumber = self.port.writeStringReadFirstMatchingGroup('sn?\r', replyPattern='(\\d+)') def doTurnOn(self): -# self.port.writeStringExpectMatchingString('@cobas 0\r', '') - self.port.writeStringExpectMatchingString('l1\r', '') + self.port.writeString('l1\r') def doTurnOff(self): # The laser needs to be unplug after using this function or it doesn't restart using doTurnOn?? - self.port.writeStringExpectMatchingString('l0\r', '') + self.port.writeString('l0\r') def doSetPower(self, powerInWatts): command = 'p {0:0.3f}\r'.format(powerInWatts) - self.port.writeStringExpectMatchingString(command, '') - - def doGetPower(self): - self.port.writeStringExpectMatchingString('pa?\r', replyPattern='\d.\d+') + self.port.writeString(command) + def doGetPower(self) -> float: + value = self.port.writeStringReadFirstMatchingGroup('pa?\r', replyPattern='(\\d.\\d+)') + return float(value) if __name__ == "__main__": + laser = None try: - laser = CoboltDevice("COM5") + laser = CoboltDevice("debug") except: - exit("No laser found on COM5") + exit("No laser") + + laser.initializeDevice() + laser.turnOn() + laser.setPower(0.1) + print(laser.power()) + laser.setPower(0.01) + print(laser.power()) + laser.setPower(0.2) + print(laser.power()) + laser.turnOff() + laser.shutdownDevice() + -laser.doTurnOn() #laser.doTurnOn() #laser.doGetLaserSerialNumber() #laser.doGetPower() diff --git a/CommunicationPort.py b/CommunicationPort.py index ead2464e..b1b58a88 100644 --- a/CommunicationPort.py +++ b/CommunicationPort.py @@ -1,8 +1,8 @@ import serial import re -from threading import Thread, RLock import time import random +from threading import Thread, RLock class CommunicationReadTimeout(serial.SerialException): pass @@ -12,28 +12,47 @@ class CommunicationReadNoMatch(Exception): class CommunicationPort: """CommunicationPort class with basic application-level protocol - functions to write strings and read strings""" - port = None - def __init__(self, bsdPath=None, vendorId=None, - productId=None, serialNumber=None): + functions to write strings and read strings, and abstract away + the details of the communication. + + Two strategies to initialize the CommunicationPort: + 1. with a bsdPath/port name (i.e. "COM1" or "/dev/cu.serial") + 2. with an instance of SerialPort() that will support the same + functions as pyserial.Serial() (open, close, read, write, readline) + """ + + def __init__(self, bsdPath=None, port = None): self.bsdPath = bsdPath - self.vendorId = vendorId - self.productId = productId - self.serialNumber = serialNumber + + if port is not None and port.is_open: + port.close() + self.port = port # direct port , must be closed. + self.portLock = RLock() self.transactionLock = RLock() + @property + def isOpen(self): + return self.port.is_open def open(self): - self.port = serial.Serial(self.bsdPath, 19200, timeout=0.3) + if self.port is None: + self.port = serial.Serial(self.bsdPath, 19200, timeout=0.3) + else: + self.port.open() def close(self): self.port.close() - def bytesAvailable(self): + def bytesAvailable(self) -> int: return self.port.in_waiting - def readData(self, length): + def flush(self): + if self.isOpen: + self.port.reset_input_buffer() + self.port.reset_output_buffer() + + def readData(self, length) -> bytearray: with self.portLock: data = self.port.read(length) if len(data) != length: @@ -41,7 +60,7 @@ def readData(self, length): return data - def writeData(self, data): + def writeData(self, data) -> int: with self.portLock: nBytesWritten = self.port.write(data) if nBytesWritten != len(data): @@ -50,7 +69,7 @@ def writeData(self, data): return nBytesWritten - def readString(self): + def readString(self) -> str: with self.portLock: byte = None data = bytearray(0) @@ -64,9 +83,9 @@ def readString(self): return string - def writeString(self, string): + def writeString(self, string) -> int: + nBytes = 0 with self.portLock: - nBytes = 0 data = bytearray(string, "utf-8") nBytes = self.writeData(data) @@ -78,7 +97,7 @@ def writeStringExpectMatchingString(self, string, replyPattern): reply = self.readString() match = re.search(replyPattern, reply) if match is None: - raise CommunicationReadNoMatch("No match") + raise CommunicationReadNoMatch("No match, expected {0} read {1}".format(replyPattern, reply)) return reply @@ -94,44 +113,12 @@ def writeStringReadMatchingGroups(self, string, replyPattern): with self.transactionLock: self.writeString(string) reply = self.readString() + match = re.search(replyPattern, reply) if match is not None: - print(match.groups()) + return match.groups() else: raise CommunicationReadNoMatch("Unable to match pattern:'{0}' in reply:'{1}'".format(replyPattern, reply)) -class DebugEchoCommunicationPort(CommunicationPort): - def __init__(self, delay=0): - self.buffer = bytearray() - self.delay = delay - super(DebugEchoCommunicationPort, 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) - - return len(data) diff --git a/DebugEchoCommunicationPort.py b/DebugEchoCommunicationPort.py new file mode 100644 index 00000000..8cad0692 --- /dev/null +++ b/DebugEchoCommunicationPort.py @@ -0,0 +1,48 @@ +from CommunicationPort import * + +class DebugEchoCommunicationPort(CommunicationPort): + def __init__(self, delay=0): + self.buffer = bytearray() + self.delay = delay + self._isOpen = False + super(DebugEchoCommunicationPort, self).__init__() + + @property + def isOpen(self): + return self._isOpen + + def open(self): + if self._isOpen: + raise Exception() + + self._isOpen = True + return + + def close(self): + self._isOpen = False + return + + def bytesAvailable(self): + return len(self.buffer) + + def flush(self): + self.buffer = bytearray() + + 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) + + return len(data) diff --git a/LaserSourceDevice.py b/LaserSourceDevice.py new file mode 100644 index 00000000..3e4cc46b --- /dev/null +++ b/LaserSourceDevice.py @@ -0,0 +1,17 @@ + +class LaserSourceDevice: + + def turnOn(self): + self.doTurnOn() + + def turnOff(self): + self.doTurnOff() + + def setPower(self, power:float): + self.doSetPower(power) + + def power(self) -> float: + return self.doGetPower() + + def interlock(self) -> bool: + return self.doGetInterlockState() diff --git a/PhysicalDevice.py b/PhysicalDevice.py index 6d040646..ef3d339e 100644 --- a/PhysicalDevice.py +++ b/PhysicalDevice.py @@ -8,6 +8,9 @@ class DeviceState(Enum): Recognized = 2 # Initialization has succeeded, but currently shutdown Unrecognized = 3 # Initialization failed +class PhysicalDeviceUnableInitialize(Exception): + pass + class PhysicalDevice: def __init__(self, serialNumber:str, productId:np.uint32, vendorId:np.uint32): diff --git a/README.md b/README.md new file mode 100644 index 00000000..002f5131 --- /dev/null +++ b/README.md @@ -0,0 +1,11 @@ +# PyHardwareLibrary +A simple device-oriented library with CommunicationPort for controlling devices + +## Introduction +You will find a simple, trivial script named `cobolt.py` to change the power of the Cobolt laser. There are three versions: you should read the three examples : + +1. `1-simple`: a very trivial implementation with simple commands in sequence +2. `2-class`: a class implementation of `CoboltLaser` that partially encapsulates the details and exposes a few functions: `setPower()` and `power()` +3. `3-class+debugPort`: a class implementation with a debug port that mimicks the real device + +More to come. diff --git a/testCoboltDevice.py b/testCoboltDevice.py new file mode 100644 index 00000000..d474f08e --- /dev/null +++ b/testCoboltDevice.py @@ -0,0 +1,62 @@ +import unittest +import time +from threading import Thread, Lock + +from serial import * +from CommunicationPort import * +from DebugEchoCommunicationPort import * +from CoboltDevice import * + +class BaseTestCases: + device:CoboltDevice = None + + class TestCobolt(unittest.TestCase): + + def testCreate(self): + self.assertIsNotNone(self.device) + + def testInitializeShutdown(self): + self.device.initializeDevice() + self.device.shutdownDevice() + + def testTurnOn(self): + self.device.turnOn() + + def testTurnOff(self): + self.device.turnOff() + + def testTurnOnOff(self): + self.device.turnOn() + self.device.setPower(0.01) + self.assertTrue(self.device.power() == 0.01) + + def testInterloc(self): + self.assertTrue(self.device.interlock()) + +class TestDebugCobolt(BaseTestCases.TestCobolt): + + def setUp(self): + self.device = CoboltDevice(bsdPath="debug") + self.assertIsNotNone(self.device) + self.device.initializeDevice() + + def tearDown(self): + self.device.shutdownDevice() + return + +@unittest.skip +class TestRealCobolt(BaseTestCases.TestCobolt): + + def setUp(self): + self.device = CoboltDevice(bsdPath="COM5") + self.assertIsNotNone(self.device) + self.device.initializeDevice() + + def tearDown(self): + self.device.shutdownDevice() + return + + + +if __name__ == '__main__': + unittest.main() diff --git a/testCoboltSerial.py b/testCoboltSerial.py new file mode 100644 index 00000000..10785600 --- /dev/null +++ b/testCoboltSerial.py @@ -0,0 +1,86 @@ +import unittest +import time +from threading import Thread, Lock + +from serial import * +from CommunicationPort import * +from CoboltDebugSerial import * + +class BaseTestCases: + + class TestCoboltSerialPort(unittest.TestCase): + port = None + + def testCreate(self): + self.assertIsNotNone(self.port) + + def testCantReopen(self): + self.assertTrue(self.port.isOpen) + with self.assertRaises(Exception) as context: + self.port.open() + + def testCloseReopen(self): + self.assertTrue(self.port.isOpen) + self.port.close() + self.port.open() + + def testCloseTwice(self): + self.assertTrue(self.port.isOpen) + self.port.close() + self.port.close() + + def testLaserOn(self): + self.port.writeString('l1\r') + with self.assertRaises(CommunicationReadTimeout) as context: + self.port.readString() + + def testLaserOff(self): + self.port.writeString('l0\r') + with self.assertRaises(CommunicationReadTimeout) as context: + self.port.readString() + + def testReadPower(self): + self.port.writeStringExpectMatchingString('pa?\r',replyPattern='\\d+.\\d+') + + def testReadSetPower(self): + self.port.writeStringExpectMatchingString('p?\r',replyPattern='\\d+.\\d+') + + def testReadInterlock(self): + self.port.writeStringExpectMatchingString('p?\r',replyPattern='\\d+.\\d+') + + def testWriteSetPower(self): + self.port.writeString('p 0.001\r') + with self.assertRaises(CommunicationReadTimeout) as context: + self.port.readString() + + def testReadSerialNumber(self): + self.port.writeStringExpectMatchingString('sn?\r',replyPattern='\\d+') + + def testUnknownCommand(self): + with self.assertRaises(Exception) as context: + self.port.writeString('nothing useful\r') + + +class TestDebugCoboltSerialPort(BaseTestCases.TestCoboltSerialPort): + + def setUp(self): + self.port = CommunicationPort(port=CoboltDebugSerial()) + self.assertIsNotNone(self.port) + self.port.open() + + def tearDown(self): + self.port.close() + +@unittest.skip +class TestRealCoboltSerialPort(BaseTestCases.TestCoboltSerialPort): + + def setUp(self): + self.port = CommunicationPort(port="COM5") + self.assertIsNotNone(self.port) + self.port.open() + + def tearDown(self): + self.port.close() + +if __name__ == '__main__': + unittest.main() diff --git a/testEcho.py b/testEcho.py index 08a0376b..0d8d1fb7 100644 --- a/testEcho.py +++ b/testEcho.py @@ -1,9 +1,11 @@ import unittest -from serial import * -from CommunicationPort import * import time from threading import Thread, Lock +from serial import * +from CommunicationPort import * +from DebugEchoCommunicationPort import * + payloadData = b'1234' payloadString = '1234\n' globalLock = Lock() @@ -17,6 +19,24 @@ class TestEchoPort(unittest.TestCase): def testCreate(self): self.assertIsNotNone(self.port) + def testIsOpenOnCreation(self): + self.assertTrue(self.port.isOpen) + + def testCantReopen(self): + self.assertTrue(self.port.isOpen) + with self.assertRaises(Exception) as context: + self.port.open() + + def testCloseReopen(self): + self.assertTrue(self.port.isOpen) + self.port.close() + self.port.open() + + def testCloseTwice(self): + self.assertTrue(self.port.isOpen) + self.port.close() + self.port.close() + def testWriteData(self): nBytes = self.port.writeData(payloadData) self.assertTrue(nBytes == len(payloadData)) @@ -30,24 +50,27 @@ def testWriteDataReadEcho(self): self.assertTrue(nBytes == len(payloadData)) data = self.port.readData(length=len(payloadData)) - self.assertTrue(data == payloadData) + self.assertTrue(data == payloadData, "Data {0}, payload:{1}".format(data, payloadData)) def testWriteDataReadEchoSequence(self): nBytes = self.port.writeData(payloadData) + self.assertTrue(nBytes == len(payloadData)) nBytes = self.port.writeData(payloadData) + self.assertTrue(nBytes == len(payloadData)) data = self.port.readData(length=len(payloadData)) - self.assertTrue(data == payloadData) + self.assertTrue(data == payloadData, "Data {0}, payload:{1}".format(data, payloadData)) data = self.port.readData(length=len(payloadData)) - self.assertTrue(data == payloadData) + self.assertTrue(data == payloadData, "Data {0}, payload:{1}".format(data, payloadData)) def testWriteDataReadEchoLarge(self): for i in range(100): nBytes = self.port.writeData(payloadData) + self.assertTrue(nBytes == len(payloadData)) for i in range(100): data = self.port.readData(length=len(payloadData)) - self.assertTrue(data == payloadData) + self.assertTrue(data == payloadData, "Data {0}, payload:{1}".format(data, payloadData)) def testWriteString(self): nBytes = self.port.writeString(payloadString) @@ -62,7 +85,9 @@ def testWriteStringReadEcho(self): def testWriteStringReadEchoSequence(self): nBytes = self.port.writeString(payloadString) + self.assertTrue(nBytes == len(payloadString)) nBytes = self.port.writeString(payloadString) + self.assertTrue(nBytes == len(payloadString)) string = self.port.readString() self.assertTrue(string == payloadString) @@ -72,10 +97,11 @@ def testWriteStringReadEchoSequence(self): def testWriteStringReadEchoLarge(self): for i in range(100): nBytes = self.port.writeString(payloadString) + self.assertTrue(nBytes == len(payloadString)) for i in range(100): string = self.port.readString() - self.assertTrue(string == payloadString) + self.assertTrue(string == payloadString,"{0} is not {1}".format(string, payloadString)) def testTimeoutReadData(self): with self.assertRaises(CommunicationReadTimeout) as context: @@ -170,17 +196,19 @@ def threadReadWrite(port, index): threadFailed = index + class TestDebugEchoPort(BaseTestCases.TestEchoPort): def setUp(self): self.port = DebugEchoCommunicationPort() self.assertIsNotNone(self.port) self.port.open() + self.assertTrue(self.port.isOpen) + self.port.flush() def tearDown(self): self.port.close() - - + self.assertFalse(self.port.isOpen) class TestSlowDebugEchoPort(BaseTestCases.TestEchoPort): @@ -189,6 +217,7 @@ def setUp(self): self.assertIsNotNone(self.port) self.port.delay = 0.01 self.port.open() + self.port.flush() def tearDown(self): self.port.close() @@ -201,11 +230,13 @@ def setUp(self): self.port = CommunicationPort("/dev/cu.usbserial-ftDXIKC4") self.assertIsNotNone(self.port) self.port.open() + self.port.flush() except: self.fail("Unable to setUp serial port") def tearDown(self): + self.port.flush() self.port.close()