200 lines
8.7 KiB
Python
200 lines
8.7 KiB
Python
#!/usr/bin/env python3
|
|
# testStrap.py - the $DF00 strap option, end to end.
|
|
#
|
|
# python3 testStrap.py <disk.d64> <shotPrefix>
|
|
#
|
|
# A real SwiftLink has a jumper that moves the 6551 from I/O1 ($DE00) to I/O2 ($DF00), and some
|
|
# Ultimate configurations default to the second setting. The driver probes $DE00 first and falls
|
|
# back to $DF00, but every run before this one answered at $DE00, so the fallback had never executed.
|
|
#
|
|
# This script starts VICE with the cartridge at $DF00 only and watches BOTH pages at once:
|
|
#
|
|
# * $DE00-$DE03 - where the probe looks first. With nothing mapped there VICE returns open bus,
|
|
# so the write of the probe pattern and the read that fails to match should both show up here,
|
|
# and nothing else ever should.
|
|
# * $DF00-$DF03 - the real registers. Every ACIA access the driver makes has to land here.
|
|
# * $DE04-$DEFF and $DF04-$DFFF - the register mirrors. A real SwiftLink decodes only A0 and A1
|
|
# inside its page, so all 64 four-byte windows of the page are the same four registers; VICE maps
|
|
# only the first four bytes, which makes these ranges a clean tripwire: a stop anywhere in them is
|
|
# an instruction reaching into the cartridge page that has no business being there, and on real
|
|
# hardware would be a register access. The stock driver's 'sta $DF59,y' is exactly that case.
|
|
#
|
|
# It then reads back the six self-modified accessors at $E551-$E56E, which are the driver's whole
|
|
# record of where the hardware is.
|
|
import os
|
|
import re
|
|
import sys
|
|
import time
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from viceHarness import ViceSession, aciaArgs, readByte, ACIA_BASE, SCRATCH
|
|
from testModemSelect import SerialSink, collectHits
|
|
|
|
SHOTS = os.path.join(os.path.dirname(os.path.abspath(__file__)), "shots")
|
|
LOGS = os.path.join(os.path.dirname(os.path.abspath(__file__)), "testLogs")
|
|
|
|
# The six one-instruction ACIA accessors. Each is a three-byte absolute instruction whose operand
|
|
# high byte aciaSetPage patches; the low byte says which register it is.
|
|
ACCESSORS = [(0xE554, "aciaCmdWrite sta $xx02 command"),
|
|
(0xE558, "aciaStatusRead lda $xx01 status"),
|
|
(0xE55F, "aciaDataRead lda $xx00 receive"),
|
|
(0xE563, "aciaDataWrite sta $xx00 transmit"),
|
|
(0xE567, "aciaCtrlWrite sta $xx03 control"),
|
|
(0xE56B, "aciaCtrlRead lda $xx03 control read-back / the probe")]
|
|
|
|
STOP = re.compile(r"#(\d+)\s+\(Stop on\s+(\w+)\s+([0-9a-f]{4})\)")
|
|
DISASM = re.compile(r"\.C:([0-9a-f]{4})\s+((?:[0-9A-F]{2} )+)\s*(\S+)\s+(\S*)\s*-\s*A:([0-9A-Fa-f]{2})")
|
|
|
|
|
|
def describe(blob):
|
|
# Turn the monitor's watchpoint stops into an ordered list of "who touched what, with what".
|
|
out = []
|
|
lines = blob.splitlines()
|
|
for index, line in enumerate(lines):
|
|
stop = STOP.search(line)
|
|
if not stop:
|
|
continue
|
|
detail = ""
|
|
for follow in lines[index:index + 4]:
|
|
hit = DISASM.search(follow)
|
|
if hit:
|
|
detail = (f"{hit.group(3)} {hit.group(4)} at ${hit.group(1)} "
|
|
f"(A=${hit.group(5).upper()})")
|
|
break
|
|
out.append((int(stop.group(1)), stop.group(2), int(stop.group(3), 16), detail))
|
|
return out
|
|
|
|
|
|
def readAccessors(session):
|
|
# Each accessor is one three-byte absolute instruction, so the operand high byte - the only place
|
|
# the driver records which page the cartridge is on - is at the instruction address plus two.
|
|
values = {}
|
|
for addr, name in ACCESSORS:
|
|
values[addr] = (name, readByte(session, addr + 2))
|
|
return values
|
|
|
|
|
|
def showAccessors(values, note):
|
|
print(f"--- the six ACIA accessors' operand high bytes ({note}) ---", flush=True)
|
|
for addr, name in ACCESSORS:
|
|
_, value = values[addr]
|
|
page = "??" if value is None else f"${value:02X}00"
|
|
print(f" ${addr:04X} {name:42s} high byte at ${addr + 2:04X} = "
|
|
f"${'??' if value is None else '%02X' % value} -> page {page}", flush=True)
|
|
|
|
|
|
def main():
|
|
disk = os.path.abspath(sys.argv[1])
|
|
prefix = sys.argv[2] if len(sys.argv) > 2 else "strap"
|
|
os.makedirs(SHOTS, exist_ok=True)
|
|
os.makedirs(LOGS, exist_ok=True)
|
|
sink = SerialSink()
|
|
print(f"serial sink listening on 127.0.0.1:{sink.port}", flush=True)
|
|
args = aciaArgs(rsDevAddress=f"127.0.0.1:{sink.port}", baud=2400)
|
|
print(f"acia base for this run: ${ACIA_BASE:04X}", flush=True)
|
|
print("acia args:", args, flush=True)
|
|
session = ViceSession(disk, f"{SCRATCH}/{prefix}.vice.log", args, label=prefix)
|
|
try:
|
|
session.connect()
|
|
session.bootPastLoader()
|
|
session.findWindow()
|
|
session.focus()
|
|
time.sleep(8)
|
|
session.shot(f"{SHOTS}/{prefix}01menu.png")
|
|
|
|
session.enterMonitor()
|
|
row = readByte(session, 0x91D5)
|
|
print(f"menu row before input: {row}", flush=True)
|
|
session.sock.sendall(b"x\n")
|
|
session.recv(3)
|
|
for _ in range(3):
|
|
session.enterMonitor()
|
|
row = readByte(session, 0x91D5)
|
|
session.sock.sendall(b"x\n")
|
|
session.recv(3)
|
|
if row == 0:
|
|
break
|
|
session.hold("KP_8", 250)
|
|
time.sleep(0.8)
|
|
print(f"menu row after input: {row}", flush=True)
|
|
|
|
# Both pages watched before the module can touch anything.
|
|
session.enterMonitor()
|
|
for kind in ("store", "load"):
|
|
session.mon(f"watch {kind} $de00 $de03") # where the probe looks first
|
|
session.mon(f"watch {kind} $df00 $df03") # the real registers
|
|
session.mon(f"watch {kind} $de04 $deff") # mirrors: nothing should ever land here
|
|
session.mon(f"watch {kind} $df04 $dfff")
|
|
session.sock.sendall(b"x\n")
|
|
session.recv(2)
|
|
|
|
session.hold("KP_0", 300)
|
|
hits = []
|
|
for attempt in range(24):
|
|
session.hold("a", 250)
|
|
session.hold("space", 250)
|
|
hits += collectHits(session, 4, maxHits=400)
|
|
if hits:
|
|
print(f"first cartridge-page access after {attempt + 1} attempt(s)", flush=True)
|
|
break
|
|
hits += collectHits(session, 25, maxHits=400)
|
|
blob = "".join(hits)
|
|
open(f"{LOGS}/{prefix}.hits.txt", "w").write(blob)
|
|
events = describe(blob)
|
|
print(f"=== {len(events)} watchpoint stops, in order ===", flush=True)
|
|
for number, kind, addr, detail in events[:80]:
|
|
page = "$DE00 probe" if 0xDE00 <= addr <= 0xDE03 else \
|
|
"$DF00 registers" if 0xDF00 <= addr <= 0xDF03 else "MIRROR - unexpected"
|
|
print(f" {number:4d} {kind:5s} ${addr:04X} [{page}] {detail}", flush=True)
|
|
mirrors = [event for event in events if not (0xDE00 <= event[2] <= 0xDE03 or
|
|
0xDF00 <= event[2] <= 0xDF03)]
|
|
print(f"accesses to the mirror ranges $DE04-$DEFF / $DF04-$DFFF: {len(mirrors)}", flush=True)
|
|
for event in mirrors[:40]:
|
|
print(f" MIRROR {event}", flush=True)
|
|
|
|
session.enterMonitor()
|
|
showAccessors(readAccessors(session), "after the link was opened")
|
|
session.mon("m e540 e56f")
|
|
session.mon("m e030 e05f")
|
|
session.mon("m e5b8 e5bf")
|
|
session.mon("m df00 df03")
|
|
session.mon("m de00 de03")
|
|
session.mon("d fffa fffb")
|
|
session.sock.sendall(b"x\n")
|
|
session.recv(3)
|
|
time.sleep(8)
|
|
session.shot(f"{SHOTS}/{prefix}02afterOpen.png")
|
|
|
|
# Leave only the mirror tripwires armed and let the game run on, so anything the rest of the
|
|
# program does to the cartridge page while the link is live is caught.
|
|
session.enterMonitor()
|
|
session.mon("del")
|
|
for kind in ("store", "load"):
|
|
session.mon(f"watch {kind} $de04 $deff")
|
|
session.mon(f"watch {kind} $df04 $dfff")
|
|
session.sock.sendall(b"x\n")
|
|
session.recv(2)
|
|
late = collectHits(session, 45, maxHits=200)
|
|
lateEvents = describe("".join(late))
|
|
print(f"=== mirror-only watch, 45 s of running: {len(lateEvents)} stops ===", flush=True)
|
|
for event in lateEvents[:40]:
|
|
print(f" MIRROR {event}", flush=True)
|
|
open(f"{LOGS}/{prefix}.mirror.txt", "w").write("".join(late))
|
|
|
|
session.enterMonitor()
|
|
session.mon("del")
|
|
session.sock.sendall(b"x\n")
|
|
session.recv(3)
|
|
time.sleep(6)
|
|
session.shot(f"{SHOTS}/{prefix}03later.png")
|
|
finally:
|
|
session.close()
|
|
time.sleep(1)
|
|
sink.stop()
|
|
print(f"serial sink connected={sink.connected} bytes={len(sink.data)}", flush=True)
|
|
print("serial bytes:", sink.data[:400].hex(" "), flush=True)
|
|
print("as text:", repr(bytes(sink.data[:400])), flush=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|