Александр Петров
58 lines
2.3 KiB
Python
58 lines
2.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Живой MCP 2.x пробник экрана выполняющегося hello во время getchar()."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
from pathlib import Path
|
|
import shutil
|
|
import sys
|
|
|
|
from mcp import Client, StdioServerParameters
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
async def probe(socket: str, proof: Path | None) -> None:
|
|
parameters = StdioServerParameters(
|
|
command=sys.executable,
|
|
args=[str(ROOT / 'toolchain/sdbg_mcp.py'), '--socket', socket])
|
|
async with Client(parameters) as client:
|
|
async def call(name: str, arguments: dict | None = None) -> dict:
|
|
result = await client.call_tool(name, arguments or {})
|
|
if result.is_error or result.structured_content is None:
|
|
raise RuntimeError(name + ': ' + repr(result.content))
|
|
return result.structured_content
|
|
|
|
status = await call('session_status')
|
|
if not status['running']:
|
|
raise RuntimeError('Ожидался выполняющийся hello в getchar()')
|
|
pixels = await call('read_screen_pixels', {
|
|
'x': 0, 'y': 0, 'width': 8, 'height': 8})
|
|
if pixels['stale_frame'] or len(pixels['hex']) != 256:
|
|
raise RuntimeError('Нет свежих пикселей выполняющегося hello')
|
|
shot = await call('screenshot')
|
|
source = Path(shot['path'])
|
|
if shot['stale_frame'] or source.read_bytes()[:8] != b'\x89PNG\r\n\x1a\n':
|
|
raise RuntimeError('Нет PNG выполняющегося hello')
|
|
if proof:
|
|
proof.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copyfile(source, proof)
|
|
print(json.dumps({'event': 'mcp_hello_running_screen',
|
|
'frame': shot['frame'], 'size': shot['size'],
|
|
'proof': str(proof) if proof else str(source)},
|
|
ensure_ascii=False), flush=True)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--socket', required=True)
|
|
parser.add_argument('--proof', type=Path)
|
|
args = parser.parse_args()
|
|
asyncio.run(probe(args.socket, args.proof))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|