842 lines
23 KiB
Python
Executable File
842 lines
23 KiB
Python
Executable File
# Python >= 3.6
|
|
|
|
import struct
|
|
import random
|
|
import socket
|
|
import array
|
|
import time
|
|
import sys
|
|
import socket
|
|
|
|
now = time.time # now(): return current time in seconds as decimal
|
|
|
|
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 # seconds
|
|
|
|
#---- NOTES ----
|
|
# Header Format: !2cH6s2x
|
|
# ! : Big Endian, std format
|
|
# 2c: 2 bytes
|
|
# H : unsigned short 0-65535 (2 bytes)
|
|
# 6s: 6 byte array
|
|
# 2x: 2 pad (null) bytes
|
|
#
|
|
# NF_MAGIC: 1 byte
|
|
# COMMAND: 1 byte
|
|
# SEQUENCE: 2 bytes
|
|
# ETH_ADDR (MAC): 6 bytes
|
|
# PAD: 2 bytes
|
|
# Total: 12 bytes
|
|
#
|
|
# Example: b'Z\x00V\x81\xff\xff\xff\xff\xff\xff\x00\x00'
|
|
# NF_MAGIC: b'Z' (HEX: 5A, DEC: 90)
|
|
# COMMAND: b'\x00' (NF_IDENTIFY)
|
|
# SEQUENCE: b'V\x81' (22145)
|
|
# ETH_ADDR (MAC): b'\xff\xff\xff\xff\xff\xff' (FF:FF:FF:FF:FF:FF)
|
|
# PAD: b'\x00\x00'
|
|
|
|
|
|
#-----------------------------------------------------------------------------
|
|
def MkSeq():
|
|
"""Generate a random int for the SEQ parameter, used as a type of CRC."""
|
|
return random.randint(1, 65535)
|
|
|
|
|
|
#-----------------------------------------------------------------------------
|
|
def MkHeader(header, seq, eth_addr):
|
|
"""Generate a network header packet. (12 bytes)
|
|
|
|
:param header: The header byte that identifies the payload
|
|
:type header: byte
|
|
:param seq: The SEQ parameter, used as a simple CRC for the packet.
|
|
:type seq: int
|
|
:param eth_addr: The ethernet address (MAC) of the sender.
|
|
:type eth_addr: bytes
|
|
:returns: a bytestring representing the header data
|
|
:rtype: bytes
|
|
"""
|
|
|
|
return struct.pack(
|
|
HEADER_FMT,
|
|
bytes([NF_MAGIC]),
|
|
bytes([header]),
|
|
seq,
|
|
eth_addr
|
|
)
|
|
|
|
|
|
#-----------------------------------------------------------------------------
|
|
def MkIdentify(seq):
|
|
"""Generate an Identify packet."""
|
|
return MkHeader(NF_IDENTIFY, seq, bytes([0xFF]) * 6)
|
|
|
|
|
|
#-----------------------------------------------------------------------------
|
|
def MkIdentifyReply(seq, vDev):
|
|
"""Generate an IdentifyReply packet.
|
|
|
|
:param seq: The SEQ parameter. Should match the SEQ that was sent in
|
|
initial Identify request.
|
|
:type seq: int
|
|
:param vDev:
|
|
:type vDev:
|
|
"""
|
|
|
|
# Header Format: !2cH6s2x
|
|
# ! : Big Endian, std format
|
|
# 2c: 2 bytes
|
|
# H : unsigned short 0-65535 (2 bytes)
|
|
# 6s: 6 byte array
|
|
# 2x: 2 pad (null) bytes
|
|
#
|
|
# NF_MAGIC: 1 byte
|
|
# COMMAND: 1 byte
|
|
# SEQUENCE: 2 bytes
|
|
# ETH_ADDR (MAC): 6 bytes
|
|
# PAD: 2 bytes
|
|
# Total: 12 bytes
|
|
#
|
|
# Example: b'Z\x00V\x81\xff\xff\xff\xff\xff\xff\x00\x00'
|
|
# NF_MAGIC: b'Z' (HEX: 5A, DEC: 90)
|
|
# COMMAND: b'\x00' (NF_IDENTIFY)
|
|
# SEQUENCE: b'V\x81' (22145)
|
|
# ETH_ADDR (MAC): b'\xff\xff\xff\xff\xff\xff' (FF:FF:FF:FF:FF:FF)
|
|
# PAD: b'\x00\x00'
|
|
|
|
# header: 12 bytes (See MkHeader for details)
|
|
# IdentifyReplyData Format: !H6c4s4s4s4s4s4s32s
|
|
# ! : Network (Big Endian) byte order
|
|
# H : unsigned short - Int 0-65535 (2 bytes) [uptime_days]
|
|
# 6c : 6 bytes [uptime_hrs; uptime_mins; uptime_secs;
|
|
# mode; alert; ip_type]
|
|
# 4s : 4 byte array [ip_addr]
|
|
# 4s : 4 byte array [ip_netmask]
|
|
# 4s : 4 byte array [ip_gw]
|
|
# 4s : 4 byte array [app_ver]
|
|
# 4s : 4 byte array [boot_ver]
|
|
# 4s : 4 byte array [hw_ver]
|
|
# 32s: 32 byte array [name]
|
|
# IdentifyReplyData
|
|
# uptime_days [0-65535]
|
|
# uptime_hrs [0-23]
|
|
# uptime_mins [0-59]
|
|
# uptime_secs [0-59]
|
|
# mode [0: Bootloader mode; 1: Application (operating) mode]
|
|
# alert [0: OK; 1: WARNING; 255: ERROR]
|
|
# ip_type [static|dynamic]
|
|
# ip_addr [255.255.255.255] (IP addr of device)
|
|
# ip_netmask [255.255.255.255] (Netmask of device)
|
|
# ip_gw [255.255.255.255] (gateway of davice)
|
|
# app_ver [255.255.255.255] (application firmware version)
|
|
# boot_ver [255.255.255.255] (bootloader firmware version)
|
|
# hw_ver [255.255.255.255] (hardware version)
|
|
# name [32 chars] (Hostname)
|
|
# todo: setup vDev class, VirtualDevice, which will provide the answers
|
|
# for the reply data
|
|
return MkHeader(NF_IDENTIFY_REPLY, seq, vDev.addr) + \
|
|
struct.pack(
|
|
IDENTIFY_REPLY_FMT,
|
|
chr(vDev.updays),
|
|
chr(vDev.uphrs),
|
|
chr(vDev.upmins),
|
|
chr(vDev.upsecs),
|
|
chr(vDev.mode),
|
|
chr(vDev.alert),
|
|
chr(vDev.iptype),
|
|
socket.inet_aton(vDev.ipaddr),
|
|
socket.inet_aton(vDev.ipnetmask),
|
|
socket.inet_aton(vDev.ipgw),
|
|
vDev.version_binary,
|
|
vDev.bootver_binary,
|
|
vDev.hwver_binary,
|
|
vDev.name_binary
|
|
)
|
|
|
|
#-----------------------------------------------------------------------------
|
|
def MkAssignment(seq, eth_addr, ip_type, ip_addr, netmask, gateway):
|
|
"""Generate a network configuration packet.
|
|
|
|
The network configuration packet will configure the network settings of a
|
|
Prologix Ethernet GPIB device.
|
|
|
|
Parameters
|
|
----------
|
|
eth_addr : bytes
|
|
The ethernet address (MAC) of the target device to configure
|
|
ip_type : int
|
|
The type of IP configuration; either NF_IP_STATIC or NF_IP_DYNAMIC
|
|
ip_addr : str
|
|
The static IP address to set on the target device
|
|
netmask : str
|
|
The netmask to set on the target device
|
|
gateway : str
|
|
The network gateway to set on the target device
|
|
|
|
Returns
|
|
-------
|
|
bytes
|
|
A bytestring representing the data packet to send to the target device
|
|
"""
|
|
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 as 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 = now() + MAX_TIMEOUT
|
|
while now() < 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 = now() + 2 # timeout
|
|
|
|
while now() < exp:
|
|
# wait for a valid response
|
|
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 = now() + 10 # Flash erase could take a while
|
|
|
|
while now() < 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 = now() + MAX_TIMEOUT
|
|
|
|
while now() < 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" % (a[0],
|
|
a[1],
|
|
a[2],
|
|
a[3],
|
|
a[4],
|
|
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:", end = " ")
|
|
if d['mode'] == NF_MODE_BOOTLOADER:
|
|
print('Bootloader')
|
|
elif d['mode'] == NF_MODE_APPLICATION:
|
|
print('Application')
|
|
else:
|
|
print('Unknown')
|
|
|
|
|
|
class ProGPIB:
|
|
"""
|
|
=====
|
|
ProGPIB
|
|
=====
|
|
|
|
Provides an interface to Prologix GPIB adapters
|
|
"""
|
|
|
|
from bits import Bytes
|
|
import time
|
|
|
|
def __init__(self, MAC="02:01:01:01:01:01", Virtual=False, IP="0.0.0.0"):
|
|
"""
|
|
MAC: Media Access Control (MAC) Address of device
|
|
Virtual: (bool) Set True to emulate a Prologix GPIB device
|
|
IP: The IPv4 address of the device
|
|
"""
|
|
self.__start = time.time()
|
|
|