Александр Петров
247 lines
14 KiB
Python
247 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""Живой raw MCP-пробник на C-приложении hello и подготовленной DSS-среде."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
|
|
from mcp import ClientSession, StdioServerParameters
|
|
from mcp.client.stdio import stdio_client
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
MAME_REPO = ROOT.parent / "MAME.HT"
|
|
sys.path.insert(0, str(ROOT / "toolchain"))
|
|
|
|
from make_disk import create_floppy_image # noqa: E402
|
|
from mame_interactive import basename_83, build_events # noqa: E402
|
|
from mame_profile import write_keyboard_config # noqa: E402
|
|
from sdbg.image import read_ihx # noqa: E402
|
|
from sdbg.model import DebugMap # noqa: E402
|
|
from sdbg_launcher import lua_script # noqa: E402
|
|
|
|
|
|
def answer_text(result) -> str:
|
|
return "".join(item.text for item in result.content if item.type == "text")
|
|
|
|
|
|
async def call(session: ClientSession, name: str, arguments=None) -> str:
|
|
result = await session.call_tool(name, arguments or {})
|
|
value = answer_text(result)
|
|
if result.isError or value.startswith("ERROR:"):
|
|
raise AssertionError(f"{name}: {value}")
|
|
return value
|
|
|
|
|
|
async def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--show-seconds", type=float, default=0,
|
|
help="Сколько секунд оставить видимым экран hello перед вводом x")
|
|
parser.add_argument("--proof", type=Path,
|
|
help="Сохранить снимок работающего hello вне временной сессии")
|
|
args = parser.parse_args()
|
|
home = Path(os.environ["MAME_HOME"]).resolve()
|
|
binary = Path(os.environ.get("MAME_BIN", MAME_REPO / "sprinter")).resolve()
|
|
rompath = Path(os.environ.get("MAME_ROMPATH", home / "roms")).resolve()
|
|
package = ROOT / "tests/hello/.sprinter-cc-hello"
|
|
model = DebugMap(package)
|
|
model.verify_executable()
|
|
executable = Path(model.manifest["executable_path"])
|
|
function = next(item for item in model.functions if item["name"] == "main")
|
|
main_address = function["start"]
|
|
image = read_ihx(package / (executable.stem + ".ihx"))
|
|
signature = bytes(image[addr] for addr in range(
|
|
function["start"], min(function["end"], function["start"] + 24)))
|
|
variable = next(item for item in model.variables if item["name"] == "errno")
|
|
raw_address = 0x10000 + variable["logical_address"]
|
|
|
|
with tempfile.TemporaryDirectory(prefix="sprinter-raw-hello-") as directory:
|
|
state = Path(directory)
|
|
for name in ("ipc", "cfg", "nvram", "diff", "snapshot"):
|
|
(state / name).mkdir()
|
|
write_keyboard_config(state / "cfg")
|
|
shutil.copyfile(home / "IMG/sp_hdd_sys.chd", state / "system.chd")
|
|
disk = state / "debug.img"
|
|
if not create_floppy_image(str(disk),
|
|
[(basename_83(executable.name), str(executable))]):
|
|
raise RuntimeError("Не удалось создать тестовую дискету")
|
|
ready = state / "main.json"
|
|
lua = state / "launch.lua"
|
|
events = build_events([(0, "a:\\" + basename_83(executable.name).replace(" ", "") + "\n")])
|
|
lua_script(lua, ready, main_address, signature, events, 0, 30)
|
|
env = dict(os.environ, MAME_MCP_DIR=str(state / "ipc"),
|
|
MAME_MCP_SNAP_DIR=str(state / "snapshot"),
|
|
SDL_NO_SIGNAL_HANDLERS="1")
|
|
command = [str(binary), "sprinter", "-noreadconfig", "-rompath", str(rompath),
|
|
"-bios", "v3.06", "-kbd", "ms_naturl,bios=sp2k",
|
|
"-video", "soft", "-window", "-sound", "none", "-skip_gameinfo",
|
|
"-beta:wd179x:0", "35hd", "-beta:wd179x:1", "35hd",
|
|
"-flop1", str(disk), "-flop2", str(home / "IMG/dss171u.img"),
|
|
"-hard1", str(state / "system.chd"),
|
|
"-debug", "-debugger", "sdbg", "-plugin", "mamebridge",
|
|
"-pluginspath", str(MAME_REPO / "plugins"),
|
|
"-autoboot_delay", "0", "-autoboot_script", str(lua)]
|
|
for name in ("cfg", "nvram", "diff", "snapshot"):
|
|
command.extend(["-" + name + "_directory", str(state / name)])
|
|
log_path = state / "mame.log"
|
|
with log_path.open("w") as log:
|
|
machine = subprocess.Popen(command, cwd=state, env=env,
|
|
stdout=log, stderr=subprocess.STDOUT)
|
|
try:
|
|
params = StdioServerParameters(
|
|
command=sys.executable, args=[str(MAME_REPO / "src/mame_mcp.py")],
|
|
env=env)
|
|
async with stdio_client(params) as (read, write):
|
|
async with ClientSession(read, write) as session:
|
|
deadline = time.monotonic() + 75
|
|
await session.initialize()
|
|
while not ready.exists():
|
|
if machine.poll() is not None:
|
|
raise RuntimeError("MAME завершился до main")
|
|
if (state / "main.json.error").exists():
|
|
raise RuntimeError((state / "main.json.error").read_text())
|
|
if time.monotonic() > deadline:
|
|
raise TimeoutError("MAME не дошёл до main")
|
|
await asyncio.sleep(0.1)
|
|
status = await call(session, "status")
|
|
if not status.startswith("state=stop"):
|
|
raise AssertionError(f"Ожидалась остановка в main: {status}")
|
|
logical = await call(session, "read_logical_memory",
|
|
{"address": hex(variable["logical_address"]),
|
|
"length": 1})
|
|
original = await call(session, "read_memory",
|
|
{"address": hex(raw_address), "length": 1})
|
|
if logical.strip() != original.strip():
|
|
raise AssertionError(f"logical/raw memory diverged: {logical}/{original}")
|
|
changed = int(original, 16) ^ 1
|
|
await call(session, "write_memory",
|
|
{"address": hex(raw_address), "hex_bytes": f"{changed:02X}"})
|
|
actual = await call(session, "read_memory",
|
|
{"address": hex(raw_address), "length": 1})
|
|
if actual.strip() != f"{changed:02X}":
|
|
raise AssertionError(f"write_memory failed: {actual}")
|
|
await call(session, "write_memory",
|
|
{"address": hex(raw_address), "hex_bytes": original.strip()})
|
|
restored = await call(session, "read_memory",
|
|
{"address": hex(raw_address), "length": 1})
|
|
if restored.strip() != original.strip():
|
|
raise AssertionError("scratch byte not restored")
|
|
|
|
shares = await call(session, "list_shares")
|
|
if "vram" not in shares:
|
|
raise AssertionError(f"VRAM share missing: {shares}")
|
|
await call(session, "read_vram", {"address": "0", "length": 4})
|
|
await call(session, "read_share",
|
|
{"name": "vram", "address": "0", "length": 4})
|
|
pixels = await call(session, "read_screen_pixels",
|
|
{"x": 0, "y": 0, "width": 2, "height": 2})
|
|
if not pixels.strip():
|
|
raise AssertionError("screen pixels are empty")
|
|
await call(session, "list_ports")
|
|
await call(session, "disassemble",
|
|
{"address": hex(0x10000 + main_address), "num_bytes": 16})
|
|
shot = await call(session, "screenshot", {"name": "raw-hello.png"})
|
|
if "raw-hello.png" not in shot or not (state / "snapshot/raw-hello.png").is_file():
|
|
raise AssertionError(f"screenshot path missing: {shot}")
|
|
|
|
point = await call(session, "set_watchpoint",
|
|
{"address": hex(raw_address), "length": 2,
|
|
"access": "w", "space": "program"})
|
|
match = re.search(r"Watchpoint (\d+) set", point)
|
|
if not match:
|
|
raise AssertionError(f"watchpoint ID missing: {point}")
|
|
wp_id = int(match.group(1))
|
|
await call(session, "resume")
|
|
deadline = time.monotonic() + 8
|
|
while time.monotonic() < deadline:
|
|
status = await call(session, "status")
|
|
if status.startswith("state=stop"):
|
|
break
|
|
await asyncio.sleep(0.1)
|
|
else:
|
|
raise TimeoutError("errno watchpoint did not stop CPU")
|
|
await call(session, "clear_watchpoint", {"index": wp_id})
|
|
after_getchar = model.line_locations("hello.c", 63)["locations"][0]["logical_address"]
|
|
point = await call(session, "set_breakpoint",
|
|
{"address": hex(after_getchar)})
|
|
match = re.search(r"Breakpoint (\d+) set", point)
|
|
if not match:
|
|
raise AssertionError(f"breakpoint ID missing: {point}")
|
|
bp_id = int(match.group(1))
|
|
await call(session, "resume")
|
|
await asyncio.sleep(0.5)
|
|
running_shot = await call(session, "screenshot",
|
|
{"name": "hello-running.png"})
|
|
running_image = state / "snapshot/hello-running.png"
|
|
if "hello-running.png" not in running_shot or not running_image.is_file():
|
|
raise AssertionError(f"working hello screenshot missing: {running_shot}")
|
|
if args.proof:
|
|
args.proof.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copyfile(running_image, args.proof)
|
|
print(f"Снимок работающего hello: {args.proof}", flush=True)
|
|
if args.show_seconds:
|
|
await asyncio.sleep(args.show_seconds)
|
|
await call(session, "move_mouse", {"dx": 1, "dy": 0, "frames": 2})
|
|
await call(session, "click_mouse", {"button": "left", "frames": 2})
|
|
await call(session, "press_input",
|
|
{"port": ":JOY1", "mask": "0x400", "frames": 2})
|
|
await call(session, "set_input",
|
|
{"port": ":JOY1", "mask": "0x400", "value": 1,
|
|
"frames": 2})
|
|
await call(session, "press_key", {"key": "x", "frames": 3})
|
|
deadline = time.monotonic() + 8
|
|
while time.monotonic() < deadline:
|
|
status = await call(session, "status")
|
|
if status.startswith("state=stop"):
|
|
break
|
|
await asyncio.sleep(0.1)
|
|
else:
|
|
raise TimeoutError("press_key did not leave getchar")
|
|
if f"PC=0x{after_getchar:X}" not in status:
|
|
raise AssertionError(f"stopped before line 63: {status}")
|
|
await call(session, "clear_breakpoint", {"index": bp_id})
|
|
await call(session, "step_over", {"count": 1})
|
|
deadline = time.monotonic() + 8
|
|
while time.monotonic() < deadline:
|
|
status = await call(session, "status")
|
|
if status.startswith("state=stop"):
|
|
break
|
|
await asyncio.sleep(0.1)
|
|
else:
|
|
raise TimeoutError("step_over did not stop")
|
|
await call(session, "step_out")
|
|
deadline = time.monotonic() + 8
|
|
while time.monotonic() < deadline:
|
|
status = await call(session, "status")
|
|
if status.startswith("state=stop"):
|
|
break
|
|
await asyncio.sleep(0.1)
|
|
else:
|
|
raise TimeoutError("step_out did not stop")
|
|
await call(session, "resume")
|
|
await call(session, "type_string", {"text": "a"})
|
|
await call(session, "type_text", {"text": "b"})
|
|
print(f"main=0x{main_address:04X}; errno=0x{raw_address:X}; "
|
|
f"write/read-back OK; watchpoint={wp_id}; "
|
|
f"keyboard→line63=0x{after_getchar:04X}; "
|
|
f"pixels={pixels.strip()[:50]}; screenshot={running_shot.strip()}")
|
|
except BaseException:
|
|
print(log_path.read_text(errors="replace")[-4000:], file=sys.stderr)
|
|
raise
|
|
finally:
|
|
machine.terminate()
|
|
machine.wait(timeout=8)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|