Rework of teensy progress, adding descriptions, handling for various illegal user flows

Author: malcolm-sparkfun

MicroPython_Firmware_Uploader/MicroPython_Firmware_Uploader.py

@@ -14,7 +14,7 @@
     AUxEsptoolEraseFlash
 from .mpremote_utils import MPRemoteSession
 from .au_act_rp2 import AUxRp2UploadRp2
-from .au_act_teensy import AUxTeensyUploadTeensy, TeensyProgress
+from .au_act_teensy import AUxTeensyUploadTeensy
 from .pyqt_utils import PopupListButton
 from .firmware_utils import FirmwareFile, resource_path, GithubFirmware
 
@@ -97,8 +97,6 @@ def __init__(self, parent: QWidget = None) -> None:
         # ---------------------------- Variables/Objects ------------------
         self.flashSize = 0
 
-        self.teensyProgress = TeensyProgress() # Progress tracking for Teensy uploads. Its done in a different way for ESP32 and RP2...
-
         self.githubFirmware = GithubFirmware(_RELEASE_REPO, _BOARD_MANIFEST_FILE, _RESOURCE_DIRECTORY)
 
         if self.githubFirmware.offline:
@@ -144,13 +142,16 @@ def __init__(self, parent: QWidget = None) -> None:
 
         # Populate the device button with the list of devices from the latest release
         # Add an automatic device detection option to the list of devices
-        self.device_button.addIconItem(_AUTO_DETECT_DECORATOR, "Automatically detect the device", resource_path("auto_detect.png", _RESOURCE_DIRECTORY))
+        self.device_button.addIconItem(_AUTO_DETECT_DECORATOR, "If your board is already running MicroPython, automatically selects board.", resource_path("auto_detect.png", _RESOURCE_DIRECTORY))
 
-        for device, imagePath in self.githubFirmware.get_all_board_image_paths():
-            self.device_button.addIconItem(device, "", imagePath)
+        for device, description, imagePath in self.githubFirmware.get_all_board_icon_info():
+            self.device_button.addIconItem(device, description, imagePath)
 
         # Add the local firmware option to the list of firmware files
-        self.firmware_button.addIconItem(_LOCAL_DECORATOR, "Local firmware file", resource_path("fw_local.png", _RESOURCE_DIRECTORY))
+        self.firmware_button.addIconItem(_LOCAL_DECORATOR, 
+                                         "Manually select a firmware file from your local machine.", 
+                                         resource_path("fw_local.png",_RESOURCE_DIRECTORY)
+                                        )
 
         # Attach the browse callback to the firmware button's textChanged signal
         self.firmware_button.textChanged.connect(self.on_fw_button_pressed)
@@ -296,13 +297,15 @@ def update_firmware_list(self) -> None:
 
         self.firmware_button.items.clear()
         # Add the local firmware option to the list of firmware files
-        self.firmware_button.addIconItem(_LOCAL_DECORATOR, "Local firmware file", resource_path("fw_local.png", _RESOURCE_DIRECTORY))
+        self.firmware_button.addIconItem(_LOCAL_DECORATOR, 
+                                         "Manually select a firmware file from your local machine.", 
+                                         resource_path("fw_local.png",_RESOURCE_DIRECTORY)
+                                        )
 
         # Check the currently selected device item and update the firmware list accordingly
         self.firmware_button.setText(_FIRMWARE_CHOICE_DECORATOR)
         currentDevice = self.device_button.text()
         for fw in self.githubFirmware.deviceDict[currentDevice]:
-            # print("Adding firmware: ", fw.displayName, " for device: ", currentDevice)
             self.firmware_button.addIconItem(fw.displayName, fw.description(), fw.fw_image_path())
 
     def show_user_message(self, message: str, windowTitle: str, warning = False) -> None:
@@ -369,16 +372,8 @@ def parse_progress(self, msg: str) -> None:
         """Parse the progress message."""
         if self.is_esp32_upload():
             self.parse_esp32_progress(msg)
-        elif self.is_teensy_upload():
-            progress = self.teensyProgress.parse_message(msg)
-            # update the progress bar with the extracted percentage
-            if progress is not None:
-                if progress > 0 and progress < 100:
-                    # emit the progress signal to update the progress bar in the GUI
-                    self.sig_progress.emit(progress)
 
-        # rp2 progress is handled down a few levels while the firmware is being uploaded.
-        # this is because it is not a parsing of the output message but rather a manual tracking of the copy progress.
+        # rp2 progress and teensy progress are handled down a few levels while the firmware is being uploaded.
 
     @pyqtSlot(str)
     def appendMessage(self, msg: str) -> None:
@@ -432,6 +427,19 @@ def progress_bar_start(self, label) -> None:
 
     @pyqtSlot(int)
     def on_progress(self, progress: int) -> None:
+        if progress < 0: 
+            # This indicates an error from the worker. We can handle differently for different platforms.
+            # For now, this will only happen when teensy times out waiting for the bootloader
+            if self.is_teensy_upload():
+                # Open a popup warning the user that the upload failed.
+                # Also end the upload and allow them to try again by re-enabling the back button (interface)
+                error_message = "Teensy upload timed out waiting for the bootloader to be ready. Please try again."
+                # If on MacOS, also inform user they need to check their "Input Monitoring" security permissions
+                if platform.system() == "Darwin":
+                    error_message += "\n\nIf you are on MacOS, please also check your \"Input Monitoring\" security permissions for this application."
+                self.show_user_message(error_message, "Teensy Upload Error", warning=True)
+                self.end_upload_with_message(error_message)
+
         self.progress_bar.setValue(progress)
 
     #--------------------------------------------------------------
@@ -465,15 +473,15 @@ def on_finished(self, status, action_type, job_id) -> None:
             self.writeMessage("Reset complete...")
             # Emit a signal to update the progress bar to 100%
             self.sig_progress.emit(100)
-            self.writeMessage("DONE: Firmware file copied to ESP32 device.\n")
+            self.writeMessage("DONE: ESP32 Upload Job Complete.\n")
             self.disable_interface(False)
 
         if action_type == AUxRp2UploadRp2.ACTION_ID:
-            self.end_upload_with_message("DONE: Firmware file copied to RP2 device.\n")
+            self.end_upload_with_message("DONE: RP2 Upload Job Complete.\n")
 
         if action_type == AUxTeensyUploadTeensy.ACTION_ID:
             self.sig_progress.emit(100)
-            self.end_upload_with_message("DONE: Firmware file copied to Teensy device.\n")
+            self.end_upload_with_message("DONE: Teensy Upload Job Complete\n")
 
     # --------------------------------------------------------------
     # on_port_combobox()
@@ -782,17 +790,10 @@ def end_upload_with_message(self, msg: str) -> None:
         """End the upload with a message."""
         self.writeMessage(msg)
         self.cleanup_temp()
-        # self.switch_to_page_one()
         self.disable_interface(False)
-        # self.progress_bar.hide()
-        # self.progress_label.hide()
 
     def get_mpremote_session(self) -> MPRemoteSession:
         """Get an mpremote session for the current port."""
-        # portName = self.port_combobox.currentText() # This will be something like "USB Serial Device (COM3)" 
-        # portName = portName.split("(")[1].split(")")[0].strip() # This will be something like "COM3" and is what mpremote wants
-        # print("Port: ", self.port_button.text())
-
         # check if we are on mac or linux and if so, we need to add the /dev/ prefix to the port name
         if platform.system() == "Darwin" or platform.system().startswith("Linux"):
             portName = "/dev/" + self.port_button.text()
@@ -820,10 +821,17 @@ def do_upload_rp2(self, fwFile = None) -> None:
 
         # Start an mpremote session for selected port
         mpr = self.get_mpremote_session()
+        
+        mpr_base_platform = None
+
+        # This means we have a valid mpremote session
+        if mpr is not None:
+            mpr_base_platform = mpr.get_base_platform()
+            self.writeMessage("MPRemote session validated. Platform: " + mpr_base_platform + "\n")
 
         # If we are able to use mpremote to enter bootloader mode, do so
         # Otherwise, prompt the user to press the correct button sequence to enter bootloader mode
-        if mpr is not None:
+        if mpr_base_platform == "rp2":
             self.writeMessage("Able to automatically enter bootloader mode...\n")
             mpr.enter_bootloader()
             self.writeMessage("Entering bootloader mode...\n")
@@ -905,12 +913,12 @@ def do_upload_teensy(self, fwFile = None, boardName = None) -> None:
         theJob = AxJob(AUxTeensyUploadTeensy.ACTION_ID, {"loader": resource_path("teensy_loader_cli.exe", _RESOURCE_DIRECTORY),
                                                          "mcu":"imxrt1062", 
                                                          "board": boardName,
-                                                         "file": fwFile
+                                                         "file": fwFile,
+                                                         "size": self.get_current_firmware_file_size(),
                                                          })
 
 
-        # Show to progress bar and set it to 0. Reset our teensy progress bar and save the file size
-        self.teensyProgress.reset(self.get_current_firmware_file_size())
+        # Show progress bar and set it to 0.
         self.progress_bar_start("Teensy Upload Progress:")
 
         self._worker.add_job(theJob)

MicroPython_Firmware_Uploader/au_act_teensy.py

@@ -1,55 +1,29 @@
 from .au_action import AxAction, AxJob
 import shutil
-from os import system
+from os import system, set_blocking
 import subprocess
 import sys
-
-class AUxTeensyUploadTeensy(AxAction):
-    ACTION_ID = "upload-teensy"
-    NAME = "Teensy Upload"
-
-    def __init__(self) -> None:
-        super().__init__(self.ACTION_ID, self.NAME)
-    
-    def run_job(self, job:AxJob, **kwargs):
-        # Ensure stdout of our subprocess gets routed through our stdout to the wedge thingy and then the UI. 
-        # Basic os.system() and subprocess.run() don't do this.
-
-        # Some platforms (Windows) have CREATE_NO_WINDOW, but others (linux) don't. 
-        # So we need to check if the flag exists and only use it if it does.
-        if hasattr(subprocess, 'CREATE_NO_WINDOW'):
-            teensy_proc = subprocess.Popen([job.loader, f"--mcu={job.mcu}", "-v", "-w", job.board, job.file], 
-                                            creationflags=subprocess.CREATE_NO_WINDOW,
-                                            stdout=subprocess.PIPE, 
-                                            stderr=subprocess.PIPE)
-        else:
-            teensy_proc = subprocess.Popen([job.loader, f"--mcu={job.mcu}", "-v", "-w", job.board, job.file], 
-                                            stdout=subprocess.PIPE, 
-                                            stderr=subprocess.PIPE)
-        
-        while True:
-            output = teensy_proc.stdout.read(1)  # Read a single character
-            if teensy_proc.poll() is not None and not output:
-                break
-            if output:
-                sys.stdout.write(output.decode())  # Write the character to stdout
-                sys.stdout.flush()  # Ensure it appears immediately
-
-        teensy_proc.wait()
-        print("\nTeensy Upload Done.\n")
+from time import perf_counter
+import select
 
 class TeensyProgress(object):
     """Class to keep track of the Teensy progress."""
     # See https://github.com/PaulStoffregen/teensy_loader_cli/blob/master/teensy_loader_cli.c
     # a "." character is printed for every block of data written to the teensy. For TEENSY4.0 andd TEENSY4.1 a block is 1024
     kTeensyBlockSize = 1024 
+    # This is the maximum time we will wait to receive the "Programming" message from the teensy loader
+    # before setting the timeout flag
+    # TODO: What is the absolute minimum times we can have here? This was arbitrary...
+    kMaximumWaitForBootloader = 5 # seconds
 
     def __init__(self) -> None:
         self.progSeen = False
         self.dotsWritten = 0
         self.currentMessage = ""
         self.percent = 0
         self.size = 0
+        self.startTime = None
+        self.timeout = False
 
     def reset(self, size = 0) -> None:
         """Reset the Teensy progress."""
@@ -58,6 +32,28 @@ def reset(self, size = 0) -> None:
         self.currentMessage = ""
         self.percent = 0
         self.size = size
+        self.timeout = False
+        # Create a new start time
+        self.startTime = perf_counter()
+    
+    def elapsed_time(self) -> float:
+        """Get the elapsed time since the start time."""
+        if self.startTime is None:
+            return 0
+        
+        return perf_counter() - self.startTime
+
+    def check_timeout(self) -> bool:
+        """Check if the Teensy progress has timed out."""
+        if self.startTime is None:
+            return False
+
+        elapsed_time = self.elapsed_time()
+        if (elapsed_time > self.kMaximumWaitForBootloader) and not self.progSeen:
+            self.timeout = True
+            return True
+
+        return False
 
     def dots_to_percent(self, numDots: int) -> int:
         """Convert the number of dots to a percentage."""
@@ -91,5 +87,82 @@ def parse_message(self, msg: str) -> int:
             self.dotsWritten += self.currentMessage.count(".")
             self.currentMessage = ""
             self.percent = self.dots_to_percent(self.dotsWritten)
+        
+        else:
+            # If we haven't seen the "Programming" message yet, we need to check against our start time to see if we should time out
+            if self.check_timeout():
+                self.percent = 0
 
         return self.percent
+
+class AUxTeensyUploadTeensy(AxAction):
+    ACTION_ID = "upload-teensy"
+    NAME = "Teensy Upload"
+
+    def __init__(self) -> None:
+        super().__init__(self.ACTION_ID, self.NAME)
+        self.report_progress = None
+        self.__teensy_prog = TeensyProgress()
+
+    def run_job(self, job:AxJob, **kwargs):
+        self.report_progress = kwargs["worker_cb"]
+
+        self.__teensy_prog.reset(job.size)
+        # Ensure stdout of our subprocess gets routed through our stdout to the wedge thingy and then the UI. 
+        # Basic os.system() and subprocess.run() don't do this.
+
+        # Some platforms (Windows) have CREATE_NO_WINDOW, but others (linux) don't. 
+        # So we need to check if the flag exists and only use it if it does.
+        if hasattr(subprocess, 'CREATE_NO_WINDOW'):
+            teensy_proc = subprocess.Popen([job.loader, f"--mcu={job.mcu}", "-v", "-w", job.board, job.file], 
+                                            creationflags=subprocess.CREATE_NO_WINDOW,
+                                            stdout=subprocess.PIPE, 
+                                            stderr=subprocess.PIPE)
+        else:
+            teensy_proc = subprocess.Popen([job.loader, f"--mcu={job.mcu}", "-v", "-w", job.board, job.file], 
+                                            stdout=subprocess.PIPE, 
+                                            stderr=subprocess.PIPE)
+        
+        while not self.__teensy_prog.timeout:
+            # Until we are programming, we can use communicate to get the output with a timeout
+            if not self.__teensy_prog.progSeen:
+                # NOTE: THIS REQUIRES PYTHON 3.12+ ON WINDOWS OR 3.5+ ON UNIX-LIKE
+                set_blocking(teensy_proc.stdout.fileno(), False)  # Set the stdout to non-blocking
+                
+                # Read a chunk of output (polling, since select.select does not work with pipes on Windows)
+                output = teensy_proc.stdout.read(1024)
+                
+                # Feed output to the progress class and to the console
+                if output:
+                    sys.stdout.write(output.decode())
+                    sys.stdout.flush()
+                    # Also feed the output to the progress class
+                    self.__teensy_prog.parse_message(output.decode())
+                
+                # If we have timed out waiting for the bootloader, we need to break out of the loop
+                if self.__teensy_prog.check_timeout():
+                    # Kill the process
+                    teensy_proc.kill()
+                    self.report_progress(-1)  # Report -1 to indicate a timeout
+                    sys.stdout.write("\nTeensy Upload Error: Timed Out...\n")
+                    sys.stdout.flush()
+                    break
+                
+            if self.__teensy_prog.progSeen:
+                # Now, make read blocking again (Likely not necessary, but it's how I thought about it originally)
+                set_blocking(teensy_proc.stdout.fileno(), True)
+                # After we see the "Programming" message, we need to read the output one character at a time
+                # to get the progress percentage and display interactive content to the user with blocking reads and writes
+                output = teensy_proc.stdout.read(1)  # Read a single character
+                if teensy_proc.poll() is not None and not output:
+                    break
+                if output:
+                    sys.stdout.write(output.decode())  # Write the character to stdout
+                    sys.stdout.flush()  # Ensure it appears immediately
+                    # Also feed the output to the progress class
+                    percent = self.__teensy_prog.parse_message(output.decode())
+                    if percent > 0 and percent < 100:
+                        self.report_progress(percent)
+
+        teensy_proc.wait()
+        print("\nTeensy Upload Done.\n")