start of tape loading. hw works and have a generation script
This commit is contained in:
+35
-12
@@ -6,6 +6,7 @@ from __future__ import annotations
|
||||
import argparse
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
try:
|
||||
@@ -132,7 +133,8 @@ def make_blocks(memory: dict[int, int], block_size: int) -> list[tuple[int, byte
|
||||
|
||||
|
||||
class Monitor:
|
||||
def __init__(self, port: str, baud: int, timeout: float):
|
||||
def __init__(self, port: str, baud: int, timeout: float, char_delay: float):
|
||||
self.char_delay = char_delay
|
||||
try:
|
||||
self.serial = serial.Serial(
|
||||
port=port,
|
||||
@@ -151,18 +153,32 @@ class Monitor:
|
||||
|
||||
def command(self, command: str) -> str:
|
||||
try:
|
||||
self.serial.write(command.encode("ascii") + b"\r")
|
||||
self.serial.flush()
|
||||
for value in command.encode("ascii") + b"\r":
|
||||
self.serial.write(bytes((value,)))
|
||||
self.serial.flush()
|
||||
if self.char_delay:
|
||||
time.sleep(self.char_delay)
|
||||
response = self.serial.read_until(b"> ")
|
||||
except serial.SerialException as exc:
|
||||
raise ProgramError(f"serial communication failed: {exc}") from exc
|
||||
if not response.endswith(b"> "):
|
||||
raise ProgramError(f"timeout waiting for monitor after '{command}'")
|
||||
text = response.decode("ascii", errors="replace")
|
||||
raise ProgramError(
|
||||
f"timeout waiting for monitor after '{command}'; partial={text!r}"
|
||||
)
|
||||
return response.decode("ascii", errors="replace")
|
||||
|
||||
def synchronize(self) -> None:
|
||||
try:
|
||||
self.serial.reset_input_buffer()
|
||||
# reset_input_buffer() cannot discard bytes which are still queued
|
||||
# in the computer or USB adapter. Read until the line has stayed
|
||||
# quiet before generating a fresh, unambiguous prompt.
|
||||
normal_timeout = self.serial.timeout
|
||||
self.serial.timeout = min(normal_timeout, 0.2)
|
||||
while self.serial.read(256):
|
||||
pass
|
||||
self.serial.timeout = normal_timeout
|
||||
except serial.SerialException as exc:
|
||||
raise ProgramError(f"serial communication failed: {exc}") from exc
|
||||
response = self.command("")
|
||||
@@ -187,14 +203,14 @@ def write_and_verify(monitor: Monitor, start: int, data: bytes) -> tuple[bool, s
|
||||
values = " ".join(f"{value:02X}" for value in data)
|
||||
response = monitor.command(f"W {start:04X} {values}")
|
||||
if not response_has_ok(response):
|
||||
return False, "monitor did not acknowledge the write"
|
||||
return False, f"monitor did not acknowledge the write; response={response!r}"
|
||||
|
||||
end = start + len(data) - 1
|
||||
expected = sum(data) & 0xFFFF
|
||||
response = monitor.command(f"C {start:04X} {end:04X}")
|
||||
actual = response_checksum(response)
|
||||
if actual is None:
|
||||
return False, "monitor did not return a checksum"
|
||||
return False, f"monitor did not return a checksum; response={response!r}"
|
||||
if actual != expected:
|
||||
return False, f"checksum mismatch (expected {expected:04X}, received {actual:04X})"
|
||||
return True, ""
|
||||
@@ -225,6 +241,12 @@ def parse_args() -> argparse.Namespace:
|
||||
default=5.0,
|
||||
help="serial response timeout in seconds (default: 5)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--char-delay",
|
||||
type=float,
|
||||
default=0.005,
|
||||
help="delay after each transmitted character in seconds (default: 0.005)",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
if not 1 <= args.block_size <= MAX_BLOCK_SIZE:
|
||||
parser.error(f"--block-size must be between 1 and {MAX_BLOCK_SIZE}")
|
||||
@@ -234,6 +256,8 @@ def parse_args() -> argparse.Namespace:
|
||||
parser.error("--baud must be positive")
|
||||
if args.timeout <= 0:
|
||||
parser.error("--timeout must be positive")
|
||||
if args.char_delay < 0:
|
||||
parser.error("--char-delay cannot be negative")
|
||||
return args
|
||||
|
||||
|
||||
@@ -249,7 +273,7 @@ def main() -> int:
|
||||
f"({min(memory):04X}-{max(memory):04X})"
|
||||
)
|
||||
|
||||
monitor = Monitor(args.port, args.baud, args.timeout)
|
||||
monitor = Monitor(args.port, args.baud, args.timeout, args.char_delay)
|
||||
monitor.synchronize()
|
||||
|
||||
for number, (start, data) in enumerate(blocks, 1):
|
||||
@@ -263,9 +287,8 @@ def main() -> int:
|
||||
success, reason = False, str(exc)
|
||||
if success:
|
||||
print(
|
||||
f"\rBlock {number}/{len(blocks)} {start:04X}-{end:04X} verified",
|
||||
end="",
|
||||
flush=True,
|
||||
f"Block {number}/{len(blocks)} "
|
||||
f"{start:04X}-{end:04X} verified"
|
||||
)
|
||||
break
|
||||
if attempt == args.retries:
|
||||
@@ -274,12 +297,12 @@ def main() -> int:
|
||||
f"{args.retries + 1} attempts: {reason}"
|
||||
)
|
||||
print(
|
||||
f"\nRetrying {start:04X}-{end:04X} "
|
||||
f"Retrying {start:04X}-{end:04X} "
|
||||
f"({attempt + 1}/{args.retries}): {reason}",
|
||||
file=sys.stderr,
|
||||
)
|
||||
|
||||
print(f"\nProgrammed and verified {len(memory)} bytes")
|
||||
print(f"Programmed and verified {len(memory)} bytes")
|
||||
return 0
|
||||
except ProgramError as exc:
|
||||
print(f"error: {exc}", file=sys.stderr)
|
||||
|
||||
Reference in New Issue
Block a user