From 298c1ea81ba1f8cdb838aaaa1454a5c40404bd4a Mon Sep 17 00:00:00 2001 From: Daniel Cote Date: Wed, 22 May 2019 07:53:12 -0400 Subject: [PATCH 01/24] Simple script to test cobolt laser --- cobolt.py | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) create mode 100644 cobolt.py diff --git a/cobolt.py b/cobolt.py new file mode 100644 index 00000000..a8bd3cfe --- /dev/null +++ b/cobolt.py @@ -0,0 +1,19 @@ +import serial + +port = Serial("COM1") + +# Read power +port.write(b'pa?\r') +reply = port.readline() +print(reply) + +# Set power +port.write(b'p 0.01\r') +# There is no reply + +# Read power again +port.write(b'pa?\r') +reply = port.readline() +print(reply) + +port.close() \ No newline at end of file From fb62a99bf82435eb5f2cf43f7dbdbe10e512770e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20C=C3=B4t=C3=A9?= Date: Wed, 22 May 2019 08:00:27 -0400 Subject: [PATCH 02/24] Create README.md --- README.md | 7 +++++++ 1 file changed, 7 insertions(+) create mode 100644 README.md diff --git a/README.md b/README.md new file mode 100644 index 00000000..5b92c6d0 --- /dev/null +++ b/README.md @@ -0,0 +1,7 @@ +# 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. + +More to come. From d053f5a671ea6242b01d7fd10a1e0f20d5a56773 Mon Sep 17 00:00:00 2001 From: Daniel Cote Date: Wed, 22 May 2019 10:28:05 -0400 Subject: [PATCH 03/24] Reorganizatio of files for teaching --- 1-simple/cobolt.py | 32 ++++++++++++++++++++++++++++++++ cobolt.py | 19 ------------------- 2 files changed, 32 insertions(+), 19 deletions(-) create mode 100644 1-simple/cobolt.py delete mode 100644 cobolt.py 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/cobolt.py b/cobolt.py deleted file mode 100644 index a8bd3cfe..00000000 --- a/cobolt.py +++ /dev/null @@ -1,19 +0,0 @@ -import serial - -port = Serial("COM1") - -# Read power -port.write(b'pa?\r') -reply = port.readline() -print(reply) - -# Set power -port.write(b'p 0.01\r') -# There is no reply - -# Read power again -port.write(b'pa?\r') -reply = port.readline() -print(reply) - -port.close() \ No newline at end of file From b8ab17c02392c0e23081c9d09fad6583ddad6ef6 Mon Sep 17 00:00:00 2001 From: Daniel Cote Date: Wed, 22 May 2019 10:40:36 -0400 Subject: [PATCH 04/24] Simple class-based strategy, for teaching purposes --- 2-class/cobolt.py | 35 +++++++++++++++++++++++++++++++++++ 1 file changed, 35 insertions(+) create mode 100644 2-class/cobolt.py diff --git a/2-class/cobolt.py b/2-class/cobolt.py new file mode 100644 index 00000000..f159dab2 --- /dev/null +++ b/2-class/cobolt.py @@ -0,0 +1,35 @@ +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 = SP.Serial(portName) + + def __del__(self): + 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): + port.write(b'pa?\r') + reply = port.readline() + power = float(reply.decode("utf-8") ) + return power + + +if __name__ = "__main__": + laser = CoboltLaser("COM1") + laser.setPower(powerInWatts=0.1) + print(laser.power()) + From 14e1a05e208bb04e4d855c35bea41e26102a26fe Mon Sep 17 00:00:00 2001 From: Daniel Cote Date: Wed, 22 May 2019 11:27:10 -0400 Subject: [PATCH 05/24] Created a debug port for cobolt.py, alllowed correction for CoboltLaser class --- 3-class+debugPort/cobolt.py | 79 +++++++++++++++++++++++++++++++++++++ 1 file changed, 79 insertions(+) create mode 100644 3-class+debugPort/cobolt.py diff --git a/3-class+debugPort/cobolt.py b/3-class+debugPort/cobolt.py new file mode 100644 index 00000000..8c7b291b --- /dev/null +++ b/3-class+debugPort/cobolt.py @@ -0,0 +1,79 @@ +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'\n' + self.power = 0.1 + + def open(self): + return + + def close(self): + return + + 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 (\\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() + for c in self.outputBuffer: + data.append(c) + if c == self.lineEnding: + 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() + print(reply) + power = float(reply.decode()) + return power + + +if __name__ == "__main__": + laser = CoboltLaser("debug") + laser.setPower(powerInWatts=0.1) + print(laser.power()) + From 17a7422f8656042af13f6a9e07fec9fa0fc10fb6 Mon Sep 17 00:00:00 2001 From: Daniel Cote Date: Wed, 22 May 2019 11:28:32 -0400 Subject: [PATCH 06/24] Moved modifications to section 2-class --- 2-class/cobolt.py | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/2-class/cobolt.py b/2-class/cobolt.py index f159dab2..b522610f 100644 --- a/2-class/cobolt.py +++ b/2-class/cobolt.py @@ -14,7 +14,8 @@ def __init__(self, portName): self.port = SP.Serial(portName) def __del__(self): - self.port.close() + if self.port is not None: + self.port.close() def setPower(self, powerInWatts): command = "p {0:0.3f}\r".format(powerInWatts) @@ -22,14 +23,14 @@ def setPower(self, powerInWatts): self.port.write(data) def power(self): - port.write(b'pa?\r') - reply = port.readline() - power = float(reply.decode("utf-8") ) + self.port.write(b'pa?\r') + reply = self.port.readline() + print(reply) + power = float(reply.decode()) return power -if __name__ = "__main__": - laser = CoboltLaser("COM1") +if __name__ == "__main__": + laser = CoboltLaser("debug") laser.setPower(powerInWatts=0.1) print(laser.power()) - From 9cb148e6e357381c7adbade93b140ea01ab65523 Mon Sep 17 00:00:00 2001 From: Daniel Cote Date: Wed, 22 May 2019 11:37:45 -0400 Subject: [PATCH 07/24] Better printout, typo in 2-constructor fixed. --- 2-class/cobolt.py | 9 +++++++-- 3-class+debugPort/cobolt.py | 9 ++++++--- 2 files changed, 13 insertions(+), 5 deletions(-) diff --git a/2-class/cobolt.py b/2-class/cobolt.py index b522610f..2a428279 100644 --- a/2-class/cobolt.py +++ b/2-class/cobolt.py @@ -11,6 +11,7 @@ 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): @@ -31,6 +32,10 @@ def power(self): if __name__ == "__main__": - laser = CoboltLaser("debug") + try: + laser = CoboltLaser("COM1") + except: + exit("No laser found on COM1") + laser.setPower(powerInWatts=0.1) - print(laser.power()) + print("Power is {0:0.3f} W".format(laser.power())) diff --git a/3-class+debugPort/cobolt.py b/3-class+debugPort/cobolt.py index 8c7b291b..04db405d 100644 --- a/3-class+debugPort/cobolt.py +++ b/3-class+debugPort/cobolt.py @@ -67,13 +67,16 @@ def setPower(self, powerInWatts): def power(self): self.port.write(b'pa?\r') reply = self.port.readline() - print(reply) power = float(reply.decode()) return power if __name__ == "__main__": - laser = CoboltLaser("debug") + try: + laser = CoboltLaser("COM1") + except: + laser = CoboltLaser("debug") + laser.setPower(powerInWatts=0.1) - print(laser.power()) + print("Power is {0:0.3f} W".format(laser.power())) From 07ecb03714269e40a9df681614cac36d865e87d0 Mon Sep 17 00:00:00 2001 From: Daniel Cote Date: Wed, 22 May 2019 11:44:52 -0400 Subject: [PATCH 08/24] Update README.md --- README.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/README.md b/README.md index 5b92c6d0..7c5f10a5 100644 --- a/README.md +++ b/README.md @@ -2,6 +2,13 @@ A simple device-oriented library with CommunicationPort for controlling devices ## Introduction +You should read the three examples `cobolt.py` : + +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 +4. `4-testing`: Examples of how testing can be done for a serial port and a class + You will find a simple, trivial script named `cobolt.py` to change the power of the Cobolt laser. More to come. From dff2a74a60ef671f72c586ada4bb2ce4caa2c5f8 Mon Sep 17 00:00:00 2001 From: Daniel Cote Date: Wed, 22 May 2019 11:45:55 -0400 Subject: [PATCH 09/24] Update README.md --- README.md | 1 - 1 file changed, 1 deletion(-) diff --git a/README.md b/README.md index 7c5f10a5..f2b49ba9 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,6 @@ You should read the three examples `cobolt.py` : 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 -4. `4-testing`: Examples of how testing can be done for a serial port and a class You will find a simple, trivial script named `cobolt.py` to change the power of the Cobolt laser. From f636106ab0ca00b219a6518f66cf53d1dc1dd995 Mon Sep 17 00:00:00 2001 From: Daniel Cote Date: Wed, 22 May 2019 11:46:47 -0400 Subject: [PATCH 10/24] Update README.md --- README.md | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/README.md b/README.md index f2b49ba9..002f5131 100644 --- a/README.md +++ b/README.md @@ -2,12 +2,10 @@ A simple device-oriented library with CommunicationPort for controlling devices ## Introduction -You should read the three examples `cobolt.py` : +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 -You will find a simple, trivial script named `cobolt.py` to change the power of the Cobolt laser. - More to come. From ee78a452971d0c4e4cbb6e164e9e417a78e1513f Mon Sep 17 00:00:00 2001 From: Daniel Cote Date: Wed, 22 May 2019 18:01:41 -0400 Subject: [PATCH 11/24] Fixed buffer that was not emptying --- 3-class+debugPort/cobolt.py | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/3-class+debugPort/cobolt.py b/3-class+debugPort/cobolt.py index 04db405d..1d493440 100644 --- a/3-class+debugPort/cobolt.py +++ b/3-class+debugPort/cobolt.py @@ -7,7 +7,7 @@ class CoboltDebugSerial: def __init__(self): self.outputBuffer = bytearray() - self.lineEnding = b'\n' + self.lineEnding = b'\r' self.power = 0.1 def open(self): @@ -34,12 +34,15 @@ def write(self, data:bytearray) -> int : def readline(self) -> bytearray: data = bytearray() - for c in self.outputBuffer: - data.append(c) - if c == self.lineEnding: + byte = b'' + while byte != self.lineEnding: + if len(self.outputBuffer) > 0: + byte = self.outputBuffer.pop(0) + data.append(byte) + else: break - return data + return data """ Class-oriented strategy to talk to the Cobolt laser, change its power. See manual in ../manuals/ page 27""" @@ -79,4 +82,8 @@ def power(self): 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())) + print("Power is {0:0.3f} W".format(laser.power())) From 36f834dfb1ad514b15c37d320f32010cd2ea709f Mon Sep 17 00:00:00 2001 From: Daniel Cote Date: Wed, 22 May 2019 18:02:07 -0400 Subject: [PATCH 12/24] Update cobolt.py --- 3-class+debugPort/cobolt.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/3-class+debugPort/cobolt.py b/3-class+debugPort/cobolt.py index 1d493440..2d821241 100644 --- a/3-class+debugPort/cobolt.py +++ b/3-class+debugPort/cobolt.py @@ -85,5 +85,7 @@ def power(self): 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())) From c636e02d8e9c88c4b5807f19ff198a5a58574c80 Mon Sep 17 00:00:00 2001 From: Daniel Cote Date: Wed, 22 May 2019 18:07:29 -0400 Subject: [PATCH 13/24] Completed commands (p?) and added read() to DebugPort --- 3-class+debugPort/cobolt.py | 32 +++++++++++++++++++++++++++++++- 1 file changed, 31 insertions(+), 1 deletion(-) diff --git a/3-class+debugPort/cobolt.py b/3-class+debugPort/cobolt.py index 2d821241..9e92ce6b 100644 --- a/3-class+debugPort/cobolt.py +++ b/3-class+debugPort/cobolt.py @@ -9,13 +9,37 @@ 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') @@ -25,6 +49,12 @@ def write(self, data:bytearray) -> int : 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] @@ -85,7 +115,7 @@ def power(self): 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())) From 282b3511defb39262dfa48ae12d424e0caebb056 Mon Sep 17 00:00:00 2001 From: Daniel Cote Date: Wed, 22 May 2019 19:10:10 -0400 Subject: [PATCH 14/24] Fixed up PhysicalDevice, CoboltLaserDevice and LaserDevice so they are functional. Finally completed the code for PhysicalDevice implementation. --- CoboltDebugSerial.py | 86 ++++++++++++++++++++++++++++++++++++++++++++ CoboltDevice.py | 71 +++++++++++++++++++++++++----------- CommunicationPort.py | 11 +++--- 3 files changed, 142 insertions(+), 26 deletions(-) create mode 100644 CoboltDebugSerial.py diff --git a/CoboltDebugSerial.py b/CoboltDebugSerial.py new file mode 100644 index 00000000..a122104b --- /dev/null +++ b/CoboltDebugSerial.py @@ -0,0 +1,86 @@ +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 + + def open(self): + if self.isOpen: + raise IOError("port is already open") + else: + self.isOpen = True + + return + + def close(self): + if not self.isOpen: + raise IOError("port is not open") + else: + 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}\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}\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 = match.groups()[0] + return len(data) + + match = re.search("l1\r", string) + if match is not None: + self.isOn = True + return len(data) + + match = re.search("l0\r", string) + if match is not None: + self.isOn = False + 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/CoboltDevice.py b/CoboltDevice.py index 166675a6..2da900e4 100644 --- a/CoboltDevice.py +++ b/CoboltDevice.py @@ -1,12 +1,10 @@ from PhysicalDevice import * from CommunicationPort import * import numpy as np - +from CoboltDebugSerial import * +import re class LaserSourceDevice: - def __init__(self): - self.requestedPower = 0 - self.power = 0 def turnOn(self): self.doTurnOn() @@ -24,20 +22,38 @@ def power(self) -> float: 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=bsdPath) + + # print("Port {0}".format(self.port)) + # if self.port is None: + # raise ValueError() + self.port.open() - self.doGetLaserSerialNumber() except Exception as error: self.port.close() raise error @@ -46,34 +62,47 @@ def doShutdownDevice(self): # Doesn't seem to do anything 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) def doGetLaserSerialNumber(self): - self.laserSerialNumber = self.port.writeStringExpectMatchingString('sn?\r', replyPattern='(\\d+)') + 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, '') + self.port.writeString(command) def doGetPower(self): - self.port.writeStringExpectMatchingString('pa?\r', replyPattern='\d.\d+') - + 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") + print("No laser found on COM5. Using debug") + laser = CoboltDevice("debug") + + 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..ba354498 100644 --- a/CommunicationPort.py +++ b/CommunicationPort.py @@ -13,8 +13,8 @@ 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, + + def __init__(self, bsdPath=None, port = None, vendorId=None, productId=None, serialNumber=None): self.bsdPath = bsdPath self.vendorId = vendorId @@ -22,10 +22,11 @@ def __init__(self, bsdPath=None, vendorId=None, self.serialNumber = serialNumber self.portLock = RLock() self.transactionLock = RLock() - + self.port = port # direct port assignment 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) def close(self): self.port.close() @@ -97,7 +98,7 @@ def writeStringReadMatchingGroups(self, string, replyPattern): 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)) From 510b7fac84b992d087d754162370ec369f8970c2 Mon Sep 17 00:00:00 2001 From: Daniel Cote Date: Wed, 22 May 2019 21:04:46 -0400 Subject: [PATCH 15/24] Separate files, reorganized, started unit tests on Devices --- CoboltDevice.py | 38 ++++++-------------- CommunicationPort.py | 68 +++++++++++------------------------ DebugEchoCommunicationPort.py | 35 ++++++++++++++++++ LaserSourceDevice.py | 17 +++++++++ testEcho.py | 6 ++-- 5 files changed, 88 insertions(+), 76 deletions(-) create mode 100644 DebugEchoCommunicationPort.py create mode 100644 LaserSourceDevice.py diff --git a/CoboltDevice.py b/CoboltDevice.py index 2da900e4..04c1c32d 100644 --- a/CoboltDevice.py +++ b/CoboltDevice.py @@ -1,25 +1,14 @@ from PhysicalDevice import * +from LaserSourceDevice import * + from CommunicationPort import * -import numpy as np from CoboltDebugSerial import * -import re - -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() +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): @@ -28,13 +17,11 @@ def __init__(self, bsdPath=None, serialNumber: str = None, self.interlockState = True self.laserSerialNumber = None - self.bsdPath = bsdPath PhysicalDevice.__init__(self, serialNumber, vendorId, productId) LaserSourceDevice.__init__(self) self.port = None - def __del__(self): try: self.port.close() @@ -49,16 +36,14 @@ def doInitializeDevice(self): else: self.port = CommunicationPort(bsdPath=bsdPath) - # print("Port {0}".format(self.port)) - # if self.port is None: - # raise ValueError() - self.port.open() + self.doGetLaserSerialNumber() + except Exception as error: self.port.close() raise error - def doShutdownDevice(self): # Doesn't seem to do anything + def doShutdownDevice(self): self.port.close() return @@ -66,7 +51,7 @@ def doGetInterlockState(self) -> bool: value = self.port.writeStringExpectMatchingString('ilk?\r', replyPattern='([0|1])') self.interlockState = bool(value) - def doGetLaserSerialNumber(self): + def doGetLaserSerialNumber(self) -> str: self.laserSerialNumber = self.port.writeStringReadFirstMatchingGroup('sn?\r', replyPattern='(\\d+)') def doTurnOn(self): @@ -79,7 +64,7 @@ def doSetPower(self, powerInWatts): command = 'p {0:0.3f}\r'.format(powerInWatts) self.port.writeString(command) - def doGetPower(self): + def doGetPower(self) -> float: value = self.port.writeStringReadFirstMatchingGroup('pa?\r', replyPattern='(\\d.\\d+)') return float(value) @@ -88,8 +73,7 @@ def doGetPower(self): try: laser = CoboltDevice("debug") except: - print("No laser found on COM5. Using debug") - laser = CoboltDevice("debug") + exit("No laser") laser.initializeDevice() laser.turnOn() diff --git a/CommunicationPort.py b/CommunicationPort.py index ba354498..487db9db 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,29 +12,35 @@ class CommunicationReadNoMatch(Exception): class CommunicationPort: """CommunicationPort class with basic application-level protocol - functions to write strings and read strings""" + functions to write strings and read strings, and abstract away + the details of the communication. + + Three 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, vendorId=None, - productId=None, serialNumber=None): + def __init__(self, bsdPath=None, port = None): self.bsdPath = bsdPath - self.vendorId = vendorId - self.productId = productId - self.serialNumber = serialNumber + self.port = port # direct port , must be closed. + self.portLock = RLock() self.transactionLock = RLock() - self.port = port # direct port assignment def open(self): 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 readData(self, length) -> bytearray: with self.portLock: data = self.port.read(length) if len(data) != length: @@ -42,7 +48,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): @@ -51,7 +57,7 @@ def writeData(self, data): return nBytesWritten - def readString(self): + def readString(self) -> str: with self.portLock: byte = None data = bytearray(0) @@ -65,9 +71,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) @@ -95,6 +101,7 @@ def writeStringReadMatchingGroups(self, string, replyPattern): with self.transactionLock: self.writeString(string) reply = self.readString() + match = re.search(replyPattern, reply) if match is not None: @@ -103,36 +110,3 @@ def writeStringReadMatchingGroups(self, string, replyPattern): 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..2f5f09b5 --- /dev/null +++ b/DebugEchoCommunicationPort.py @@ -0,0 +1,35 @@ +from CommunicationPort import * + +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/LaserSourceDevice.py b/LaserSourceDevice.py new file mode 100644 index 00000000..8e88ae9c --- /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.doGetInterlock() diff --git a/testEcho.py b/testEcho.py index 08a0376b..bfeecac6 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() From f212f79dc9b61d9d9855badc3770daae097be35d Mon Sep 17 00:00:00 2001 From: Daniel Cote Date: Wed, 22 May 2019 21:39:00 -0400 Subject: [PATCH 16/24] Fixed testEcho and several bugs in DebugEchoCommPort re-opening is an error, but closing twice is not (in pySerial). We do the same. --- CommunicationPort.py | 6 +++++- DebugEchoCommunicationPort.py | 10 ++++++++++ testEcho.py | 32 +++++++++++++++++++++++++++----- 3 files changed, 42 insertions(+), 6 deletions(-) diff --git a/CommunicationPort.py b/CommunicationPort.py index 487db9db..75d55f3d 100644 --- a/CommunicationPort.py +++ b/CommunicationPort.py @@ -15,7 +15,7 @@ class CommunicationPort: functions to write strings and read strings, and abstract away the details of the communication. - Three strategies to initialize the CommunicationPort: + 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) @@ -28,6 +28,10 @@ def __init__(self, bsdPath=None, port = None): self.portLock = RLock() self.transactionLock = RLock() + @property + def isOpen(self): + return self.port.is_open + def open(self): if self.port is None: self.port = serial.Serial(self.bsdPath, 19200, timeout=0.3) diff --git a/DebugEchoCommunicationPort.py b/DebugEchoCommunicationPort.py index 2f5f09b5..defde5b8 100644 --- a/DebugEchoCommunicationPort.py +++ b/DebugEchoCommunicationPort.py @@ -4,12 +4,22 @@ 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): diff --git a/testEcho.py b/testEcho.py index bfeecac6..99bcd192 100644 --- a/testEcho.py +++ b/testEcho.py @@ -19,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)) @@ -44,10 +62,10 @@ def testWriteDataReadEchoSequence(self): self.assertTrue(data == payloadData) def testWriteDataReadEchoLarge(self): - for i in range(100): + for i in range(1000): nBytes = self.port.writeData(payloadData) - for i in range(100): + for i in range(1000): data = self.port.readData(length=len(payloadData)) self.assertTrue(data == payloadData) @@ -72,12 +90,12 @@ def testWriteStringReadEchoSequence(self): self.assertTrue(string == payloadString) def testWriteStringReadEchoLarge(self): - for i in range(100): + for i in range(1000): nBytes = self.port.writeString(payloadString) - for i in range(100): + for i in range(1000): 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: @@ -172,18 +190,22 @@ def threadReadWrite(port, index): threadFailed = index +#@unittest.skip class TestDebugEchoPort(BaseTestCases.TestEchoPort): def setUp(self): self.port = DebugEchoCommunicationPort() self.assertIsNotNone(self.port) self.port.open() + self.assertTrue(self.port.isOpen) def tearDown(self): self.port.close() + self.assertFalse(self.port.isOpen) +@unittest.skip class TestSlowDebugEchoPort(BaseTestCases.TestEchoPort): def setUp(self): From 5c65a292a163cf35592007d1c5a4873e86b84fc9 Mon Sep 17 00:00:00 2001 From: Daniel Cote Date: Wed, 22 May 2019 21:43:57 -0400 Subject: [PATCH 17/24] Update testEcho.py Added comments to errors when tests fail --- testEcho.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/testEcho.py b/testEcho.py index 99bcd192..c9e64e63 100644 --- a/testEcho.py +++ b/testEcho.py @@ -50,16 +50,16 @@ 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) nBytes = self.port.writeData(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(1000): @@ -67,7 +67,7 @@ def testWriteDataReadEchoLarge(self): for i in range(1000): 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) From d0e62b78e7a832e53530788f54286066526fcb3a Mon Sep 17 00:00:00 2001 From: Daniel Cote Date: Wed, 22 May 2019 21:47:14 -0400 Subject: [PATCH 18/24] Update testEcho.py More assertion to catch the occasional test failure: written data is not read back with real device. --- testEcho.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/testEcho.py b/testEcho.py index c9e64e63..779e2546 100644 --- a/testEcho.py +++ b/testEcho.py @@ -54,7 +54,9 @@ def testWriteDataReadEcho(self): 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, "Data {0}, payload:{1}".format(data, payloadData)) @@ -64,6 +66,7 @@ def testWriteDataReadEchoSequence(self): def testWriteDataReadEchoLarge(self): for i in range(1000): nBytes = self.port.writeData(payloadData) + self.assertTrue(nBytes == len(payloadData)) for i in range(1000): data = self.port.readData(length=len(payloadData)) @@ -82,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) @@ -92,6 +97,7 @@ def testWriteStringReadEchoSequence(self): def testWriteStringReadEchoLarge(self): for i in range(1000): nBytes = self.port.writeString(payloadString) + self.assertTrue(nBytes == len(payloadString)) for i in range(1000): string = self.port.readString() From b12012d8a2cfaa0e518a50224d215153cb560840 Mon Sep 17 00:00:00 2001 From: Daniel Cote Date: Wed, 22 May 2019 21:54:04 -0400 Subject: [PATCH 19/24] Added a flush() method to flush (i.e. discard) all buffers. Language is flush() to drop, drain() to wait until all trasnmitted/received. --- CommunicationPort.py | 4 ++++ DebugEchoCommunicationPort.py | 3 +++ testEcho.py | 3 +++ 3 files changed, 10 insertions(+) diff --git a/CommunicationPort.py b/CommunicationPort.py index 75d55f3d..3aac06e5 100644 --- a/CommunicationPort.py +++ b/CommunicationPort.py @@ -44,6 +44,10 @@ def close(self): def bytesAvailable(self) -> int: return self.port.in_waiting + def flush(self): + self.port.reset_input_buffer() + self.port.reset_output_buffer() + def readData(self, length) -> bytearray: with self.portLock: data = self.port.read(length) diff --git a/DebugEchoCommunicationPort.py b/DebugEchoCommunicationPort.py index defde5b8..8cad0692 100644 --- a/DebugEchoCommunicationPort.py +++ b/DebugEchoCommunicationPort.py @@ -25,6 +25,9 @@ def close(self): 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()) diff --git a/testEcho.py b/testEcho.py index 779e2546..9997cbca 100644 --- a/testEcho.py +++ b/testEcho.py @@ -204,6 +204,7 @@ def setUp(self): self.assertIsNotNone(self.port) self.port.open() self.assertTrue(self.port.isOpen) + self.port.flush() def tearDown(self): self.port.close() @@ -219,6 +220,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() @@ -231,6 +233,7 @@ 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") From ff383b2a7f2b5275d667b1e0c5efe92b0a466fd3 Mon Sep 17 00:00:00 2001 From: Daniel Cote Date: Wed, 22 May 2019 22:03:01 -0400 Subject: [PATCH 20/24] Before flushing() check port is open --- CommunicationPort.py | 5 +++-- testEcho.py | 14 ++++++-------- 2 files changed, 9 insertions(+), 10 deletions(-) diff --git a/CommunicationPort.py b/CommunicationPort.py index 3aac06e5..2cbe5a9e 100644 --- a/CommunicationPort.py +++ b/CommunicationPort.py @@ -45,8 +45,9 @@ def bytesAvailable(self) -> int: return self.port.in_waiting def flush(self): - self.port.reset_input_buffer() - self.port.reset_output_buffer() + if self.isOpen: + self.port.reset_input_buffer() + self.port.reset_output_buffer() def readData(self, length) -> bytearray: with self.portLock: diff --git a/testEcho.py b/testEcho.py index 9997cbca..0d8d1fb7 100644 --- a/testEcho.py +++ b/testEcho.py @@ -64,11 +64,11 @@ def testWriteDataReadEchoSequence(self): self.assertTrue(data == payloadData, "Data {0}, payload:{1}".format(data, payloadData)) def testWriteDataReadEchoLarge(self): - for i in range(1000): + for i in range(100): nBytes = self.port.writeData(payloadData) self.assertTrue(nBytes == len(payloadData)) - for i in range(1000): + for i in range(100): data = self.port.readData(length=len(payloadData)) self.assertTrue(data == payloadData, "Data {0}, payload:{1}".format(data, payloadData)) @@ -95,11 +95,11 @@ def testWriteStringReadEchoSequence(self): self.assertTrue(string == payloadString) def testWriteStringReadEchoLarge(self): - for i in range(1000): + for i in range(100): nBytes = self.port.writeString(payloadString) self.assertTrue(nBytes == len(payloadString)) - for i in range(1000): + for i in range(100): string = self.port.readString() self.assertTrue(string == payloadString,"{0} is not {1}".format(string, payloadString)) @@ -196,7 +196,7 @@ def threadReadWrite(port, index): threadFailed = index -#@unittest.skip + class TestDebugEchoPort(BaseTestCases.TestEchoPort): def setUp(self): @@ -210,9 +210,6 @@ def tearDown(self): self.port.close() self.assertFalse(self.port.isOpen) - - -@unittest.skip class TestSlowDebugEchoPort(BaseTestCases.TestEchoPort): def setUp(self): @@ -239,6 +236,7 @@ def setUp(self): def tearDown(self): + self.port.flush() self.port.close() From 663e8c87ca2235771b1ee4e3db32f3ca44e27d23 Mon Sep 17 00:00:00 2001 From: Daniel Cote Date: Wed, 22 May 2019 23:07:34 -0400 Subject: [PATCH 21/24] CoboltDebugSerial tested. readline removed. --- 3-class+debugPort/CoboltDebugSerial.py | 69 ++++++++++++++++++++++++++ CoboltDebugSerial.py | 48 ++++++++++-------- CommunicationPort.py | 5 +- testCoboltDevice.py | 55 ++++++++++++++++++++ testCoboltSerial.py | 68 +++++++++++++++++++++++++ 5 files changed, 224 insertions(+), 21 deletions(-) create mode 100644 3-class+debugPort/CoboltDebugSerial.py create mode 100644 testCoboltDevice.py create mode 100644 testCoboltSerial.py 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/CoboltDebugSerial.py b/CoboltDebugSerial.py index a122104b..f901a3f3 100644 --- a/CoboltDebugSerial.py +++ b/CoboltDebugSerial.py @@ -8,21 +8,22 @@ def __init__(self): self.power = 0.1 self.isOn = True self.requestedPower = 0 - self.isOpen = True + self._isOpen = True + @property + def is_open(self): + return self._isOpen + def open(self): - if self.isOpen: + if self._isOpen: raise IOError("port is already open") else: - self.isOpen = True + self._isOpen = True return def close(self): - if not self.isOpen: - raise IOError("port is not open") - else: - self.isOpen = False + self._isOpen = False return @@ -46,13 +47,13 @@ def write(self, data:bytearray) -> int : match = re.search("pa\\?", string) if match is not None: - replyData = bytearray("{0}\r\n".format(self.power), encoding='utf-8') + 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}\r\n".format(self.requestedPower), encoding='utf-8') + replyData = bytearray("{0:0.4f}\r\n".format(self.requestedPower), encoding='utf-8') self.outputBuffer.extend(replyData) return len(data) @@ -71,16 +72,23 @@ def write(self, data:bytearray) -> int : self.isOn = False return len(data) - raise ValueError("Unknown command: {0}".format(data)) + match = re.search("sn\\?\r", string) + if match is not None: + replyData = bytearray("123456\r\n", 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 len(data) - return 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/CommunicationPort.py b/CommunicationPort.py index 2cbe5a9e..b1b58a88 100644 --- a/CommunicationPort.py +++ b/CommunicationPort.py @@ -23,6 +23,9 @@ class CommunicationPort: def __init__(self, bsdPath=None, port = None): self.bsdPath = bsdPath + + if port is not None and port.is_open: + port.close() self.port = port # direct port , must be closed. self.portLock = RLock() @@ -94,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 diff --git a/testCoboltDevice.py b/testCoboltDevice.py new file mode 100644 index 00000000..a55c8f2d --- /dev/null +++ b/testCoboltDevice.py @@ -0,0 +1,55 @@ +import unittest +import time +from threading import Thread, Lock + +from serial import * +from CommunicationPort import * +from DebugEchoCommunicationPort import * + +class BaseTestCases: + device:CoboltDevice = None + + class TestCobolt(unittest.TestCase): + + def create(self): + + + + +class TestDebugCobolt(BaseTestCases.TestCobolt): + + def setUp(self): + self.port = () + self.assertIsNotNone(self.port) + self.port.open() + + def tearDown(self): + self.port.close() + + +class TestRealCobolt(BaseTestCases.TestCobolt): + + def setUp(self): + self.device = + + def tearDown(self): + self.port.close() + + +class TestRealEchoPort(BaseTestCases.TestEchoPort): + + def setUp(self): + try: + self.port = CommunicationPort("/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() diff --git a/testCoboltSerial.py b/testCoboltSerial.py new file mode 100644 index 00000000..a3058df8 --- /dev/null +++ b/testCoboltSerial.py @@ -0,0 +1,68 @@ +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 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+') + + + +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() From ddf070dfb19ddf08be95a55f94b29a3b00524a12 Mon Sep 17 00:00:00 2001 From: Daniel Cote Date: Wed, 22 May 2019 23:10:34 -0400 Subject: [PATCH 22/24] Test port completely, 100% coverage --- testCoboltSerial.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/testCoboltSerial.py b/testCoboltSerial.py index a3058df8..10785600 100644 --- a/testCoboltSerial.py +++ b/testCoboltSerial.py @@ -14,6 +14,21 @@ class TestCoboltSerialPort(unittest.TestCase): 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: @@ -41,6 +56,9 @@ def testWriteSetPower(self): 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): From db3be918ad06090e53d83f06ccfb42c608ca8c1d Mon Sep 17 00:00:00 2001 From: Daniel Cote Date: Wed, 22 May 2019 23:44:21 -0400 Subject: [PATCH 23/24] coboltDevice test completed. --- CoboltDebugSerial.py | 9 +++++++- CoboltDevice.py | 12 +++++++---- LaserSourceDevice.py | 2 +- PhysicalDevice.py | 3 +++ testCoboltDevice.py | 51 +++++++++++++++++++++++++------------------- 5 files changed, 49 insertions(+), 28 deletions(-) diff --git a/CoboltDebugSerial.py b/CoboltDebugSerial.py index f901a3f3..b7c74fe9 100644 --- a/CoboltDebugSerial.py +++ b/CoboltDebugSerial.py @@ -59,7 +59,7 @@ def write(self, data:bytearray) -> int : match = re.search("p (\\d+\\.?\\d+)\r", string) if match is not None: - self.power = match.groups()[0] + self.power = float(match.groups()[0]) return len(data) match = re.search("l1\r", string) @@ -79,6 +79,13 @@ def write(self, data:bytearray) -> int : 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) + raise ValueError("Unknown command: {0}".format(data)) # def readline(self) -> bytearray: diff --git a/CoboltDevice.py b/CoboltDevice.py index 04c1c32d..7347922b 100644 --- a/CoboltDevice.py +++ b/CoboltDevice.py @@ -34,14 +34,17 @@ def doInitializeDevice(self): if self.bsdPath == "debug": self.port = CommunicationPort(port=CoboltDebugSerial()) else: - self.port = CommunicationPort(bsdPath=bsdPath) - + 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): self.port.close() @@ -50,6 +53,7 @@ def doShutdownDevice(self): def doGetInterlockState(self) -> bool: value = self.port.writeStringExpectMatchingString('ilk?\r', replyPattern='([0|1])') self.interlockState = bool(value) + return self.interlockState def doGetLaserSerialNumber(self) -> str: self.laserSerialNumber = self.port.writeStringReadFirstMatchingGroup('sn?\r', replyPattern='(\\d+)') diff --git a/LaserSourceDevice.py b/LaserSourceDevice.py index 8e88ae9c..3e4cc46b 100644 --- a/LaserSourceDevice.py +++ b/LaserSourceDevice.py @@ -14,4 +14,4 @@ def power(self) -> float: return self.doGetPower() def interlock(self) -> bool: - return self.doGetInterlock() + 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/testCoboltDevice.py b/testCoboltDevice.py index a55c8f2d..d474f08e 100644 --- a/testCoboltDevice.py +++ b/testCoboltDevice.py @@ -5,50 +5,57 @@ from serial import * from CommunicationPort import * from DebugEchoCommunicationPort import * +from CoboltDevice import * class BaseTestCases: device:CoboltDevice = None class TestCobolt(unittest.TestCase): - def create(self): + 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.port = () - self.assertIsNotNone(self.port) - self.port.open() + self.device = CoboltDevice(bsdPath="debug") + self.assertIsNotNone(self.device) + self.device.initializeDevice() def tearDown(self): - self.port.close() - + self.device.shutdownDevice() + return +@unittest.skip class TestRealCobolt(BaseTestCases.TestCobolt): def setUp(self): - self.device = + self.device = CoboltDevice(bsdPath="COM5") + self.assertIsNotNone(self.device) + self.device.initializeDevice() def tearDown(self): - self.port.close() - - -class TestRealEchoPort(BaseTestCases.TestEchoPort): + self.device.shutdownDevice() + return - def setUp(self): - try: - self.port = CommunicationPort("/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__': From 7010ef301ecf738e9f315b23135e107312e8cd6f Mon Sep 17 00:00:00 2001 From: Daniel Cote Date: Thu, 23 May 2019 10:36:15 -0400 Subject: [PATCH 24/24] Update CoboltDebugSerial.py Update to match documentation: OK when command does not return any values, Syntax error: cmd when command not recognized. All tests currently fail. --- CoboltDebugSerial.py | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/CoboltDebugSerial.py b/CoboltDebugSerial.py index b7c74fe9..2534e25c 100644 --- a/CoboltDebugSerial.py +++ b/CoboltDebugSerial.py @@ -60,23 +60,28 @@ def write(self, data:bytearray) -> int : 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) @@ -86,7 +91,9 @@ def write(self, data:bytearray) -> int : return len(data) - raise ValueError("Unknown command: {0}".format(data)) + # Error: + replyData = bytearray("Syntax error: {0}", encoding='utf-8') + self.outputBuffer.extend(replyData) # def readline(self) -> bytearray: # data = bytearray()