Send a debug command and return parsed response.
(cmd, timeout=10)
| 33 | RESET = "\033[0m" |
| 34 | |
| 35 | def debug_cmd(cmd, timeout=10): |
| 36 | """Send a debug command and return parsed response.""" |
| 37 | sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) |
| 38 | sock.settimeout(timeout) |
| 39 | try: |
| 40 | sock.connect(DEBUG_SOCK) |
| 41 | req = {"type": "debug_command", "id": 1, "command": cmd} |
| 42 | sock.send((json.dumps(req) + "\n").encode()) |
| 43 | |
| 44 | # Read until we get a complete JSON response |
| 45 | buf = b"" |
| 46 | while True: |
| 47 | chunk = sock.recv(65536) |
| 48 | if not chunk: |
| 49 | break |
| 50 | buf += chunk |
| 51 | try: |
| 52 | return json.loads(buf.decode()) |
| 53 | except json.JSONDecodeError: |
| 54 | continue |
| 55 | except Exception as e: |
| 56 | return {"error": str(e)} |
| 57 | finally: |
| 58 | sock.close() |
| 59 | |
| 60 | def get_server_pid(): |
| 61 | """Get the jcode server PID.""" |