Add proper timeouts and error handling to fx2tool.

software/fx2/__init__.py

@@ -7,10 +7,10 @@
 VID_CYPRESS = 0x04B4
 PID_FX2     = 0x8613
 
-_CMD_RW_RAM       = 0xA0
-_CMD_RW_EEPROM_SB = 0xA2
-_CMD_RENUM        = 0xA8
-_CMD_RW_EEPROM_DB = 0xA9
+CMD_RW_RAM       = 0xA0
+CMD_RW_EEPROM_SB = 0xA2
+CMD_RENUM        = 0xA8
+CMD_RW_EEPROM_DB = 0xA9
 
 
 class FX2DeviceError(Exception):
@@ -25,6 +25,7 @@ class FX2Device:
     or raises a :exc:`FX2DeviceError`.
     """
     def __init__(self, vid=VID_CYPRESS, pid=PID_FX2):
+        self._timeout = 1000
         self._context = usb1.USBContext()
         self._device = self._context.openByVendorIDAndProductID(vid, pid)
         if self._device is None:
@@ -33,22 +34,36 @@ def __init__(self, vid=VID_CYPRESS, pid=PID_FX2):
 
         self._eeprom_size = None
 
+    def _control_read(self, request_type, request, value, index, length,
+                      timeout=None):
+        if timeout is None:
+            timeout = self._timeout
+        return self._device.controlRead(request_type, request, value, index, length, timeout)
+
+    def _control_write(self, request_type, request, value, index, data,
+                      timeout=None):
+        if timeout is None:
+            timeout = self._timeout
+        self._device.controlWrite(request_type, request, value, index, data, timeout)
+
     def read_ram(self, addr, length):
         """
         Read ``length`` bytes at ``addr`` from internal RAM.
         Note that not all memory can be addressed this way; consult the TRM.
         """
         if addr & 1: # unaligned
-            return self._device.controlRead(0x40, _CMD_RW_RAM, addr, 0, length + 1)[1:]
+            return self._control_read(usb1.REQUEST_TYPE_VENDOR,
+                                      CMD_RW_RAM, addr, 0, length + 1)[1:]
         else:
-            return self._device.controlRead(0x40, _CMD_RW_RAM, addr, 0, length)
+            return self._control_read(usb1.REQUEST_TYPE_VENDOR,
+                                      CMD_RW_RAM, addr, 0, length)
 
     def write_ram(self, addr, data):
         """
         Write ``data`` to ``addr`` to internal RAM.
         Note that not all memory can be addressed this way; consult the TRM.
         """
-        self._device.controlWrite(0x40, _CMD_RW_RAM, addr, 0, data)
+        self._control_write(usb1.REQUEST_TYPE_VENDOR, CMD_RW_RAM, addr, 0, data)
 
     def cpu_reset(self, is_reset):
         """Bring CPU in or out of reset."""
@@ -57,21 +72,23 @@ def cpu_reset(self, is_reset):
     @staticmethod
     def _eeprom_cmd(addr_width):
         if addr_width == 1:
-            return _CMD_RW_EEPROM_SB
+            return CMD_RW_EEPROM_SB
         elif addr_width == 2:
-            return _CMD_RW_EEPROM_DB
+            return CMD_RW_EEPROM_DB
         else:
             raise ValueError("Address width {addr_width} is not supported"
                              .format(addr_width=addr_width))
 
     def read_eeprom(self, addr, length, addr_width):
         """Read ``length`` bytes at ``addr`` from boot EEPROM."""
-        return self._device.controlRead(0x40, self._eeprom_cmd(addr_width), addr, 0, length)
+        return self._control_read(usb1.REQUEST_TYPE_VENDOR,
+                                  self._eeprom_cmd(addr_width), addr, 0, length)
 
     def write_eeprom(self, addr, data, addr_width):
         """Write ``data`` to ``addr`` in boot EEPROM."""
-        self._device.controlWrite(0x40, self._eeprom_cmd(addr_width), addr, 0, data)
+        self._control_write(usb1.REQUEST_TYPE_VENDOR,
+                            self._eeprom_cmd(addr_width), addr, 0, data)
 
     def reenumerate(self):
         """Trigger re-enumeration."""
-        self._device.controlWrite(0x40, _CMD_RENUM, 0, 0, [])
+        self._control_write(usb1.REQUEST_TYPE_VENDOR, CMD_RENUM, 0, 0, [])

software/fx2/fx2tool.py

@@ -5,6 +5,7 @@
 import collections
 import argparse
 import textwrap
+import usb1
 
 from . import VID_CYPRESS, PID_FX2, FX2Device, FX2DeviceError
 
@@ -310,36 +311,49 @@ def main():
     except FX2DeviceError as e:
         raise SystemExit(e)
 
-    if args.bootloader:
-        bootloader_ihex = os.path.join(resource_dir, "bootloader.ihex")
-        load(device, "auto", open(bootloader_ihex))
-    elif args.stage2:
-        load(device, "auto", args.stage2)
-
-    if args.action == "load":
-        load(device, args.format, args.firmware)
-
-    elif args.action == "read_ram":
-        device.cpu_reset(True)
-        data = device.read_ram(args.address, args.length)
-        output_data(args.format, args.file, data, args.address)
-
-    elif args.action == "write_ram":
-        data = input_data(args.format, args.file, args.data, args.offset)
-        device.cpu_reset(True)
-        for address, chunk in data:
-            device.write_ram(address, chunk)
-
-    elif args.action == "read_eeprom":
-        device.cpu_reset(False)
-        data = device.read_eeprom(args.address, args.length, args.address_width)
-        output_data(args.format, args.file, data, args.address)
-
-    elif args.action == "write_eeprom":
-        data = input_data(args.format, args.file, args.data, args.offset)
-        device.cpu_reset(False)
-        for address, chunk in data:
-            device.write_eeprom(address, chunk, args.address_width)
+    try:
+        if args.bootloader:
+            bootloader_ihex = os.path.join(resource_dir, "bootloader.ihex")
+            load(device, "auto", open(bootloader_ihex))
+        elif args.stage2:
+            load(device, "auto", args.stage2)
+
+        if args.action == "load":
+            load(device, args.format, args.firmware)
+
+        elif args.action == "read_ram":
+            device.cpu_reset(True)
+            data = device.read_ram(args.address, args.length)
+            output_data(args.format, args.file, data, args.address)
+
+        elif args.action == "write_ram":
+            data = input_data(args.format, args.file, args.data, args.offset)
+            device.cpu_reset(True)
+            for address, chunk in data:
+                device.write_ram(address, chunk)
+
+        elif args.action == "read_eeprom":
+            device.cpu_reset(False)
+            data = device.read_eeprom(args.address, args.length, args.address_width)
+            output_data(args.format, args.file, data, args.address)
+
+        elif args.action == "write_eeprom":
+            data = input_data(args.format, args.file, args.data, args.offset)
+            device.cpu_reset(False)
+            for address, chunk in data:
+                device.write_eeprom(address, chunk, args.address_width)
+
+    except usb1.USBErrorPipe:
+        if args.action in ["read_eeprom", "write_eeprom"]:
+            raise SystemExit("Command not acknowledged (wrong address width?)")
+        else:
+            raise SystemExit("Command not acknowledged")
+
+    except usb1.USBErrorTimeout:
+        if args.action in ["read_eeprom", "write_eeprom"]:
+            raise SystemExit("Command timeout (bootloader not loaded?)")
+        else:
+            raise SystemExit("Command timeout")
 
 
 if __name__ == "__main__":