OXYGENHUB
# 398756c0e9b2 retro-remote-debug-controller / client / emu_control.py
8.3 KiB · Python · last changed 0.4.0: pointer injection (POST/GET /pointer) 3 months ago source blame history raw
1 
"""emu_control.py — shared client for the Retro Remote Debug Controller.
2
3 
One client for every port's pytest suite. Speaks the HTTP contract in
4 
../SPEC.md (0.4.x). Depends only on the stdlib; Pillow is optional and only
5 
needed for .screenshot().
6
7 
    from emu_control import EmuControl
8
9 
    emu = EmuControl(port=8080)      # emulator started with -controlport 8080
10 
    emu.wait_ready()
11 
    emu.step(frames=120)
12 
    assert emu.regs()["a"] == 0
13 
    data = emu.mem(0x0400, 256, bank=1)
14 
    img = emu.screenshot()           # PIL.Image (requires Pillow)
15 
"""
16
17 
from __future__ import annotations
18
19 
import io
20 
import json
21 
import time
22 
import urllib.error
23 
import urllib.request
24
25 
CONTRACT_MAJOR = 0
26
27
28 
class EmuControlError(RuntimeError):
29 
    pass
30
31
32 
def parse_ppm_header(data: bytes):
33 
    """Parse a binary PPM (P6) header. Returns (width, height, maxval, offset)
34 
    where offset is the index of the first pixel byte. Raises on malformed
35 
    input. Stdlib-only — no Pillow needed."""
36 
    if data[:2] != b"P6":
37 
        raise EmuControlError(f"not a P6 PPM (magic {data[:2]!r})")
38 
    fields, i, n = [], 2, len(data)
39 
    while len(fields) < 3 and i < n:
40 
        while i < n and data[i:i + 1].isspace():
41 
            i += 1
42 
        if i < n and data[i:i + 1] == b"#":            # comment to end of line
43 
            while i < n and data[i:i + 1] not in (b"\n", b"\r"):
44 
                i += 1
45 
            continue
46 
        start = i
47 
        while i < n and not data[i:i + 1].isspace():
48 
            i += 1
49 
        fields.append(int(data[start:i]))
50 
    if len(fields) < 3:
51 
        raise EmuControlError("truncated PPM header")
52 
    return fields[0], fields[1], fields[2], i + 1        # +1: single whitespace
53
54
55 
class EmuControl:
56 
    def __init__(self, port: int, host: str = "127.0.0.1", timeout: float = 5.0):
57 
        self.base = f"http://{host}:{port}"
58 
        self.timeout = timeout
59
60 
    # -- transport ---------------------------------------------------------
61 
    def _get(self, path: str) -> tuple[bytes, str]:
62 
        try:
63 
            with urllib.request.urlopen(self.base + path, timeout=self.timeout) as r:
64 
                return r.read(), r.headers.get_content_type()
65 
        except urllib.error.HTTPError as e:
66 
            raise EmuControlError(f"GET {path} -> {e.code}: {e.read()!r}") from e
67
68 
    def _post(self, path: str, body: bytes = b"") -> bytes:
69 
        req = urllib.request.Request(self.base + path, method="POST", data=body)
70 
        try:
71 
            with urllib.request.urlopen(req, timeout=self.timeout) as r:
72 
                return r.read()
73 
        except urllib.error.HTTPError as e:
74 
            raise EmuControlError(f"POST {path} -> {e.code}: {e.read()!r}") from e
75
76 
    # -- endpoints ---------------------------------------------------------
77 
    def status(self) -> dict:
78 
        """GET /status — liveness + contract dict; raises on a MAJOR mismatch."""
79 
        body, _ = self._get("/status")
80 
        st = json.loads(body)
81 
        major = int(str(st.get("contract", "0")).split(".")[0])
82 
        if major != CONTRACT_MAJOR:
83 
            raise EmuControlError(
84 
                f"contract major mismatch: emulator {st.get('contract')} "
85 
                f"vs client {CONTRACT_MAJOR}.x"
86 
            )
87 
        return st
88
89 
    def frame(self) -> int:
90 
        """Current completed-frame counter (the determinism anchor)."""
91 
        return int(self.status()["frame"])
92
93 
    def regs(self) -> dict:
94 
        """GET /regs — CPU registers as a dict (keys are platform-specific)."""
95 
        body, _ = self._get("/regs")
96 
        return json.loads(body)
97
98 
    def mem(self, addr: int, length: int, bank: int | None = None) -> bytes:
99 
        """GET /mem — read `length` bytes at `addr` (optional `bank`)."""
100 
        path = f"/mem?addr={addr}&len={length}"
101 
        if bank is not None:
102 
            path += f"&bank={bank}"
103 
        body, _ = self._get(path)
104 
        return body
105
106 
    def step(self, frames: int = 1) -> int:
107 
        """POST /step — advance `frames` complete frames, halt; returns the new
108 
        frame counter. Meaningful once paused."""
109 
        body = self._post(f"/step?frames={frames}")
110 
        return int(json.loads(body)["frame"])
111
112 
    def pause(self) -> None:
113 
        """POST /pause — halt the machine (advances only via step())."""
114 
        self._post("/pause")
115
116 
    def resume(self) -> None:
117 
        """POST /resume — return the machine to free-running."""
118 
        self._post("/resume")
119
120 
    def key(self, text: str | None = None, code: int | None = None,
121 
            down: bool | None = None) -> None:
122 
        """Inject a key (contract 0.2). Provide `text` (a single character) or
123 
        `code` (a raw platform key code). down=True presses/holds, False
124 
        releases, None taps (press then release). Hold across step() frames for
125 
        a program that samples input each frame."""
126 
        parts = []
127 
        if text is not None:
128 
            parts.append(f"text={text}")
129 
        if code is not None:
130 
            parts.append(f"code={code}")
131 
        if down is not None:
132 
            parts.append(f"down={1 if down else 0}")
133 
        if not parts:
134 
            raise EmuControlError("key() needs text= or code=")
135 
        self._post("/key?" + "&".join(parts))
136
137 
    def reset(self) -> None:
138 
        """Soft/cold reset the machine (contract 0.2)."""
139 
        self._post("/reset")
140
141 
    def mem_write(self, addr: int, data: bytes, bank: int | None = None) -> int:
142 
        """Write/poke bytes at addr (contract 0.3). Returns the number written."""
143 
        path = f"/mem?addr={addr}"
144 
        if bank is not None:
145 
            path += f"&bank={bank}"
146 
        return int(json.loads(self._post(path, bytes(data)))["written"])
147
148 
    def audio(self) -> bytes:
149 
        """Raw WAV (PCM) of audio synthesized since the last call (contract 0.3).
150 
        Drain regularly (e.g. after each step) so the ring doesn't overflow."""
151 
        body, _ = self._get("/audio")
152 
        return body
153
154 
    def pointer(self, x: int | None = None, y: int | None = None,
155 
                dx: int | None = None, dy: int | None = None,
156 
                buttons: int | None = None) -> None:
157 
        """Inject a pointer move/click (contract 0.4). Give absolute (x, y) or
158 
        relative (dx, dy). `buttons` is a bitmask (bit0=primary, bit1=secondary);
159 
        omit it for a pure move (leaves the button state unchanged). Hold buttons
160 
        across step() frames for a program that samples the pointer each frame."""
161 
        parts = []
162 
        if x is not None and y is not None:
163 
            parts.append(f"x={x}")
164 
            parts.append(f"y={y}")
165 
        elif dx is not None or dy is not None:
166 
            parts.append(f"dx={dx if dx is not None else 0}")
167 
            parts.append(f"dy={dy if dy is not None else 0}")
168 
        else:
169 
            raise EmuControlError("pointer() needs x= and y=, or dx=/dy=")
170 
        if buttons is not None:
171 
            parts.append(f"buttons={buttons}")
172 
        self._post("/pointer?" + "&".join(parts))
173
174 
    def pointer_get(self) -> dict:
175 
        """GET /pointer — the current pointer as {x, y, buttons} (contract 0.4).
176 
        `buttons` uses the same bitmask as pointer()."""
177 
        body, _ = self._get("/pointer")
178 
        return json.loads(body)
179
180 
    def screenshot_ppm(self) -> bytes:
181 
        """Return the live screen as raw PPM (P6) bytes. Stdlib-only."""
182 
        body, _ = self._get("/screenshot")
183 
        return body
184
185 
    def screenshot(self):
186 
        """Return the live screen as a PIL.Image (requires Pillow)."""
187 
        body = self.screenshot_ppm()
188 
        try:
189 
            from PIL import Image
190 
        except ImportError as e:  # pragma: no cover
191 
            raise EmuControlError("screenshot() needs Pillow (pip install pillow)") from e
192 
        return Image.open(io.BytesIO(body))
193
194 
    # -- helpers -----------------------------------------------------------
195 
    def wait_ready(self, tries: int = 100, delay: float = 0.05) -> dict:
196 
        """Poll /status until the server answers (emulator boot race)."""
197 
        last = None
198 
        for _ in range(tries):
199 
            try:
200 
                return self.status()
201 
            except (EmuControlError, urllib.error.URLError, ConnectionError) as e:
202 
                last = e
203 
                time.sleep(delay)
204 
        raise EmuControlError(f"emulator control port never came up: {last}")
205
206 
    def run_to_frame(self, target: int) -> int:
207 
        """Step until the frame counter reaches >= target."""
208 
        cur = self.frame()
209 
        if target > cur:
210 
            self.step(frames=target - cur)
211 
        return self.frame()