242 lines
11 KiB
Python
242 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
# test1200AutoBaud.py - the two things about 1200 baud that two linked machines cannot show.
|
|
#
|
|
# python3 test1200AutoBaud.py <disk.d64> [tag]
|
|
#
|
|
# 1. The auto-baud path. One machine, its serial line wired to a fake Hayes modem written in this
|
|
# script rather than to a peer. The fake modem answers the module's AT string with a verbose
|
|
# "CONNECT 1200" result code, which is exactly what serviceModemInput $E799 sniffs for: the CR
|
|
# that ends the line arrives with lastModemChar $EB02 = '0', so $E7B0 loads X = 3 and $E7B2 calls
|
|
# loadBaudParameters. What the ACIA control register holds afterwards is the whole question.
|
|
#
|
|
# 2. The same entry point driven directly. With the emulator stopped, the ACIA is forced back to
|
|
# 300 baud, PC is put on $E7B2 and X is loaded by hand, and the monitor steps over the JSR. Doing
|
|
# it for X = 0, 3 and 6 shows the three-byte stride of baudEntryTable $E6FD behaving, not just the
|
|
# one index the sniffer uses.
|
|
#
|
|
# Nothing here trusts a comment: every number below is read back out of the running machine.
|
|
import os
|
|
import re
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from viceHarness import ViceSession, aciaArgs, pickFreePort, SCRATCH
|
|
from testTwoMachines import pickModemOpponent
|
|
from testRealtime import (command, flush, pauseAll, resumeAll, readRange, snapshot, showState,
|
|
decodeControl, SHOTS, LOGS)
|
|
|
|
|
|
class FakeModem:
|
|
# A TCP listener that behaves like a Hayes modem far enough to exercise the result-code sniffer:
|
|
# it watches for the CR that ends the module's AT string and answers with a verbose CONNECT 1200.
|
|
# VICE's -rsdev1 makes the emulator the client, so this end listens.
|
|
def __init__(self, reply=b"\r\nCONNECT 1200\r"):
|
|
self.port = pickFreePort()
|
|
self.reply = reply
|
|
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(2)
|
|
self.rx = bytearray()
|
|
self.events = []
|
|
self.sent = 0
|
|
self.armed = True
|
|
self.running = True
|
|
self.t0 = time.time()
|
|
threading.Thread(target=self.serve, daemon=True).start()
|
|
|
|
def note(self, text):
|
|
self.events.append((round(time.time() - self.t0, 3), text))
|
|
print(f"[modem +{time.time() - self.t0:7.3f}s] {text}", flush=True)
|
|
|
|
def serve(self):
|
|
self.listener.settimeout(0.5)
|
|
conn = None
|
|
while self.running:
|
|
if conn is None:
|
|
try:
|
|
conn, _ = self.listener.accept()
|
|
conn.settimeout(0.2)
|
|
self.note("the emulator opened the serial line")
|
|
except socket.timeout:
|
|
continue
|
|
except OSError:
|
|
return
|
|
try:
|
|
chunk = conn.recv(4096)
|
|
except socket.timeout:
|
|
continue
|
|
except OSError:
|
|
chunk = b""
|
|
if not chunk:
|
|
continue
|
|
self.rx += chunk
|
|
self.note(f"from the C64: {bytes(chunk)!r}")
|
|
# The AT string is CR "ATQ0V1X1A" CR (answer) or ... "D" CR (originate). Answer the CR
|
|
# that ends it, once, as fast as a real modem never could - the module leaves terminal
|
|
# mode 121 frames after that CR, so the reply has about two emulated seconds to land.
|
|
if self.armed and chunk.rstrip(b"\x00").endswith(b"\r") and b"AT" in bytes(self.rx):
|
|
self.armed = False
|
|
try:
|
|
conn.sendall(self.reply)
|
|
self.sent += 1
|
|
self.note(f"answered {self.reply!r}")
|
|
except OSError:
|
|
pass
|
|
|
|
def stop(self):
|
|
self.running = False
|
|
try:
|
|
self.listener.close()
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def waitForModule(session, tag, tries=60):
|
|
# The opponent module is only fetched off track 34 once COMPETE WITH MODEM OPPONENT is picked,
|
|
# and at true C64 speed that 1541 load takes the best part of a minute. Until it lands,
|
|
# $E000-$EFFF still holds whatever the previous overlay left there, so every module address reads
|
|
# garbage - including $E5BE, which is what testTwoMachines.answerModemPrompts uses as its "the
|
|
# link is open" test, and which read $AC out of the wreckage in the first attempt at this test.
|
|
# The module's own jump table is the honest signal: sixteen bytes that cannot be anything else.
|
|
want = [0x4C, 0x11, 0xE1, 0x4C, 0x9A, 0xE2, 0x4C, 0xA9, 0xE0, 0x4C, 0xF6, 0xE0, 0x4C, 0xCB,
|
|
0xE3, 0x60]
|
|
for attempt in range(tries):
|
|
pauseAll([(session, tag)])
|
|
got = readRange(session, 0xE000, 16)
|
|
resumeAll([(session, tag)])
|
|
if got == want:
|
|
print(f"[{tag}] the module is resident: $E000 jump table matches", flush=True)
|
|
return True
|
|
print(f"[{tag}] waiting for the module to load: $E000 = "
|
|
f"{' '.join('??' if v is None else f'{v:02X}' for v in got)}", flush=True)
|
|
if attempt % 6 == 5:
|
|
# The fire press that starts the load does not always land - it did not in one run of
|
|
# this test - and there is no harm in offering it again while the options menu is still
|
|
# up, because the row is already on COMPETE WITH MODEM OPPONENT.
|
|
print(f"[{tag}] no sign of the load - pressing fire again", flush=True)
|
|
session.focus()
|
|
session.hold("KP_0", 300)
|
|
time.sleep(3)
|
|
return False
|
|
|
|
|
|
def openLink(session, tag, answerKey, tries=6):
|
|
# Answer both modem prompts and then leave the emulator completely alone for eight seconds: the
|
|
# AT string, the fake modem's answer, the result-code sniff and the 121-frame terminal-mode
|
|
# timeout all have to happen in emulated real time, and every monitor command stops the clock.
|
|
# $DE03 is the test - the ACIA control register is $00 until configureUserPortLines writes it.
|
|
for attempt in range(tries):
|
|
session.focus()
|
|
session.hold(answerKey, 250)
|
|
time.sleep(0.8)
|
|
session.hold("space", 250)
|
|
time.sleep(8)
|
|
state = snapshot(session, tag)
|
|
showState(state)
|
|
if state["aciaControl"]:
|
|
print(f"[{tag}] the link opened on attempt {attempt + 1}", flush=True)
|
|
return state
|
|
return state
|
|
|
|
|
|
def forceEntry(session, tag, x, expect):
|
|
# Drive $E7B2 with a chosen X and read back what the 6551 was given. The ACIA is put on 300 baud
|
|
# first so that a 1200-baud answer cannot be the value that was already there, and txCharActive
|
|
# is set so that aciaSetControlIdle's clear is visible too.
|
|
pauseAll([(session, tag)])
|
|
command(session, "sidefx on")
|
|
command(session, "> de03 15")
|
|
command(session, "sidefx off")
|
|
command(session, "> e056 00")
|
|
command(session, "> e5bb 01")
|
|
before = {"$DE03": readRange(session, 0xDE03, 1)[0], "$E055": readRange(session, 0xE055, 1)[0],
|
|
"$E056": readRange(session, 0xE056, 1)[0], "$E5BB": readRange(session, 0xE5BB, 1)[0]}
|
|
saved = command(session, "r")
|
|
command(session, f"r pc=e7b2")
|
|
command(session, f"r x={x:02x}")
|
|
regs = command(session, "r")
|
|
stepped = command(session, "n") # step over JSR loadBaudParameters $E353
|
|
after = {"$DE03": readRange(session, 0xDE03, 1)[0], "$E055": readRange(session, 0xE055, 1)[0],
|
|
"$E056": readRange(session, 0xE056, 1)[0], "$E5BB": readRange(session, 0xE5BB, 1)[0]}
|
|
result = {"x": x, "before": before, "after": after, "expectControl": expect,
|
|
"regsBefore": saved.strip().splitlines()[-2:], "regsSet": regs.strip().splitlines()[-2:],
|
|
"stopped": stepped.strip().splitlines()[-2:]}
|
|
ok = after["$DE03"] == expect and after["$E056"] == expect
|
|
result["ok"] = ok
|
|
print(f"\n*** forced entry at $E7B2 with X={x}: $DE03 ${before['$DE03']:02X} -> "
|
|
f"${after['$DE03']:02X}, $E056 ${before['$E056']:02X} -> ${after['$E056']:02X}, "
|
|
f"$E5BB ${before['$E5BB']:02X} -> ${after['$E5BB']:02X}, baudIndex $E055 "
|
|
f"${before['$E055']:02X} -> ${after['$E055']:02X}; expected control ${expect:02X}; "
|
|
f"{'PASS' if ok else 'FAIL'}", flush=True)
|
|
print(f" {decodeControl(after['$DE03'])}", flush=True)
|
|
return result
|
|
|
|
|
|
def main():
|
|
disk = os.path.abspath(sys.argv[1])
|
|
tag = sys.argv[2] if len(sys.argv) > 2 else "ab"
|
|
os.makedirs(SHOTS, exist_ok=True)
|
|
os.makedirs(LOGS, exist_ok=True)
|
|
modem = FakeModem()
|
|
print(f"fake modem listening on 127.0.0.1:{modem.port}", flush=True)
|
|
args = aciaArgs(rsDevAddress=f"127.0.0.1:{modem.port}", baud=2400)
|
|
a = ViceSession(disk, f"{SCRATCH}/ab.a.vice.log", args, label="A", warp=False)
|
|
sessions = [(a, "A")]
|
|
try:
|
|
a.connect()
|
|
command(a, "warp on")
|
|
a.bootPastLoader(waitSecs=900)
|
|
pauseAll(sessions)
|
|
command(a, "warp off")
|
|
print(f"[A] {command(a, 'warp').strip()}", flush=True)
|
|
resumeAll(sessions)
|
|
a.findWindow()
|
|
a.focus()
|
|
time.sleep(25)
|
|
a.shot(f"{SHOTS}/ab{tag}01Menu.png")
|
|
pickModemOpponent(a, "A", "a")
|
|
waitForModule(a, "A")
|
|
|
|
state = snapshot(a, "A")
|
|
showState(state)
|
|
print(f"before the link opens: {decodeControl(state['aciaControl'])}", flush=True)
|
|
|
|
after = openLink(a, "A", "a")
|
|
a.shot(f"{SHOTS}/ab{tag}02AfterConnect.png")
|
|
print(f"\n*** after the fake modem said CONNECT 1200: "
|
|
f"{decodeControl(after['aciaControl'])}", flush=True)
|
|
print(f" bitPeriodLo $E056 = ${(after['bitPeriodLo'] or 0):02X}, "
|
|
f"baudIndex $E055 = {after['baudIndex']}, connectionPhase $E040 = "
|
|
f"{after['connectionPhase']}, lastModemChar $EB02 = "
|
|
f"{readRange(a, 0xEB02, 1)[0]}", flush=True)
|
|
autoOk = after["aciaControl"] == 0x17 and after["bitPeriodLo"] == 0x17
|
|
print(f" auto-baud to 1200: {'PASS' if autoOk else 'FAIL'}", flush=True)
|
|
print(f" the fake modem sent its result code {modem.sent} time(s); it saw "
|
|
f"{bytes(modem.rx[:64])!r}", flush=True)
|
|
|
|
forced = [forceEntry(a, "A", 0, 0x15), forceEntry(a, "A", 3, 0x17),
|
|
forceEntry(a, "A", 6, 0x18)]
|
|
|
|
print("\n================ SUMMARY ================", flush=True)
|
|
print(f"auto-baud CONNECT 1200 -> control ${(after['aciaControl'] or 0):02X}: "
|
|
f"{'PASS' if autoOk else 'FAIL'}", flush=True)
|
|
for item in forced:
|
|
print(f"forced $E7B2 with X={item['x']}: $DE03 = ${item['after']['$DE03']:02X}, "
|
|
f"expected ${item['expectControl']:02X}: "
|
|
f"{'PASS' if item['ok'] else 'FAIL'}", flush=True)
|
|
open(f"{LOGS}/autoBaud1200.{tag}.modem.txt", "w").write(
|
|
"\n".join(f"{t}\t{text}" for t, text in modem.events) + "\n")
|
|
finally:
|
|
a.close()
|
|
time.sleep(1)
|
|
modem.stop()
|
|
print(f"[modem] total {len(modem.rx)} bytes from the C64: {bytes(modem.rx)!r}", flush=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|