diff --git a/Firmware/CHANGELOG.md b/Firmware/CHANGELOG.md index ce21cf1e..2f243136 100644 --- a/Firmware/CHANGELOG.md +++ b/Firmware/CHANGELOG.md @@ -11,7 +11,10 @@ Please add a note of your changes below this heading if you make a Pull Request. * System stats (e.g. stack usage) are exposed under `.system_stats` ### Changed -* The DFU script now verifies the flash after writing +* DFU script updates + * Verify the flash after writing + * Automatically download firmware from GitHub releases if no file is provided + * Retain configuration during firmware updates * Refactor python tools * The scripts `explore_odrive.py`, `liveplotter.py`, `drv_status.py` and `rate_test.py` have been merged into one single `odrivetool` script. Running this script without any arguments provides the shell that `explore_odrive.py` used to provide. * The command line options of `odrivetool` have changed compared to the original `explore_odrive.py`. See `odrivetool --help` for more details. @@ -19,6 +22,7 @@ Please add a note of your changes below this heading if you make a Pull Request. * No need to restart the `odrivetool` shell when devices get disconnected and reconnected * ODrive accesses from within python tools are now thread-safe. That means you can read from the same remote property from multiple threads concurrently. * The liveplotter (`odrivetool liveplotter`, formerly `liveplotter.py`) does no longer steal focus and closes as expected + * Add commands `odrivetool backup-config` and `odrivetool restore-config` * (experimental: start liveplotter from `odrivetool` shell by typing `start_liveplotter(lambda: odrv0.motor0.encoder.encoder_state)`) * `make write_otp` command to burn the board version onto the ODrive's one-time programmable memory. If you have an ODrive v3.4 or older, you can run this once for a better firmware update user experience in the future. Run the command without any options for more details. Once set, the board version is exposed through the `hw_version_[...]` properties. * bake Git-derived firmware version into firmware binary. The firmware version is exposed through the `fw_version_[...]` properties. diff --git a/Firmware/MotorControl/main.cpp b/Firmware/MotorControl/main.cpp index ea3c155c..80013331 100644 --- a/Firmware/MotorControl/main.cpp +++ b/Firmware/MotorControl/main.cpp @@ -33,6 +33,8 @@ void save_configuration(void) { &motor_configs, &axis_configs)) { //printf("saving configuration failed\r\n"); osDelay(5); + } else { + user_config_loaded_ = true; } } diff --git a/tools/odrive/configuration.py b/tools/odrive/configuration.py new file mode 100644 index 00000000..8f297249 --- /dev/null +++ b/tools/odrive/configuration.py @@ -0,0 +1,82 @@ + +import json +import os +import tempfile +import odrive.remote_object +from odrive.utils import OperationAbortedException + +def get_dict(obj, is_config_object): + result = {} + for (k,v) in obj._remote_attributes.items(): + if isinstance(v, odrive.remote_object.RemoteProperty) and is_config_object: + result[k] = v.get_value() + elif isinstance(v, odrive.remote_object.RemoteObject): + sub_dict = get_dict(v, k == 'config') + if sub_dict != {}: + result[k] = sub_dict + return result + +def set_dict(obj, path, config_dict): + errors = [] + for (k,v) in config_dict.items(): + name = path + ("." if path != "" else "") + k + if not k in obj._remote_attributes: + errors.append("Could not restore {}: property not found on device".format(name)) + continue + remote_attribute = obj._remote_attributes[k] + if isinstance(remote_attribute, odrive.remote_object.RemoteObject): + errors += set_dict(remote_attribute, name, v) + else: + try: + remote_attribute.set_value(v) + except Exception as ex: + errors.append("Could not restore {}: {}".format(name, str(ex))) + return errors + +def get_temp_config_filename(device): + serial_number = odrive.utils.get_serial_number_str(device) + safe_serial_number = ''.join(filter(str.isalnum, serial_number)) + return os.path.join(tempfile.gettempdir(), 'odrive-config-{}.json'.format(safe_serial_number)) + +def backup_config(device, filename, logger): + """ + Exports the configuration of an ODrive to a JSON file. + If no file name is provided, the file is placed into a + temporary directory. + """ + + if filename is None: + filename = get_temp_config_filename(device) + + logger.info("Saving configuration to {}...".format(filename)) + + if os.path.exists(filename): + if not odrive.utils.yes_no_prompt("The file {} already exists. Do you want to override it?".format(filename), True): + raise OperationAbortedException() + + data = get_dict(device, False) + with open(filename, 'w') as file: + json.dump(data, file) + logger.info("Configuration saved.") + +def restore_config(device, filename, logger): + """ + Restores the configuration stored in a file + """ + + if filename is None: + filename = get_temp_config_filename(device) + + with open(filename) as file: + data = json.load(file) + + logger.info("Restoring configuration from {}...".format(filename)) + errors = odrive.configuration.set_dict(device, "", data) + + for error in errors: + logger.info(error) + if errors: + logger.warn("Some of the configuration could not be restored.") + + device.save_configuration() + logger.info("Configuration restored.") diff --git a/tools/odrive/dfu.py b/tools/odrive/dfu.py index 1efd4592..0f7b458d 100755 --- a/tools/odrive/dfu.py +++ b/tools/odrive/dfu.py @@ -9,11 +9,13 @@ import time import threading import platform import struct -import array -import fractions +import requests +import re +import io +import os import usb.core import odrive.discovery -from odrive.utils import Event +from odrive.utils import Event, OperationAbortedException from odrive.dfuse import * try: @@ -24,46 +26,17 @@ except: sys.exit(1) -SIZE_MULTIPLIERS = {' ': 1, 'K': 1024, 'M' : 1024*1024} -MAX_TRANSFER_SIZE = 2048 +def get_fw_version_string(fw_version): + if (fw_version[0], fw_version[1], fw_version[2]) == (0, 0, 0): + return "[unknown version]" + else: + return "v{}.{}.{}{}".format(fw_version[0], fw_version[1], fw_version[2], "-dev" if fw_version[3] else "") - -def get_device_sectors(dfudev): - """ - 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 dfudev.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 get_hw_version_string(hw_version): + if hw_version == (0, 0, 0): + return "[unknown version]" + else: + return "v{}.{}{}".format(hw_version[0], hw_version[1], ("-" + str(hw_version[2]) + "V") if hw_version[2] > 0 else "") def populate_sectors(sectors, hexfile): """ @@ -88,66 +61,6 @@ def populate_sectors(sectors, hexfile): # TODO: verify if the section is writable yield (sector, hexfile.tobinarray(addr, addr + size - 1)) -def set_alternate_safe(dfudev, alt): - dfudev.set_alternate(alt) - if dfudev.get_state() == DfuState.DFU_ERROR: - dfudev.clear_status() - dfudev.wait_while_state(DfuState.DFU_ERROR) - -#def clear_error(dfudev) -def set_address_safe(dfudev, addr): - dfudev.set_address(addr) - status = dfudev.wait_while_state(DfuState.DFU_DOWNLOAD_BUSY) - if status[1] != DfuState.DFU_DOWNLOAD_IDLE: - raise RuntimeError("An error occured. Device Status: %r" % status) - # take device out of DFU_DOWNLOAD_SYNC and into DFU_IDLE - dfudev.abort() - status = dfudev.wait_while_state(DfuState.DFU_DOWNLOAD_SYNC) - if status[1] != DfuState.DFU_IDLE: - raise RuntimeError("An error occured. Device Status: %r" % status) - - -def erase(dfudev, sector): - set_alternate_safe(dfudev, sector['alt']) - dfudev.erase(sector['addr']) - status = dfudev.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" % status) - -def flash(dfudev, sector, data): - set_alternate_safe(dfudev, sector['alt']) - set_address_safe(dfudev, 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))) - dfudev.write(blocknum, block) - status = dfudev.wait_while_state(DfuState.DFU_DOWNLOAD_BUSY) - if status[1] != DfuState.DFU_DOWNLOAD_IDLE: - raise RuntimeError("An error occured. Device Status: %r" % status) - -def read(dfudev, sector): - """ - Reads data from the specified sector - Returns: a byte array containing the data - """ - set_alternate_safe(dfudev, sector['alt']) - set_address_safe(dfudev, 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 = dfudev.read(blocknum, transfer_size) - data.extend(deviceBlock) - dfudev.abort() # take device into DFU_IDLE - return data def get_first_mismatch_index(array1, array2): """ @@ -161,42 +74,132 @@ def get_first_mismatch_index(array1, array2): return pos return None - -def jump_to_application(dfudev, address): - set_address_safe(dfudev, address) - #dfudev.set_address(address) - #status = dfudev.wait_while_state(DfuState.DFU_DOWNLOAD_BUSY) - #if status[1] != DfuState.DFU_DOWNLOAD_IDLE: - # raise RuntimeError("An error occured. Device Status: {}".format(status[1])) - - dfudev.leave() - status = dfudev.wait_while_state(DfuState.DFU_MANIFEST_SYNC) - if status[1] != DfuState.DFU_MANIFEST: - raise RuntimeError("An error occured. Device Status: {}".format(status[1])) - - -def dump_otp(): +def dump_otp(dfudev): """ Dumps the contents of the one-time-programmable - memory. The OTP will be used in future versions of - this script to determine the board version. + memory for debugging purposes. + The OTP is used to determine the board version. """ # 512 Byte OTP - otp_sector = [s for s in sectors if s['name'] == 'OTP Memory' and s['addr'] == 0x1fff7800][0] - data = read(dfudev, otp_sector) + otp_sector = [s for s in dfudev.sectors if s['name'] == 'OTP Memory' and s['addr'] == 0x1fff7800][0] + data = dfudev.read_sector(otp_sector) print(' '.join('{:02X}'.format(x) for x in data)) # 16 lock bytes - otp_lock_sector = [s for s in sectors if s['name'] == 'OTP Memory' and s['addr'] == 0x1fff7A00][0] - data = read(dfudev, otp_lock_sector) + otp_lock_sector = [s for s in dfudev.sectors if s['name'] == 'OTP Memory' and s['addr'] == 0x1fff7A00][0] + data = dfudev.read_sector(otp_lock_sector) print(' '.join('{:02X}'.format(x) for x in data)) +class Firmware(): + def __init__(self): + self.fw_version = (0, 0, 0, True) + self.hw_version = (0, 0, 0) + + @staticmethod + def is_newer(a, b): + a_num = (a[0], a[1], a[2]) + b_num = (b[0], b[1], b[2]) + if a_num == (0, 0, 0) or b_num == (0, 0, 0): + return False # Cannot compare unknown versions + return a_num > b_num or (a_num == b_num and not a[3] and b[3]) + + def __gt__(self, other): + """ + Compares two firmware versions. If both versions are equal, the + prerelease version is considered older than the release version. + """ + if not isinstance(other, tuple): + other = other.fw_version + return Firmware.is_newer(self.fw_version, other) + + def __lt__(self, other): + """ + Compares two firmware versions. If both versions are equal, the + prerelease version is considered older than the release version. + """ + if not isinstance(other, tuple): + other = other.fw_version + return Firmware.is_newer(other, self.fw_version) + + def is_compatible(self, hw_version): + """ + Determines if this firmware is compatible + with the specified hardware version + """ + return self.hw_version == hw_version + +class FirmwareFromGithub(Firmware): + """ + Represents a firmware asset + """ + def __init__(self, release_json, asset_json): + Firmware.__init__(self) + if release_json['draft'] or release_json['prerelease']: + release_json['tag_name'] += "*" + self.fw_version = odrive.version.version_str_to_tuple(release_json['tag_name']) + + hw_version_regex = r'.*v([0-9]+).([0-9]+)(-(?P[0-9]+)V)?.hex' + hw_version_match = re.search(hw_version_regex, asset_json['name']) + self.hw_version = (int(hw_version_match[1]), + int(hw_version_match[2]), + int(hw_version_match.groupdict().get('voltage') or 0)) + self.github_asset_id = asset_json['id'] + self.hex = None + # no technical reason to fetch this - just interesting + self.download_count = asset_json['download_count'] + + def get_as_hex(self): + """ + Returns the content of the firmware in as a binary array in Intel Hex format + """ + if self.hex is None: + print("Downloading firmware {}...".format(get_fw_version_string(self.fw_version))) + response = requests.get('https://api.github.com/repos/madcowswe/ODrive/releases/assets/' + str(self.github_asset_id), + headers={'Accept': 'application/octet-stream'}) + if response.status_code != 200: + raise Exception("failed to download firmware") + self.hex = response.content + return io.StringIO(self.hex.decode('utf-8')) + +class FirmwareFromFile(Firmware): + def __init__(self, file): + Firmware.__init__(self) + self._file = file + def get_as_hex(self): + return self._file + +def get_all_github_firmwares(): + response = requests.get('https://api.github.com/repos/madcowswe/ODrive/releases') + if response.status_code != 200: + raise Exception("could not fetch releases") + response_json = response.json() + + for release_json in response_json: + for asset_json in release_json['assets']: + try: + if asset_json['name'].lower().endswith('.hex'): + fw = FirmwareFromGithub(release_json, asset_json) + yield fw + except Exception as ex: + print(ex) + +def get_newest_firmware(hw_version): + """ + Returns the newest available firmware for the specified hardware version + """ + firmwares = get_all_github_firmwares() + firmwares = filter(lambda fw: not fw.fw_version[3], firmwares) # ignore prereleases + firmwares = filter(lambda fw: fw.hw_version == hw_version, firmwares) + firmwares = list(firmwares) + firmwares.sort() + return firmwares[-1] if len(firmwares) else None + def show_deferred_message(message, cancellation_token): """ Shows a message after 10s, unless cancellation_token gets set. """ def show_message_thread(message, cancellation_token): - for i in range(1,10): + for _ in range(1,10): if cancellation_token.is_set(): return time.sleep(1) @@ -206,104 +209,157 @@ def show_deferred_message(message, cancellation_token): t.daemon = True t.start() -def put_odrive_into_dfu_mode(my_drive, cancellation_token): +def put_into_dfu_mode(device, cancellation_token): """ Puts the specified device into DFU mode """ - if not hasattr(my_drive, "enter_dfu_mode"): + if not hasattr(device, "enter_dfu_mode"): print("The firmware on device {} does not support DFU. You need to \n" "flash the firmware once using STLink (`make flash`), after that \n" "DFU with this script should work fine." - .format(my_drive.__channel__.usb_device.serial_number)) + .format(device.__channel__.usb_device.serial_number)) return - hw_version_major = my_drive.hw_version_major if hasattr(my_drive, 'hw_version_major') else 3 - hw_version_minor = my_drive.hw_version_minor if hasattr(my_drive, 'hw_version_minor') else 4 - if hw_version_major == 3 and hw_version_minor >= 5: - print("Putting device {} into DFU mode...".format(my_drive.__channel__.usb_device.serial_number)) - try: - my_drive.enter_dfu_mode() - except odrive.protocol.ChannelBrokenException: - pass # this is expected because the device reboots - if platform.system() == "Windows": - show_deferred_message("Still waiting for the device to reappear.\n" - "Use the Zadig utility to set the driver of 'STM32 BOOTLOADER' to libusb-win32.", - cancellation_token) - else: - print("Found device {}".format(my_drive.__channel__.usb_device.serial_number)) + hw_version_major = device.hw_version_major if hasattr(device, 'hw_version_major') else 3 + hw_version_minor = device.hw_version_minor if hasattr(device, 'hw_version_minor') else 4 + if hw_version_major == 3 and hw_version_minor < 5: print(" DFU mode is not supported on board version 3.4 or earlier.") print(" This is because entering DFU mode on such a device would") print(" break the brake resistor FETs under some circumstances.") + raise Exception("not supported") + + print("Putting device {} into DFU mode...".format(device.__channel__.usb_device.serial_number)) + try: + device.enter_dfu_mode() + except odrive.protocol.ChannelBrokenException: + pass # this is expected because the device reboots + if platform.system() == "Windows": + show_deferred_message("Still waiting for the device to reappear.\n" + "Use the Zadig utility to set the driver of 'STM32 BOOTLOADER' to libusb-win32.", + cancellation_token) -def launch_dfu(args, app_shutdown_token): +def find_device_in_dfu_mode(serial_number, cancellation_token): """ - Waits for a device that matches args.path and args.serial_number - and then upgrades the device's firmware. + Polls libusb until a device in DFU mode is found """ + while not cancellation_token.is_set(): + params = {} if serial_number == None else {'serial_number': serial_number} + stm_device = usb.core.find(idVendor=0x0483, idProduct=0xdf11, **params) + if stm_device != None: + return stm_device + time.sleep(1) + return None + +def update_device(device, firmware, logger, cancellation_token): + """ + Updates the specified device with the specified firmware. + The device passed to this function can either be in + normal mode or in DFU mode. + The firmware should be an instance of Firmware or None. + If firmware is None, the newest firmware for the device is + downloaded from GitHub releases. + """ + + if isinstance(device, usb.core.Device): + serial_number = device.serial_number + dfudev = DfuDevice(device) + if (logger._verbose): + logger.debug("OTP:") + dump_otp(dfudev) + + # Read hardware version from one-time-programmable memory + otp_sector = [s for s in dfudev.sectors if s['name'] == 'OTP Memory' and s['addr'] == 0x1fff7800][0] + otp_data = dfudev.read_sector(otp_sector) + if otp_data[0] == 0: + otp_data = otp_data[16:] + if otp_data[0] == 0xfe: + hw_version = (otp_data[3], otp_data[4], otp_data[5]) + else: + hw_version = (0, 0, 0) + else: + serial_number = device.__channel__.usb_device.serial_number + dfudev = None + + # Read hardware version as reported from firmware + hw_version_major = device.hw_version_major if hasattr(device, 'hw_version_major') else 0 + hw_version_minor = device.hw_version_minor if hasattr(device, 'hw_version_minor') else 0 + hw_version_variant = device.hw_version_variant if hasattr(device, 'hw_version_variant') else 0 + hw_version = (hw_version_major, hw_version_minor, hw_version_variant) + + fw_version_major = device.fw_version_major if hasattr(device, 'fw_version_major') else 0 + fw_version_minor = device.fw_version_minor if hasattr(device, 'fw_version_minor') else 0 + fw_version_revision = device.fw_version_revision if hasattr(device, 'fw_version_revision') else 0 + fw_version_prerelease = device.fw_version_prerelease if hasattr(device, 'fw_version_prerelease') else True + fw_version = (fw_version_major, fw_version_minor, fw_version_revision, fw_version_prerelease) + + print("Found ODrive {} ({}) with firmware {}{}".format( + serial_number, + get_hw_version_string(hw_version), + get_fw_version_string(fw_version), + " in DFU mode" if dfudev is not None else "")) + + if firmware is None: + if hw_version == (0, 0, 0): + if dfudev is None: + suggestion = 'You have to manually flash an up-to-date firmware to make automatic checks work. Run `odrivetool dfu --help` for more info.' + else: + suggestion = 'Run "make write_otp" to program the board version.' + raise Exception('Cannot check online for new firmware because the board version is unknown. ' + suggestion) + print("Checking online for newest firmware...", end='') + firmware = get_newest_firmware(hw_version) + if firmware is None: + raise Exception("could not find any firmware release for this board version") + print(" found {}".format(get_fw_version_string(firmware.fw_version))) + + if firmware < fw_version: + print("Warning: you are about to flash firmware {} which is older than the firmware on the device ({}).".format( + get_fw_version_string(firmware.fw_version), + get_fw_version_string(fw_version))) + if not odrive.utils.yes_no_prompt("Do you want to flash this firmware anyway?", True): + raise OperationAbortedException() # load hex file # TODO: Either use the elf format or pack a custom format with a manifest. # This way we can for instance verify the target board version and only - # have to publish one file for every board. - hexfile = IntelHex(args.file) + # have to publish one file for every board (instead of elf AND hex files). + hexfile = IntelHex(firmware.get_as_hex()) - if (args.verbose): - print("Contiguous segments in hex file:") - for start, end in hexfile.segments(): - print(" {:08X} to {:08X}".format(start, end - 1)) + logger.debug("Contiguous segments in hex file:") + for start, end in hexfile.segments(): + logger.debug(" {:08X} to {:08X}".format(start, end - 1)) - serial_number = args.serial_number + # Back up configuration + if dfudev is None: + did_backup_config = device.user_config_loaded if hasattr(device, 'user_config_loaded') else False + if did_backup_config: + odrive.configuration.backup_config(device, None, logger) + elif not odrive.utils.yes_no_prompt("The configuration cannot be backed up because the device is already in DFU mode. The configuration may be lost after updating. Do you want to continue anyway?", True): + raise OperationAbortedException() - find_odrive_cancellation_token = Event(app_shutdown_token) + # Put the device into DFU mode if it's not already in DFU mode + if dfudev is None: + put_into_dfu_mode(device, cancellation_token) + stm_device = find_device_in_dfu_mode(serial_number, cancellation_token) + dfudev = DfuDevice(stm_device) - print("Waiting for ODrive...") - - # Scan for ODrives not in DFU mode and put them into DFU mode once they appear - # We only scan on USB because DFU is only possible over USB - odrive.discovery.find_all(args.path, serial_number, - lambda dev: put_odrive_into_dfu_mode(dev, find_odrive_cancellation_token), - find_odrive_cancellation_token, app_shutdown_token) - - # Poll libUSB until a device in DFU mode is found - while not app_shutdown_token.is_set(): - params = {} if serial_number == None else {'serial_number': serial_number} - stm_device = usb.core.find(idVendor=0x0483, idProduct=0xdf11, **params) - if stm_device != None: - break - time.sleep(1) - find_odrive_cancellation_token.set() # we don't need this thread anymore - if app_shutdown_token.is_set(): - sys.exit(1) - print("Found device {} in DFU mode".format(stm_device.serial_number)) - - dfudev = DfuDevice(stm_device) - - sectors = list(get_device_sectors(dfudev)) - - if (args.verbose): - print("Sectors on device: ") - for sector in sectors: - print(" {:08X} to {:08X} ({})".format( - sector['addr'], - sector['addr'] + sector['len'] - 1, - sector['name'])) + logger.debug("Sectors on device: ") + for sector in dfudev.sectors: + logger.debug(" {:08X} to {:08X} ({})".format( + sector['addr'], + sector['addr'] + sector['len'] - 1, + sector['name'])) # fill sectors with data - touched_sectors = list(populate_sectors(sectors, hexfile)) + touched_sectors = list(populate_sectors(dfudev.sectors, hexfile)) - if (args.verbose): - print("The following sectors will be flashed: ") - for sector,_ in touched_sectors: - print(" {:08X} to {:08X}".format(sector['addr'], sector['addr'] + sector['len'] - 1)) - - if (args.verbose): - print("OTP:") - dump_otp() + logger.debug("The following sectors will be flashed: ") + for sector,_ in touched_sectors: + logger.debug(" {:08X} to {:08X}".format(sector['addr'], sector['addr'] + sector['len'] - 1)) # Erase try: for i, (sector, data) in enumerate(touched_sectors): print("Erasing... (sector {}/{}) \r".format(i, len(touched_sectors)), end='', flush=True) - erase(dfudev, sector) + dfudev.erase_sector(sector) print('Erasing... done \r', end='', flush=True) finally: print('', flush=True) @@ -312,7 +368,7 @@ def launch_dfu(args, app_shutdown_token): try: for i, (sector, data) in enumerate(touched_sectors): print("Flashing... (sector {}/{}) \r".format(i, len(touched_sectors)), end='', flush=True) - flash(dfudev, sector, data) + dfudev.write_sector(sector, data) print('Flashing... done \r', end='', flush=True) finally: print('', flush=True) @@ -321,7 +377,7 @@ def launch_dfu(args, app_shutdown_token): try: for i, (sector, expected_data) in enumerate(touched_sectors): print("Verifying... (sector {}/{}) \r".format(i, len(touched_sectors)), end='', flush=True) - observed_data = read(dfudev, sector) + observed_data = dfudev.read_sector(sector) mismatch_pos = get_first_mismatch_index(observed_data, expected_data) if not mismatch_pos is None: mismatch_pos -= mismatch_pos % 16 @@ -341,7 +397,47 @@ def launch_dfu(args, app_shutdown_token): # So for debugging you should comment this last part out. # Jump to application - jump_to_application(dfudev, 0x08000000) + dfudev.jump_to_application(0x08000000) + + logger.info("Waiting for the device to reappear...") + device = odrive.discovery.find_any("usb", serial_number, + cancellation_token, cancellation_token, timeout=30) + + if did_backup_config: + odrive.configuration.restore_config(device, None, logger) + os.remove(odrive.configuration.get_temp_config_filename(device)) + + logger.success("Device firmware update successful.") + +def launch_dfu(args, logger, cancellation_token): + """ + Waits for a device that matches args.path and args.serial_number + and then upgrades the device's firmware. + """ + + serial_number = args.serial_number + find_odrive_cancellation_token = Event(cancellation_token) + + logger.info("Waiting for ODrive...") + + devices = [None, None] + + # Start background thread to scan for ODrives in DFU mode + def find_device_in_dfu_mode_thread(): + devices[0] = find_device_in_dfu_mode(serial_number, find_odrive_cancellation_token) + find_odrive_cancellation_token.set() + threading.Thread(target=find_device_in_dfu_mode_thread).start() + + # Scan for ODrives not in DFU mode + # We only scan on USB because DFU is only implemented over USB + devices[1] = odrive.discovery.find_any("usb", serial_number, + find_odrive_cancellation_token, cancellation_token) + find_odrive_cancellation_token.set() + + device = devices[0] or devices[1] + firmware = FirmwareFromFile(args.file) if args.file else None + + update_device(device, firmware, logger, cancellation_token) diff --git a/tools/odrive/dfuse/DfuDevice.py b/tools/odrive/dfuse/DfuDevice.py index b9ca449c..e7158d3b 100644 --- a/tools/odrive/dfuse/DfuDevice.py +++ b/tools/odrive/dfuse/DfuDevice.py @@ -1,5 +1,8 @@ import usb.util import time +import fractions +import array +from odrive.dfuse.DfuState import DfuState DFU_REQUEST_SEND = 0x21 DFU_REQUEST_RECEIVE = 0xa1 @@ -12,6 +15,9 @@ 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 ] @@ -24,6 +30,7 @@ class DfuDevice: 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] @@ -98,3 +105,115 @@ class DfuDevice: 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" % 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" % 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" % 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" % 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])) diff --git a/tools/odrive/discovery.py b/tools/odrive/discovery.py index a6536638..bc92ec68 100644 --- a/tools/odrive/discovery.py +++ b/tools/odrive/discovery.py @@ -62,7 +62,7 @@ def find_all(path, serial_number, return json_data = {"name": "odrive", "members": json_data} obj = odrive.remote_object.RemoteObject(json_data, None, channel, printer) - device_serial_number = format(obj.serial_number, 'x').upper() if hasattr(obj, 'serial_number') else "[unknown serial number]" + device_serial_number = odrive.utils.get_serial_number_str(obj) if serial_number != None and device_serial_number != serial_number: printer("Ignoring device with serial number {}".format(device_serial_number)) return diff --git a/tools/odrive/shell.py b/tools/odrive/shell.py index ef072010..7fc3e5ad 100644 --- a/tools/odrive/shell.py +++ b/tools/odrive/shell.py @@ -52,7 +52,7 @@ def did_discover_device(odrive, logger, app_shutdown_token): # Publish new ODrive to interactive console interactive_variables[interactive_name] = odrive globals()[interactive_name] = odrive # Add to globals so tab complete works - logger.info("{} to ODrive {:012X} as {}".format(verb, serial_number, interactive_name)) + logger.notify("{} to ODrive {:012X} as {}".format(verb, serial_number, interactive_name)) # Subscribe to disappearance of the device odrive.__channel__._channel_broken.subscribe(lambda: did_lose_device(interactive_name, logger, app_shutdown_token)) diff --git a/tools/odrive/usbbulk_transport.py b/tools/odrive/usbbulk_transport.py index 4483bb02..6d7b5cff 100644 --- a/tools/odrive/usbbulk_transport.py +++ b/tools/odrive/usbbulk_transport.py @@ -163,15 +163,18 @@ def discover_channels(path, serial_number, callback, cancellation_token, channel known_devices = [] def device_matcher(device): #print(" test {:04X}:{:04X}".format(device.idVendor, device.idProduct)) - if (device.bus, device.address) in known_devices: - return False - if bus != None and device.bus != bus: - return False - if address != None and device.address != address: - return False - if serial_number != None and device.serial_number != serial_number: - return False - if (device.idVendor, device.idProduct) not in ODRIVE_VID_PID_PAIRS: + try: + if (device.bus, device.address) in known_devices: + return False + if bus != None and device.bus != bus: + return False + if address != None and device.address != address: + return False + if serial_number != None and device.serial_number != serial_number: + return False + if (device.idVendor, device.idProduct) not in ODRIVE_VID_PID_PAIRS: + return False + except: return False return True diff --git a/tools/odrive/utils.py b/tools/odrive/utils.py index 050d4110..d73cbba2 100755 --- a/tools/odrive/utils.py +++ b/tools/odrive/utils.py @@ -25,6 +25,9 @@ data_rate = 100 plot_rate = 10 num_samples = 1000 +class OperationAbortedException(Exception): + pass + def start_liveplotter(get_var_callback): """ Starts a liveplotter. @@ -157,6 +160,11 @@ def setup_udev_rules(logger): subprocess.run(["udevadm", "trigger"], check=True) logger.info('udev rules configured successfully') +def get_serial_number_str(device): + if hasattr(device, 'serial_number'): + return format(device.serial_number, 'x').upper() + else: + return "[unknown serial number]" ## Exceptions ## @@ -357,9 +365,30 @@ class Logger(): def success(self, text): self.print_colored(self._prefix + text, Logger.COLOR_GREEN) def info(self, text): + self.print_colored(self._prefix + text, Logger.COLOR_DEFAULT) + def notify(self, text): self.print_colored(self._prefix + text, Logger.COLOR_CYAN) def warn(self, text): self.print_colored(self._prefix + text, Logger.COLOR_YELLOW) def error(self, text): # TODO: write to stderr self.print_colored(self._prefix + text, Logger.COLOR_RED) + +def yes_no_prompt(question, default=None): + if default is None: + question += " [y/n] " + elif default == True: + question += " [Y/n] " + elif default == False: + question += " [y/N] " + + while True: + print(question, end='') + + choice = input().lower() + if choice in {'yes', 'y'}: + return True + elif choice in {'no', 'n'}: + return False + elif choice == '' and default is not None: + return default diff --git a/tools/odrive/version.py b/tools/odrive/version.py index 68bad7d0..b0b43035 100644 --- a/tools/odrive/version.py +++ b/tools/odrive/version.py @@ -4,6 +4,20 @@ import subprocess import os import sys +def version_str_to_tuple(version_string): + """ + Converts a version string to a tuple of the form + (major, minor, revision, prerelease) + + Example: "fw-v0.3.6-23" => (0, 3, 6, True) + """ + regex=r'.*v([0-9a-zA-Z]+).([0-9a-zA-Z]+).([0-9a-zA-Z]+)(.*)' + return (int(re.sub(regex, r"\1", version_string)), + int(re.sub(regex, r"\2", version_string)), + int(re.sub(regex, r"\3", version_string)), + (re.sub(regex, r"\4", version_string) != "")) + + def get_version_from_git(): script_dir = os.path.dirname(os.path.realpath(__file__)) try: @@ -12,19 +26,15 @@ def get_version_from_git(): cwd=script_dir) git_tag = git_tag.decode(sys.stdout.encoding).rstrip('\n') - regex=r'.*v([0-9a-zA-Z]).([0-9a-zA-Z]).([0-9a-zA-Z])(.*)' - package_version_major = int(re.sub(regex, r"\1", git_tag)) - package_version_minor = int(re.sub(regex, r"\2", git_tag)) - package_version_revision = int(re.sub(regex, r"\3", git_tag)) - package_version_unreleased = (re.sub(regex, r"\4", git_tag) != "") + (major, minor, revision, is_prerelease) = version_str_to_tuple(git_tag) - if package_version_unreleased: - package_version_revision += 1 + if is_prerelease: + revision += 1 + return git_tag, major, minor, revision, is_prerelease except Exception as ex: print(ex) return "[unknown version]", 0, 0, 0, 1 - return git_tag, package_version_major, package_version_minor, package_version_revision, package_version_unreleased def get_version_str(git_only=False): """ diff --git a/tools/odrivetool b/tools/odrivetool index 901ebde3..54433c10 100755 --- a/tools/odrivetool +++ b/tools/odrivetool @@ -7,7 +7,8 @@ from __future__ import print_function import sys import argparse import odrive.discovery -from odrive.utils import Logger, Event +from odrive.utils import Logger, Event, OperationAbortedException +from odrive.configuration import * # Flush stdout by default # Source: @@ -33,8 +34,26 @@ shell_parser.add_argument("--no-ipython", action="store_true", "instead of the IPython shell, " "even if IPython is installed.") -dfu_parser = subparsers.add_parser('dfu', help="Upgrade the ODrive device firmware") -dfu_parser.add_argument('file', metavar='HEX', help='The .hex file to be flashed. Make sure your firmware board version matches the actual board version.') +dfu_parser = subparsers.add_parser('dfu', help="Upgrade the ODrive device firmware." + "If no serial number is specified, the first ODrive that is found is updated") +dfu_parser.add_argument('file', metavar='HEX', nargs='?', + help='The .hex file to be flashed. Make sure target board version ' + 'of the firmware file matches the actual board version. ' + 'You can download the latest release manually from ' + 'https://github.com/madcowswe/ODrive/releases. ' + 'If no file is provided, the script automatically downloads ' + 'the latest firmware.') + + +dfu_parser = subparsers.add_parser('backup-config', help="Saves the configuration of the ODrive to a JSON file") +dfu_parser.add_argument('file', nargs='?', + help="Path to the file where to store the data. " + "If no path is provided, the configuration is stored in {}.".format(tempfile.gettempdir())) + +dfu_parser = subparsers.add_parser('restore-config', help="Restores the configuration of the ODrive from a JSON file") +dfu_parser.add_argument('file', nargs='?', + help="Path to the file that contains the configuration data. " + "If no path is provided, the configuration is loaded from {}.".format(tempfile.gettempdir())) subparsers.add_parser('liveplotter', help="Upgrade the ODrive's Firmware") subparsers.add_parser('drv-status', help="Show status of the on-board DRV8301 chips (for debugging only)") @@ -110,12 +129,14 @@ try: elif args.command == 'dfu': print_version() import odrive.dfu - odrive.dfu.launch_dfu(args, app_shutdown_token) + odrive.dfu.launch_dfu(args, logger, app_shutdown_token) elif args.command == 'liveplotter': from odrive.utils import start_liveplotter print("Waiting for ODrive...") - my_odrive = odrive.discovery.find_any(path=args.path, serial_number=args.serial_number) + my_odrive = odrive.discovery.find_any(path=args.path, serial_number=args.serial_number, + search_cancellation_token=app_shutdown_token, + channel_termination_token=app_shutdown_token) # If you want to plot different values, change them here. # You can plot any number of values concurrently. @@ -125,22 +146,44 @@ try: elif args.command == 'drv-status': from odrive.utils import print_drv_regs print("Waiting for ODrive...") - my_odrive = odrive.discovery.find_any(path=args.path, serial_number=args.serial_number) + my_odrive = odrive.discovery.find_any(path=args.path, serial_number=args.serial_number, + search_cancellation_token=app_shutdown_token, + channel_termination_token=app_shutdown_token) print_drv_regs("Motor 0", my_odrive.axis0.motor) print_drv_regs("Motor 1", my_odrive.axis1.motor) elif args.command == 'rate-test': from odrive.utils import rate_test print("Waiting for ODrive...") - my_odrive = odrive.discovery.find_any(path=args.path, serial_number=args.serial_number) + my_odrive = odrive.discovery.find_any(path=args.path, serial_number=args.serial_number, + search_cancellation_token=app_shutdown_token, + channel_termination_token=app_shutdown_token) rate_test(my_odrive) elif args.command == 'udev-setup': from odrive.utils import setup_udev_rules setup_udev_rules(logger) + elif args.command == 'backup-config': + from odrive.configuration import backup_config + print("Waiting for ODrive...") + my_odrive = odrive.discovery.find_any(path=args.path, serial_number=args.serial_number, + search_cancellation_token=app_shutdown_token, + channel_termination_token=app_shutdown_token) + backup_config(my_odrive, args.file, logger) + + elif args.command == 'restore-config': + from odrive.configuration import restore_config + print("Waiting for ODrive...") + my_odrive = odrive.discovery.find_any(path=args.path, serial_number=args.serial_number, + search_cancellation_token=app_shutdown_token, + channel_termination_token=app_shutdown_token) + restore_config(my_odrive, args.file, logger) + else: raise Exception("unknown command: " + args.command) +except OperationAbortedException: + logger.info("Operation aborted.") finally: app_shutdown_token.set() diff --git a/tools/run_tests.py b/tools/run_tests.py index 1cb1bcd8..447546e9 100755 --- a/tools/run_tests.py +++ b/tools/run_tests.py @@ -136,7 +136,7 @@ try: if isinstance(test, ODriveTest): def odrv_test_thread(odrv_name): odrv_ctx = odrives_by_name[odrv_name] - logger.info('* running {} on {}...'.format(type(test).__name__, odrv_name)) + logger.notify('* running {} on {}...'.format(type(test).__name__, odrv_name)) try: test.check_preconditions(odrv_ctx, logger.indent(' {}: '.format(odrv_name))) @@ -165,7 +165,7 @@ try: try: if not app_shutdown_token.is_set(): # Run test on this axis - logger.info('* running {} on {}...'.format(type(test).__name__, axis_name)) + logger.notify('* running {} on {}...'.format(type(test).__name__, axis_name)) try: test.check_preconditions(axis_ctx, logger.indent(' {}: '.format(axis_name))) @@ -197,7 +197,7 @@ try: try: if not app_shutdown_token.is_set(): # Run test on this axis - logger.info('* running {} on {}...'.format(type(test).__name__, coupling_name)) + logger.notify('* running {} on {}...'.format(type(test).__name__, coupling_name)) try: test.check_preconditions(coupled_axes[0], coupled_axes[1], logger.indent(' {}: '.format(coupling_name)))