tigertool: cleanup

Fix serial number detect
new python3.9 incompatibilities.
cros lint to the extent possible

BUG=b:216199797
TEST=./tigertest.py -c 10; servo_updater -b servo_v4
BRANCH=none

Signed-off-by: Nick Sanders <nsanders@chromium.org>
Change-Id: I2ec628389f63711a540223fc9004c9fff7155c1e
Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/platform/ec/+/3615478
Reviewed-by: Ruben Rodriguez Buchillon <coconutruben@chromium.org>

4 files changed, 117 insertions, 123 deletions

diff --git a/extra/tigertool/ecusb/tiny_servo_common.py b/extra/tigertool/ecusb/tiny_servo_common.py
index e27736a9dc..726e2e64b7 100644
--- a/extra/tigertool/ecusb/tiny_servo_common.py
+++ b/extra/tigertool/ecusb/tiny_servo_common.py
@@ -11,15 +11,11 @@
 # Note: This is a py2/3 compatible file.
 
 import datetime
-import errno
-import os
-import re
-import subprocess
 import sys
 import time
-import usb
 
 import six
+import usb
 
 from . import pty_driver
 from . import stm32uart
@@ -52,20 +48,13 @@ def check_usb(vidpid, serialname=None):
     vidpid: string representation of the usb vid:pid, eg. '18d1:2001'
     serialname: serialname if specified.
 
-  Returns: True if found, False, otherwise.
+  Returns:
+    True if found, False, otherwise.
   """
-  if serialname:
-    output = subprocess.check_output(['lsusb', '-v', '-d', vidpid],
-                                     **get_subprocess_args())
-    m = re.search(r'^\s*iSerial\s+\d+\s+%s$' % serialname, output, flags=re.M)
-    if m:
-      return True
+  if get_usb_dev(vidpid, serialname):
+    return True
 
-    return False
-  else:
-    if subprocess.call(['lsusb', '-d', vidpid], stdout=open('/dev/null', 'w')):
-      return False
-  return True
+  return False
 
 def check_usb_sn(vidpid):
   """Return the serial number
@@ -77,13 +66,15 @@ def check_usb_sn(vidpid):
   Args:
     vidpid: string representation of the usb vid:pid, eg. '18d1:2001'
 
-  Returns: string serial number if found, None otherwise.
+  Returns:
+    string serial number if found, None otherwise.
   """
-  output = subprocess.check_output(['lsusb', '-v', '-d', vidpid],
-                                   **get_subprocess_args())
-  m = re.search(r'^\s*iSerial\s+(.*)$', output, flags=re.M)
-  if m:
-    return m.group(1)
+  dev = get_usb_dev(vidpid)
+
+  if dev:
+    dev_serial = usb.util.get_string(dev, dev.iSerialNumber)
+
+    return dev_serial
 
   return None
 
@@ -98,13 +89,13 @@ def get_usb_dev(vidpid, serialname=None):
     vidpid: string representation of the usb vid:pid, eg. '18d1:2001'
     serialname: serialname if specified.
 
-  Returns: pyusb device if found, None otherwise.
+  Returns:
+    pyusb device if found, None otherwise.
   """
   vidpidst = vidpid.split(':')
   vid = int(vidpidst[0], 16)
   pid = int(vidpidst[1], 16)
 
-
   dev_g = usb.core.find(idVendor=vid, idProduct=pid, find_all=True)
   dev_list = list(dev_g)
 
@@ -140,7 +131,8 @@ def check_usb_dev(vidpid, serialname=None):
     vidpid: string representation of the usb vid:pid, eg. '18d1:2001'
     serialname: serialname if specified.
 
-  Returns: usb device number if found, None otherwise.
+  Returns:
+    usb device number if found, None otherwise.
   """
   dev = get_usb_dev(vidpid, serialname=serialname)
 
@@ -173,7 +165,7 @@ def wait_for_usb(vidpid, serialname=None, timeout=None, desiredpresence=True):
   if timeout:
     finish = datetime.datetime.now() + datetime.timedelta(seconds=timeout)
   while check_usb(vidpid, serialname) != desiredpresence:
-    time.sleep(.01)
+    time.sleep(.1)
     if timeout:
       if datetime.datetime.now() > finish:
         raise TinyServoError('Timeout', 'Timeout waiting for USB %s' % vidpid)
@@ -194,8 +186,8 @@ def do_serialno(serialno, pty):
     TinyServoError: on failure to set.
     ptyError: on command interface error.
   """
-  cmd = 'serialno set %s' % serialno
-  regex = 'Serial number:\s+(\S+)'
+  cmd = r'serialno set %s' % serialno
+  regex = r'Serial number:\s+(\S+)'
 
   results = pty._issue_cmd_get_results(cmd, [regex])[0]
   sn = results[1].strip().strip('\n\r')
@@ -221,7 +213,8 @@ def setup_tinyservod(vidpid, interface, serialname=None, debuglog=False):
     serialname: string serial name of device requested, optional.
     debuglog: chatty printout (boolean)
 
-  Returns: pty object
+  Returns:
+    pty object
 
   Raises:
     UsbError, SusbError: on device not found
diff --git a/extra/tigertool/make_pkg.sh b/extra/tigertool/make_pkg.sh
index 5a63862242..12db65afbd 100755
--- a/extra/tigertool/make_pkg.sh
+++ b/extra/tigertool/make_pkg.sh
@@ -3,6 +3,8 @@
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
+set -e
+
 # Make sure we are in the correct dir.
 cd "$( dirname "${BASH_SOURCE[0]}" )" || exit
 
@@ -21,10 +23,11 @@ cp tigertest.py "${DEST}"
 cp README.md "${DEST}"
 
 cp -r ecusb "${DEST}"
-cp -r ../../../../../chroot/usr/lib64/python2.7/site-packages/usb "${DEST}"
+# Not compatible with glinux as of 4/28/2022.
+# cp -r ../../../../../chroot/usr/lib64/python3.6/site-packages/usb "${DEST}"
 find "${DEST}" -name "*.py[co]" -delete
 cp -r ../usb_serial "${DEST}"
 
-(cd build; tar -czf tigertool_${DATE}.tgz tigertool)
+(cd build && tar -czf tigertool_"${DATE}".tgz tigertool)
 
 echo "Done packaging tigertool_${DATE}.tgz"
diff --git a/extra/usb_serial/console.py b/extra/usb_serial/console.py
index d06b33ce23..7211dceff6 100755
--- a/extra/usb_serial/console.py
+++ b/extra/usb_serial/console.py
@@ -1,4 +1,4 @@
-#!/usr/bin/env python
+#!/usr/bin/env python3
 # Copyright 2016 The Chromium OS Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
@@ -19,15 +19,14 @@ import sys
 import termios
 import threading
 import time
-import traceback
 import tty
 try:
   import usb
-except:
-  print("import usb failed")
-  print("try running these commands:")
-  print(" sudo apt-get install python-pip")
-  print(" sudo pip install --pre pyusb")
+except ModuleNotFoundError:
+  print('import usb failed')
+  print('try running these commands:')
+  print(' sudo apt-get install python-pip')
+  print(' sudo pip install --pre pyusb')
   print()
   sys.exit(-1)
 
@@ -77,9 +76,9 @@ class Susb():
     Discovers and connects to USB endpoints.
 
     Args:
-      vendor    : usb vendor id of device
-      product   : usb product id of device
-      interface : interface number ( 1 - 8 ) of device to use
+      vendor: usb vendor id of device
+      product: usb product id of device
+      interface: interface number ( 1 - 8 ) of device to use
       serialname: string of device serialnumber.
 
     Raises:
@@ -89,7 +88,7 @@ class Susb():
     dev_g = usb.core.find(idVendor=vendor, idProduct=product, find_all=True)
     dev_list = list(dev_g)
     if dev_list is None:
-      raise SusbError("USB device not found")
+      raise SusbError('USB device not found')
 
     # Check if we have multiple devices.
     dev = None
@@ -104,7 +103,7 @@ class Susb():
           dev = d
           break
       if dev is None:
-        raise SusbError("USB device(%s) not found" % (serialname,))
+        raise SusbError('USB device(%s) not found' % (serialname,))
     else:
       try:
         dev = dev_list[0]
@@ -112,7 +111,7 @@ class Susb():
         try:
           dev = dev_list.next()
         except:
-          raise SusbError("USB device %04x:%04x not found" % (vendor, product))
+          raise SusbError('USB device %04x:%04x not found' % (vendor, product))
 
     # If we can't set configuration, it's already been set.
     try:
@@ -126,7 +125,7 @@ class Susb():
     self._intf = intf
 
     if not intf:
-      raise SusbError("Interface not found")
+      raise SusbError('Interface not found')
 
     # Detach raiden.ko if it is loaded.
     if dev.is_kernel_driver_active(intf.bInterfaceNumber) is True:
@@ -181,7 +180,7 @@ class Suart():
     """
     self._done = threading.Event()
     self._susb = Susb(vendor=vendor, product=product,
-        interface=interface, serialname=serialname)
+                      interface=interface, serialname=serialname)
 
   def wait_until_done(self, timeout=None):
     return self._done.wait(timeout=timeout)
@@ -199,7 +198,7 @@ class Suart():
             # If we miss some characters on pty disconnect, that's fine.
             # ep.read() also throws USBError on timeout, which we discard.
             if not isinstance(e, (OSError, usb.core.USBError)):
-              print("rx %s" % e)
+              print('rx %s' % e)
     finally:
       self._done.set()
 
@@ -208,19 +207,18 @@ class Suart():
       while True:
           try:
             r = GetBuffer(sys.stdin).read(1)
-            if not r or r == b"\x03":
+            if not r or r == b'\x03':
               break
             if r:
               self._susb._write_ep.write(array.array('B', r),
                                          self._susb.TIMEOUT_MS)
           except Exception as e:
-            print("tx %s" % e)
+            print('tx %s' % e)
     finally:
       self._done.set()
 
   def run(self):
-    """Creates pthreads to poll USB & PTY for data.
-    """
+    """Creates pthreads to poll USB & PTY for data."""
     self._exit = False
 
     self._rx_thread = threading.Thread(target=self.run_rx_thread)
@@ -239,18 +237,17 @@ class Suart():
   Ctrl-C exits.
 """
 
-parser = argparse.ArgumentParser(description="Open a console to a USB device")
+parser = argparse.ArgumentParser(description='Open a console to a USB device')
 parser.add_argument('-d', '--device', type=str,
-    help="vid:pid of target device", default="18d1:501c")
+                    help='vid:pid of target device', default='18d1:501c')
 parser.add_argument('-i', '--interface', type=int,
-    help="interface number of console", default=0)
+                    help='interface number of console', default=0)
 parser.add_argument('-s', '--serialno', type=str,
-    help="serial number of device", default="")
+                    help='serial number of device', default='')
 parser.add_argument('-S', '--notty-exit-sleep', type=float, default=0.2,
-    help="When stdin is *not* a TTY, wait this many seconds after EOF from "
-    "stdin before exiting, to give time for receiving a reply from the USB "
-    "device.")
-
+                    help='When stdin is *not* a TTY, wait this many seconds '
+                    'after EOF from stdin before exiting, to give time for '
+                    'receiving a reply from the USB device.')
 
 def runconsole():
   """Run the usb console code
@@ -280,7 +277,7 @@ def main():
   stdin_isatty = sys.stdin.isatty()
   if stdin_isatty:
     fd = sys.stdin.fileno()
-    os.system("stty -echo")
+    os.system('stty -echo')
     old_settings = termios.tcgetattr(fd)
 
   try:
@@ -288,7 +285,7 @@ def main():
   finally:
     if stdin_isatty:
       termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
-      os.system("stty echo")
+      os.system('stty echo')
     # Avoid having the user's shell prompt start mid-line after the final output
     # from this program.
     print()
diff --git a/extra/usb_updater/servo_updater.py b/extra/usb_updater/servo_updater.py
index fa0d21670c..432ee120de 100755
--- a/extra/usb_updater/servo_updater.py
+++ b/extra/usb_updater/servo_updater.py
@@ -9,15 +9,15 @@
 
 # Note: This is a py2/3 compatible file.
 
+"""USB updater tool for servo and similar boards."""
+
 from __future__ import print_function
 
 import argparse
-import errno
 import os
 import re
 import subprocess
 import time
-import tempfile
 
 import json
 
@@ -62,14 +62,15 @@ TEST_IMAGE_BASE_PATH = '/usr/local/'
 
 COMMON_PATH = 'share/servo_updater'
 
-FIRMWARE_DIR = "firmware/"
-CONFIGS_DIR = "configs/"
+FIRMWARE_DIR = 'firmware/'
+CONFIGS_DIR = 'configs/'
 
 RETRIES_COUNT = 10
 RETRIES_DELAY = 1
 
 def do_with_retries(func, *args):
-  """
+  """Try a function several times
+
   Call function passed as argument and check if no error happened.
   If exception was raised by function,
   it will be retried up to RETRIES_COUNT times.
@@ -88,21 +89,21 @@ def do_with_retries(func, *args):
     try:
       return func(*args)
     except Exception as e:
-      print("Retrying function %s: %s" % (func.__name__, e))
+      print('Retrying function %s: %s' % (func.__name__, e))
       retry = retry + 1
       time.sleep(RETRIES_DELAY)
       continue
 
-  raise Exception("'{}' failed after {} retries".format(func.__name__, RETRIES_COUNT))
+  raise Exception("'{}' failed after {} retries".format(
+      func.__name__, RETRIES_COUNT))
 
 def flash(brdfile, serialno, binfile):
-  """
-  Call fw_update to upload to updater USB endpoint.
+  """Call fw_update to upload to updater USB endpoint.
 
   Args:
-    brdfile:  path to board configuration file
+    brdfile: path to board configuration file
     serialno: device serial number
-    binfile:  firmware file
+    binfile: firmware file
   """
 
   p = fw_update.Supdate()
@@ -113,29 +114,28 @@ def flash(brdfile, serialno, binfile):
   # Start transfer and erase.
   p.start()
   # Upload the bin file
-  print("Uploading %s" % binfile)
+  print('Uploading %s' % binfile)
   p.write_file()
 
   # Finalize
-  print("Done. Finalizing.")
+  print('Done. Finalizing.')
   p.stop()
 
 def flash2(vidpid, serialno, binfile):
-  """
-  Call fw update via usb_updater2 commandline.
+  """Call fw update via usb_updater2 commandline.
 
   Args:
-    vidpid:   vendor id and product id of device
+    vidpid: vendor id and product id of device
     serialno: device serial number (optional)
-    binfile:  firmware file
+    binfile: firmware file
   """
 
   tool = 'usb_updater2'
-  cmd = "%s -d %s" % (tool, vidpid)
+  cmd = '%s -d %s' % (tool, vidpid)
   if serialno:
-    cmd += " -S %s" % serialno
-  cmd += " -n"
-  cmd += " %s" % binfile
+    cmd += ' -S %s' % serialno
+  cmd += ' -n'
+  cmd += ' %s' % binfile
 
   print(cmd)
   help_cmd = '%s --help' % tool
@@ -151,27 +151,28 @@ def flash2(vidpid, serialno, binfile):
   if res in (0, 1, 2):
     return res
   else:
-    raise ServoUpdaterException("%s exit with res = %d" % (cmd, res))
+    raise ServoUpdaterException('%s exit with res = %d' % (cmd, res))
 
 def select(tinys, region):
-  """
+  """Jump to specified boot region
+
   Ensure the servo is in the expected ro/rw region.
   This function jumps to the required region and verify if jump was
   successful by executing 'sysinfo' command and reading current region.
   If response was not received or region is invalid, exception is raised.
 
   Args:
-    tinys:  TinyServod object
+    tinys: TinyServod object
     region: region to jump to, only "rw" and "ro" is allowed
   """
 
-  if region not in ["rw", "ro"]:
-    raise Exception("Region must be ro or rw")
+  if region not in ['rw', 'ro']:
+    raise Exception('Region must be ro or rw')
 
-  if region is "ro":
-    cmd = "reboot"
+  if region == 'ro':
+    cmd = 'reboot'
   else:
-    cmd = "sysjump %s" % region
+    cmd = 'sysjump %s' % region
 
   tinys.pty._issue_cmd(cmd)
 
@@ -179,10 +180,10 @@ def select(tinys, region):
   time.sleep(2)
   tinys.reinitialize()
 
-  res = tinys.pty._issue_cmd_get_results("sysinfo", ["Copy:[\s]+(RO|RW)"])
+  res = tinys.pty._issue_cmd_get_results('sysinfo', [r'Copy:[\s]+(RO|RW)'])
   current_region = res[0][1].lower()
   if current_region != region:
-    raise Exception("Invalid region: %s/%s" % (current_region, region))
+    raise Exception('Invalid region: %s/%s' % (current_region, region))
 
 def do_version(tinys):
   """Check version via ec console 'pty'.
@@ -198,8 +199,8 @@ def do_version(tinys):
   # ...
   # Build:   tigertail_v1.1.6749-74d1a312e
   """
-  cmd = '\r\nversion\r\n'
-  regex = 'Build:\s+(\S+)[\r\n]+'
+  cmd = 'version'
+  regex = r'Build:\s+(\S+)[\r\n]+'
 
   results = tinys.pty._issue_cmd_get_results(cmd, [regex])[0]
 
@@ -219,9 +220,9 @@ def do_updater_version(tinys):
   # Servo versions below 58 are from servo-9040.B. Versions starting with _v2
   # are newer than anything _v1, no need to check the exact number. Updater
   # version is not directly queryable.
-  if re.search('_v[2-9]\.\d', vers):
+  if re.search(r'_v[2-9]\.\d', vers):
     return 6
-  m = re.search('_v1\.1\.(\d\d\d\d)', vers)
+  m = re.search(r'_v1\.1\.(\d\d\d\d)', vers)
   if m:
     version_number = int(m.group(1))
     if version_number < 5800:
@@ -310,7 +311,7 @@ def get_files_and_version(cname, fname=None, channel=DEFAULT_CHANNEL):
       cname = newname
     else:
       # Try appending ".json" to convert board name to config file.
-      cname = newname + ".json"
+      cname = newname + '.json'
     if not os.path.isfile(cname):
       raise ServoUpdaterException("Can't find config file: %s." % cname)
 
@@ -344,26 +345,26 @@ def get_files_and_version(cname, fname=None, channel=DEFAULT_CHANNEL):
   return cname, fname, binvers
 
 def main():
-  parser = argparse.ArgumentParser(description="Image a servo device")
+  parser = argparse.ArgumentParser(description='Image a servo device')
   parser.add_argument('-p', '--print', dest='print_only', action='store_true',
                       default=False,
                       help='only print available firmware for board/channel')
   parser.add_argument('-s', '--serialno', type=str,
-                      help="serial number to program", default=None)
+                      help='serial number to program', default=None)
   parser.add_argument('-b', '--board', type=str,
-                      help="Board configuration json file",
+                      help='Board configuration json file',
                       default=DEFAULT_BOARD, choices=BOARDS)
   parser.add_argument('-c', '--channel', type=str,
-                      help="Firmware channel to use",
+                      help='Firmware channel to use',
                       default=DEFAULT_CHANNEL, choices=CHANNELS)
   parser.add_argument('-f', '--file', type=str,
-                      help="Complete ec.bin file", default=None)
-  parser.add_argument('--force', action="store_true",
-                      help="Update even if version match", default=False)
-  parser.add_argument('-v', '--verbose', action="store_true",
-                      help="Chatty output")
-  parser.add_argument('-r', '--reboot', action="store_true",
-                      help="Always reboot, even after probe.")
+                      help='Complete ec.bin file', default=None)
+  parser.add_argument('--force', action='store_true',
+                      help='Update even if version match', default=False)
+  parser.add_argument('-v', '--verbose', action='store_true',
+                      help='Chatty output')
+  parser.add_argument('-r', '--reboot', action='store_true',
+                      help='Always reboot, even after probe.')
 
   args = parser.parse_args()
 
@@ -384,34 +385,34 @@ def main():
   with open(brdfile) as data_file:
       data = json.load(data_file)
   vid, pid = int(data['vid'], 0), int(data['pid'], 0)
-  vidpid = "%04x:%04x" % (vid, pid)
+  vidpid = '%04x:%04x' % (vid, pid)
   iface = int(data['console'], 0)
   boardname = data['board']
 
   # Make sure device is up.
-  print("===== Waiting for USB device =====")
+  print('===== Waiting for USB device =====')
   c.wait_for_usb(vidpid, serialname=serialno)
   # We need a tiny_servod to query some information. Set it up first.
   tinys = tiny_servod.TinyServod(vid, pid, iface, serialno, args.verbose)
 
   if not args.force:
     vers = do_version(tinys)
-    print("Current %s version is   %s" % (boardname, vers))
-    print("Available %s version is %s" % (boardname, newvers))
+    print('Current %s version is   %s' % (boardname, vers))
+    print('Available %s version is %s' % (boardname, newvers))
 
     if newvers == vers:
-      print("No version update needed")
+      print('No version update needed')
       if args.reboot:
         select(tinys, 'ro')
       return
     else:
-      print("Updating to recommended version.")
+      print('Updating to recommended version.')
 
   # Make sure the servo MCU is in RO
-  print("===== Jumping to RO =====")
+  print('===== Jumping to RO =====')
   do_with_retries(select, tinys, 'ro')
 
-  print("===== Flashing RW =====")
+  print('===== Flashing RW =====')
   vers = do_with_retries(do_updater_version, tinys)
   # To make sure that the tiny_servod here does not interfere with other
   # processes, close it out.
@@ -430,10 +431,10 @@ def main():
   tinys.reinitialize()
 
   # Make sure the servo MCU is in RW
-  print("===== Jumping to RW =====")
+  print('===== Jumping to RW =====')
   do_with_retries(select, tinys, 'rw')
 
-  print("===== Flashing RO =====")
+  print('===== Flashing RO =====')
   vers = do_with_retries(do_updater_version, tinys)
 
   if vers == 2:
@@ -444,13 +445,13 @@ def main():
     raise ServoUpdaterException("Can't detect updater version")
 
   # Make sure the servo MCU is in RO
-  print("===== Rebooting =====")
+  print('===== Rebooting =====')
   do_with_retries(select, tinys, 'ro')
   # Perform additional reboot to free USB/UART resources, taken by tiny servod.
   # See https://issuetracker.google.com/196021317 for background.
-  tinys.pty._issue_cmd("reboot")
+  tinys.pty._issue_cmd('reboot')
 
-  print("===== Finished =====")
+  print('===== Finished =====')
 
-if __name__ == "__main__":
+if __name__ == '__main__':
   main()