Method run

ci/util/serial_monitor.py:291–358

Run the serial monitor until timeout or handler requests stop. Returns: True if all handlers succeeded, False otherwise

(self)

        self.mHandlers.append(handler)

    def run(self) -> bool:
        """Run the serial monitor until timeout or handler requests stop.

        Returns:
            True if all handlers succeeded, False otherwise
        """
        # Open serial connection
        try:
            self.mSerialPort = serial.Serial(self.mPort, self.mBaudRate, timeout=1)
            time.sleep(0.5)  # Let serial connection stabilize
            print(f"{Fore.GREEN}✅ Serial connection established{Style.RESET_ALL}\n")
        except serial.SerialException as e:
            print(f"{Fore.RED}❌ Failed to open serial port: {e}{Style.RESET_ALL}")
            return False

        # Initialize handlers that need the serial port
        for handler in self.mHandlers:
            if isinstance(handler, InputTriggerHandler):
                handler.set_serial_port(self.mSerialPort)

        start_time = time.time()
        should_continue = True

        try:
            while should_continue:
                # Check timeout (unless in streaming mode)
                if not self.mStream:
                    elapsed = time.time() - start_time
                    if elapsed >= self.mTimeoutSeconds:
                        print(
                            f"\n⏱️  Timeout reached ({self.mTimeoutSeconds}s), stopping monitor..."
                        )
                        break

                # Read available serial data
                if self.mSerialPort.in_waiting:
                    try:
                        line = (
                            self.mSerialPort.readline()
                            .decode("utf-8", errors="ignore")
                            .strip()
                        )
                        if line:
                            print(line)  # Display line

                            # Dispatch to all handlers
                            for handler in self.mHandlers:
                                if not handler.handle_line(line):
                                    # Handler requested stop
                                    should_continue = False
                                    break

                    except UnicodeDecodeError:
                        # Skip lines with decode errors
                        pass

                time.sleep(0.01)  # Small delay to prevent busy-waiting