395 lines
14 KiB
Python
395 lines
14 KiB
Python
import json
|
|
import opticalinterface.opticalcommands as opticalcommands
|
|
import opticalinterface.opticalexceptions as opticalexceptions
|
|
import os
|
|
import struct
|
|
import math
|
|
import logging
|
|
import time
|
|
|
|
|
|
class BreezeOptical(object):
|
|
|
|
SEEK_SET = 0
|
|
SEEK_CUR = 1
|
|
SEEK_END = 2
|
|
|
|
# maximum number of attempts to login
|
|
LOGIN_ATTEMP_MAX = 5
|
|
|
|
def __init__(self, port, irda=False):
|
|
# import register definitions
|
|
DIRNAME = os.path.split(__file__)[0]
|
|
registers_file = os.path.join(DIRNAME, "registers.json")
|
|
skeleton_file = os.path.join(DIRNAME, "skeletonkeys.json")
|
|
|
|
with open(registers_file) as fp:
|
|
self.register_dict = json.load(fp)
|
|
with open(skeleton_file) as fp:
|
|
self.keys = json.load(fp)
|
|
|
|
# strip out comments at start and end
|
|
self.register_dict = {r: self.register_dict[r] for r in
|
|
self.register_dict if not r.startswith("_")}
|
|
|
|
# use the type field to select the data type class to use
|
|
for r in self.register_dict:
|
|
reg_type = self.register_dict[r]['type']
|
|
if reg_type == "int8_t":
|
|
reg_type_class = opticalcommands.Int8
|
|
elif reg_type == "int16_t":
|
|
reg_type_class = opticalcommands.Int16
|
|
elif reg_type == "int32_t":
|
|
reg_type_class = opticalcommands.Int32
|
|
elif reg_type == "int64_t":
|
|
reg_type_class = opticalcommands.Int64
|
|
elif reg_type == "uint8_t":
|
|
reg_type_class = opticalcommands.UInt8
|
|
elif reg_type == "uint16_t":
|
|
reg_type_class = opticalcommands.UInt16
|
|
elif reg_type == "uint32_t":
|
|
reg_type_class = opticalcommands.UInt32
|
|
elif reg_type == "uint64_t":
|
|
reg_type_class = opticalcommands.UInt64
|
|
elif reg_type == "uint96_t":
|
|
reg_type_class = opticalcommands.UInt96
|
|
elif reg_type == "bool_t":
|
|
reg_type_class = opticalcommands.Bool
|
|
elif reg_type == "enum8":
|
|
reg_type_class = opticalcommands.Enum8
|
|
elif reg_type == "bitmask8":
|
|
reg_type_class = opticalcommands.Bitmask8
|
|
elif reg_type == "string":
|
|
reg_type_class = opticalcommands.String
|
|
elif reg_type == "RPC":
|
|
reg_type_class = opticalcommands.RPC
|
|
elif reg_type == "status_t":
|
|
reg_type_class = opticalcommands.Status
|
|
elif reg_type == "time_t":
|
|
reg_type_class = opticalcommands.Time
|
|
else:
|
|
raise Exception("Unknown data type {}".format(reg_type))
|
|
self.register_dict[r]['type_class'] = reg_type_class
|
|
|
|
self.o = opticalcommands.OpticalCommands(port, irda=irda)
|
|
|
|
def list_registers(self):
|
|
registers = list(self.register_dict.keys())
|
|
registers.sort()
|
|
logging.info("\n".join(registers))
|
|
|
|
def __getattr__(self, name):
|
|
# use this so we can access registers as if they were
|
|
# members of OpticalApp
|
|
register = self.register_dict[name]
|
|
|
|
return self.o.read_reg(register['id'], register['type_class'])
|
|
|
|
def __setattr__(self, name, value):
|
|
if name != "register_dict" and name in self.register_dict:
|
|
# use this so we can access registers as if they were
|
|
# members of OpticalApp
|
|
register = self.register_dict[name]
|
|
return self.o.write_reg(register['id'],
|
|
register['type_class'], value)
|
|
else:
|
|
super(BreezeOptical, self).__setattr__(name, value)
|
|
|
|
def login(self, level, password=None):
|
|
# logout, this prevents cycling error on PCBSerialNumber when login
|
|
# is called repeatedly
|
|
self.CONFIGEXCHANGE_Privilege = 0
|
|
|
|
if level == 0:
|
|
logging.info("Logout succeeded")
|
|
else:
|
|
if not password:
|
|
# password hasn't been supplied,
|
|
# try to find it in the list of skeleton keys
|
|
|
|
logged_in = False
|
|
retry_count = 1
|
|
while not logged_in and retry_count <= self.LOGIN_ATTEMP_MAX:
|
|
if retry_count > 1:
|
|
logging.warn("login attempt (%d of %d)" % (retry_count, self.LOGIN_ATTEMP_MAX))
|
|
errstr = ""
|
|
try:
|
|
serial_number = self.CONFIGEXCHANGE_PCBSerialNumber
|
|
password = self.keys[serial_number]
|
|
logging.info("No password supplied, "
|
|
"found one in the list of skeleton keys")
|
|
logged_in = True
|
|
break
|
|
|
|
except KeyError:
|
|
errstr = "KeyError"
|
|
|
|
except opticalexceptions.ReadFailedException as e:
|
|
errstr = str(e)
|
|
|
|
except Exception as e:
|
|
errstr = str(e)
|
|
|
|
retry_count += 1
|
|
logging.warn("Error Logging into device (%s)" % errstr)
|
|
self.CONFIGEXCHANGE_Privilege = 0
|
|
|
|
# couldn't log in, short delay to allow meter to
|
|
# finish any tasks.
|
|
time.sleep(0.2)
|
|
|
|
if not logged_in:
|
|
logging.warn("login failed - raising exception")
|
|
raise opticalexceptions.LoginException()
|
|
|
|
self.CONFIGEXCHANGE_Privilege = level
|
|
self.CONFIGEXCHANGE_Password = password
|
|
# Check whether we successfully logged in
|
|
achieved_privilege = self.CONFIGEXCHANGE_Privilege
|
|
|
|
if achieved_privilege != level:
|
|
# Exception("Login Failed (check password)")
|
|
logging.warn("opticalapp.login failed. achieved_privilege %d, requested privilege: %d" % (achieved_privilege, level))
|
|
raise opticalexceptions.LoginException()
|
|
|
|
logging.info("Login succeeded")
|
|
|
|
def login_pcb(self, level, pcbsn):
|
|
"""
|
|
Login using a PCB serial number
|
|
Assume the skeleton key is in the list in CVS
|
|
(true for alpha2 and 2.1 boards)
|
|
"""
|
|
self.login(level, self.keys[pcbsn])
|
|
|
|
def changeBaudRate(self, newbaudrate):
|
|
# First query the capabilities
|
|
baud_cap, packet_cap, version = self.o.query_caps()
|
|
# Now enter training mode with desired baud rate
|
|
self.o.train(0x76, 5000, newbaudrate)
|
|
# After training baud goes back to base
|
|
self.o.BAUD_RATE = 9600
|
|
# Now attempt to change baud rate
|
|
try:
|
|
self.o.set_caps(newbaudrate, packet_cap)
|
|
self.o.BAUD_RATE = newbaudrate
|
|
except opticalexceptions.TrainingFailedException:
|
|
logging.info("Training failed, baud rate is now 9600")
|
|
raise opticalexceptions.BaudRateException
|
|
|
|
def fopen(self, filename, mode):
|
|
# make sure Null terminators go in
|
|
fn = filename.encode("utf=8")
|
|
if fn[-1] != '\0':
|
|
fn += '\0'
|
|
md = mode.encode("utf=8")
|
|
if md[-1] != '\0':
|
|
md += '\0'
|
|
# make sure each string is padded to 32bit words
|
|
padlen = 4 - len(fn) % 4
|
|
padlen = 0 if padlen == 4 else padlen
|
|
fn += '\0' * padlen
|
|
# concatenate the strings
|
|
rpc = fn + md
|
|
# write RPC to the FOpen register
|
|
self.CONFIGEXCHANGE_FOpen = (rpc)
|
|
# read FOpen register for the file handler and return it
|
|
return self.CONFIGEXCHANGE_FOpen
|
|
|
|
def fclose(self, handler):
|
|
# convert handler to a 16bit packed value
|
|
h = struct.pack('<I', handler)
|
|
# write the packed handler value to the FClose register
|
|
self.CONFIGEXCHANGE_FClose = h
|
|
# read the result from FClose register and return it
|
|
return self.CONFIGEXCHANGE_FClose
|
|
|
|
def fwrite(self, data, size, count, handle):
|
|
# pack the integer params
|
|
rpc = struct.pack('<III', size, count, handle)
|
|
# append the data to be written and then write to the FWrite register
|
|
self.CONFIGEXCHANGE_FWrite = rpc + data
|
|
# read FWrite for the number of elements written and return it
|
|
return self.CONFIGEXCHANGE_FWrite
|
|
|
|
def fread(self, size, count, handle):
|
|
total_read = 0
|
|
# pack our request into RPC params
|
|
rpc = struct.pack('<III', size, count, handle)
|
|
self.CONFIGEXCHANGE_FRead = rpc
|
|
# first read back the data length being returned
|
|
count_ret = self.CONFIGEXCHANGE_FRead
|
|
# read back data
|
|
data = ''
|
|
while (count_ret != 0): # The last word will be zero'd
|
|
register = self.register_dict['CONFIGEXCHANGE_FRead']
|
|
# calculate the length of the read back in words
|
|
reads = math.ceil((size*count_ret)/4.0)
|
|
data += self.o.read_reg_multiple(register['id'], reads)
|
|
total_read += count_ret
|
|
# Check if more data is comming
|
|
count_ret = self.CONFIGEXCHANGE_FRead
|
|
# return the number actually read and data
|
|
return total_read, data[:int(total_read * size)]
|
|
|
|
def ftell(self, handle):
|
|
# pack the handle into rpc and write to FTell register
|
|
self.CONFIGEXCHANGE_FTell = struct.pack('<I', handle)
|
|
# read and return the FTell register
|
|
return self.CONFIGEXCHANGE_FTell
|
|
|
|
def fseek(self, handle, offset, whence):
|
|
# pack the integer params into rpc
|
|
rpc = struct.pack('<Iii', handle, offset, whence)
|
|
# write rpc to FSeek register
|
|
self.CONFIGEXCHANGE_FSeek = rpc
|
|
# read and return the return code
|
|
return self.CONFIGEXCHANGE_FSeek
|
|
|
|
def fremove(self, filename):
|
|
# make sure Null terminators go in
|
|
rpc = filename.encode("utf=8")
|
|
if rpc[-1] != '\0':
|
|
rpc += '\0'
|
|
# write RPC to the Remove register
|
|
self.CONFIGEXCHANGE_Remove = rpc
|
|
# read FOpen register for the file handler and return it
|
|
return self.CONFIGEXCHANGE_Remove
|
|
|
|
def fflush(self, handle):
|
|
# pack the handle into rpc and write to FFlush register
|
|
self.CONFIGEXCHANGE_FFlush = struct.pack('<I', handle)
|
|
# read and return FFlush register
|
|
return self.CONFIGEXCHANGE_FFlush
|
|
|
|
def catalogue(self, wildcard, key=0):
|
|
# make sure Null terminators go in
|
|
wc = wildcard.encode("utf=8")
|
|
if wc[-1] != '\0':
|
|
wc += '\0'
|
|
ky = struct.pack('<I', key)
|
|
# make sure each string is padded to 32bit words
|
|
padlen = 4 - len(wc) % 4
|
|
padlen = 0 if padlen == 4 else padlen
|
|
wc += '\0' * padlen
|
|
# concatenate the strings
|
|
rpc = wc + ky
|
|
# write RPC to the FOpen register
|
|
self.CONFIGEXCHANGE_Catalogue = (rpc)
|
|
# read until a NULL terminator is encountered
|
|
data = ""
|
|
ret_key = None
|
|
while '\0' not in data:
|
|
data += struct.pack('<I', self.CONFIGEXCHANGE_Catalogue)
|
|
if data:
|
|
ret_key = self.CONFIGEXCHANGE_Catalogue
|
|
# garbage may be appended to the data string, lopp it off!
|
|
return data.split('\0')[0], ret_key
|
|
|
|
def feof(self, handle):
|
|
self.CONFIGEXCHANGE_FEOF = struct.pack('<I', handle)
|
|
return self.CONFIGEXCHANGE_FEOF
|
|
|
|
|
|
if __name__ == '__main__':
|
|
import logging
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description="Poke a wired Genesis"
|
|
" 'optical' port")
|
|
parser.add_argument('serialport', help='what serial port to use')
|
|
|
|
args = parser.parse_args()
|
|
|
|
# Open the serial port
|
|
logging.basicConfig(level=logging.DEBUG)
|
|
o = BreezeOptical(args.serialport, irda=True)
|
|
#o.list_registers()
|
|
|
|
|
|
o.login(8)
|
|
print("Pressure - {}".format(o.FUNCTEST_Pressure))
|
|
print("Monotonic seconds: {}".format(o.SYSTEM_MonotonicSeconds))
|
|
print("Serial number: {}".format(o.SYSTEM_PCBSerialNumber))
|
|
|
|
# o.FUNCTEST_Iloop = 25000
|
|
# print(o.FUNCTEST_Iloop)
|
|
|
|
print("Genesis flow sample rate = {}".format(o.GENESISFLOW_SampleRate))
|
|
print("Calendar seconds: {}".format(o.SYSTEM_CalendarSeconds))
|
|
print("Privilege {}".format(o.CONFIGEXCHANGE_Privilege))
|
|
o.SYSTEM_DriveCapacity = 0 # select drive
|
|
print("Drive Capacity: {}".format(o.SYSTEM_DriveCapacity))
|
|
|
|
# open a new file called test on drive 0 in write mode
|
|
handle = o.fopen("0\\test", "w")
|
|
if not handle:
|
|
print("Failed to open file")
|
|
|
|
# write soem data to said file
|
|
data = """This is a test file
|
|
Some data, blah, blah, blah,
|
|
filling some space up with absolute rubbish!!!,
|
|
woop woop!!!!!!!"""
|
|
print("data len: {}".format(len(data)))
|
|
write_ret = o.fwrite(data, 1, len(data), handle)
|
|
if not write_ret:
|
|
print("Failed to write file")
|
|
|
|
# close said file
|
|
close_ret = o.fclose(handle)
|
|
if close_ret:
|
|
print("Failed to close file")
|
|
|
|
# Look for a file with name starting with t on drive 0
|
|
name, key = o.catalogue("0\\t*")
|
|
|
|
# open said file in read mode
|
|
handle = o.fopen("0\\" + name, "r")
|
|
if not handle:
|
|
print("Failed to open file")
|
|
|
|
# read 32 bytes from said file
|
|
read_code, read_data = o.fread(1, 32, handle)
|
|
|
|
print("read code: {}".format(read_code))
|
|
print("read data: {}".format(read_data))
|
|
|
|
# move file pointer forwards 5 bytes
|
|
print("fseek code: {}".format(o.fseek(handle, 5, BreezeOptical.SEEK_CUR)))
|
|
|
|
# print the file pointer position
|
|
print("FTell: {}".format(o.ftell(handle)))
|
|
|
|
# read another 32 bytes
|
|
read_num, read_data = o.fread(1, 32, handle)
|
|
|
|
print("read number: {}".format(read_num))
|
|
print("read data: {}".format(read_data))
|
|
|
|
# Check if we have read past the end of the file
|
|
if o.feof(handle):
|
|
print("Read past end of file")
|
|
|
|
# close said file
|
|
close_ret = o.fclose(handle)
|
|
if close_ret:
|
|
print("Failed to close file")
|
|
|
|
# delete said file
|
|
remove_ret = o.remove("0\\" + name)
|
|
if remove_ret:
|
|
print("Failed to remove file")
|
|
|
|
# print the directory of drive 0
|
|
key = 0 # start with key 0
|
|
prev_key = None
|
|
directory = []
|
|
while key != prev_key: # keep looping until the keys match
|
|
prev_key = key
|
|
name, key = o.catalogue("0\\*", key)
|
|
# if the keys dont match it must be a new filename
|
|
if key != prev_key:
|
|
directory.append("0\\" + name)
|
|
print(directory)
|