Mercurial > ~darius > hgwebdir.cgi > pyinst
changeset 71:00800345fbae default tip
Python3ise RSIB code
| author | Daniel O'Connor <doconnor@gsoft.com.au> |
|---|---|
| date | Thu, 24 Aug 2023 16:52:10 +0930 |
| parents | 6ffa6fcf278e |
| children | |
| files | rs_fsp7_example.py rsib.py scpi.py |
| diffstat | 3 files changed, 40 insertions(+), 42 deletions(-) [+] |
line wrap: on
line diff
--- a/rs_fsp7_example.py Mon Aug 30 12:40:43 2021 +0930 +++ b/rs_fsp7_example.py Thu Aug 24 16:52:10 2023 +0930 @@ -33,7 +33,7 @@ def test(r): # ID instrument r.write('*IDN?') - print("ID is " + r.read(5)) + print("ID is " + r.read(5).decode('ascii')) # Reset to defaults r.write("*RST")
--- a/rsib.py Mon Aug 30 12:40:43 2021 +0930 +++ b/rsib.py Thu Aug 24 16:52:10 2023 +0930 @@ -77,7 +77,6 @@ MSG_CMD = 0x90 # Command to the instrument MSG_SRQ = 0x91 # Query SRQ -import exceptions import socket class RSIBDevice(object): @@ -93,12 +92,12 @@ s1 = socket.socket() s1.settimeout(5) s1.connect((self.host, self.port)) - s1.send('\x00\x00\x00\x40') - rx = '' + s1.send(b'\x00\x00\x00\x40') + rx = b'' while len(rx) < 8: rx = rx + s1.recv(8) - if rx == '': - raise exceptions.IOError("EOF from device") + if rx == b'': + raise IOError("EOF from device") s2 = socket.socket() s2.settimeout(5) @@ -114,11 +113,11 @@ self.s1.settimeout(timeout) if len(cmd) > 0x99: - raise exceptions.ValueError("Command too long") + raise ValueError("Command too long") # Pre-increment for easy comparison in read self.tag = (self.tag + 1) & 0xff - msg = '\x00\x00\x00' + chr(len(cmd)) + '\x90\x00' + chr(self.tag) + cmd + msg = b'\x00\x00\x00' + bytes([len(cmd)]) + b'\x90\x00' + bytes([self.tag]) + cmd.encode('ascii') self.s1.send(msg) def read(self, timeout = 0.5): @@ -127,67 +126,67 @@ self.s1.settimeout(timeout) # Fetch the header - rx = '' + rx = b'' while len(rx) < 7: rx = self.s1.recv(7) - if rx == '': - raise exceptions.IOError("EOF from device") + if rx == b'': + raise IOError("EOF from device") - if self.tag != ord(rx[6]): - raise exceptions.IOError("Reply out of order, got 0x%02x expected 0x%02x" % (ord(rx[6]), self.tag)) + if self.tag != rx[6]: + raise IOError("Reply out of order, got 0x%02x expected 0x%02x" % (ord(rx[6]), self.tag)) - rxlen = ord(rx[0]) << 24 | ord(rx[1]) << 16 | ord(rx[2]) << 8 | ord(rx[3]) - #print "Msg ID 0x%02x" % (ord(rx[4])) - if False and ord(rx[4]) != MSG_CMDREP: - raise exceptions.IOError("Unexpected Msg ID 0x%02x" % (ord(rx[4]))) + rxlen = rx[0] << 24 | rx[1] << 16 | rx[2] << 8 | rx[3] + #print "Msg ID 0x%02x" % (rx[4]) + if False and rx[4] != MSG_CMDREP: + raise IOError("Unexpected Msg ID 0x%02x" % (rx[4])) - if ord(rx[5]) != 0: - print("Mystery byte %d != 0" % (ord(rx[5]))) + if rx[5] != 0: + print("Mystery byte %d != 0" % (rx[5])) # Fetch the actual reply now we know the length - reply = '' + reply = b'' while len(reply) < rxlen: rx = self.s1.recv(rxlen) - if rx == '': - raise exceptions.IOError("EOF from device") + if rx == b'': + raise IOError("EOF from device") reply += rx return(reply) def queryrsb(self): - msg = '\x00\x00\x00\x00\xd1\x18\x00' + msg = b'\x00\x00\x00\x00\xd1\x18\x00' self.s2.send(msg) - reply = '' + reply = b'' while len(reply) < 7: rx = self.s2.recv(7) - if rx == '': - raise exceptions.IOError("EOF from device") + if rx == b'': + raise IOError("EOF from device") reply += rx # '\x00\x00\x00\x00\x80\x04\x01' => STB = 4 - if rx[4] != '\x80': - raise exceptions.IOError("Incorrect Msg ID in response to STB query") + if rx[4] != b'\x80': + raise IOError("Incorrect Msg ID in response to STB query") - return ord(rx[5]) + return rx[5] def waitsrq(self): - msg = '\x00\x00\x00\x00\xb1\x00\x00' + msg = b'\x00\x00\x00\x00\xb1\x00\x00' def testsrq(self): - msg = '\x00\x00\x00\x00\x91\x00\x00' + msg = b'\x00\x00\x00\x00\x91\x00\x00' self.s2.send(msg) - reply = '' + reply = b'' while len(reply) < 7: rx = self.s2.recv(7) - if rx == '': - raise exceptions.IOError("EOF from device") + if rx == b'': + raise IOError("EOF from device") reply += rx # 00 00 00 00 80 14 08 - SRQ = 0 # 00 00 00 00 a0 64 07 - SRQ = 1 - if rx == '\x00\x00\x00\x00\x80\x14\x08': + if rx == b'\x00\x00\x00\x00\x80\x14\x08': return False - elif rx == '\x00\x00\x00\x00\xa0\x64\x07': + elif rx == b'\x00\x00\x00\x00\xa0\x64\x07': return True else: - raise exceptions.IOError("Unknown SRQ byte sequence - " + str(list(map(ord, rx)))) + raise IOError("Unknown SRQ byte sequence - " + str(rx))
--- a/scpi.py Mon Aug 30 12:40:43 2021 +0930 +++ b/scpi.py Thu Aug 24 16:52:10 2023 +0930 @@ -25,7 +25,6 @@ # SUCH DAMAGE. # -import exceptions import numpy import re @@ -56,7 +55,7 @@ data = clean(data) if data[0] != '#': - raise exceptions.ValueError('data length header incorrect') + raise ValueError('data length header incorrect') dlenlen = int(data[1]) if dlenlen == 0: return numpy.array([]) @@ -71,7 +70,7 @@ data = data[0:-1] if len(data) != wanted: - raise exceptions.ValueError('length mismatch, header says %d, actually got %d bytes' % (wanted, len(data))) + raise ValueError('length mismatch, header says %d, actually got %d bytes' % (wanted, len(data))) return numpy.frombuffer(data[2 + dlenlen:], dtype = dtype) def ascdecode(data, dtype = numpy.float32, sep = None): @@ -96,7 +95,7 @@ def instURL(url): m = insturlre.match(url) if m == None: - raise exceptions.ValueError('Unable to parse URL') + raise ValueError('Unable to parse URL') proto, xxx, user, xxx, pwd, address, xxx, port = m.groups() if proto == 'rsib': import rsib @@ -113,6 +112,6 @@ import scpisock return scpisock.SCPISockDevice(address, port) else: - raise exceptions.NotImplementedError("unknown protocol " + proto) + raise NotImplementedError("unknown protocol " + proto)