94 lines
3.4 KiB
Python
Executable file
94 lines
3.4 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Host end of the X68000 serial gate.
|
|
|
|
MAME's null_modem socket form CONNECTS outward, so this side listens. Once the
|
|
guest's SERIAL.X is up it echoes every byte it receives, so sending a probe
|
|
string and reading it back proves both directions of the link.
|
|
|
|
The guest takes a while to appear: Human68k boots, then COMMAND.X runs
|
|
AUTOEXEC.BAT, then SERIAL.X initialises. Rather than guess, this retransmits
|
|
the probe periodically until the echo comes back or the deadline passes.
|
|
"""
|
|
|
|
import argparse
|
|
import socket
|
|
import sys
|
|
import time
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--port", type=int, default=6800)
|
|
ap.add_argument("--probe", default="JOEYLIB-X68K")
|
|
ap.add_argument("--timeout", type=float, default=420.0)
|
|
ap.add_argument("--until", default="",
|
|
help="listen-only: keep collecting until this text arrives")
|
|
ap.add_argument("--listen-only", action="store_true",
|
|
help="never transmit; pass if ANY bytes arrive. Isolates the\n guest TX path (e.g. keyboard -> jlSerialWrite) from echo.")
|
|
args = ap.parse_args()
|
|
|
|
probe = args.probe.encode("ascii")
|
|
deadline = time.time() + args.timeout
|
|
|
|
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
srv.bind(("127.0.0.1", args.port))
|
|
srv.listen(1)
|
|
srv.settimeout(max(1.0, args.timeout))
|
|
print(f"X68SER listening on 127.0.0.1:{args.port}")
|
|
|
|
try:
|
|
conn, peer = srv.accept()
|
|
except socket.timeout:
|
|
print("X68SER FAIL: MAME never connected")
|
|
return 1
|
|
print(f"X68SER connected from {peer}")
|
|
conn.settimeout(2.0)
|
|
|
|
received = bytearray()
|
|
sent_total = 0
|
|
last_send = 0.0
|
|
|
|
while time.time() < deadline:
|
|
# Retransmit periodically: the guest is not listening until SERIAL.X
|
|
# has booted and opened the port, and we cannot see when that happens.
|
|
if not args.listen_only and time.time() - last_send > 5.0:
|
|
try:
|
|
conn.sendall(probe)
|
|
sent_total += len(probe)
|
|
last_send = time.time()
|
|
except OSError as exc:
|
|
print(f"X68SER FAIL: send error {exc}")
|
|
return 1
|
|
try:
|
|
chunk = conn.recv(256)
|
|
except socket.timeout:
|
|
continue
|
|
except OSError as exc:
|
|
print(f"X68SER FAIL: recv error {exc}")
|
|
return 1
|
|
if not chunk:
|
|
print("X68SER FAIL: link closed by MAME")
|
|
return 1
|
|
received += chunk
|
|
print(f"X68SER rx {len(chunk)} bytes: {chunk!r}")
|
|
if args.listen_only and args.until and args.until.encode() in received:
|
|
print(f"X68SER PASS: guest transmitted {len(received)} bytes "
|
|
f"unprompted: {bytes(received)!r}")
|
|
return 0
|
|
if probe in received:
|
|
elapsed = args.timeout - (deadline - time.time())
|
|
print(f"X68SER PASS: probe echoed back after {elapsed:.1f}s "
|
|
f"(sent {sent_total} bytes, received {len(received)})")
|
|
return 0
|
|
|
|
if args.listen_only and received:
|
|
print(f"X68SER COLLECTED {len(received)} bytes: {bytes(received)!r}")
|
|
return 0
|
|
print(f"X68SER FAIL: timeout. sent={sent_total} received={len(received)} "
|
|
f"bytes: {bytes(received[:64])!r}")
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|