JSON-RPC send/receive helper (mirrors test_lpc_pin_toggle_rx.py).
(
s: "serial.Serial",
method: str,
args: list[Any] | None = None,
request_id: int = 1,
timeout: float = 10.0,
)
| 71 | |
| 72 | |
| 73 | def send_rpc( |
| 74 | s: "serial.Serial", |
| 75 | method: str, |
| 76 | args: list[Any] | None = None, |
| 77 | request_id: int = 1, |
| 78 | timeout: float = 10.0, |
| 79 | ) -> dict[str, Any] | None: |
| 80 | """JSON-RPC send/receive helper (mirrors test_lpc_pin_toggle_rx.py).""" |
| 81 | params = args if args is not None else [] |
| 82 | req = {"jsonrpc": "2.0", "method": method, "params": params, "id": request_id} |
| 83 | s.write((json.dumps(req, separators=(",", ":")) + "\n").encode("ascii")) |
| 84 | s.flush() |
| 85 | |
| 86 | deadline = time.time() + timeout |
| 87 | buf = b"" |
| 88 | while time.time() < deadline: |
| 89 | chunk = s.read(s.in_waiting or 1) |
| 90 | if not chunk: |
| 91 | continue |
| 92 | buf += chunk |
| 93 | while b"\n" in buf: |
| 94 | line, _, buf = buf.partition(b"\n") |
| 95 | text = line.decode("ascii", errors="replace").strip() |
| 96 | for prefix in ("REMOTE: ", "RESULT: "): |
| 97 | if text.startswith(prefix): |
| 98 | text = text[len(prefix) :] |
| 99 | break |
| 100 | if not (text.startswith("{") and text.endswith("}")): |
| 101 | continue |
| 102 | try: |
| 103 | msg = json.loads(text) |
| 104 | except json.JSONDecodeError: |
| 105 | continue |
| 106 | if isinstance(msg, dict) and msg.get("id") == request_id: |
| 107 | return msg |
| 108 | return None |
| 109 | |
| 110 | |
| 111 | def parse_csv_result(csv: str) -> dict[str, int]: |