260 lines
9.2 KiB
Python
260 lines
9.2 KiB
Python
from __future__ import print_function
|
|
import opticalinterface.opticalapp as oa
|
|
import opticalinterface.opticalexceptions as opticalexceptions
|
|
import struct
|
|
import time
|
|
|
|
class BinGen(object):
|
|
def __init__(self, hexfile):
|
|
import subprocess
|
|
import re
|
|
import crc16
|
|
import os
|
|
|
|
# convert hex file to binary data. Go via binary file to avoid shell expanding line endings
|
|
binary = subprocess.check_output(["srec_cat.exe", hexfile, "-Intel", "-o", "mybin.bin", "-Binary"], shell=True)
|
|
with open('mybin.bin', 'rb') as f:
|
|
binary = f.read()
|
|
os.remove("mybin.bin")
|
|
|
|
# split whole binary in the apps
|
|
apps = re.split(r'APP>', binary)
|
|
|
|
# keep trying until we're out of apps
|
|
while len(apps):
|
|
try:
|
|
# check we haven't overdone the splitting
|
|
appno = 0
|
|
good_apps = list()
|
|
while appno < len(apps):
|
|
# pop the APP> back on
|
|
a = 'APP>' + apps[appno]
|
|
appno += 1
|
|
|
|
# get the application length
|
|
header_format = "<IHHHBBIIIIIIIBBBB"
|
|
sig, crc, version, _, _, _, stack, rw_len, ro_len, zi_len, fn_table, ev_table, flags, appid, fn_table_len, _, _ = struct.unpack_from(header_format, a)
|
|
applen = rw_len + ro_len
|
|
|
|
# join bits back together until it is long enough
|
|
while (len(a) < applen):
|
|
a += 'APP>'
|
|
a += apps[appno]
|
|
appno += 1
|
|
|
|
# trim the padding
|
|
a = a[:applen]
|
|
|
|
# now check the crc
|
|
calc_crc = crc16.crc16xmodem(a[struct.calcsize("<IH"):], 0xffff)
|
|
if (calc_crc != crc):
|
|
raise Exception("Unpacking hex file failed, app CRC doesn't match")
|
|
good_apps.append(a)
|
|
# finished, exit the while loop nastily
|
|
apps = list()
|
|
|
|
except Exception as e:
|
|
# Something went wrong. The beginning of the hex file has non-applications in it
|
|
if len(good_apps):
|
|
# we're not at the start, barf
|
|
raise(e)
|
|
# ditch the first chunk and try again
|
|
apps = apps[1:]
|
|
|
|
self.filelist = list()
|
|
for i, a in enumerate(good_apps):
|
|
filename = 'mybins{}.bin'.format(i)
|
|
with open(filename, 'wb') as f:
|
|
f.write(a)
|
|
self.filelist.append(filename)
|
|
|
|
def binfiles(self):
|
|
return self.filelist
|
|
|
|
def tidy(self):
|
|
for file in self.filelist:
|
|
os.remove(file)
|
|
|
|
|
|
|
|
class Modify(object):
|
|
def __init__(self, appid, function):
|
|
self.function = function
|
|
self.appid = appid
|
|
|
|
def download(self, comms):
|
|
# move along, nothing to see here
|
|
pass
|
|
|
|
class Upgrade(Modify):
|
|
def __init__(self, filename):
|
|
self.filename = filename
|
|
appid, self.version = self.parsebinary(self.filename)
|
|
super(Upgrade, self).__init__(appid, 'upgrade')
|
|
|
|
@staticmethod
|
|
def parsebinary(filename):
|
|
with open(filename, 'rb') as f:
|
|
header_format = "<IHHHBBIIIIIIIBBBB"
|
|
header = f.read(struct.calcsize(header_format))
|
|
sig, crc, version, _, _, _, stack, rw_len, ro_len, zi_len, fn_table, ev_table, flags, appid, fn_table_len, _, _ = struct.unpack(header_format, header)
|
|
if (sig != struct.unpack("<I", "APP>")[0]):
|
|
raise Exception("Header format for file {} not valid".format(filename))
|
|
|
|
return appid, version
|
|
|
|
def download(self, comms):
|
|
chunksize = 100
|
|
with open(self.filename, 'rb') as f:
|
|
meter_file = "1\\upg{:02X}".format(self.appid)
|
|
mf = comms.fopen(meter_file, 'w')
|
|
if not mf:
|
|
raise Exception("upgrade application file {:02X} failed to open".format(self.appid))
|
|
written = 0
|
|
while (1):
|
|
# work in chunks
|
|
chunk = f.read(chunksize)
|
|
if chunk == '':
|
|
# end of file
|
|
break
|
|
comms.fwrite(chunk, 1, len(chunk), mf)
|
|
written += len(chunk)
|
|
print('{} bytes written\r'.format(written), end='')
|
|
comms.fclose(mf)
|
|
print('{} bytes written\r'.format(written))
|
|
print('Complete!')
|
|
|
|
def __repr__(self):
|
|
return "{:02X}:".format(self.appid)
|
|
|
|
class Remove(Modify):
|
|
def __init__(self, appid):
|
|
super(Remove, self).__init__(appid, 'remove')
|
|
|
|
def __repr__(self):
|
|
return "{:02X}*".format(self.appid)
|
|
|
|
def write_config(comms, upgrades):
|
|
mf = comms.fopen("1\\upgrade", "w")
|
|
if not mf:
|
|
raise Exception("upgrade file failed to open")
|
|
for u in upgrades:
|
|
comms.fwrite(repr(u), 3, 1, mf)
|
|
comms.fwrite("\r", 1, 1, mf)
|
|
comms.fclose(mf)
|
|
|
|
if __name__ == '__main__':
|
|
import logging
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description="Perform a firmware update")
|
|
parser.add_argument('serialport', help='what serial port to use')
|
|
parser.add_argument('binaries', nargs='+', help='list of application binaries to upgrade')
|
|
parser.add_argument('--irda', action='store_true', default=False, help='use IrDA rather than wired UART')
|
|
parser.add_argument('--fullupgrade', action='store_true', default=False, help='remove all existing applications and replace with binaries provided rather than just adding/upgrading the applications provided')
|
|
parser.add_argument('--hex', action='store_true', default=False, help='applications supplied as a single hex file, assumes --fullupgrade (note this requires srec_cat utility to be available on the path)')
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.hex:
|
|
if len(args.binaries) > 1:
|
|
raise Exception("Expecting just one hex file")
|
|
|
|
bingen = BinGen(args.binaries[0])
|
|
args.binaries = bingen.binfiles()
|
|
args.fullupgrade = True
|
|
|
|
# Open the serial port
|
|
#logging.basicConfig(level=logging.DEBUG)
|
|
o = oa.BreezeOptical(args.serialport, irda=args.irda)
|
|
|
|
# Get the existing application list
|
|
before_apps = dict()
|
|
o.login(8)
|
|
print("Applications already on the meter")
|
|
MAX_APPLICATION_ID = 30 #todo this is a bit of a bodge
|
|
for appid in range(MAX_APPLICATION_ID):
|
|
o.SYSTEM_CheckPresence = struct.pack("<I", appid)
|
|
try:
|
|
v = o.SYSTEM_CheckPresence
|
|
print("Application {:02X}, version {:04X}".format(appid, v))
|
|
before_apps["{}".format(appid)] = v
|
|
except opticalexceptions.ReadFailedException:
|
|
# app not present but that's ok
|
|
pass
|
|
# and now the core revision
|
|
print ("Core: {}".format(o.SYSTEM_CoreRevision))
|
|
print("")
|
|
|
|
# Loop through the binaries and find their details
|
|
upgrades = list()
|
|
dontbother = list()
|
|
print("Applications to upgrade")
|
|
for new_app in args.binaries:
|
|
u = Upgrade(new_app)
|
|
if ("{}".format(u.appid) not in before_apps) or (u.version != before_apps["{}".format(u.appid)]):
|
|
upgrades.append(u)
|
|
print("Application {:02X}, version {:04X}".format(u.appid, u.version))
|
|
else:
|
|
dontbother.append(u)
|
|
print("Skipping application {:02X}, this version is already there".format(u.appid))
|
|
print("")
|
|
|
|
# If this ia a full upgrade then work out what to remove
|
|
if args.fullupgrade:
|
|
print("Applications to remove")
|
|
all_binaries = [x.appid for x in upgrades+dontbother]
|
|
for app in before_apps:
|
|
appid = int(app)
|
|
if appid not in all_binaries:
|
|
upgrades.append(Remove(appid))
|
|
print("Application {:02X}".format(appid))
|
|
print("")
|
|
|
|
# Write config file to the meter
|
|
start = time.time()
|
|
print("Writing the config file")
|
|
write_config(o, upgrades)
|
|
|
|
# Write upgrades to the meter
|
|
print("Writing the application binaries")
|
|
for u in upgrades:
|
|
print("Write binary for application {:02X}".format(u.appid))
|
|
u.download(o)
|
|
|
|
# don't need any intermediate binaries any more
|
|
if args.hex:
|
|
bingen.tidy()
|
|
|
|
# Trigger upgrade
|
|
print("Triggering the upgrade")
|
|
try:
|
|
o.SYSTEM_TriggerUpgrade = 1
|
|
except Exception as e:
|
|
print("Upgrade rejected backtrace may provide a useful error code")
|
|
raise(e)
|
|
end = time.time()
|
|
|
|
print("Upgrade process took {:.0f} seconds".format(end - start))
|
|
|
|
# Now read and print the application list again
|
|
print("Now waiting for reboot and logging back in")
|
|
for i in range(15):
|
|
try:
|
|
o.login(8)
|
|
except:
|
|
# didn't work, wait a second and try again
|
|
time.sleep(1)
|
|
|
|
print("Final applications on the meter")
|
|
for appid in range(MAX_APPLICATION_ID):
|
|
o.SYSTEM_CheckPresence = struct.pack("<I", appid)
|
|
try:
|
|
v = o.SYSTEM_CheckPresence
|
|
print("Application {:02X}, version {:04X}".format(appid, v))
|
|
except opticalexceptions.ReadFailedException:
|
|
# app not present but that's ok
|
|
pass
|
|
print ("Core: {}".format(o.SYSTEM_CoreRevision))
|
|
print("")
|