Fetch schema from device via getSchema RPC call
(self)
| 73 | self._fetch_schema() |
| 74 | |
| 75 | def _fetch_schema(self): |
| 76 | """Fetch schema from device via getSchema RPC call""" |
| 77 | try: |
| 78 | with serial.Serial(self.port, self.baudrate, timeout=self.timeout) as ser: |
| 79 | time.sleep(2.0) # Wait for device application to fully initialize |
| 80 | ser.reset_input_buffer() |
| 81 | |
| 82 | # Send rpc.discover request (built-in RPC schema method) |
| 83 | request: dict[str, Any] = { |
| 84 | "method": "rpc.discover", |
| 85 | "params": [], |
| 86 | "id": 1, |
| 87 | "jsonrpc": "2.0", |
| 88 | } |
| 89 | ser.write((json.dumps(request) + "\n").encode()) |
| 90 | ser.flush() |
| 91 | |
| 92 | # Read response (with timeout) |
| 93 | start = time.time() |
| 94 | lines_read = 0 |
| 95 | while time.time() - start < self.timeout: |
| 96 | if ser.in_waiting: |
| 97 | line = ser.readline().decode("utf-8", errors="ignore").strip() |
| 98 | lines_read += 1 |
| 99 | |
| 100 | # Debug: print first few lines if not getting REMOTE: response |
| 101 | if lines_read <= 10 and not line.startswith("REMOTE:"): |
| 102 | print(f" [debug] Line {lines_read}: {line[:80]}") |
| 103 | |
| 104 | if line.startswith("REMOTE:"): |
| 105 | # Parse JSON-RPC response |
| 106 | json_str = line[len("REMOTE:") :].strip() |
| 107 | response = json.loads(json_str) |
| 108 | |
| 109 | if "result" in response: |
| 110 | self.schema = RpcSchema(**response["result"]) |
| 111 | print( |
| 112 | f"✅ Fetched schema: {len(self.schema.methods)} methods" |
| 113 | ) |
| 114 | return |
| 115 | elif "error" in response: |
| 116 | raise RuntimeError( |
| 117 | f"Schema fetch failed: {response['error']}" |
| 118 | ) |
| 119 | |
| 120 | time.sleep(0.01) |
| 121 | |
| 122 | raise TimeoutError( |
| 123 | f"Timeout waiting for schema response (read {lines_read} lines)" |
| 124 | ) |
| 125 | |
| 126 | except KeyboardInterrupt as ki: |
| 127 | handle_keyboard_interrupt(ki) |
| 128 | raise |
| 129 | except Exception as e: |
| 130 | raise RuntimeError(f"Failed to fetch RPC schema: {e}") |
| 131 | |
| 132 | def validate_request( |