205 lines
7 KiB
Python
205 lines
7 KiB
Python
#!/usr/bin/env python3
|
|
# testTwoMachines.py - run two copies of the patched game, each with its own SwiftLink, and cross
|
|
# connect their serial lines so whatever one transmits the other receives.
|
|
#
|
|
# python3 testTwoMachines.py <disk.d64>
|
|
#
|
|
# VICE's -rsdev1 "host:port" makes the emulator the TCP *client*, so the two instances cannot be
|
|
# pointed at each other directly. A relay in this script accepts both connections and forwards bytes
|
|
# between them - a null modem cable made of sockets. It also logs every byte in both directions,
|
|
# which is what makes the exchange observable.
|
|
import os
|
|
import select
|
|
import socket
|
|
import sys
|
|
import threading
|
|
import time
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from viceHarness import ViceSession, aciaArgs, pickFreePort, readByte, SCRATCH
|
|
|
|
SHOTS = os.path.join(os.path.dirname(os.path.abspath(__file__)), "shots")
|
|
|
|
|
|
class NullModemRelay:
|
|
def __init__(self):
|
|
self.port = pickFreePort()
|
|
self.listener = socket.socket()
|
|
self.listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
self.listener.bind(("127.0.0.1", self.port))
|
|
self.listener.listen(4)
|
|
self.conns = []
|
|
self.log = [bytearray(), bytearray()]
|
|
self.running = True
|
|
threading.Thread(target=self.serve, daemon=True).start()
|
|
|
|
def serve(self):
|
|
self.listener.settimeout(0.5)
|
|
while self.running:
|
|
readable = list(self.conns)
|
|
if len(self.conns) < 2:
|
|
try:
|
|
conn, _ = self.listener.accept()
|
|
conn.setblocking(False)
|
|
self.conns.append(conn)
|
|
print(f"[relay] connection {len(self.conns)}", flush=True)
|
|
continue
|
|
except socket.timeout:
|
|
pass
|
|
except OSError:
|
|
return
|
|
if not readable:
|
|
time.sleep(0.1)
|
|
continue
|
|
ready, _, _ = select.select(readable, [], [], 0.2)
|
|
for conn in ready:
|
|
try:
|
|
chunk = conn.recv(4096)
|
|
except OSError:
|
|
chunk = b""
|
|
if not chunk:
|
|
continue
|
|
index = self.conns.index(conn)
|
|
self.log[index] += chunk
|
|
for other in self.conns:
|
|
if other is not conn:
|
|
try:
|
|
other.sendall(chunk)
|
|
except OSError:
|
|
pass
|
|
|
|
def stop(self):
|
|
self.running = False
|
|
try:
|
|
self.listener.close()
|
|
except OSError:
|
|
pass
|
|
for conn in self.conns:
|
|
try:
|
|
conn.close()
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def pickModemOpponent(session, tag, answerKey):
|
|
# Options menu: the highlighted row lives in $91D5, row 0 = COMPETE WITH MODEM OPPONENT. Fire
|
|
# picks it; the game then asks A/O (answer or originate) and then for space before it opens the
|
|
# link, so keep offering both until the ACIA command register stops reading back zero.
|
|
for _ in range(4):
|
|
session.enterMonitor()
|
|
row = readByte(session, 0x91D5)
|
|
session.sock.sendall(b"x\n")
|
|
session.recv(3)
|
|
print(f"[{tag}] menu row = {row}", flush=True)
|
|
if row == 0:
|
|
break
|
|
session.hold("KP_8", 250)
|
|
time.sleep(0.8)
|
|
session.hold("KP_0", 300)
|
|
time.sleep(2)
|
|
return answerKey
|
|
|
|
|
|
def answerModemPrompts(session, tag, answerKey):
|
|
# aciaCommandShadow ($E5BE) is zero until configureUserPortLines has programmed the 6551, so it
|
|
# is the cheapest "is the link open yet" test there is.
|
|
for attempt in range(24):
|
|
session.hold(answerKey, 250)
|
|
session.hold("space", 250)
|
|
time.sleep(1.5)
|
|
session.enterMonitor()
|
|
shadow = readByte(session, 0xE5BE)
|
|
session.sock.sendall(b"x\n")
|
|
session.recv(2)
|
|
if shadow:
|
|
print(f"[{tag}] link opened after {attempt + 1} attempt(s), "
|
|
f"command register = ${shadow:02X}", flush=True)
|
|
return True
|
|
print(f"[{tag}] link never opened", flush=True)
|
|
return False
|
|
|
|
|
|
def forceCarrier(session, tag):
|
|
# C= + C toggles carrierOverrideFlags ($E04A) bit 6, which makes the driver believe DCD is up.
|
|
# Two machines on a direct cable have no carrier signal, so without this they sit in the Hayes
|
|
# dialogue for ever. VICE maps the Commodore key to left Ctrl.
|
|
for _ in range(4):
|
|
session.key("ctrl+c")
|
|
time.sleep(1)
|
|
session.enterMonitor()
|
|
flags = readByte(session, 0xE04A)
|
|
phase = readByte(session, 0xE040)
|
|
session.sock.sendall(b"x\n")
|
|
session.recv(2)
|
|
print(f"[{tag}] carrierOverrideFlags = {flags}, connectionPhase = {phase}", flush=True)
|
|
if flags:
|
|
return True
|
|
return False
|
|
|
|
|
|
def main():
|
|
disk = os.path.abspath(sys.argv[1])
|
|
os.makedirs(SHOTS, exist_ok=True)
|
|
relay = NullModemRelay()
|
|
print(f"relay on 127.0.0.1:{relay.port}", flush=True)
|
|
args = aciaArgs(rsDevAddress=f"127.0.0.1:{relay.port}", baud=2400)
|
|
a = ViceSession(disk, f"{SCRATCH}/two.a.vice.log", args, label="A")
|
|
b = ViceSession(disk, f"{SCRATCH}/two.b.vice.log", args, label="B")
|
|
try:
|
|
a.connect()
|
|
b.connect()
|
|
a.bootPastLoader()
|
|
b.bootPastLoader()
|
|
a.findWindow()
|
|
b.findWindow()
|
|
a.focus()
|
|
b.focus()
|
|
time.sleep(8)
|
|
a.shot(f"{SHOTS}/two01aMenu.png")
|
|
b.shot(f"{SHOTS}/two01bMenu.png")
|
|
|
|
# A answers, B originates - the two halves of the Hayes dialogue the driver expects.
|
|
pickModemOpponent(a, "A", "a")
|
|
pickModemOpponent(b, "B", "o")
|
|
answerModemPrompts(a, "A", "a")
|
|
answerModemPrompts(b, "B", "o")
|
|
time.sleep(10)
|
|
a.shot(f"{SHOTS}/two02aLinkOpen.png")
|
|
b.shot(f"{SHOTS}/two02bLinkOpen.png")
|
|
report(relay)
|
|
|
|
# There is no carrier on a null modem cable, so tell both drivers to pretend there is one.
|
|
forceCarrier(a, "A")
|
|
forceCarrier(b, "B")
|
|
time.sleep(20)
|
|
a.shot(f"{SHOTS}/two03aCarrierForced.png")
|
|
b.shot(f"{SHOTS}/two03bCarrierForced.png")
|
|
report(relay)
|
|
|
|
for session, tag in ((a, "A"), (b, "B")):
|
|
session.enterMonitor()
|
|
session.mon("m e030 e047")
|
|
session.mon("m e5b8 e5bf")
|
|
session.mon("m de00 de03")
|
|
session.sock.sendall(b"x\n")
|
|
session.recv(3)
|
|
time.sleep(30)
|
|
a.shot(f"{SHOTS}/two04aLater.png")
|
|
b.shot(f"{SHOTS}/two04bLater.png")
|
|
report(relay)
|
|
finally:
|
|
a.close()
|
|
b.close()
|
|
time.sleep(1)
|
|
relay.stop()
|
|
report(relay)
|
|
|
|
|
|
def report(relay):
|
|
for i, buf in enumerate(relay.log):
|
|
name = "AB"[i] if i < 2 else str(i)
|
|
print(f"[relay] from {name}: {len(buf)} bytes: {bytes(buf[-120:])!r}", flush=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|