This commit is contained in:
2025-05-13 01:34:53 +03:00
parent 427735e23d
commit 83f3f1c7d4
945 changed files with 633484 additions and 0 deletions
@@ -0,0 +1,6 @@
The Python dfuse tool was written by Paul Liétar.
Minor modifications were made for this project.
Original source: https://github.com/plietar/dfuse-tool
The license for this module is unclear.
@@ -0,0 +1,219 @@
import usb.util
import time
import fractions
import array
from odrive.dfuse.DfuState import DfuState
DFU_REQUEST_SEND = 0x21
DFU_REQUEST_RECEIVE = 0xa1
DFU_DETACH = 0x00
DFU_DNLOAD = 0x01
DFU_UPLOAD = 0x02
DFU_GETSTATUS = 0x03
DFU_CLRSTATUS = 0x04
DFU_GETSTATE = 0x05
DFU_ABORT = 0x06
SIZE_MULTIPLIERS = {' ': 1, 'K': 1024, 'M' : 1024*1024}
MAX_TRANSFER_SIZE = 2048
# Order is LSB first
def address_to_4bytes(a):
return [ a % 256, (a >> 8)%256, (a >> 16)%256, (a >> 24)%256 ]
class DfuDevice:
def __init__(self, device, timeout = None):
self.dev = device
self.timeout = timeout
self.cfg = self.dev[0]
self.intf = None
#self.dev.reset()
self.cfg.set()
self.sectors = list(self.get_device_sectors())
def alternates(self):
return [(usb.util.get_string(self.dev, intf.iInterface), intf) for intf in self.cfg]
def set_alternate(self, intf):
if isinstance(intf, tuple):
self.intf = intf[1]
else:
self.intf = intf
self.intf.set_altsetting()
def control_msg(self, requestType, request, value, buffer, timeout=None):
return self.dev.ctrl_transfer(requestType, request, value, self.intf.bInterfaceNumber, buffer, timeout=timeout)
def detach(self, timeout):
return self.control_msg(DFU_REQUEST_SEND, DFU_DETACH, timeout, None)
def dnload(self, blockNum, data):
cnt = self.control_msg(DFU_REQUEST_SEND, DFU_DNLOAD, blockNum, list(data))
return cnt
def upload(self, blockNum, size):
return self.control_msg(DFU_REQUEST_RECEIVE, DFU_UPLOAD, blockNum, size)
def get_status(self, timeout=None):
status = self.control_msg(DFU_REQUEST_RECEIVE, DFU_GETSTATUS, 0, 6, timeout=timeout)
return (status[0], status[4], status[1] + (status[2] << 8) + (status[3] << 16), status[5])
def clear_status(self):
self.control_msg(DFU_REQUEST_SEND, DFU_CLRSTATUS, 0, None)
def get_state(self):
return self.control_msg(DFU_REQUEST_RECEIVE, DFU_GETSTATE, 0, 1)[0]
def abort(self):
self.control_msg(DFU_REQUEST_RECEIVE, DFU_ABORT, 0, 0)
def set_address(self, ap):
return self.dnload(0x0, [0x21] + address_to_4bytes(ap))
def write(self, block, data):
return self.dnload(block + 2, data)
def read(self, block, size):
return self.upload(block + 2, size)
def erase(self, pa):
return self.dnload(0x0, [0x41] + address_to_4bytes(pa))
def leave(self):
return self.dnload(0x0, []) # Just send an empty data.
def wait_while_state(self, state, timeout=None):
if not isinstance(state, (list, tuple)):
states = (state,)
else:
states = state
try:
status = self.get_status()
except:
time.sleep(0.100)
status = self.get_status()
while (status[1] in states):
claimed_timeout = status[2]
actual_timeout = int(max(timeout or 0, claimed_timeout))
#print("timeout = %f, claimed = %f" % (timeout, status[2]))
#time.sleep(timeout)
status = self.get_status(timeout=actual_timeout)
return status
## High level functions ##
# by ODrive Robotics
def get_device_sectors(self):
"""
Returns a list of all sectors on the device.
Each sector is represented as a dictionary with the following keys:
- name: name of the associated memory region (e.g. "Internal Flash")
- alt: USB alternate setting associated with this memory region
- addr: Start address of the sector (e.g. 0x08004000 for the second flash sectors)
- baseaddr: Start address of the memory region associated with the sector
(e.g. 0x08000000 for all flash sectors)
- len: Number of bytes in the sector
"""
for name, alt in self.alternates():
# example for name:
# '@Internal Flash /0x08000000/04*016Kg,01*064Kg,07*128Kg'
label, baseaddr, layout = name.split('/')
baseaddr = int(baseaddr, 0) # convert hex to decimal
addr = baseaddr
for sector in layout.split(','):
repeat, size = map(int, sector[:-2].split('*'))
size *= SIZE_MULTIPLIERS[sector[-2].upper()]
mode = sector[-1]
while repeat > 0:
# TODO: verify if the section is writable
yield {
'name': label.strip().strip('@'),
'alt': alt,
'baseaddr': baseaddr,
'addr': addr,
'len': size,
'mode': mode
}
addr += size
repeat -= 1
def set_alternate_safe(self, alt):
self.set_alternate(alt)
if self.get_state() == DfuState.DFU_ERROR:
self.clear_status()
self.wait_while_state(DfuState.DFU_ERROR)
#def clear_error(self)
def set_address_safe(self, addr):
self.set_address(addr)
status = self.wait_while_state(DfuState.DFU_DOWNLOAD_BUSY)
if status[1] != DfuState.DFU_DOWNLOAD_IDLE:
raise RuntimeError("An error occured. Device Status: {!r}".format(status))
# take device out of DFU_DOWNLOAD_SYNC and into DFU_IDLE
self.abort()
status = self.wait_while_state(DfuState.DFU_DOWNLOAD_SYNC)
if status[1] != DfuState.DFU_IDLE:
raise RuntimeError("An error occured. Device Status: {!r}".format(status))
def erase_sector(self, sector):
self.set_alternate_safe(sector['alt'])
self.erase(sector['addr'])
status = self.wait_while_state(DfuState.DFU_DOWNLOAD_BUSY, timeout=sector['len']/32)
if status[1] != DfuState.DFU_DOWNLOAD_IDLE:
raise RuntimeError("An error occured. Device Status: {!r}".format(status))
def write_sector(self, sector, data):
self.set_alternate_safe(sector['alt'])
self.set_address_safe(sector['addr'])
transfer_size = fractions.gcd(sector['len'], MAX_TRANSFER_SIZE)
blocks = [data[i:i + transfer_size] for i in range(0, len(data), transfer_size)]
for blocknum, block in enumerate(blocks):
#print('write to {:08X} ({} bytes)'.format(
# sector['addr'] + blocknum * TRANSFER_SIZE, len(block)))
self.write(blocknum, block)
status = self.wait_while_state(DfuState.DFU_DOWNLOAD_BUSY)
if status[1] != DfuState.DFU_DOWNLOAD_IDLE:
raise RuntimeError("An error occured. Device Status: {!r}".format(status))
def read_sector(self, sector):
"""
Reads data from the specified sector
Returns: a byte array containing the data
"""
self.set_alternate_safe(sector['alt'])
self.set_address_safe(sector['addr'])
transfer_size = fractions.gcd(sector['len'], MAX_TRANSFER_SIZE)
#blocknum_offset = int((sector['addr'] - sector['baseaddr']) / transfer_size)
data = array.array(u'B')
for blocknum in range(int(sector['len'] / transfer_size)):
#print('read at {:08X}'.format(sector['addr'] + blocknum * TRANSFER_SIZE))
deviceBlock = self.read(blocknum, transfer_size)
data.extend(deviceBlock)
self.abort() # take device into DFU_IDLE
return data
def jump_to_application(self, address):
self.set_address_safe(address)
#self.set_address(address)
#status = self.wait_while_state(DfuState.DFU_DOWNLOAD_BUSY)
#if status[1] != DfuState.DFU_DOWNLOAD_IDLE:
# raise RuntimeError("An error occured. Device Status: {}".format(status[1]))
self.leave()
status = self.wait_while_state(DfuState.DFU_MANIFEST_SYNC)
if status[1] != DfuState.DFU_MANIFEST:
raise RuntimeError("An error occured. Device Status: {}".format(status[1]))
@@ -0,0 +1,66 @@
import argparse
import sys
import struct
import binascii
def named(tuple,names):
return dict(zip(names,tuple))
def parse(fmt,data,names):
return named(struct.unpack(fmt,data),names)
def fileunpack(f, fmt, names):
n = struct.calcsize(fmt)
return parse(fmt, f.read(n), names)
class DfuFile:
def __init__(self, path):
self.targets = list()
self.devInfo = dict()
try:
dfufile = open(path, 'rb')
except:
raise argparse.ArgumentTypeError('Could not open file %r' % path)
with dfufile:
header = fileunpack(dfufile, "<5sBLB", ('signature', 'version', 'size', 'targets'))
if header['signature'] != b'DfuSe':
raise argparse.ArgumentTypeError('File signature does not match')
if header['version'] != 1:
raise argparse.ArgumentTypeError('Unsupport DfuSe file version')
for t in range(header['targets']):
target_prefix = fileunpack(dfufile, "<6sBL255sLL", ('signature', 'alternate', 'named', 'name', 'size', 'elements'))
if target_prefix['signature'] != b'Target':
raise argparse.ArgumentTypeError('Target signature does not match')
target = {
'name': target_prefix['name'].decode('ascii').rstrip('\0'),
'alternate': target_prefix['alternate'],
'elements': list()
}
for e in range(target_prefix['elements']):
element_prefix = fileunpack(dfufile,"<LL", ('address', 'size'))
element = {
'address': element_prefix['address'],
'data': dfufile.read(element_prefix['size'])
}
target['elements'].append(element)
self.targets.append(target)
suffix = fileunpack(dfufile, "<HHHH3sBL", ('fwVersion', 'pid', 'vid', 'dfuSpec', 'signature', 'length', 'crc'))
if suffix['signature'] != b'UFD':
raise argparse.ArgumentTypeError('File\'s suffix signature does not match')
self.devInfo = dict(suffix)
del(self.devInfo['signature'])
del(self.devInfo['length'])
del(self.devInfo['crc'])
@@ -0,0 +1,13 @@
class DfuState():
APP_IDLE = 0x00
APP_DETACH = 0x01
DFU_IDLE = 0x02
DFU_DOWNLOAD_SYNC = 0x03
DFU_DOWNLOAD_BUSY = 0x04
DFU_DOWNLOAD_IDLE = 0x05
DFU_MANIFEST_SYNC = 0x06
DFU_MANIFEST = 0x07
DFU_MANIFEST_WAIT_RESET = 0x08
DFU_UPLOAD_IDLE = 0x09
DFU_ERROR = 0x0a
@@ -0,0 +1,18 @@
class DfuStatus:
OK = 0x00
ERROR_TARGET = 0x01
ERROR_FILE = 0x02
ERROR_WRITE = 0x03
ERROR_ERASE = 0x04
ERROR_CHECK_ERASED = 0x05
ERROR_PROG = 0x06
ERROR_VERIFY = 0x07
ERROR_ADDRESS = 0x08
ERROR_NOTDONE = 0x09
ERROR_FIRMWARE = 0x0a
ERROR_VENDOR = 0x0b
ERROR_USBR = 0x0c
ERROR_POR = 0x0d
ERROR_UNKNOWN = 0x0e
ERROR_STALLEDPKT = 0x0f
@@ -0,0 +1,4 @@
from .DfuDevice import DfuDevice
from .DfuStatus import DfuStatus
from .DfuState import DfuState
from .DfuFile import DfuFile