1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
|
#!/usr/bin/env python3
"""gateway.py - host-side "link cable adapter" for gbos networking.
Wraps the emulator, owns its link-port serial (stdio), speaks SLIP, and does
the real network work the Game Boy can't (DNS/TCP/HTTP). Modes:
echo - bounce every frame back (loopback test)
http - treat each frame as a URL, HTTP GET it, stream the body back
Reply frames are typed: b'D'+chunk for data, b'E' to end the stream.
usage: tools/gateway.py [--mode echo|http] [--cmd "wget example.com"] [--rom PATH]
"""
import os, sys, subprocess, threading, time, argparse, urllib.request
EMU = os.path.expanduser("~/dev/gbc/build/sl0pboy")
END, ESC, ESC_END, ESC_ESC = 0xC0, 0xDB, 0xDC, 0xDD
CHUNK = 200
LIMIT = 800 # cap the fetched body (link is ~1-32 KB/s)
def slip_encode(payload):
out = bytearray([END])
for c in payload:
if c == END: out += bytes([ESC, ESC_END])
elif c == ESC: out += bytes([ESC, ESC_ESC])
else: out.append(c)
out.append(END)
return bytes(out)
def http_fetch(payload):
url = bytes(payload).decode("latin1", "replace").strip()
if not url.startswith("http"):
url = "http://" + url
sys.stderr.write("\n[gw] GET %s\n" % url)
try:
req = urllib.request.Request(url, headers={"User-Agent": "gbos/0.1"})
with urllib.request.urlopen(req, timeout=8) as r:
body = r.read(LIMIT)
except Exception as e:
body = ("ERR: " + str(e)).encode()
sys.stderr.write("[gw] %d bytes\n" % len(body))
frames = [b"D" + body[i:i+CHUNK] for i in range(0, len(body), CHUNK)]
frames.append(b"E")
return frames
def handle(payload, mode):
if mode == "http":
return http_fetch(payload)
if mode == "chat":
msg = bytes(payload).decode("latin1", "replace")
sys.stderr.write("\n[gw] GB says: %s\n" % msg)
return [("<bot> " + msg).encode()] # a simple echo-bot peer
return [bytes(payload)] # echo
def reader(out, inp, mode):
buf, in_frame, esc = bytearray(), False, False
while True:
b = out.read(1)
if not b:
return
c = b[0]
if c == END:
if in_frame and buf:
for f in handle(buf, mode):
inp.write(slip_encode(f)); inp.flush()
buf, in_frame, esc = bytearray(), False, False
else:
buf, in_frame, esc = bytearray(), True, False
elif in_frame:
if esc:
c = {ESC_END: END, ESC_ESC: ESC}.get(c, c); esc = False; buf.append(c)
elif c == ESC:
esc = True
else:
buf.append(c)
else:
sys.stderr.write(chr(c) if 32 <= c < 127 else ("\n" if c == 10 else "."))
sys.stderr.flush()
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--mode", default="echo", choices=["echo", "http", "chat"])
ap.add_argument("--cmd", default="necho")
ap.add_argument("--rom", default=os.path.join(os.path.dirname(__file__), "..", "gbos.gb"))
ap.add_argument("--keys", default=None)
a = ap.parse_args()
cmd = [EMU, "--headless", "--uncapped"]
if a.keys: cmd += ["--keys", a.keys]
cmd.append(a.rom)
p = subprocess.Popen(cmd,
stdin=subprocess.PIPE, stdout=subprocess.PIPE, bufsize=0)
threading.Thread(target=reader, args=(p.stdout, p.stdin, a.mode), daemon=True).start()
time.sleep(1.0)
p.stdin.write((a.cmd + "\n").encode()); p.stdin.flush()
if a.mode == "chat":
time.sleep(1.0)
for m in [b"welcome to sl0pchat", b"<alice> hey gameboy!", b"<bob> nice link cable"]:
p.stdin.write(slip_encode(m)); p.stdin.flush(); time.sleep(0.6)
time.sleep(4.0)
p.stdin.write(b"exit\n"); p.stdin.flush()
try: p.wait(timeout=3)
except Exception: p.kill()
if __name__ == "__main__":
main()
|
