Method run

ci/util/serial_monitor.py:415–523

Run the PIO device 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 PIO device monitor until timeout or handler requests stop.

        Returns:
            True if all handlers succeeded, False otherwise
        """
        cmd = [
            "pio",
            "device",
            "monitor",
            "--project-dir",
            str(self.mBuildDir),
        ]
        if self.mEnvironment:
            cmd.extend(["--environment", self.mEnvironment])
        if self.mMonitorPort:
            cmd.extend(["--port", self.mMonitorPort])
        if self.mVerbose:
            cmd.append("--verbose")

        formatter = TimestampFormatter()
        proc = RunningProcess(
            cmd,
            cwd=self.mBuildDir,
            auto_run=True,
            output_formatter=formatter,
            env=get_utf8_env(),
        )

        start_time = time.time()
        timeout_reached = False
        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..."
                        )
                        timeout_reached = True
                        proc.terminate()
                        break

                # Read next line with 30 second read timeout
                try:
                    line = proc.get_next_line(timeout=30)
                    if isinstance(line, EndOfStream):
                        break
                    if line:
                        print(line)  # Real-time output

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