Method connect

ci/rpc_client.py:171–281

Open serial connection (async). Args: boot_wait: Maximum time to wait for device boot (seconds). When ``boot_signals`` is non-empty, this is the *deadline* for the spin-poll — the helper returns as soon as any signal appears, other

(
        self,
        boot_wait: float = 3.0,
        drain_boot: bool = True,
        boot_signals: "list[str] | tuple[str, ...] | None" = None,
    )

        return self._serial is not None and self._connected

    async def connect(
        self,
        boot_wait: float = 3.0,
        drain_boot: bool = True,
        boot_signals: "list[str] | tuple[str, ...] | None" = None,
    ) -> None:
        """Open serial connection (async).

        Args:
            boot_wait: Maximum time to wait for device boot (seconds). When
                ``boot_signals`` is non-empty, this is the *deadline* for the
                spin-poll — the helper returns as soon as any signal appears,
                otherwise it falls back to waiting the full duration (matching
                the legacy fixed-sleep behavior so slow devices still work).
            drain_boot: Whether to drain boot output after connecting. Lines
                consumed while spinning for ``boot_signals`` count toward the
                drain, so this second pass only catches leftovers.
            boot_signals: Substrings to look for in serial output during the
                boot wait. When a match is found, ``connect()`` returns
                immediately instead of sleeping the full ``boot_wait``. Pass
                an empty list to opt out of spin-polling (legacy sleep path).
                Defaults to ``BOOT_SIGNALS_ANY`` which covers both ROM-loader
                and Arduino ``setup()`` banners.
        """
        if self._connected:
            return

        # Create serial interface if not provided externally (defaults to fbuild)
        if self._serial is None:
            from ci.util.serial_interface import create_serial_interface

            self._serial = create_serial_interface(
                port=self.port,
                baud_rate=self.baudrate,
            )
            self._owns_serial = True

        backend_name = type(self._serial).__name__
        if self.verbose:
            print(f"🔌 [RPC] Connecting to {self.port} @ {self.baudrate} baud...")
            print(f"🔌 [RPC] Using {backend_name} backend")

        await self._serial.connect()
        self._connected = True

        if self.verbose:
            print(f"✅ [RPC] Serial connection established")

        # Resolve default boot signals lazily so tests that don't touch
        # serial I/O don't import the util module.
        effective_signals: tuple[str, ...] | list[str]
        if boot_signals is None:
            from ci.util.boot_wait import BOOT_SIGNALS_ANY

            effective_signals = BOOT_SIGNALS_ANY
        else:
            effective_signals = boot_signals