Groesz.org
/
progpib
1
0
Fork 0
Code Issues 3 Pull Requests Releases Wiki Activity

add vendor sample programs

This commit is contained in:
S Groesz 2020-09-09 05:06:47 +00:00
parent 11aed31d31
commit 63554da593
6 changed files with 1469 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

159
vendortools/arb.py Executable file
View File

@ -0,0 +1,159 @@
import os.path
import sys
import array
import time
import serial
# Special characters to be escaped
LF = 0x0A
CR = 0x0D
ESC = 0x1B
PLUS = 0x2B
#==============================================================================
def IsSpecial(data):
return data in (LF, CR, ESC, PLUS)
#==============================================================================
def CheckError():
ser.write("SYST:ERR?\n")
ser.write("++read eoi\n")
s = ser.read(100)
print s
#==============================================================================
if __name__ == '__main__':
if len( sys.argv ) != 4:
print "Usage: ", os.path.basename( sys.argv[0] ), "<COM port> <HP33120A GPIB address> <points>"
sys.exit(1)
# Prologix GPIB-USB Controller serial port
comport = sys.argv[1];
# HP33120A GPIB address
addr = sys.argv[2]
# Number of waveform data points
points = int(sys.argv[3])
# 8-16000 points required
if points < 8:
print "Too few points."
exit(1)
if points > 16000:
print "Too many points."
exit(1)
try:
# Waveform points are in short int (16-bit) array
data = array.array('H');
# Create waveform data. Simple ramp up and down.
for i in range(points/2):
data.append((i * 2047)/(points/2))
for i in range(points/2):
data.append((((points/2)-i) * 2047)/(points/2))
# Swap bytes so MSB is first. (Required on Windows)
data.byteswap()
# Output data is in byte array
outdata = array.array('B');
# Build output data, escaping all special characters
for byte in data.tostring():
if IsSpecial(ord(byte)):
outdata.append(ESC)
outdata.append(ord(byte))
# Open serial port
ser = serial.Serial( comport, 9600, timeout=0.5 )
# Set mode as CONTROLLER
ser.write("++mode 1\n")
# Set HP33120A address
ser.write("++addr " + addr + "\n")
# Turn off read-after-write to avoid "Query Unterminated" errors
ser.write("++auto 0\n")
# Do not append CR or LF to GPIB data
ser.write("++eos 3\n")
# Assert EOI with last byte to indicate end of data
ser.write("++eoi 1\n")
# Reset AWG
cmd = "*RST"
print cmd
ser.write(cmd + "\n")
time.sleep(1.0)
CheckError()
# Format output data command. Use length of points array, NOT output array
datalen = len(data) * 2
cmd = "DATA:DAC VOLATILE, #" + str(len(str(datalen))) + str(datalen)
print cmd
# Write binary block data
ser.write(cmd)
ser.write(outdata.tostring())
# Terminate USB command string
ser.write("\n")
time.sleep(0.5)
CheckError()
cmd = "DATA:COPY PULSE, VOLATILE"
print cmd
ser.write(cmd + "\n")
time.sleep(2.0)
CheckError()
cmd = "FUNC:USER PULSE"
print cmd
ser.write(cmd + "\n")
time.sleep(1)
CheckError()
cmd = "FUNC:SHAP USER"
print cmd
ser.write(cmd + "\n")
time.sleep(0.5)
CheckError()
cmd = "OUTP:LOAD 50"
print cmd
ser.write(cmd + "\n")
time.sleep(0.5)
CheckError()
cmd = "FREQ 5000;VOLT 0.5"
print cmd
ser.write(cmd + "\n")
time.sleep(0.5)
CheckError()
except serial.SerialException, e:
print e

177
vendortools/arb_eth.py Executable file
View File

@ -0,0 +1,177 @@
import os.path
import sys
import array
import time
import socket
# Special characters to be escaped
LF = 0x0A
CR = 0x0D
ESC = 0x1B
PLUS = 0x2B
#==============================================================================
def IsSpecial(data):
return data in (LF, CR, ESC, PLUS)
#==============================================================================
def CheckError():
sock.send("SYST:ERR?\n")
sock.send("++read eoi\n")
s = None
try:
s = sock.recv(100)
except socket.timeout:
s = ""
print s
#==============================================================================
if __name__ == '__main__':
if len( sys.argv ) != 4:
print "Usage: ", os.path.basename( sys.argv[0] ), "<GPIB-ETHERNET IP address> <HP33120A GPIB address> <points>"
sys.exit(1)
# IP address of GPIB-ETHERNET
ip = sys.argv[1]
# HP33120A GPIB address
addr = sys.argv[2]
# Number of waveform data points
points = int(sys.argv[3])
# 8-16000 points required
if points < 8:
print "Too few points."
exit(1)
if points > 16000:
print "Too many points."
exit(1)
try:
# Waveform points are in short int (16-bit) array
data = array.array('H');
# Create waveform data. Simple ramp up and down.
for i in range(points/2):
data.append((i * 2047)/(points/2))
for i in range(points/2):
data.append((((points/2)-i) * 2047)/(points/2))
# Swap bytes so MSB is first. (Required on Windows)
data.byteswap()
# Output data is in byte array
outdata = array.array('B');
# Build output data, escaping all special characters
for byte in data.tostring():
if IsSpecial(ord(byte)):
outdata.append(ESC)
outdata.append(ord(byte))
# Open TCP connect to poet 1234 of GPIB-ETHERNET
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP)
sock.settimeout(0.1)
sock.connect((ip, 1234))
# Set mode as CONTROLLER
sock.send("++mode 1\n")
# Set HP33120A address
sock.send("++addr " + addr + "\n")
# Turn off read-after-write to avoid "Query Unterminated" errors
sock.send("++auto 0\n")
# Read timeout is 500 msec
sock.send("++read_tmo_ms 500\n")
# Do not append CR or LF to GPIB data
sock.send("++eos 3\n")
# Assert EOI with last byte to indicate end of data
sock.send("++eoi 1\n")
# Reset AWG
cmd = "*RST"
print cmd
sock.send(cmd + "\n")
time.sleep(1.0)
CheckError()
# Format output data command. Use length of points array, NOT output array
datalen = len(data) * 2
cmd = "DATA:DAC VOLATILE, #" + str(len(str(datalen))) + str(datalen)
print cmd
# Increase host TCP timeout when sending large data
sock.settimeout(10.0)
# Write binary block data
sock.send(cmd)
sock.send(outdata.tostring())
# Terminate ETHERNET data
sock.send("\n")
time.sleep(0.5)
sock.settimeout(0.5)
CheckError()
cmd = "DATA:COPY PULSE, VOLATILE"
print cmd
sock.send(cmd + "\n")
time.sleep(2.0)
CheckError()
cmd = "FUNC:USER PULSE"
print cmd
sock.send(cmd + "\n")
time.sleep(1)
CheckError()
cmd = "FUNC:SHAP USER"
print cmd
sock.send(cmd + "\n")
time.sleep(0.5)
CheckError()
cmd = "OUTP:LOAD 50"
print cmd
sock.send(cmd + "\n")
time.sleep(0.5)
CheckError()
cmd = "FREQ 5000;VOLT 0.5"
print cmd
sock.send(cmd + "\n")
time.sleep(0.5)
CheckError()
sock.close()
except socket.error, e:
print e

65
vendortools/enumip.py Executable file
View File

@ -0,0 +1,65 @@
import socket
import struct
import array
IOCTL_SIOCGIFCONF = 0x8912
# /*
# * Structure used in SIOCGIFCONF request.
# * Used to retrieve interface configuration
# * for machine (useful for programs which
# * must know all networks accessible).
# */
#
# struct ifconf {
# int ifc_len; /* size of associated buffer */
# union {
# caddr_t ifcu_buf;
# struct ifreq *ifcu_req;
# } ifc_ifcu;
#
# /*
# * Interface request structure used for socket
# * ioctl's.
# */
#
# struct ifreq {
# char ifr_name[16]; /* if name, e.g. "en0" */
# union {
# struct sockaddr ifru_addr;
# struct sockaddr ifru_dstaddr;
# struct sockaddr ifru_broadaddr;
# short ifru_flags;
# int ifru_metric;
# caddr_t ifru_data;
# } ifr_ifru;
#
#
# /* Socket address, DARPA Internet style */
# struct sockaddr_in {
# short sin_family;
# unsigned short sin_port;
# struct in_addr sin_addr;
# char sin_zero[8];
# };
#
#
# ioctl (fd, SIOCGIFCONF, &ifconf)
def enumIpUnix():
import fcntl
inbytes = 128 * 32 # Maximum 128 interfaces
ifreq = array.array('B', '\0' * inbytes)
ifconf = struct.pack('iL', inbytes, ifreq.buffer_info()[0])
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
ifconf = fcntl.ioctl(s.fileno(), IOCTL_SIOCGIFCONF, ifconf)
outbytes = struct.unpack('iL', ifconf)[0]
iplist = [socket.inet_ntoa(ifreq[i+20:i+24]) for i in range(0, outbytes, 32)]
return [ip for ip in iplist if ip != '127.0.0.1']

355
vendortools/nfcli.py Executable file
View File

@ -0,0 +1,355 @@
import getopt
import socket
import sys
import os
import platform
from nfutil import *
from enumip import *
#-----------------------------------------------------------------------------
def usage():
print
print "Usage: nfcli --list --eth_addr=ADDR --ip_type=TYPE --ip_addr=IP, --netmask=MASK, --gateway=GW"
print "Search and configure Prologix GPIB-ETHERNET Controllers."
print "--help : display this help"
print "--list : search for controllers"
print "--eth_addr=ADDR : configure controller with Ethernet address ADDR"
print "--ip_type=TYPE : set controller ip address type to TYPE (\"static\" or \"dhcp\")"
print "--ip_addr=IP : set controller address to IP"
print "--netmask=MASK : set controller network mask to MASK"
print "--gateway=GW : set controller default gateway to GW"
#-----------------------------------------------------------------------------
def enumIp():
if platform.system() in ('Windows', 'Microsoft'):
return socket.gethostbyname_ex(socket.gethostname())[2];
return enumIpUnix()
#-----------------------------------------------------------------------------
def ValidateNetParams(ip_str, mask_str, gw_str):
try:
ip = socket.inet_aton(ip_str)
except:
print "IP address is invalid."
return False
try:
mask = socket.inet_aton(mask_str)
except:
print "Network mask is invalid."
return False
try:
gw = socket.inet_aton(gw_str)
except:
print "Gateway address is invalid."
return False
# Validate network mask
# Convert to integer from byte array
mask = struct.unpack("!L", mask)[0]
# Exclude restricted masks
if (mask == 0) or (mask == 0xFFFFFFFF):
print "Network mask is invalid."
return False
# Exclude non-left-contiguous masks
if (((mask + (mask & -mask)) & 0xFFFFFFFF) != 0):
print "Network mask is not contiguous."
return False
# Validate gateway address
octet1 = ord(gw[0])
# Convert to integer from byte array
gw = struct.unpack("!L", gw)[0]
# Exclude restricted addresses
# 0.0.0.0 is valid
if ((gw != 0) and ((octet1 == 0) or (octet1 == 127) or (octet1 > 223))):
print "Gateway address is invalid."
return False
# Validate IP address
octet1 = ord(ip[0])
# Convert to integer from byte array
ip = struct.unpack("!L", ip)[0]
# Exclude restricted addresses
if ((octet1 == 0) or (octet1 == 127) or (octet1 > 223)):
print "IP address is invalid."
return False
# Exclude subnet network address
if ((ip & ~mask) == 0):
print "IP address is invalid."
return False
# Exclude subnet broadcast address
if ((ip & ~mask) == (0xFFFFFFFF & ~mask)):
print "IP address is invalid."
return False
return True
#-----------------------------------------------------------------------------
#def ValidateAddress(address):
# if address is None:
# return False
# parts = address.split(".")
# if len(parts) != 4:
# return False
# try:
# for item in parts:
# if not 0 <= int(item) <= 255:
# return False
# except:
# return False
# return True
#-----------------------------------------------------------------------------
def main():
invalid_args = False
showhelp = False
search = False
ip_type = None
ip_addr = None
netmask = None
gateway = None
eth_addr = None
try:
opts, args = getopt.getopt(sys.argv[1:], '', ['help', 'list', 'eth_addr=', 'ip_type=', 'ip_addr=', 'netmask=', 'gateway='])
except getopt.GetoptError, err:
print str(err)
sys.exit(1)
# Check for unparsed parameters
if len(args) != 0:
usage()
sys.exit(1)
for o, a in opts:
if o == "--help":
showhelp = True
elif o == "--list":
search = True
elif o == "--eth_addr":
eth_addr = a
elif o == "--ip_type":
ip_type = a
elif o == "--ip_addr":
ip_addr = a
elif o == "--netmask":
netmask = a
elif o == "--gateway":
gateway = a
if (len(opts) == 0) or (showhelp):
usage()
sys.exit(1)
if search:
if not eth_addr is None:
print "--list and --eth_addr are not compatible."
invalid_args = True
if not ip_type is None:
print "--list and --ip_type are not compatible."
invalid_args = True
if not ip_addr is None:
print "--list and --ip_addr are not compatible."
invalid_args = True
if not netmask is None:
print "--list and --netmask are not compatible."
invalid_args = True
if not gateway is None:
print "--list and --gateway are not compatible."
invalid_args = True
else:
try:
eth_addr = eth_addr.strip().replace(":", "").replace("-", "")
eth_addr = eth_addr.decode('hex')
except:
print "Invalid Ethernet address."
sys.exit(1)
if len(eth_addr) != 6:
print "Invalid Ethernet address."
sys.exit(1)
if ip_type in ["Static", "static"]:
ip_type = NF_IP_STATIC
elif ip_type in ["Dynamic", "dynamic", "Dhcp", "dhcp"]:
ip_type = NF_IP_DYNAMIC
else:
print "--ip_type must be 'static' or 'dhcp'."
sys.exit(1)
if ip_type == NF_IP_STATIC:
if not ValidateNetParams(ip_addr, netmask, gateway):
invalid_args = True
# if not ValidateIP(ip_addr):
# print "Invalid, or no, IP address specified."
# invalid_args = True
# if not ValidateIP(netmask):
# print "Invalid, or no, netmask specified."
# invalid_args = True
# if not ValidateIP(gateway):
# print "Invalid, or no, gateway address specified."
# invalid_args = True
else:
if ip_addr is None:
ip_addr = "0.0.0.0"
else:
print "--ip_addr not allowed when --ip_type=dhcp."
invalid_args = True
if netmask is None:
netmask = "0.0.0.0"
else:
print "--netmask not allowed when --ip_type=dhcp."
invalid_args = True
if gateway is None:
gateway = "0.0.0.0"
else:
print "--gateway not allowed when --ip_type=dhcp."
invalid_args = True
if invalid_args:
sys.exit(1)
global seq
sys.stdout = os.fdopen(sys.stdout.fileno(), 'w', 0)
iplist = enumIp();
if len(iplist) == 0:
print "Host has no IP address."
sys.exit(1)
devices = {}
for ip in iplist:
print "Searching through network interface:", ip
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
port = 0
try:
s.bind((ip, port))
except socket.error, e:
print "Bind error on send socket:", e
sys.exit(1)
port = s.getsockname()[1]
r = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
r.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
r.setblocking(1)
r.settimeout(0.100)
try:
r.bind(('', port))
except socket.error, e:
print "Bind error on receive socket:", e
sys.exit(1)
d = Discover(s, r)
print
for k in d:
d[k]['host_ip'] = ip
devices[k] = d[k]
s.close()
r.close()
if search:
print
print "Found", len(devices), "Prologix GPIB-ETHERNET Controller(s)."
for key in devices:
PrintDetails(devices[key])
print
else:
if eth_addr in devices:
print "Updating network settings of Prologix GPIB-ETHERNET Controller", FormatEthAddr(eth_addr)
device = devices[eth_addr]
if (device['ip_type'] == NF_IP_STATIC) or (ip_type == NF_IP_STATIC):
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
port = 0
try:
s.bind((device['host_ip'], port))
except socket.error, e:
print "Bind error on send socket:", e
sys.exit(1)
port = s.getsockname()[1]
r = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
r.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
r.setblocking(1)
r.settimeout(0.100)
try:
r.bind(('', port))
except socket.error, e:
print "Bind error on receive socket:", e
sys.exit(1)
result = Assignment(s, r, eth_addr, ip_type, ip_addr, netmask, gateway)
print
if len(result) == 0:
print "Network settings update failed."
else:
if result['result'] == NF_SUCCESS:
print "Network settings updated successfully."
else:
print "Network settings update failed."
else:
print "Prologix GPIB-ETHERNET Controller", FormatEthAddr(eth_addr), "already configured for DHCP."
else:
print "Prologix GPIB-ETHERNET Controller", FormatEthAddr(eth_addr), "not found."
#-----------------------------------------------------------------------------
if __name__ == "__main__":
main()

648
vendortools/nfutil.py Executable file
View File

@ -0,0 +1,648 @@
import struct
import random
import socket
import array
import time
import sys
import socket
NETFINDER_SERVER_PORT = 3040
NF_IDENTIFY = 0
NF_IDENTIFY_REPLY = 1
NF_ASSIGNMENT = 2
NF_ASSIGNMENT_REPLY = 3
NF_FLASH_ERASE = 4
NF_FLASH_ERASE_REPLY = 5
NF_BLOCK_SIZE = 6
NF_BLOCK_SIZE_REPLY = 7
NF_BLOCK_WRITE = 8
NF_BLOCK_WRITE_REPLY = 9
NF_VERIFY = 10
NF_VERIFY_REPLY = 11
NF_REBOOT = 12
NF_SET_ETHERNET_ADDRESS = 13
NF_SET_ETHERNET_ADDRESS_REPLY = 14
NF_TEST = 15
NF_TEST_REPLY = 16
NF_SUCCESS = 0
NF_CRC_MISMATCH = 1
NF_INVALID_MEMORY_TYPE = 2
NF_INVALID_SIZE = 3
NF_INVALID_IP_TYPE = 4
NF_MAGIC = 0x5A
NF_IP_DYNAMIC = 0
NF_IP_STATIC = 1
NF_ALERT_OK = 0x00
NF_ALERT_WARN = 0x01
NF_ALERT_ERROR = 0xFF
NF_MODE_BOOTLOADER = 0
NF_MODE_APPLICATION = 1
NF_MEMORY_FLASH = 0
NF_MEMORY_EEPROM = 1
NF_REBOOT_CALL_BOOTLOADER = 0
NF_REBOOT_RESET = 1
HEADER_FMT = "!2cH6s2x"
IDENTIFY_FMT = HEADER_FMT
IDENTIFY_REPLY_FMT = "!H6c4s4s4s4s4s4s32s"
ASSIGNMENT_FMT = "!3xc4s4s4s32x"
ASSIGNMENT_REPLY_FMT = "!c3x"
FLASH_ERASE_FMT = HEADER_FMT
FLASH_ERASE_REPLY_FMT = HEADER_FMT
BLOCK_SIZE_FMT = HEADER_FMT
BLOCK_SIZE_REPLY_FMT = "!H2x"
BLOCK_WRITE_FMT = "!cxHI"
BLOCK_WRITE_REPLY_FMT = "!c3x"
VERIFY_FMT = HEADER_FMT
VERIFY_REPLY_FMT = "!c3x"
REBOOT_FMT = "!c3x"
SET_ETHERNET_ADDRESS_FMT = "!6s2x"
SET_ETHERNET_ADDRESS_REPLY_FMT = HEADER_FMT
TEST_FMT = HEADER_FMT
TEST_REPLY_FMT = "!32s"
MAX_ATTEMPTS = 10
MAX_TIMEOUT = 0.5
#-----------------------------------------------------------------------------
def MkHeader(id, seq, eth_addr):
return struct.pack(
HEADER_FMT,
chr(NF_MAGIC),
chr(id),
seq,
eth_addr
);
#-----------------------------------------------------------------------------
def MkIdentify(seq):
return MkHeader(NF_IDENTIFY, seq, '\xFF\xFF\xFF\xFF\xFF\xFF')
#-----------------------------------------------------------------------------
def MkAssignment(seq, eth_addr, ip_type, ip_addr, netmask, gateway):
return MkHeader(NF_ASSIGNMENT, seq, eth_addr) + \
struct.pack(
ASSIGNMENT_FMT,
chr(ip_type),
socket.inet_aton(ip_addr),
socket.inet_aton(netmask),
socket.inet_aton(gateway)
)
#-----------------------------------------------------------------------------
def MkFlashErase(seq, eth_addr):
return MkHeader(NF_FLASH_ERASE, seq, eth_addr)
#-----------------------------------------------------------------------------
def MkBlockSize(seq, eth_addr):
return MkHeader(NF_BLOCK_SIZE, seq, eth_addr)
#-----------------------------------------------------------------------------
def MkBlockWrite(seq, eth_addr, memtype, addr, data):
return MkHeader(NF_BLOCK_WRITE, seq, eth_addr) + \
struct.pack(
BLOCK_WRITE_FMT,
chr(memtype),
len(data),
addr,
) + \
data
#-----------------------------------------------------------------------------
def MkVerify(seq, eth_addr):
return MkHeader(NF_VERIFY, seq, eth_addr)
#-----------------------------------------------------------------------------
def MkReboot(seq, eth_addr, reboottype):
return MkHeader(NF_REBOOT, seq, eth_addr) + \
struct.pack(
REBOOT_FMT,
chr(reboottype)
);
#-----------------------------------------------------------------------------
def MkTest(seq, eth_addr):
return MkHeader(NF_TEST, seq, eth_addr)
#-----------------------------------------------------------------------------
def MkSetEthernetAddress(seq, eth_addr, new_eth_addr):
return MkHeader(NF_SET_ETHERNET_ADDRESS, seq, eth_addr) + \
struct.pack(
SET_ETHERNET_ADDRESS_FMT,
new_eth_addr
);
#-----------------------------------------------------------------------------
def UnMkHeader(msg):
params = struct.unpack(
HEADER_FMT,
msg
);
d = {}
d['magic'] = ord(params[0])
d['id'] = ord(params[1])
d['sequence'] = params[2]
d['eth_addr'] = params[3]
return d
#-----------------------------------------------------------------------------
def UnMkIdentifyReply(msg):
hdrlen = struct.calcsize(HEADER_FMT)
d = UnMkHeader(msg[0:hdrlen])
params = struct.unpack(
IDENTIFY_REPLY_FMT,
msg[hdrlen:]
);
d['uptime_days'] = params[0]
d['uptime_hrs'] = ord(params[1])
d['uptime_min'] = ord(params[2])
d['uptime_secs'] = ord(params[3])
d['mode'] = ord(params[4])
d['alert'] = ord(params[5])
d['ip_type'] = ord(params[6])
d['ip_addr'] = params[7]
d['ip_netmask'] = params[8]
d['ip_gw'] = params[9]
d['app_ver'] = params[10]
d['boot_ver'] = params[11]
d['hw_ver'] = params[12]
d['name'] = params[13]
return d
#-----------------------------------------------------------------------------
def UnMkAssignmentReply(msg):
hdrlen = struct.calcsize(HEADER_FMT)
d = UnMkHeader(msg[0:hdrlen])
params = struct.unpack(
ASSIGNMENT_REPLY_FMT,
msg[hdrlen:]
);
d['result'] = ord(params[0])
return d
#-----------------------------------------------------------------------------
def UnMkFlashEraseReply(msg):
return UnMkHeader(msg)
#-----------------------------------------------------------------------------
def UnMkBlockSizeReply(msg):
hdrlen = struct.calcsize(HEADER_FMT)
d = UnMkHeader(msg[0:hdrlen])
params = struct.unpack(
BLOCK_SIZE_REPLY_FMT,
msg[hdrlen:]
);
d['size'] = params[0]
return d
#-----------------------------------------------------------------------------
def UnMkBlockWriteReply(msg):
hdrlen = struct.calcsize(HEADER_FMT)
d = UnMkHeader(msg[0:hdrlen])
params = struct.unpack(
BLOCK_WRITE_REPLY_FMT,
msg[hdrlen:]
);
d['result'] = ord(params[0])
return d
#-----------------------------------------------------------------------------
def UnMkVerifyReply(msg):
hdrlen = struct.calcsize(HEADER_FMT)
d = UnMkHeader(msg[0:hdrlen])
params = struct.unpack(
VERIFY_REPLY_FMT,
msg[hdrlen:]
);
d['result'] = ord(params[0])
return d
#-----------------------------------------------------------------------------
def UnMkTestReply(msg):
hdrlen = struct.calcsize(HEADER_FMT)
d = UnMkHeader(msg[0:hdrlen])
params = struct.unpack(
TEST_REPLY_FMT,
msg[hdrlen:]
);
result = ''
for i in params[0]:
if ord(i) == 0:
break
result = result + i
d['result'] = result
return d
#-----------------------------------------------------------------------------
def UnMkSetEthernetAddressReply(msg):
return UnMkHeader(msg)
#-----------------------------------------------------------------------------
def SendMsg(s, msg):
try:
s.sendto(msg, ('<broadcast>', NETFINDER_SERVER_PORT))
except socket.error, e:
print e
return False;
return True;
#-----------------------------------------------------------------------------
def RecvMsg(s):
try:
return s.recv(256)
except socket.error: # ignore socket errors
return ''
#-----------------------------------------------------------------------------
def Discover(s, r):
devices = {}
attempts = 2
while attempts > 0:
attempts = attempts - 1
seq = random.randint(1, 65535)
msg = MkIdentify(seq)
if (SendMsg(s, msg)):
exp = time.time() + MAX_TIMEOUT
while time.time() < exp:
sys.stdout.write('.')
reply = RecvMsg(r)
if len(reply) != struct.calcsize(HEADER_FMT) + struct.calcsize(IDENTIFY_REPLY_FMT):
continue
d = UnMkIdentifyReply(reply)
if d['magic'] != NF_MAGIC:
continue
if d['id'] != NF_IDENTIFY_REPLY:
continue
if d['sequence'] != seq:
continue
devices[d['eth_addr']] = d
return devices
#-----------------------------------------------------------------------------
def Identify(s, r, eth_addr):
attempts = 2
while attempts > 0:
attempts = attempts - 1
seq = random.randint(1, 65535)
msg = MkIdentify(seq)
if (SendMsg(s, msg)):
exp = time.time() + 2 # Longer timeout
while time.time() < exp:
sys.stdout.write('.')
reply = RecvMsg(r)
if len(reply) != struct.calcsize(HEADER_FMT) + struct.calcsize(IDENTIFY_REPLY_FMT):
continue
d = UnMkIdentifyReply(reply)
if d['magic'] != NF_MAGIC:
continue
if d['id'] != NF_IDENTIFY_REPLY:
continue
if d['sequence'] != seq:
continue
if d['eth_addr'] != eth_addr:
continue
return d
return {}
#-----------------------------------------------------------------------------
def Assignment(s, r, eth_addr, ip_type, ip_addr, netmask, gateway):
attempts = MAX_ATTEMPTS
while attempts > 0:
attempts = attempts - 1
seq = random.randint(1, 65535)
msg = MkAssignment(seq, eth_addr, ip_type, ip_addr, netmask, gateway)
if (SendMsg(s, msg)):
exp = time.time() + MAX_TIMEOUT
while time.time() < exp:
sys.stdout.write('.')
reply = RecvMsg(r)
if len(reply) != struct.calcsize(HEADER_FMT) + struct.calcsize(ASSIGNMENT_REPLY_FMT):
continue
d = UnMkAssignmentReply(reply)
if d['magic'] != NF_MAGIC:
continue
if d['id'] != NF_ASSIGNMENT_REPLY:
continue
if d['sequence'] != seq:
continue
if d['eth_addr'] != eth_addr:
continue
return d
return {}
#-----------------------------------------------------------------------------
def FlashErase(s, r, eth_addr):
attempts = MAX_ATTEMPTS
while attempts > 0:
attempts = attempts - 1
seq = random.randint(1, 65535)
msg = MkFlashErase(seq, eth_addr)
if (SendMsg(s, msg)):
exp = time.time() + 10 # Flash erase could take a while
while time.time() < exp:
sys.stdout.write('.')
reply = RecvMsg(r)
if len(reply) != struct.calcsize(HEADER_FMT):
continue
d = UnMkFlashEraseReply(reply)
if d['magic'] != NF_MAGIC:
continue
if d['id'] != NF_FLASH_ERASE_REPLY:
continue
if d['sequence'] != seq:
continue
if d['eth_addr'] != eth_addr:
continue
return d
return {}
#-----------------------------------------------------------------------------
def BlockSize(s, r, eth_addr):
attempts = MAX_ATTEMPTS
while attempts > 0:
attempts = attempts - 1
seq = random.randint(1, 65535)
msg = MkBlockSize(seq, eth_addr)
if (SendMsg(s, msg)):
exp = time.time() + MAX_TIMEOUT
while time.time() < exp:
sys.stdout.write('.')
reply = RecvMsg(r)
if len(reply) != struct.calcsize(HEADER_FMT) + struct.calcsize(BLOCK_SIZE_REPLY_FMT):
continue
d = UnMkBlockSizeReply(reply)
if d['magic'] != NF_MAGIC:
continue
if d['id'] != NF_BLOCK_SIZE_REPLY:
continue
if d['sequence'] != seq:
continue
if d['eth_addr'] != eth_addr:
continue
return d
return {}
#-----------------------------------------------------------------------------
def BlockWrite(s, r, eth_addr, memtype, addr, data):
attempts = MAX_ATTEMPTS
while attempts > 0:
attempts = attempts - 1
seq = random.randint(1, 65535)
msg = MkBlockWrite(seq, eth_addr, memtype, addr, data)
if (SendMsg(s, msg)):
exp = time.time() + MAX_TIMEOUT
while time.time() < exp:
#sys.stdout.write('.'),
reply = RecvMsg(r)
if len(reply) != struct.calcsize(HEADER_FMT) + struct.calcsize(BLOCK_WRITE_REPLY_FMT):
continue
d = UnMkBlockWriteReply(reply)
if d['magic'] != NF_MAGIC:
continue
if d['id'] != NF_BLOCK_WRITE_REPLY:
continue
if d['sequence'] != seq:
continue
if d['eth_addr'] != eth_addr:
continue
return d
return {}
#-----------------------------------------------------------------------------
def Verify(s, r, eth_addr):
attempts = MAX_ATTEMPTS
while attempts > 0:
attempts = attempts - 1
seq = random.randint(1, 65535)
msg = MkVerify(seq, eth_addr)
if (SendMsg(s, msg)):
exp = time.time() + MAX_TIMEOUT
while time.time() < exp:
sys.stdout.write('.')
reply = RecvMsg(r)
if len(reply) != struct.calcsize(HEADER_FMT) + struct.calcsize(VERIFY_REPLY_FMT):
continue
d = UnMkVerifyReply(reply)
if d['magic'] != NF_MAGIC:
continue
if d['id'] != NF_VERIFY_REPLY:
continue
if d['sequence'] != seq:
continue
if d['eth_addr'] != eth_addr:
continue
return d
return {}
#-----------------------------------------------------------------------------
def Reboot(s, eth_addr, reboottype):
seq = random.randint(1, 65535)
msg = MkReboot(seq, eth_addr, reboottype)
if (SendMsg(s, msg)):
return
#-----------------------------------------------------------------------------
def Test(s, r, eth_addr):
seq = random.randint(1, 65535)
msg = MkTest(seq, eth_addr)
if (SendMsg(s, msg)):
exp = time.time() + MAX_TIMEOUT
while time.time() < exp:
sys.stdout.write('.')
reply = RecvMsg(r)
if len(reply) != struct.calcsize(HEADER_FMT) + struct.calcsize(TEST_REPLY_FMT):
continue
d = UnMkTestReply(reply)
if d['magic'] != NF_MAGIC:
continue
if d['id'] != NF_TEST_REPLY:
continue
if d['sequence'] != seq:
continue
if d['eth_addr'] != eth_addr:
continue
return d
return {}
#-----------------------------------------------------------------------------
def SetEthernetAddress(s, r, eth_addr, new_eth_addr):
seq = random.randint(1, 65535)
msg = MkSetEthernetAddress(seq, eth_addr, new_eth_addr)
if (SendMsg(s, msg)):
exp = time.time() + MAX_TIMEOUT
while time.time() < exp:
sys.stdout.write('.')
reply = RecvMsg(r)
if len(reply) != struct.calcsize(HEADER_FMT):
continue
d = UnMkSetEthernetAddressReply(reply)
if d['magic'] != NF_MAGIC:
continue
if d['id'] != NF_SET_ETHERNET_ADDRESS_REPLY:
continue
if d['sequence'] != seq:
continue
if d['eth_addr'] != eth_addr:
continue
return d
return {}
#-----------------------------------------------------------------------------
def FormatEthAddr(a):
return "%02X-%02X-%02X-%02X-%02X-%02X" % (ord(a[0]), ord(a[1]), ord(a[2]), ord(a[3]), ord(a[4]), ord(a[5]))
#-----------------------------------------------------------------------------
def PrintDetails(d):
print
print "Ethernet Address:", FormatEthAddr(d['eth_addr'])
print "Hardware:", socket.inet_ntoa(d['hw_ver']), "Bootloader:", socket.inet_ntoa(d['boot_ver']), "Application:", socket.inet_ntoa(d['app_ver'])
print "Uptime:", d['uptime_days'], 'days', d['uptime_hrs'], 'hours', d['uptime_min'], 'minutes', d['uptime_secs'], 'seconds'
if d['ip_type'] == NF_IP_STATIC:
print "Static IP"
elif d['ip_type'] == NF_IP_DYNAMIC:
print "Dynamic IP"
else:
print "Unknown IP type"
print "IP Address:", socket.inet_ntoa(d['ip_addr']), "Mask:", socket.inet_ntoa(d['ip_netmask']), "Gateway:", socket.inet_ntoa(d['ip_gw'])
print "Mode:",
if d['mode'] == NF_MODE_BOOTLOADER:
print 'Bootloader'
elif d['mode'] == NF_MODE_APPLICATION:
print 'Application'
else:
print 'Unknown'

65
vendortools/pxread.py Executable file
View File

@ -0,0 +1,65 @@
import os.path
import serial
import sys
if __name__ == '__main__':
if len( sys.argv ) != 3:
print "Usage: ", os.path.basename( sys.argv[0] ), "<COM port> <GPIB address>"
sys.exit(1)
comport = sys.argv[1];
addr = sys.argv[2];
ser = serial.Serial()
try:
success = True
ser = serial.Serial( '\\\\.\\'+sys.argv[1], 9600, timeout=0.5 )
cmd = '++mode 1'
print 'Sending:', cmd
ser.write(cmd + '\n')
s = ser.read(256);
if len(s) > 0:
print s
cmd = '++addr ' + addr
print 'Sending:', cmd
ser.write(cmd + '\n')
s = ser.read(256);
if len(s) > 0:
print s
cmd = '++auto 1'
print 'Sending:', cmd
ser.write(cmd + '\n')
s = ser.read(256);
if len(s) > 0:
print s
cmd = 'plot;'
print 'Sending:', cmd
ser.write(cmd + '\n')
f = open("plot.bin", "wb")
while (1):
s = ser.read(1000)
if len(s) > 0:
f.write(s)
else:
break
f.close()
except serial.SerialException, e:
print e
f.close()
except KeyboardInterrupt, e:
ser.close()
f.close()