Method send_and_match

ci/rpc_client.py:403–556

Send RPC and wait for response containing specific key with mandatory ID correlation (async). Useful when device may send multiple response types and you need to match a specific one. Args: function: RPC function name args: Function arguments

(
        self,
        function: str,
        args: list[Any] | None = None,
        match_key: str | None = None,
        timeout: float | None = None,
        retries: int = 3,
    )

        )

    async def send_and_match(
        self,
        function: str,
        args: list[Any] | None = None,
        match_key: str | None = None,
        timeout: float | None = None,
        retries: int = 3,
    ) -> RpcResponse:
        """Send RPC and wait for response containing specific key with mandatory ID correlation (async).

        Useful when device may send multiple response types and you need
        to match a specific one.

        Args:
            function: RPC function name
            args: Function arguments
            match_key: Key that must be present in response (e.g., "connected")
            timeout: Override default timeout
            retries: Number of retry attempts

        Returns:
            RpcResponse containing the match_key and matching request ID (always includes ID)

        Raises:
            RpcTimeoutError: If matching response not received
        """
        if not self.is_connected:
            raise RpcError("Not connected")

        assert self._serial is not None

        # Generate unique request ID for JSON-RPC 2.0 correlation (mandatory)
        request_id = self._next_id
        self._next_id = (self._next_id + 1) & 0xFFFFFFFF  # Wrap at uint32 max

        # Wrap args in array to pass as single Json parameter to RPC functions
        # RPC functions expect signature: (const fl::json& args)
        # So we need to pass the entire args as one parameter
        wrapped_args: list[Any] = [args] if args is not None else [{}]
        cmd: dict[str, Any] = {
            "method": function,
            "params": wrapped_args,
            "id": request_id,  # ID is mandatory
        }
        cmd_str = json.dumps(cmd, separators=(",", ":"))

        if self.verbose:
            print(f"📤 [RPC] Sending: {cmd_str}")

        effective_timeout = timeout if timeout is not None else self.timeout
        attempt_timeout = effective_timeout / max(retries, 1)

        for _attempt in range(retries):
            self._check_interrupt()  # Check before each attempt

            # Write command via serial interface
            await self._serial.write(cmd_str + "\n")