From 5c8ed3c5f6ba0b63ea13c2e95dd8eeb17daf9ef8 Mon Sep 17 00:00:00 2001 From: Justin Applegate Date: Thu, 15 May 2025 15:43:38 -0600 Subject: [PATCH 1/6] Starting on documentation for NAEUSB_Backend --- .../chipwhisperer/hardware/naeusb/naeusb.py | 36 +++++++++++++++++-- 1 file changed, 33 insertions(+), 3 deletions(-) diff --git a/software/chipwhisperer/hardware/naeusb/naeusb.py b/software/chipwhisperer/hardware/naeusb/naeusb.py index cdd5970071..9d1657340a 100644 --- a/software/chipwhisperer/hardware/naeusb/naeusb.py +++ b/software/chipwhisperer/hardware/naeusb/naeusb.py @@ -370,6 +370,17 @@ def make_len_addr(dlen, addr): class NAEUSB_Backend: """ Backend to talk to the USB device. + + Attributes: + _usbdev (usb1.USBDeviceHandle): The USB device handle. + _timeout (int): Timeout for USB operations. + device (usb1.USBDevice): The USB device object. + handle (usb1.USBDeviceHandle): The USB device handle. + usb_ctx (usb1.USBContext): The USB context. + sn (str): Serial number of the device. + pid (int): Product ID of the device. + rep (int): Read endpoint address. + wep (int): Write endpoint address. """ CMD_READMEM_BULK = 0x10 @@ -379,6 +390,9 @@ class NAEUSB_Backend: CMD_MEMSTREAM = 0x14 def __init__(self): + """Initializes the USB backend with default values. + + The timeout is set to 500ms and a USB context is created.""" self._usbdev = None self._timeout = 500 self.device = None @@ -395,12 +409,28 @@ def __init__(self): self.device = None def usbdev(self) -> usb1.USBDeviceHandle: - """Safely get USB device, throwing error if not connected""" + """Safely get USB device, throwing an error if not connected. + + Returns: + A usb1.USBDeviceHandle object. + + Raises: + OSError: If the USB device is not connected.""" if not self._usbdev: raise OSError("USB Device not found. Did you connect it first?") return self._usbdev - def is_accessable(self, dev : usb1.USBDevice) -> bool: + def is_accessible(self, dev : usb1.USBDevice) -> bool: + """Notes whether the device can be accessed or not. + + Calls the :code:`getSerialNumber()` device attribute. + + Args: + dev (usb1.USBDevice): The device to check. + + Returns: + bool: True if no errors are thrown, False otherwise. + """ try: dev.getSerialNumber() return True @@ -481,7 +511,7 @@ def __del__(self): self.close() def close(self): - # """Close the USB connection""" + """Close the USB connection and clear attributes.""" if self.device: del self.device self.device = None From db49564284fb8b2dc98c3837dc1c7f343ea6c2dc Mon Sep 17 00:00:00 2001 From: Justin Applegate Date: Thu, 15 May 2025 17:40:39 -0600 Subject: [PATCH 2/6] Finished documentation and type hints for NAEUSB_Backend class --- .../chipwhisperer/hardware/naeusb/naeusb.py | 278 +++++++++++++----- 1 file changed, 198 insertions(+), 80 deletions(-) diff --git a/software/chipwhisperer/hardware/naeusb/naeusb.py b/software/chipwhisperer/hardware/naeusb/naeusb.py index 9d1657340a..5cc0085e1a 100644 --- a/software/chipwhisperer/hardware/naeusb/naeusb.py +++ b/software/chipwhisperer/hardware/naeusb/naeusb.py @@ -191,16 +191,33 @@ class CWFirmwareError(Exception): } } -def quick_firmware_erase(product_id, serial_number=None): +def quick_firmware_erase(product_id : int, serial_number : str=None): + """Quickly erase the firmware on a device by entering bootloader mode. + + Args: + product_id (int): The product ID of the device. + serial_number (str, optional): The serial number of the device. If not provided, the function will attempt to find the device by product ID.""" naeusb = NAEUSB() naeusb.con(serial_number=serial_number, idProduct=[product_id]) naeusb.enterBootloader(True) - -def _check_sam_feature(feature, fw_version, prod_id): +def _check_sam_feature(feature : str, fw_version : str, prod_id : int) -> bool: + """Checks if a feature is available for a given firmware version and product ID. + + Args: + feature (str): The feature name to check, defined in :code:`SAM_FW_FEATURES`. + fw_version (str): The firmware version to check against. + prod_id (int): The product ID of the device. + + Returns: + bool: True if the feature is available for the given firmware version and product ID, False otherwise. + + Raises: + ValueError: If the feature is not recognized. + """ if prod_id not in SAM_FW_FEATURE_BY_DEVICE: naeusb_logger.debug("Features for ProdID {:04X} not stored, skipping...".format(prod_id)) - return + return False if feature not in SAM_FW_FEATURES: raise ValueError("Unknown feature {}".format(feature)) feature_set = SAM_FW_FEATURE_BY_DEVICE[prod_id] @@ -213,13 +230,19 @@ def _check_sam_feature(feature, fw_version, prod_id): return True -def _WINDOWS_USB_CHECK_DRIVER(device) -> Optional[str]: - """Checks which driver device is using +def _WINDOWS_USB_CHECK_DRIVER(device : usb1.USBDevice) -> Optional[str]: + """Checks which driver the device is using. - Checks whether the device is connected to the PC (harder than you'd think) + Checks whether the device is connected to the PC (harder than you'd think). Does not check the actual driver in use for custom interfaces in composite devices. Instead, it just resolves to usbcggp, which is the composite device driver for Windows. + + Args: + device (usb1.USBDevice): The USB device to check. + + Returns: + Optional[str]: The driver if found, None otherwise. """ try: import winreg @@ -227,7 +250,7 @@ def _WINDOWS_USB_CHECK_DRIVER(device) -> Optional[str]: subkey = r"ControlSet001\Enum\USB" subkey += "\\VID_{:04X}&PID_{:04X}".format(device.getVendorID(), device.getProductID()) - def get_enum_by_name(handle, name): + def get_enum_by_name(handle : winreg.PyHKEY, name : str): try: cnt = 0 enum_name = "" @@ -308,14 +331,14 @@ def get_enum_by_name(handle, name): naeusb_logger.warning("Could not check driver ({}), assuming WINUSB is used".format(str(e))) return None -def packuint32(data): - """Converts a 32-bit integer into format expected by USB firmware""" +def packuint32(data : int) -> List[int]: + """Converts a 32-bit integer into format expected by USB firmware.""" data = int(data) return [data & 0xff, (data >> 8) & 0xff, (data >> 16) & 0xff, (data >> 24) & 0xff] -def unpackuint32(buf): - """"Converts an array into a 32-bit integer""" +def unpackuint32(buf : List[int]) -> int: + """"Converts an array into a 32-bit integer.""" pint = buf[0] pint |= buf[1] << 8 @@ -323,23 +346,24 @@ def unpackuint32(buf): pint |= buf[3] << 24 return pint -def packuint16(data): - """Converts a 16-bit integer into format expected by USB firmware""" +def packuint16(data : int) -> List[int]: + """Converts a 16-bit integer into format expected by USB firmware.""" data = int(data) - return [data & 0xff, (data >> 8) & 0xff, (data >> 16) & 0xff, (data >> 24) & 0xff] LEN_ADDR_HDR_SIZE = 8 -def set_len_addr(buf, dlen, addr): +def set_len_addr(buf : bytearray, dlen : int, addr : int): """Populates a buffer with the command header. + + Sets the length and address in the buffer using little-endian format. Modifies + the buffer in place. """ - # Little endian util.pack_u32_into(buf, 0, dlen) util.pack_u32_into(buf, 4, addr) -def make_len_addr(dlen, addr): +def make_len_addr(dlen : int, addr : int) -> bytearray: """Creates a command header buffer. Return: @@ -352,19 +376,19 @@ def make_len_addr(dlen, addr): NAEUSB_CTRL_IO_MAX = 128 NAEUSB_CTRL_IO_THRESHOLD = 48 -#List of all NewAE PID's +# List of all NewAE PIDs NEWAE_VID = 0x2B3E NEWAE_PIDS = { - 0xACE2: {'name': "ChipWhisperer-Lite", 'fwver': fwver("cwlite")}, - 0xACE3: {'name': "ChipWhisperer-CW1200", 'fwver': fwver("cw1200")}, - 0xC305: {'name': "CW305 Artix FPGA Board", 'fwver': fwver("cw305")}, - 0xC310: {'name': "CW305 Artix FPGA Board", 'fwver': fwver("cwbergen")}, - 0xC340: {'name': "CW305 Artix FPGA Board", 'fwver': fwver("cwluna")}, - 0xACE0: {'name': "ChipWhisperer-Nano", 'fwver': fwver("cwnano")}, - 0xACE5: {'name': "ChipWhisperer-Husky", 'fwver': fwver("cwhusky")}, - 0xACE6: {'name': "ChipWhisperer-Husky-Plus", 'fwver': fwver("cwhuskyplus")}, - 0xC521: {'name': "CW521 Ballistic-Gel", 'fwver': None}, - 0xC610: {'name': "PhyWhisperer-USB", 'fwver': None}, + 0xACE2: {'name': "ChipWhisperer-Lite", 'fwver': fwver("cwlite")}, + 0xACE3: {'name': "ChipWhisperer-CW1200", 'fwver': fwver("cw1200")}, + 0xC305: {'name': "CW305 Artix FPGA Board", 'fwver': fwver("cw305")}, + 0xC310: {'name': "CW305 Artix FPGA Board", 'fwver': fwver("cwbergen")}, + 0xC340: {'name': "CW305 Artix FPGA Board", 'fwver': fwver("cwluna")}, + 0xACE0: {'name': "ChipWhisperer-Nano", 'fwver': fwver("cwnano")}, + 0xACE5: {'name': "ChipWhisperer-Husky", 'fwver': fwver("cwhusky")}, + 0xACE6: {'name': "ChipWhisperer-Husky-Plus", 'fwver': fwver("cwhuskyplus")}, + 0xC521: {'name': "CW521 Ballistic-Gel", 'fwver': None}, + 0xC610: {'name': "PhyWhisperer-USB", 'fwver': None}, } class NAEUSB_Backend: @@ -392,7 +416,8 @@ class NAEUSB_Backend: def __init__(self): """Initializes the USB backend with default values. - The timeout is set to 500ms and a USB context is created.""" + The timeout is set to 500ms and a USB context is created. + """ self._usbdev = None self._timeout = 500 self.device = None @@ -415,7 +440,8 @@ def usbdev(self) -> usb1.USBDeviceHandle: A usb1.USBDeviceHandle object. Raises: - OSError: If the USB device is not connected.""" + OSError: If the USB device is not connected. + """ if not self._usbdev: raise OSError("USB Device not found. Did you connect it first?") return self._usbdev @@ -439,6 +465,24 @@ def is_accessible(self, dev : usb1.USBDevice) -> bool: def find(self, serial_number : Optional[str]=None, idProduct : Optional[List[int]]=None, hw_location : Optional[Tuple[int, int]]=None) -> usb1.USBDevice: + """Find a ChipWhisperer device by serial number, product ID, or hardware location tuple. + + All arguments are optional. If only one ChipWhisperer is connected, it will be returned. + If multiple are connected but no arguments were provided, or if no devices are found, + an exception will be raised. + + Args: + serial_number (str, optional): The serial number of the device to find. + idProduct (list, optional): The product ID(s) to match. + hw_location (tuple, optional): The hardware location tuple (bus number, device address). + + Returns: + usb1.USBDevice: The found USB device. + + Raises: + OSError: If no devices are found or if the specified device cannot be accessed. + Warning: If multiple devices are found and no serial number is provided. + """ # check if we got anything dev_list = self.get_possible_devices(idProduct, attempt_access=(not hw_location)) if len(dev_list) == 0: @@ -451,6 +495,7 @@ def find(self, serial_number : Optional[str]=None, idProduct : Optional[List[int if len(dev_list) != 1: raise OSError("Unable to find ChipWhisperer with hw_location {}, got {}".format(hw_location, dev_list)) return dev_list[0] + sns = ["{}:{}".format(dev.getProduct(), dev.getSerialNumber()) for dev in dev_list] if (len(dev_list) > 1) and (serial_number is None): if len(dev_list) > 1: @@ -468,11 +513,26 @@ def find(self, serial_number : Optional[str]=None, idProduct : Optional[List[int # finally, we know we have the right device and can return return dev_list[0] - def open(self, serial_number : Optional[str]=None, idProduct : Optional[List[int]]=None, - connect_to_first : bool =False, hw_location : Optional[Tuple[int, int]]=None) -> Optional[usb1.USBDeviceHandle]: - """ - Connect to device using default VID/PID + connect_to_first : bool=False, hw_location : Optional[Tuple[int, int]]=None) -> Optional[usb1.USBDeviceHandle]: + """Connect to device using serial number, product ID, or hardware location tuple. + + If :code:`connect_to_first` is set to False, then :code:`self.device` will be + set and None will be returned. Otherwise, the device will be opened and the + handle will be returned. + + Args: + serial_number (str, optional): The serial number of the device to connect to. + idProduct (list, optional): The product ID(s) to match. + connect_to_first (bool, optional): If True, open the device and return the handle. + hw_location (tuple, optional): The hardware location tuple (bus number, device address). + + Returns: + usb1.USBDeviceHandle: The opened USB device handle if :code:`connect_to_first` is True. + None: If :code:`connect_to_first` is False. + + Raises: + usb1.USBError: If the device cannot be opened. """ self.device = self.find(serial_number, idProduct, hw_location=hw_location) @@ -484,7 +544,7 @@ def open(self, serial_number : Optional[str]=None, idProduct : Optional[List[int naeusb_logger.error("Could not open USB device.") if e.value == -3: naeusb_logger.error("Check that the ChipWhisperer is not already connected") - naeusb_logger.error("Or that you have the proper permissions to access it") + naeusb_logger.error("And that you have the proper permissions to access it") raise self._usbdev = self.handle if os.name == "nt" or sys.platform == "darwin": @@ -524,15 +584,20 @@ def close(self): def get_possible_devices(self, idProduct : Optional[List[int]]=None, dictonly : bool=True, attempt_access : bool=False) -> List[usb1.USBDevice]: """Get list of USB devices that match NewAE vendor ID (0x2b3e) and - optionally a product ID + optionally a product ID. + + Checks the VendorID, then ensures the devices are accessible. - Checks VendorID, then makes sure the devices are accessable Args: - idProduct (list of int, optional): If not None, the product ID to match - sn (string, optional): If not None, + idProduct (list, optional): If not None, the product ID to match + sn (string, optional): If not None, the serial number to match + Returns: List of USBDevice that match Vendor/Product IDs - """ + + Raises: + OSError: If no devices are found or if the specified device cannot be accessed. + """ dev_list = [dev for dev in self.usb_ctx.getDeviceIterator(skip_on_error=True) if dev.getVendorID() == 0x2b3e] naeusb_logger.info("Found NAEUSB devices {}".format(dev_list)) @@ -569,8 +634,12 @@ def get_possible_devices(self, idProduct : Optional[List[int]]=None, dictonly : return dev_list def sendCtrl(self, cmd : int, value : int=0, data : bytearray=bytearray()): - """ - Send data over control endpoint + """Send data over control endpoint. + + Args: + cmd (int): The command to send. + value (int, optional): The value to send. Defaults to 0. + data (bytearray, optional): The data to send. Defaults to an empty bytearray. """ # Vendor-specific, OUT, interface control transfer naeusb_logger.debug("WRITE_CTRL: bmRequestType: {:02X}, \ @@ -582,8 +651,15 @@ def sendCtrl(self, cmd : int, value : int=0, data : bytearray=bytearray()): #return self.usbdev().ctrl_transfer(0x41, cmd, value, 0, data, timeout=self._timeout) def readCtrl(self, cmd : int, value : int=0, dlen : int=0) -> bytearray: - """ - Read data from control endpoint + """Read data from control endpoint. + + Args: + cmd (int): The command to read. + value (int, optional): The value to read. Defaults to 0. + dlen (int, optional): The length of the data to read. Defaults to 0. + + Returns: + bytearray: The received data. """ # Vendor-specific, IN, interface control transfer if dlen > NAEUSB_CTRL_IO_MAX: @@ -594,34 +670,49 @@ def readCtrl(self, cmd : int, value : int=0, dlen : int=0) -> bytearray: value, 0, dlen, response)) return response - def _get_timeout(self, timeout): + def _get_timeout(self, timeout : Union[int, float, None]=None) -> int: """Gets the default timeout if the operation caller did not specify one. Returns: - A valid timeout value. + int: A valid timeout value. """ if timeout is None: timeout = self._timeout return timeout - def _bulk_read(self, data, timeout): + def _bulk_read(self, dlen : int, timeout : Union[int, float, None]): """Reads data over the bulk-transfer endpoint. + Args: + dlen (int): The length of the data to read. + timeout (int, float, optional): The timeout for the read operation. + If None, the default timeout is used. + Returns: - The received data. + bytearray: The received data. """ timeout = self._get_timeout(timeout) - return self.handle.bulkRead(self.rep, data, timeout) + return self.handle.bulkRead(self.rep, dlen, timeout) - def _bulk_write(self, data, timeout): + def _bulk_write(self, data : bytearray, timeout : Union[int, float, None]): """Writes data over the bulk-transfer endpoint. + + Args: + data (bytearray): The data to write. + timeout (int, float, optional): The timeout for the write operation. + If None, the default timeout is used. """ timeout = self._get_timeout(timeout) self.handle.bulkWrite(self.wep, data, timeout) - def _cmd_ctrl_send_data(self, pload, cmd : int): - """Sends data over the control-transfer channel and attempts a pipe error fix if an initial - error occured. + def _cmd_ctrl_send_data(self, pload : bytearray, cmd : int): + """Sends data over the control-transfer channel. + + If a pipe error occurs, it attempts to fix it. + + Args: + pload (bytearray): The data to send. + cmd (int): The command to send. """ try: self.sendCtrl(cmd, data=pload) @@ -631,34 +722,41 @@ def _cmd_ctrl_send_data(self, pload, cmd : int): self.sendCtrl(cmd, data=pload) def _cmd_ctrl_send_header(self, addr : int, dlen : int, cmd : int): - """Sends the standard length/addr header over the control-transfer endpoint. - """ + """Sends the standard length/addr header over the control-transfer endpoint.""" # TODO: Alloc header class member? Won't hafta alloc mem every read and writectrl call... pload = make_len_addr(dlen, addr) self._cmd_ctrl_send_data(pload, cmd) - def _cmd_readmem_ctrl(self, addr : int, dlen : int): + def _cmd_readmem_ctrl(self, addr : int, dlen : int) -> bytearray: """Reads data from the external memory interface over the control-transfer endpoint. Returns: - The received data. + bytearray: The received data. """ - self._cmd_ctrl_send_header(addr, dlen, self.CMD_READMEM_CTRL); + self._cmd_ctrl_send_header(addr, dlen, self.CMD_READMEM_CTRL) return self.readCtrl(self.CMD_READMEM_CTRL, dlen=dlen) - def _cmd_readmem_bulk(self, addr : int, dlen : int): + def _cmd_readmem_bulk(self, addr : int, dlen : int) -> bytearray: """Reads data from the external memory interface over the bulk-transfer endpoint. Returns: - The received data. + bytearray: The received data. """ - self._cmd_ctrl_send_header(addr, dlen, self.CMD_READMEM_BULK); + self._cmd_ctrl_send_header(addr, dlen, self.CMD_READMEM_BULK) return self._bulk_read(dlen, None) def cmdReadMem(self, addr : int, dlen : int) -> bytearray: - """ - Send command to read over external memory interface from FPGA. Automatically - decides to use control-transfer or bulk-endpoint transfer based on data length. + """Send command to read over external memory interface from FPGA. + + It automatically decides to use control-transfer or bulk-endpoint transfer based on + data length. + + Args: + addr (int): The address to read from. + dlen (int): The length of the data to read. + + Returns: + bytearray: The received data. """ dlen = int(dlen) if dlen < NAEUSB_CTRL_IO_THRESHOLD: @@ -670,8 +768,12 @@ def cmdReadMem(self, addr : int, dlen : int) -> bytearray: .format("yes" if dlen >= NAEUSB_CTRL_IO_THRESHOLD else "no", addr, dlen, data)) return data - def _cmd_writemem_ctrl(self, addr : int, data): + def _cmd_writemem_ctrl(self, addr : int, data : bytearray): """Writes data to the external memory interface via the control-transfer endpoint. + + Args: + addr (int): The address to write to. + data (bytearray): The data to write. """ # TODO: Investigate if we don't hafta combine header with the data and can send separately. # Is this is a FW implementation or a limitation from middleware interfaces? @@ -680,16 +782,25 @@ def _cmd_writemem_ctrl(self, addr : int, data): util.bytes_fast_copy(pload, LEN_ADDR_HDR_SIZE, data) self._cmd_ctrl_send_data(pload, self.CMD_WRITEMEM_CTRL) - def _cmd_writemem_bulk(self, addr : int, data): + def _cmd_writemem_bulk(self, addr : int, data : bytearray): """Writes data to the external memory interface via the bulk-transfer endpoint. + + Args: + addr (int): The address to write to. + data (bytearray): The data to write. """ self._cmd_ctrl_send_header(addr, len(data), self.CMD_WRITEMEM_BULK) self._bulk_write(data, None) - def cmdWriteMem(self, addr : int, data): - """ - Send command to write memory over external memory interface to FPGA. Automatically - decides to use control-transfer or bulk-endpoint transfer based on data length. + def cmdWriteMem(self, addr : int, data : bytearray): + """Send command to write memory over external memory interface to FPGA. + + It automatically decides to use control-transfer or bulk-endpoint transfer based + on data length. + + Args: + addr (int): The address to write to. + data (bytearray): The data to write. """ pload = util.get_bytes_memview(data) if len(pload) < NAEUSB_CTRL_IO_THRESHOLD: @@ -700,19 +811,17 @@ def cmdWriteMem(self, addr : int, data): naeusb_logger.debug("FPGA_WRITE: bulk: {}, addr: {:08X}, dlen: {:08X}, response: {}"\ .format("yes" if len(pload) >= NAEUSB_CTRL_IO_THRESHOLD else "no", addr, len(pload), data)) - return None + def write_bulk(self, data : bytearray, timeout = None): + """Write data directly to the bulk endpoint. - def cmdWriteBulk(self, data : bytearray, timeout = None): - """ - Write data directly to the bulk endpoint. - :param data: Data to be written - :return: + Args: + data (bytearray): The data to write. + timeout (int, float, optional): The timeout for the write operation. + If None, the default timeout is used. """ naeusb_logger.debug("BULK WRITE: data = {}".format(data)) self._bulk_write(data, timeout) - writeBulk = cmdWriteBulk - def flushInput(self): """Dump all the crap left over""" try: @@ -722,6 +831,15 @@ def flushInput(self): pass def read(self, dbuf : bytearray, timeout : int) -> bytearray: + """Read data from the bulk endpoint. + + Args: + dbuf (bytearray): The buffer to read data into. + timeout (int): The timeout for the read operation. + + Returns: + bytearray: The received data. + """ resp = self._bulk_read(dbuf, timeout) naeusb_logger.debug("BULK READ: data = {}".format(dbuf)) return resp @@ -932,7 +1050,7 @@ def writeBulkEP(self, data : bytearray, timeout = None): :param data: Data to be written. :return: """ - return self.usbserializer.writeBulk(data, timeout=timeout) + return self.usbserializer.write_bulk(data, timeout=timeout) def flushInput(self): """Dump all the crap left over""" From 192bfa8311f8c3afcc4c250a221d98578941bdde Mon Sep 17 00:00:00 2001 From: Justin Applegate Date: Thu, 15 May 2025 17:48:20 -0600 Subject: [PATCH 3/6] Forgot one --- software/chipwhisperer/hardware/naeusb/naeusb.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/software/chipwhisperer/hardware/naeusb/naeusb.py b/software/chipwhisperer/hardware/naeusb/naeusb.py index 5cc0085e1a..8dc0c93de4 100644 --- a/software/chipwhisperer/hardware/naeusb/naeusb.py +++ b/software/chipwhisperer/hardware/naeusb/naeusb.py @@ -811,7 +811,7 @@ def cmdWriteMem(self, addr : int, data : bytearray): naeusb_logger.debug("FPGA_WRITE: bulk: {}, addr: {:08X}, dlen: {:08X}, response: {}"\ .format("yes" if len(pload) >= NAEUSB_CTRL_IO_THRESHOLD else "no", addr, len(pload), data)) - def write_bulk(self, data : bytearray, timeout = None): + def write_bulk(self, data : bytearray, timeout : Union[int, float, None]=None): """Write data directly to the bulk endpoint. Args: From ef144d1687258917bf433cbeba52ed16e402e8a0 Mon Sep 17 00:00:00 2001 From: Justin Applegate Date: Thu, 15 May 2025 17:56:26 -0600 Subject: [PATCH 4/6] Switched all NAEUSB_Backend functions to snake_case --- .../chipwhisperer/hardware/naeusb/naeusb.py | 38 +++++++++---------- 1 file changed, 19 insertions(+), 19 deletions(-) diff --git a/software/chipwhisperer/hardware/naeusb/naeusb.py b/software/chipwhisperer/hardware/naeusb/naeusb.py index 8dc0c93de4..800416ef84 100644 --- a/software/chipwhisperer/hardware/naeusb/naeusb.py +++ b/software/chipwhisperer/hardware/naeusb/naeusb.py @@ -633,7 +633,7 @@ def get_possible_devices(self, idProduct : Optional[List[int]]=None, dictonly : return dev_list - def sendCtrl(self, cmd : int, value : int=0, data : bytearray=bytearray()): + def send_ctrl(self, cmd : int, value : int=0, data : bytearray=bytearray()): """Send data over control endpoint. Args: @@ -650,7 +650,7 @@ def sendCtrl(self, cmd : int, value : int=0, data : bytearray=bytearray()): self.handle.controlWrite(0x41, cmd, value, 0, data, timeout=self._timeout) #return self.usbdev().ctrl_transfer(0x41, cmd, value, 0, data, timeout=self._timeout) - def readCtrl(self, cmd : int, value : int=0, dlen : int=0) -> bytearray: + def read_ctrl(self, cmd : int, value : int=0, dlen : int=0) -> bytearray: """Read data from control endpoint. Args: @@ -715,11 +715,11 @@ def _cmd_ctrl_send_data(self, pload : bytearray, cmd : int): cmd (int): The command to send. """ try: - self.sendCtrl(cmd, data=pload) + self.send_ctrl(cmd, data=pload) except usb1.USBErrorPipe: naeusb_logger.info("Attempting pipe error fix - typically safe to ignore") - self.sendCtrl(0x22, 0x11) - self.sendCtrl(cmd, data=pload) + self.send_ctrl(0x22, 0x11) + self.send_ctrl(cmd, data=pload) def _cmd_ctrl_send_header(self, addr : int, dlen : int, cmd : int): """Sends the standard length/addr header over the control-transfer endpoint.""" @@ -734,7 +734,7 @@ def _cmd_readmem_ctrl(self, addr : int, dlen : int) -> bytearray: bytearray: The received data. """ self._cmd_ctrl_send_header(addr, dlen, self.CMD_READMEM_CTRL) - return self.readCtrl(self.CMD_READMEM_CTRL, dlen=dlen) + return self.read_ctrl(self.CMD_READMEM_CTRL, dlen=dlen) def _cmd_readmem_bulk(self, addr : int, dlen : int) -> bytearray: """Reads data from the external memory interface over the bulk-transfer endpoint. @@ -745,7 +745,7 @@ def _cmd_readmem_bulk(self, addr : int, dlen : int) -> bytearray: self._cmd_ctrl_send_header(addr, dlen, self.CMD_READMEM_BULK) return self._bulk_read(dlen, None) - def cmdReadMem(self, addr : int, dlen : int) -> bytearray: + def cmd_read_mem(self, addr : int, dlen : int) -> bytearray: """Send command to read over external memory interface from FPGA. It automatically decides to use control-transfer or bulk-endpoint transfer based on @@ -792,7 +792,7 @@ def _cmd_writemem_bulk(self, addr : int, data : bytearray): self._cmd_ctrl_send_header(addr, len(data), self.CMD_WRITEMEM_BULK) self._bulk_write(data, None) - def cmdWriteMem(self, addr : int, data : bytearray): + def cmd_write_mem(self, addr : int, data : bytearray): """Send command to write memory over external memory interface to FPGA. It automatically decides to use control-transfer or bulk-endpoint transfer based @@ -822,7 +822,7 @@ def write_bulk(self, data : bytearray, timeout : Union[int, float, None]=None): naeusb_logger.debug("BULK WRITE: data = {}".format(data)) self._bulk_write(data, timeout) - def flushInput(self): + def flush_input(self): """Dump all the crap left over""" try: # TODO: This probably isn't needed, and causes slow-downs on Mac OS X. @@ -880,7 +880,7 @@ def get_possible_devices(self, idProduct : List[int]) -> usb1.USBDevice: def get_cdc_settings(self) -> list: if self.check_feature("CDC"): - return self.usbtx.readCtrl(self.CMD_CDC_SETTINGS_EN, dlen=4) + return self.usbtx.read_ctrl(self.CMD_CDC_SETTINGS_EN, dlen=4) else: return [0, 0, 0, 0] @@ -903,7 +903,7 @@ def set_cdc_settings(self, port : Tuple=(1, 1, 0, 0)): if self.check_feature("CDC"): if isinstance(port, int): port = (port, port, 0, 0) - self.usbtx.sendCtrl(self.CMD_CDC_SETTINGS_EN, (port[0]) | (port[1] << 1) | (port[2] << 2) | (port[3] << 3)) + self.usbtx.send_ctrl(self.CMD_CDC_SETTINGS_EN, (port[0]) | (port[1] << 1) | (port[2] << 2) | (port[3] << 3)) def set_smc_speed(self, val : int): """ @@ -911,12 +911,12 @@ def set_smc_speed(self, val : int): val = 1: fast read timing, should only be used for reading ADC samples; FPGA must also be set in fast FIFO read mode for this to work correctly. """ - self.usbtx.sendCtrl(self.CMD_SMC_READ_SPEED, data=[val]) + self.usbtx.send_ctrl(self.CMD_SMC_READ_SPEED, data=[val]) def get_fw_build_date(self) -> str: if self.check_feature("SAM_BUILD_DATE"): try: - build_date = bytes(self.usbtx.readCtrl(0x40, dlen=100)).decode() + build_date = bytes(self.usbtx.read_ctrl(0x40, dlen=100)).decode() return build_date except usb1.USBErrorPipe: naeusb_logger.info("Build date unavailable") @@ -927,7 +927,7 @@ def set_husky_tms_wr(self, num): # TODO: add in check_feature if self.check_feature("HUSKY_PIN_CONTROL"): num &= 0xFF - self.usbtx.sendCtrl(0x22, 0x43 | (num << 8)) + self.usbtx.send_ctrl(0x22, 0x43 | (num << 8)) else: naeusb_logger.error("Cannot set Husky TMS direction pin. SWD mode will not work! A firmware update to >=1.4 is highly recommended!") @@ -1019,14 +1019,14 @@ def sendCtrl(self, cmd : int, value : int=0, data : bytearray=bytearray()): Send data over control endpoint """ # Vendor-specific, OUT, interface control transfer - self.usbserializer.sendCtrl(cmd, value, data) + self.usbserializer.send_ctrl(cmd, value, data) def readCtrl(self, cmd : int, value : int=0, dlen : int=0) -> bytearray: """ Read data from control endpoint """ # Vendor-specific, IN, interface control transfer - return self.usbserializer.readCtrl(cmd, value, dlen) + return self.usbserializer.read_ctrl(cmd, value, dlen) def cmdReadMem(self, addr : int, dlen : int) -> bytearray: """ @@ -1034,7 +1034,7 @@ def cmdReadMem(self, addr : int, dlen : int) -> bytearray: decides to use control-transfer or bulk-endpoint transfer based on data length. """ - return self.usbserializer.cmdReadMem(addr, dlen) + return self.usbserializer.cmd_read_mem(addr, dlen) def cmdWriteMem(self, addr : int, data : bytearray): """ @@ -1042,7 +1042,7 @@ def cmdWriteMem(self, addr : int, data : bytearray): decides to use control-transfer or bulk-endpoint transfer based on data length. """ - return self.usbserializer.cmdWriteMem(addr, data) + return self.usbserializer.cmd_write_mem(addr, data) def writeBulkEP(self, data : bytearray, timeout = None): """ @@ -1054,7 +1054,7 @@ def writeBulkEP(self, data : bytearray, timeout = None): def flushInput(self): """Dump all the crap left over""" - self.usbserializer.flushInput() + self.usbserializer.flush_input() class StreamModeCaptureThreadHusky(Thread): def __init__(self, serial, dlen, segment_size, dbuf_temp, timeout_ms=2000, is_husky=False): From bd272ff93cb51e46d05317db0207ab64725ab4b0 Mon Sep 17 00:00:00 2001 From: Justin Applegate Date: Thu, 15 May 2025 18:03:19 -0600 Subject: [PATCH 5/6] Forgot one --- software/chipwhisperer/hardware/naeusb/naeusb.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/software/chipwhisperer/hardware/naeusb/naeusb.py b/software/chipwhisperer/hardware/naeusb/naeusb.py index 800416ef84..043a00b44b 100644 --- a/software/chipwhisperer/hardware/naeusb/naeusb.py +++ b/software/chipwhisperer/hardware/naeusb/naeusb.py @@ -196,7 +196,8 @@ def quick_firmware_erase(product_id : int, serial_number : str=None): Args: product_id (int): The product ID of the device. - serial_number (str, optional): The serial number of the device. If not provided, the function will attempt to find the device by product ID.""" + serial_number (str, optional): The serial number of the device. If not provided, the function will attempt to find the device by product ID. + """ naeusb = NAEUSB() naeusb.con(serial_number=serial_number, idProduct=[product_id]) naeusb.enterBootloader(True) From e523de85e834056ce611c0c9b54f84ee912304f4 Mon Sep 17 00:00:00 2001 From: Justin Applegate Date: Fri, 16 May 2025 17:40:59 -0600 Subject: [PATCH 6/6] Fixed lint errors --- .../chipwhisperer/hardware/naeusb/naeusb.py | 34 ++++++++----------- 1 file changed, 14 insertions(+), 20 deletions(-) diff --git a/software/chipwhisperer/hardware/naeusb/naeusb.py b/software/chipwhisperer/hardware/naeusb/naeusb.py index 043a00b44b..abf51adad2 100644 --- a/software/chipwhisperer/hardware/naeusb/naeusb.py +++ b/software/chipwhisperer/hardware/naeusb/naeusb.py @@ -17,22 +17,14 @@ # limitations under the License. # ========================================================================== import time -import warnings import math from threading import Thread import usb1 # type: ignore import os import sys import array -from typing import Optional, Union, List, Tuple, Dict, cast +from typing import Optional, Union, List, Tuple, Dict, Any from ...common.utils import util -from ...common.utils.util import CWByteArray # type: ignore - -from ..firmware import cwlite as fw_cwlite -from ..firmware import cw1200 as fw_cw1200 -from ..firmware import cw305 as fw_cw305 -from ..firmware import cwnano as fw_nano -from ..firmware import cwhusky as fw_cwhusky from ..firmware.open_fw import fwver @@ -191,7 +183,7 @@ class CWFirmwareError(Exception): } } -def quick_firmware_erase(product_id : int, serial_number : str=None): +def quick_firmware_erase(product_id : int, serial_number : Optional[str]=None): """Quickly erase the firmware on a device by entering bootloader mode. Args: @@ -199,7 +191,7 @@ def quick_firmware_erase(product_id : int, serial_number : str=None): serial_number (str, optional): The serial number of the device. If not provided, the function will attempt to find the device by product ID. """ naeusb = NAEUSB() - naeusb.con(serial_number=serial_number, idProduct=[product_id]) + naeusb.con(serial_number=serial_number, idProduct=(product_id,)) naeusb.enterBootloader(True) def _check_sam_feature(feature : str, fw_version : str, prod_id : int) -> bool: @@ -251,7 +243,7 @@ def _WINDOWS_USB_CHECK_DRIVER(device : usb1.USBDevice) -> Optional[str]: subkey = r"ControlSet001\Enum\USB" subkey += "\\VID_{:04X}&PID_{:04X}".format(device.getVendorID(), device.getProductID()) - def get_enum_by_name(handle : winreg.PyHKEY, name : str): + def get_enum_by_name(handle, name : str): try: cnt = 0 enum_name = "" @@ -262,7 +254,9 @@ def get_enum_by_name(handle : winreg.PyHKEY, name : str): enum_name = myenum[0] cnt += 1 naeusb_logger.debug('Found {}'.format(enum_name)) - return myenum[1] + if myenum is not None: + return myenum[1] + return None except OSError as e: return None @@ -338,7 +332,7 @@ def packuint32(data : int) -> List[int]: data = int(data) return [data & 0xff, (data >> 8) & 0xff, (data >> 16) & 0xff, (data >> 24) & 0xff] -def unpackuint32(buf : List[int]) -> int: +def unpackuint32(buf : Union[List[int], bytearray]) -> int: """"Converts an array into a 32-bit integer.""" pint = buf[0] @@ -671,7 +665,7 @@ def read_ctrl(self, cmd : int, value : int=0, dlen : int=0) -> bytearray: value, 0, dlen, response)) return response - def _get_timeout(self, timeout : Union[int, float, None]=None) -> int: + def _get_timeout(self, timeout : Union[int, float, None]=None) -> Union[int, float, Any]: """Gets the default timeout if the operation caller did not specify one. Returns: @@ -831,18 +825,18 @@ def flush_input(self): except: pass - def read(self, dbuf : bytearray, timeout : int) -> bytearray: + def read(self, dlen : int, timeout : int) -> bytearray: """Read data from the bulk endpoint. Args: - dbuf (bytearray): The buffer to read data into. + dlen (int): The length of the data to read. timeout (int): The timeout for the read operation. Returns: bytearray: The received data. """ - resp = self._bulk_read(dbuf, timeout) - naeusb_logger.debug("BULK READ: data = {}".format(dbuf)) + resp = self._bulk_read(dlen, timeout) + naeusb_logger.debug("BULK READ: data = {}".format(dlen)) return resp class NAEUSB: @@ -1288,7 +1282,7 @@ def cmdReadStream(self, is_husky : bool=False) -> Tuple[int, int]: # Ensure stream mode disabled if not is_husky: - self.sendCtrl(NAEUSB.CMD_MEMSTREAM, data=packuint32(0)) + self.sendCtrl(NAEUSB.CMD_MEMSTREAM, data=bytearray(packuint32(0))) return self.streamModeCaptureStream.drx, self.streamModeCaptureStream.timeout # def readCDCSettings(self):