Class AutoResearchAgent

ci/autoresearch_agent.py:115–468

Bidirectional JSON-RPC agent for LED autoresearch testing.

class AutoResearchAgent:
    """Bidirectional JSON-RPC agent for LED autoresearch testing."""

    REMOTE_PREFIX = "REMOTE: "

    def __init__(self, port: str, baudrate: int = 115200, timeout: float = 30.0):
        """Initialize autoresearch agent.

        Args:
            port: Serial port path (e.g., "/dev/ttyUSB0", "COM3")
            baudrate: Serial baud rate (default: 115200)
            timeout: RPC response timeout in seconds (default: 30.0)
        """
        self._monitor = SerialMonitor(
            port=port,
            baud_rate=baudrate,
            auto_reconnect=True,
            verbose=False,
        )
        self._monitor.__enter__()  # Manual context manager entry
        self.timeout = timeout

    def send_rpc(self, function: str, args: list[Any] | None = None) -> dict[str, Any]:
        """Send JSON-RPC command and wait for response.

        Args:
            function: RPC function name (e.g., "configure", "runTest")
            args: Function arguments as list (default: [])

        Returns:
            JSON response dict from device

        Raises:
            TimeoutError: If no response received within timeout
        """
        cmd = {"method": function, "params": args or []}
        cmd_str = json.dumps(cmd, separators=(",", ":"))

        # Send command (no manual buffer clearing needed with SerialMonitor)
        self._monitor.write(cmd_str + "\n")

        # Wait for REMOTE: prefixed response
        return self._wait_for_response()

    def _wait_for_response(self) -> dict[str, Any]:
        """Wait for and parse JSON-RPC response.

        Returns:
            Parsed JSON response dict

        Raises:
            TimeoutError: If no response received within timeout
        """
        start = time.time()

        while time.time() - start < self.timeout:
            # Use read_lines() iterator with short timeout
            for line in self._monitor.read_lines(timeout=0.05):