Linting

hw/production_test/Makefile

@@ -6,7 +6,6 @@ PYTHON_FILES = \
     production_tests.py \
     pybin2nvcm.py \
     pynvcm.py \
-    random_production_test.py \
     reset.py \
     usb_test.py
 
@@ -17,7 +16,7 @@ PYTHON_FILES = \
 lint:
 	autopep8 --in-place --max-line-length 70 --aggressive --aggressive  ${PYTHON_FILES}
 	mypy --disallow-untyped-defs ${PYTHON_FILES}
-	pylint --generated-member=usb1.TRANSFER_COMPLETED,usb1.USBErrorInterrupted,usb1.USBErrorIO ${PYTHON_FILES}
+	pylint --generated-member=usb1.TRANSFER_COMPLETED,usb1.USBErrorInterrupted,usb1.USBErrorIO --max-line-length 70 ${PYTHON_FILES}
 
 # Check that the NVCM generator gives a correct output for a known binary
 verify-pybin2nvcm:

hw/production_test/production_test_runner.py

@@ -2,7 +2,7 @@
 from typing import Any
 import production_tests
 
-pass_msg = '''
+PASS_MSG = '''
   _____                _____    _____
  |  __ \\      /\\      / ____|  / ____|
  | |__) |    /  \\    | (___   | (___
@@ -11,7 +11,7 @@ pass_msg = '''
  |_|      /_/    \\_\\ |_____/  |_____/
  '''
 
-fail_msg = '''
+FAIL_MSG = '''
   ______              _____   _
  |  ____|     /\\     |_   _| | |
  | |__       /  \\      | |   | |
@@ -44,6 +44,16 @@ ANSI = {
 
 
 def run_tests(test_list: list[Any]) -> bool:
+    """ Run a test list
+
+    This executes the tests from the given list in order. The test
+    sequence is stopped if a test raises an assertion, or returns
+    False. If all tests return True successfully, then the test
+    sequence is considered to have passed.
+
+    Keyword parameters:
+    test_list -- List of test functions to call
+    """
     try:
         for test in test_list:
             print("\n{:}Test step: {:}{:} ({:})".format(
@@ -66,12 +76,8 @@ def run_tests(test_list: list[Any]) -> bool:
     return True
 
 
-if __name__ == '__main__':
+def production_test_runner() -> None:
-    last_a = '1'
+    """Interactive production test environment"""
-
-    # Allow any of the settings in the parameters structure to be
-    # overridden
-    import argparse
     parser = argparse.ArgumentParser()
     parser.add_argument(
         '-l',
@@ -100,7 +106,7 @@ if __name__ == '__main__':
                 [test.__name__ for test in test_list])))
         for test in production_tests.manual_tests:
             print('{:}: {:}'.format(test.__name__, test.__doc__))
-        exit(0)
+        sys.exit(0)
 
     if args.run_test is not None:
         result = False
@@ -118,17 +124,17 @@ if __name__ == '__main__':
                     break
 
         if not found:
-            print('Test not found:{:}'.format(args.run_test))
+            raise ValueError('Test not found:{args.run_test}')
-            exit(1)
 
         if not result:
             production_tests.reset()
-            exit(1)
+            sys.exit(1)
 
-        exit(0)
+        sys.exit(0)
 
     print('\n\nProduction test system')
 
+    last_a = '1'
     while True:
         print('\n\n')
 
@@ -161,17 +167,24 @@ if __name__ == '__main__':
 
         try:
             test_sequence = options[int(a) - 1]
-        except IndexError as e:
+        except IndexError:
             print('Invalid input')
             continue
-        except ValueError as e:
+        except ValueError:
             print('Invalid input')
             continue
 
         if not run_tests(test_sequence):
-            print(ANSI['bg_red'] + fail_msg + ANSI['reset'])
+            print(ANSI['bg_red'] + FAIL_MSG + ANSI['reset'])
             production_tests.reset()
         else:
-            print(ANSI['bg_green'] + pass_msg + ANSI['reset'])
+            print(ANSI['bg_green'] + PASS_MSG + ANSI['reset'])
 
         last_a = a
+
+
+if __name__ == '__main__':
+    import argparse
+    import sys
+
+    production_test_runner()

hw/production_test/production_tests.py

@@ -54,7 +54,7 @@ parameters = {
     'pico_bootloader_target_dir': '/media/lab/RPI-RP2/'
 }
 
-tp1_pins = {
+TP1_PINS = {
     '5v_en': 7,
     'tx': 8,
     'rx': 9,
@@ -75,9 +75,9 @@ tp1_pins = {
 
 def enable_power() -> bool:
     """Enable power to the TK-1"""
-    d = IceFlasher()
+    flasher = IceFlasher()
-    d.gpio_set_direction(tp1_pins['5v_en'], True)
+    flasher.gpio_set_direction(TP1_PINS['5v_en'], True)
-    d.gpio_put(tp1_pins['5v_en'], True)
+    flasher.gpio_put(TP1_PINS['5v_en'], True)
     time.sleep(0.3)
 
     return True
@@ -86,9 +86,9 @@ def enable_power() -> bool:
 def disable_power() -> bool:
     """Disable power to the TK-1"""
     time.sleep(.1)
-    d = IceFlasher()
+    flasher = IceFlasher()
-    d.gpio_set_direction(tp1_pins['5v_en'], True)
+    flasher.gpio_set_direction(TP1_PINS['5v_en'], True)
-    d.gpio_put(tp1_pins['5v_en'], False)
+    flasher.gpio_put(TP1_PINS['5v_en'], False)
 
     return True
 
@@ -102,7 +102,7 @@ def measure_voltages(device: IceFlasher,
     sample_count -- number of samples to average
     """
     adc_vals = [0.0, 0.0, 0.0]
-    for i in range(0, sample_count):
+    for _ in range(0, sample_count):
         adc_vals = [
             total + sample for total,
             sample in zip(
@@ -119,18 +119,16 @@ def voltage_test() -> bool:
 
     enable_power()
 
-    d = IceFlasher()
+    flasher = IceFlasher()
-    vals = measure_voltages(d, 20)
+    vals = measure_voltages(flasher, 20)
-    d.close()
+    flasher.close()
 
     disable_power()
 
     print(
         'voltages:',
         ', '.join(
-            '{:}V:{:.3f}'.format(
+            f'{val[0]}V:{val[1]:.3f}' for val in vals.items()))
-                val[0],
-                val[1]) for val in vals.items()))
     if (
         (abs(vals['1.2'] - 1.2) > .2)
         | (abs(vals['2.5'] - 2.5) > .2)
@@ -142,12 +140,13 @@ def voltage_test() -> bool:
 
 
 def flash_validate_id() -> bool:
-    """Read the ID from TK-1 SPI flash, and verify that it matches the expected value"""
+    """Read the ID from TK-1 SPI flash, and compare to known values"""
     result = run([
         parameters['iceprog'],
         '-t'
     ],
-        capture_output=True)
+        capture_output=True,
+        check=True)
     disable_power()
 
     err = result.stderr.split(b'\n')
@@ -167,7 +166,7 @@ def flash_validate_id() -> bool:
             if flash_type is None:
                 print('Flash ID invalid')
                 return False
-            print('Detected flash type: {:}'.format(flash_type))
+            print(f'Detected flash type: {flash_type}')
             return True
 
     return result.returncode == 0
@@ -178,7 +177,8 @@ def flash_program() -> bool:
     result = run([
         parameters['iceprog'],
         parameters['app_gateware']
-    ])
+    ],
+        check=True)
     disable_power()
     print(result)
 
@@ -191,7 +191,8 @@ def flash_check() -> bool:
         parameters['iceprog'],
         '-c',
         parameters['app_gateware']
-    ])
+    ],
+        check=True)
     disable_power()
     print(result)
 
@@ -199,12 +200,12 @@ def flash_check() -> bool:
 
 
 def test_extra_io() -> bool:
-    """Test the TK-1 RTS, CTS, and GPIO1-4 lines by measuring a test pattern generated by the app_test gateware"""
+    """Test the TK-1 RTS, CTS, and GPIO1-4 lines"""
 
-    d = IceFlasher()
+    flasher = IceFlasher()
-    for pin in tp1_pins.values():
+    for pin in TP1_PINS.values():
-        d.gpio_set_direction(pin, False)
+        flasher.gpio_set_direction(pin, False)
-    d.close()
+    flasher.close()
 
     disable_power()
     time.sleep(1)
@@ -212,24 +213,24 @@ def test_extra_io() -> bool:
 
     time.sleep(0.2)
 
-    d = IceFlasher()
+    flasher = IceFlasher()
-    d.gpio_put(tp1_pins['rts'], False)
+    flasher.gpio_put(TP1_PINS['rts'], False)
-    d.gpio_set_direction(tp1_pins['rts'], True)
+    flasher.gpio_set_direction(TP1_PINS['rts'], True)
 
     expected_results = [1 << (i % 5) for i in range(9, -1, -1)]
 
     results = []
-    for i in range(0, 10):
+    for _ in range(0, 10):
-        vals = d.gpio_get_all()
+        vals = flasher.gpio_get_all()
         pattern = (vals >> 17) & 0b11111
         # print(f'{vals:016x} {pattern:04x}')
         results.append(pattern)
 
-        d.gpio_put(tp1_pins['rts'], True)
+        flasher.gpio_put(TP1_PINS['rts'], True)
-        d.gpio_put(tp1_pins['rts'], False)
+        flasher.gpio_put(TP1_PINS['rts'], False)
 
-    d.gpio_set_direction(tp1_pins['rts'], False)
+    flasher.gpio_set_direction(TP1_PINS['rts'], False)
-    d.close()
+    flasher.close()
 
     disable_power()
 
@@ -269,22 +270,22 @@ def inject_serial_number(
         "68de5d27-e223-4874-bc76-a54d6e84068f")
     replacement = encode_usb_strings.string_to_descriptor(serial_num)
 
-    f = bytearray(open(infile, 'rb').read())
+    with open(infile, 'rb') as fin:
+        firmware_data = bytearray(fin.read())
 
-    pos = f.find(magic)
+    pos = firmware_data.find(magic)
 
     if pos < 0:
-        print('failed to find magic string')
+        raise ValueError('failed to find magic string')
-        exit(1)
 
-    f[pos:(pos + len(magic))] = replacement
+    firmware_data[pos:(pos + len(magic))] = replacement
 
-    with open(outfile, 'wb') as of:
+    with open(outfile, 'wb') as fout:
-        of.write(f)
+        fout.write(firmware_data)
 
 
 def flash_ch552(serial_num: str) -> bool:
-    """Flash an attached CH552 device with the USB CDC firmware, injected with the given serial number"""
+    """Flash an attached CH552 device with the USB CDC firmware"""
 
     print(serial_num)
     inject_serial_number(
@@ -296,7 +297,8 @@ def flash_ch552(serial_num: str) -> bool:
     result = run([
         parameters['chprog'],
         parameters['ch552_firmware_injected']
-    ])
+    ],
+        check=True)
     print(result)
     return result.returncode == 0
 
@@ -308,7 +310,8 @@ def erase_ch552() -> bool:
     result = run([
         parameters['chprog'],
         parameters['ch552_firmware_blank']
-    ])
+    ],
+        check=True)
     print(result)
     return result.returncode == 0
 
@@ -330,7 +333,7 @@ def find_serial_device(desc: dict[str, Any]) -> str:
 
 
 def find_ch552(serial_num: str) -> bool:
-    """Search all serial devices for one that has the correct description and serial number"""
+    """Search for a serial device that has the correct description"""
     time.sleep(1)
 
     description = {
@@ -350,18 +353,18 @@ def find_ch552(serial_num: str) -> bool:
 
 
 def ch552_program() -> bool:
-    """Load the CDC ACM firmware onto a CH552 with a randomly generated serial number, and verify that it boots correctly"""
+    """Load the firmware onto a CH552, and verify that it boots"""
     if not test_found_bootloader():
         print('Error finding CH552!')
         return False
 
-    serial = str(uuid.uuid4())
+    serial_num = str(uuid.uuid4())
 
-    if not flash_ch552(serial):
+    if not flash_ch552(serial_num):
         print('Error flashing CH552!')
         return False
 
-    if not find_ch552(serial):
+    if not find_ch552(serial_num):
         print('Error finding flashed CH552!')
         return False
 
@@ -382,7 +385,7 @@ def ch552_erase() -> bool:
 
 
 def test_txrx_touchpad() -> bool:
-    """Test UART communication, RGB LED, and touchpad by asking the operator to interact with the touch pad"""
+    """Test UART communication, RGB LED, and touchpad"""
     description = {
         'vid': 0x1207,
         'pid': 0x8887,
@@ -390,36 +393,33 @@ def test_txrx_touchpad() -> bool:
         'product': 'MTA1-USB-V1'
     }
 
-    s = serial.Serial(
+    dev = serial.Serial(
         find_serial_device(description),
         9600,
         timeout=.2)
 
-    if not s.isOpen():
+    if not dev.isOpen():
         print('couldn\'t find/open serial device')
         return False
 
     for _ in range(0, 5):
         # Attempt to clear any buffered data from the serial port
-        s.write(b'0123')
+        dev.write(b'0123')
         time.sleep(0.2)
-        s.read(20)
+        dev.read(20)
 
         try:
-            s.write(b'0')
+            dev.write(b'0')
-            [count, touch_count] = s.read(2)
+            [count, touch_count] = dev.read(2)
             print(
-                'read count:{:}, touch count:{:}'.format(
+                f'read count:{count}, touch count:{touch_count}')
-                    count, touch_count))
 
             input(
                 '\n\n\nPress touch pad once and check LED, then press Enter')
-            s.write(b'0')
+            dev.write(b'0')
-            [count_post, touch_count_post] = s.read(2)
+            [count_post, touch_count_post] = dev.read(2)
             print(
-                'read count:{:}, touch count:{:}'.format(
+                'read count:{count_post}, touch count:{touch_count_post}')
-                    count_post,
-                    touch_count_post))
 
             if (count_post -
                 count != 1) or (touch_count_post -
@@ -428,8 +428,8 @@ def test_txrx_touchpad() -> bool:
                 continue
 
             return True
-        except ValueError as e:
+        except ValueError as error:
-            print(e)
+            print(error)
             continue
 
     print('Max retries exceeded, failure!')
@@ -450,9 +450,6 @@ def program_pico() -> bool:
                 parameters['pico_bootloader_target_dir'] +
                 firmware_filename)
 
-            # TODO: Test if the pico identifies as a USB-HID device
-            # after programming
-
             return True
         except FileNotFoundError:
             time.sleep(0.1)
@@ -514,35 +511,27 @@ def reset() -> None:
     """Attempt to reset the board after test failure"""
     try:
         disable_power()
-    except AttributeError as e:
+    except AttributeError:
         pass
-    except OSError as e:
+    except OSError:
         pass
-    except ValueError as e:
+    except ValueError:
         pass
 
 
-if __name__ == "__main__":
+def random_test_runner() -> None:
-    # Runs the non-interactive production tests continuously in a
+    """"Run the non-interactive production tests in a random order
-    # random order, to look for interaction bugs
 
-    import random
+    This routine is intended to be used for finding edge-cases with
-    import pynvcm
+    the production tests. It runs the non-interactive tests (as well
+    as some nondestructive tests from the nvcm module) in a random
+    order, and runs continuously.
+    """
 
     def nvcm_read_info() -> bool:
-        tp1_pins = {
+        """Check that the nvcm read info command runs"""
-            '5v_en': 7,
+        pynvcm.sleep_flash(TP1_PINS, 15)
-            'sck': 10,
+        nvcm = pynvcm.Nvcm(TP1_PINS, 15)
-            'mosi': 11,
-            'ss': 12,
-            'miso': 13,
-            'crst': 14,
-            'cdne': 15
-        }
-    
-        pynvcm.sleep_flash(tp1_pins, 15)
-    
-        nvcm = pynvcm.Nvcm(tp1_pins, 15)
         nvcm.power_on()
         nvcm.init()
         nvcm.nvcm_enable()
@@ -551,19 +540,9 @@ if __name__ == "__main__":
         return True
 
     def nvcm_verify_blank() -> bool:
-        tp1_pins = {
+        """Verify that the NVCM memory is blank"""
-            '5v_en': 7,
+        pynvcm.sleep_flash(TP1_PINS, 15)
-            'sck': 10,
+        nvcm = pynvcm.Nvcm(TP1_PINS, 15)
-            'mosi': 11,
-            'ss': 12,
-            'miso': 13,
-            'crst': 14,
-            'cdne': 15
-        }
-    
-        pynvcm.sleep_flash(tp1_pins, 15)
-    
-        nvcm = pynvcm.Nvcm(tp1_pins, 15)
         nvcm.power_on()
         nvcm.init()
         nvcm.nvcm_enable()
@@ -571,7 +550,6 @@ if __name__ == "__main__":
         nvcm.power_off()
         return True
 
-    
     tests = [
         nvcm_read_info,
         nvcm_verify_blank,
@@ -586,8 +564,18 @@ if __name__ == "__main__":
 
     parameters['iceprog'] = '/home/matt/repos/tillitis--icestorm/iceprog/iceprog'
 
+    pass_count = 0
     while True:
         i = random.randint(0, (len(tests) - 1))
-        print(f'\n\n{i} running: {tests[i].__name__}')
+        print(f'\n\n{pass_count}: running: {tests[i].__name__}')
         if not tests[i]():
-            raise Exception('oops')
+            sys.exit(1)
+        pass_count += 1
+
+
+if __name__ == "__main__":
+    import random
+    import pynvcm
+    import sys
+
+    random_test_runner()

hw/production_test/pynvcm.py

@@ -34,6 +34,13 @@ def assert_bytes_equal(
         name: str,
         expected: bytes,
         val: bytes) -> None:
+    """ Check if two bytes objects are equal
+
+    Keyword arguments:
+    name -- Description to print if the assertion fails
+    expected -- Expected value
+    val -- Value to check
+    """
     if expected != val:
         expected_str = ' '.join([f'{x:02x}' for x in expected])
         val_str = ' '.join([f'{x:02x}' for x in val])
@@ -103,40 +110,40 @@ class Nvcm():
         """Disable power to the DUT"""
         self.flasher.gpio_put(self.pins['5v_en'], False)
 
-    def enable(self, cs: bool, reset: bool = True) -> None:
+    def enable(self, chip_select: bool, reset: bool = True) -> None:
         """Set the CS and Reset pin states"""
-        self.flasher.gpio_put(self.pins['ss'], cs)
+        self.flasher.gpio_put(self.pins['ss'], chip_select)
         self.flasher.gpio_put(self.pins['crst'], reset)
 
-    def writehex(self, s: str, toggle_cs: bool = True) -> None:
+    def writehex(self, hex_data: str, toggle_cs: bool = True) -> None:
         """Write SPI data to the target device
 
         Keyword arguments:
-        s -- data to send (formatted as a string of hex data)
+        hex_data -- data to send (formatted as a string of hex data)
         toggle_cs -- If true, automatically lower the CS pin before
                      transmit, and raise it after transmit
         """
-        if self.debug and not s == "0500":
+        if self.debug and not hex_data == "0500":
-            print("TX", s)
+            print("TX", hex_data)
-        data = bytes.fromhex(s)
+        data = bytes.fromhex(hex_data)
 
         self.flasher.spi_write(data, toggle_cs)
 
-    def sendhex(self, s: str) -> bytes:
+    def sendhex(self, hex_data: str) -> bytes:
         """Perform a full-duplex write/read on the target device
 
         Keyword arguments:
         s -- data to send (formatted as a string of hex data)
         """
-        if self.debug and not s == "0500":
+        if self.debug and not hex_data == "0500":
-            print("TX", s)
+            print("TX", hex_data)
-        x = bytes.fromhex(s)
+        bytes_data = bytes.fromhex(hex_data)
 
-        b = self.flasher.spi_rxtx(x)
+        ret = self.flasher.spi_rxtx(bytes_data)
 
-        if self.debug and not s == "0500":
+        if self.debug and not hex_data == "0500":
-            print("RX", b.hex())
+            print("RX", ret.hex())
-        return b
+        return ret
 
     def delay(self, count: int) -> None:
         """'Delay' by sending clocks with CS de-asserted
@@ -151,16 +158,16 @@ class Nvcm():
         """Reboot the part and enter SPI command mode"""
         if self.debug:
             print("init")
-        self.enable(cs=True, reset=True)
+        self.enable(True, True)
-        self.enable(cs=True, reset=False)
+        self.enable(True, False)
-        self.enable(cs=False, reset=False)
+        self.enable(False, False)
-        self.enable(cs=False, reset=True)
+        self.enable(False, True)
         sleep(0.1)
-        self.enable(cs=True, reset=True)
+        self.enable(True, True)
 
     def status_wait(self) -> None:
         """Wait for the status register to clear"""
-        for i in range(0, 1000):
+        for _ in range(0, 1000):
             self.delay(1250)
             ret = self.sendhex("0500")
             status = struct.unpack('>H', ret)[0]
@@ -206,7 +213,7 @@ class Nvcm():
         """
 
         msg = ''
-        msg += ("%02x%06x" % (cmd, address))
+        msg += (f"{cmd:02x}{address:06x}")
         msg += ("00" * 9)  # dummy bytes
         msg += ("00" * length)  # read
         ret = self.sendhex(msg)
@@ -237,14 +244,14 @@ class Nvcm():
         address -- NVCM memory address to write to
         length -- Number of bytes to write
         """
-        self.writehex("%02x%06x" % (cmd, address) + data)
+        self.writehex(f"{cmd:02x}{address:06x}" + data)
 
         try:
             self.status_wait()
         except Exception as exc:
-            raise Exception(
+            raise IOError(
-                "WRITE FAILED: cmd=%02x address=%06x data=%s" %
+                f"WRITE FAILED: cmd={cmd:02x} address={address:%06x} data={data}"
-                (cmd, address, data)) from exc
+            ) from exc
 
         self.delay(2)
 
@@ -371,7 +378,7 @@ class Nvcm():
             try:
                 self.command(row)
             except Exception as exc:
-                raise Exception(
+                raise IOError(
                     "programming failed, row:{row}"
                 ) from exc
 
@@ -503,8 +510,8 @@ class Nvcm():
 
         self.bank_select('nvcm')
 
-        contents = bytearray()
+        # contents = bytearray()
-
+        #
         # for offset in range(0, length, 8):
         #    if offset % (1024 * 8) == 0:
         #        print("%6d / %6d bytes" % (offset, length))
@@ -761,6 +768,6 @@ if __name__ == "__main__":
 
     if args.do_boot:
         # hold reset low for half a second
-        nvcm.enable(cs=True, reset=False)
+        nvcm.enable(True, False)
         sleep(0.5)
-        nvcm.enable(cs=True, reset=True)
+        nvcm.enable(True, True)

hw/production_test/usb_test.py

@@ -60,14 +60,17 @@ class IceFlasher:
         bcd_device = self.handle.getDevice().getbcdDevice()
         if bcd_device != 0x0200:
             raise ValueError(
-                'Pico firmware version out of date- please upgrade it!')
+                'Pico firmware version out of date- please upgrade')
 
         self.handle.claimInterface(0)
 
+        self.cs_pin = -1
+
     def __del__(self) -> None:
         self.close()
 
     def close(self) -> None:
+        """ Release the USB device handle """
         self._wait_async()
 
         if self.handle is not None: