implement automatic firmware download

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]))

tools/odrive/utils.py

@@ -363,3 +363,22 @@ class Logger():
     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

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):
     """

tools/odrivetool

@@ -33,8 +33,15 @@ 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.')
 
 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)")