Class RpcSchemaValidator

ci/rpc_schema_validator.py:53–200

Validates JSON-RPC requests and responses against device schema. Fetches schema once from device and caches it for validation.

class RpcSchemaValidator:
    """
    Validates JSON-RPC requests and responses against device schema.

    Fetches schema once from device and caches it for validation.
    """

    def __init__(self, port: str, baudrate: int = 115200, timeout: float = 5.0):
        """
        Initialize validator and fetch schema from device.

        Args:
            port: Serial port (e.g., 'COM18' or '/dev/ttyUSB0')
            baudrate: Serial baud rate
            timeout: Timeout for schema fetch
        """
        self.port = port
        self.baudrate = baudrate
        self.timeout = timeout
        self.schema: Optional[RpcSchema] = None
        self._fetch_schema()

    def _fetch_schema(self):
        """Fetch schema from device via getSchema RPC call"""
        try:
            with serial.Serial(self.port, self.baudrate, timeout=self.timeout) as ser:
                time.sleep(2.0)  # Wait for device application to fully initialize
                ser.reset_input_buffer()

                # Send rpc.discover request (built-in RPC schema method)
                request: dict[str, Any] = {
                    "method": "rpc.discover",
                    "params": [],
                    "id": 1,
                    "jsonrpc": "2.0",
                }
                ser.write((json.dumps(request) + "\n").encode())
                ser.flush()

                # Read response (with timeout)
                start = time.time()
                lines_read = 0
                while time.time() - start < self.timeout:
                    if ser.in_waiting:
                        line = ser.readline().decode("utf-8", errors="ignore").strip()
                        lines_read += 1

                        # Debug: print first few lines if not getting REMOTE: response
                        if lines_read <= 10 and not line.startswith("REMOTE:"):
                            print(f"  [debug] Line {lines_read}: {line[:80]}")

                        if line.startswith("REMOTE:"):
                            # Parse JSON-RPC response
                            json_str = line[len("REMOTE:") :].strip()
                            response = json.loads(json_str)

                            if "result" in response:
                                self.schema = RpcSchema(**response["result"])