Enhance flash info handling and USB command retries

python3-validity.local.service

@@ -0,0 +1,13 @@
+[Unit]
+Description=python-validity driver dbus service (local checkout)
+After=open-fprintd.service
+
+[Service]
+Type=simple
+Environment=PYTHONPATH=/home/liam/src/python-validity
+ExecStart=/usr/bin/python3 /home/liam/src/python-validity/dbus_service/dbus-service --debug --configpath /home/liam/src/python-validity/etc/python-validity
+Restart=on-success
+RestartSec=3
+
+[Install]
+WantedBy=multi-user.target

validitysensor/flash.py

@@ -1,3 +1,4 @@
+import logging
 import typing
 from struct import pack, unpack
 
@@ -21,7 +22,8 @@ def __repr__(self):
 
 
 class FlashInfo:
-    def __init__(self, ic: FlashIcInfo, blocks: int, unknown0: int, blocksize: int, unknown1: int,
+    def __init__(self, ic: typing.Optional[FlashIcInfo], blocks: int, unknown0: int,
+                 blocksize: int, unknown1: int,
                  partitions: typing.Sequence[PartitionInfo]):
         self.ic = ic
         self.blocks = blocks
@@ -40,6 +42,15 @@ def get_flash_info():
     rsp = tls.cmd(unhex('3e'))
     assert_status(rsp)
     rsp = rsp[2:]
+
+    # Some sensors (e.g. 06cb:009a with an uninitialized/empty flash) reply
+    # with only the status word (0000) and no header/partition data.
+    # Treat a short response as an uninitialized flash so init_flash() can
+    # format it instead of crashing on unpack().
+    if len(rsp) < 0xe:
+        logging.info('Flash info response too short (%d bytes); assuming uninitialized flash' % len(rsp))
+        return FlashInfo(None, 0, 0, 0, 0, [])
+
     hdr = rsp[:0xe]
     rsp = rsp[0xe:]
     jid0, jid1, blocks, unknown0, blocksize, unknown1, pcnt = unpack('<HHHHHHH', hdr)

validitysensor/init.py

@@ -28,8 +28,8 @@ def close():
 
 def open_common():
     init_data_dir()
-    init_flash()
     usb.send_init()
+    init_flash()
     tls.parse_tls_flash(read_tls_flash())
     tls.open()
     upload_fwext()

validitysensor/usb.py

@@ -1,5 +1,6 @@
 import errno
 import logging
+import time
 import typing
 from binascii import hexlify, unhexlify
 from enum import Enum
@@ -11,6 +12,8 @@
 from .blobs import init_hardcoded, init_hardcoded_clean_slate
 from .util import assert_status
 
+BUSY_STATUSES = (0x0401, 0x0104)
+
 
 class SupportedDevices(Enum):
     """USB IDs for supported devices"""
@@ -76,16 +79,40 @@ def close(self):
     def usb_dev(self):
         return self.dev
 
+    def cmd_retry_busy(self, out: bytes):
+        rsp = None
+        for i in range(20):
+            try:
+                rsp = self.cmd(out)
+            except ucore.USBTimeoutError:
+                if i == 19:
+                    raise
+                time.sleep(0.5)
+                continue
+
+            s_stat, = unpack('<H', rsp[:2])
+            if s_stat in BUSY_STATUSES:
+                time.sleep(0.5)
+                continue
+
+            return rsp
+
+        return rsp
+
     def send_init(self):
         # self.dev.set_configuration()
 
         # TODO analyse responses, detect hardware type
-        assert_status(self.cmd(unhexlify('01')))  # RomInfo.get()
+        # The sensor may reply with a transient "busy" status to the
+        # first command(s) after it wakes up. Retry a few times to let it settle.
+        rsp_init = self.cmd_retry_busy(unhexlify('01'))  # RomInfo.get()
+        assert_status(rsp_init)
+
         assert_status(self.cmd(unhexlify('19')))
 
         # 43 -- get partition header(?) (02 -- fwext partition)
         # c28c745a in response is a FwextBuildtime = 0x5A748CC2
-        rsp = self.cmd(unhexlify('4302'))  # get_fw_info()
+        rsp = self.cmd_retry_busy(unhexlify('4302'))  # get_fw_info()
 
         assert_status(self.cmd(init_hardcoded))