Merge changes from 'Physical device' to 'MaxCobolt'

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()

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()))

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

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()))
+

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