UART simulation
This commit is contained in:
@@ -58,6 +58,7 @@ from .memory import MemoryAccess, MemoryMap, describe_regions
|
||||
from .peripherals import LCD
|
||||
from .runner import H8536Emulator, RunReport
|
||||
from .sci import SCI1, SciTxEvent
|
||||
from .uart import UartTiming
|
||||
|
||||
__all__ = [
|
||||
"CPUState",
|
||||
@@ -114,6 +115,7 @@ __all__ = [
|
||||
"SCI_SSR_TDRE",
|
||||
"SciTxEvent",
|
||||
"UnsupportedInstruction",
|
||||
"UartTiming",
|
||||
"VECTOR_FRT1_OCIA",
|
||||
"VECTOR_INTERVAL_TIMER",
|
||||
"VECTOR_FRT2_OCIA",
|
||||
|
||||
@@ -14,6 +14,8 @@ from .errors import UnsupportedInstruction
|
||||
from .runner import H8536Emulator
|
||||
from .rx_probe import (
|
||||
RunContext,
|
||||
UartTiming,
|
||||
_inject_frame_uart_timed,
|
||||
_interrupt_mask,
|
||||
_rx_byte_consumed,
|
||||
_rx_ready,
|
||||
@@ -88,6 +90,7 @@ class ReplayFrameResult:
|
||||
host_frame: bytes
|
||||
host_timestamp: str
|
||||
host_delta_ms: int
|
||||
rx_injection: str
|
||||
steps_before: int
|
||||
steps_during_rx: int
|
||||
emulator_gap_frames_before: tuple[bytes, ...]
|
||||
@@ -120,6 +123,7 @@ class BenchReplayResult:
|
||||
"host_timestamp": item.host_timestamp,
|
||||
"host_delta_ms": item.host_delta_ms,
|
||||
"host_frame": format_frame(item.host_frame),
|
||||
"rx_injection": item.rx_injection,
|
||||
"steps_before": item.steps_before,
|
||||
"steps_during_rx": item.steps_during_rx,
|
||||
"emulator_gap_frames_before": [format_frame(frame) for frame in item.emulator_gap_frames_before],
|
||||
@@ -150,6 +154,7 @@ class BenchReplayResult:
|
||||
lines.append(
|
||||
(
|
||||
f" [{index}] {item.host_timestamp} delta={item.host_delta_ms}ms "
|
||||
f"rx_injection={item.rx_injection} "
|
||||
f"steps_before={item.steps_before} steps_rx={item.steps_during_rx} "
|
||||
f"host={format_frame(item.host_frame)} "
|
||||
f"gap_emu={_format_frame_list(item.emulator_gap_frames_before)} "
|
||||
@@ -171,6 +176,8 @@ class ReplayConfig:
|
||||
frt1_ocia_steps: int | None = None
|
||||
frt2_ocia_steps: int | None = None
|
||||
clock_hz: int = 10_000_000
|
||||
uart_timing: bool = True
|
||||
uart_baud: int = 38_400
|
||||
p9_fast_path: bool = True
|
||||
p9_fast_input: int = 0xFF
|
||||
|
||||
@@ -256,12 +263,25 @@ def run_bench_replay(log_path: Path, *, rom_path: Path | None = None, config: Re
|
||||
steps_before = _run_cycles_for_ms(emulator, delta_ms, config.clock_hz, context)
|
||||
gap_frames = tuple(emulator.sci1.tx_frames[tx_frame_start_before_delay:])
|
||||
tx_frame_start = len(emulator.sci1.tx_frames)
|
||||
steps_during_rx = _inject_host_frame(emulator, host.frame, config.per_byte_steps, context)
|
||||
if config.uart_timing:
|
||||
timing = UartTiming(baud=config.uart_baud)
|
||||
steps_during_rx, inject_reason = _inject_frame_uart_timed(
|
||||
emulator,
|
||||
host.frame,
|
||||
timing=timing,
|
||||
max_steps_per_gap=config.per_byte_steps,
|
||||
context=context,
|
||||
)
|
||||
rx_injection = f"{timing.summary(emulator.clock_hz)} reason={inject_reason}"
|
||||
else:
|
||||
steps_during_rx = _inject_host_frame(emulator, host.frame, config.per_byte_steps, context)
|
||||
rx_injection = "polite_wait_for_rdrf_clear"
|
||||
replay_results.append(
|
||||
ReplayFrameResult(
|
||||
host_frame=host.frame,
|
||||
host_timestamp=host.timestamp,
|
||||
host_delta_ms=delta_ms,
|
||||
rx_injection=rx_injection,
|
||||
steps_before=steps_before,
|
||||
steps_during_rx=steps_during_rx,
|
||||
emulator_gap_frames_before=gap_frames,
|
||||
@@ -336,10 +356,12 @@ def build_arg_parser() -> argparse.ArgumentParser:
|
||||
parser.add_argument("log", type=Path, help="bench log produced by scripts/bench_connect_lcd_sequence.py")
|
||||
parser.add_argument("--rom", type=Path, help="ROM image path; defaults to ROM/M27C512@DIP28_1.BIN")
|
||||
parser.add_argument("--boot-steps", type=int, default=ReplayConfig.boot_steps)
|
||||
parser.add_argument("--per-byte-steps", type=int, default=ReplayConfig.per_byte_steps)
|
||||
parser.add_argument("--per-byte-steps", type=int, default=ReplayConfig.per_byte_steps, help="UART mode step limit between byte arrivals, or polite mode byte-consume limit")
|
||||
parser.add_argument("--post-log-steps", type=int, default=ReplayConfig.post_log_steps)
|
||||
parser.add_argument("--interval-steps", type=int, default=ReplayConfig.interval_steps)
|
||||
parser.add_argument("--clock-hz", type=lambda text: int(text, 0), default=ReplayConfig.clock_hz)
|
||||
parser.add_argument("--uart-baud", type=lambda text: int(text, 0), default=ReplayConfig.uart_baud, help="baud rate for bench-style UART injection")
|
||||
parser.add_argument("--polite-rx", action="store_true", help="wait for each RX byte to be consumed before injecting the next byte")
|
||||
parser.add_argument("--frt1-ocia-steps", type=int, default=ReplayConfig.frt1_ocia_steps)
|
||||
parser.add_argument("--frt2-ocia-steps", type=int, default=ReplayConfig.frt2_ocia_steps)
|
||||
parser.add_argument("--no-p9-fast-path", action="store_true", help="disable shortcut handling for known P9 routines")
|
||||
@@ -362,6 +384,8 @@ def main(argv: list[str] | None = None) -> int:
|
||||
frt1_ocia_steps=args.frt1_ocia_steps,
|
||||
frt2_ocia_steps=args.frt2_ocia_steps,
|
||||
clock_hz=args.clock_hz,
|
||||
uart_timing=not args.polite_rx,
|
||||
uart_baud=args.uart_baud,
|
||||
p9_fast_path=not args.no_p9_fast_path,
|
||||
p9_fast_input=args.p9_fast_input,
|
||||
),
|
||||
|
||||
@@ -18,6 +18,7 @@ from .constants import (
|
||||
from .errors import UnsupportedInstruction
|
||||
from .memory import MemoryAccess
|
||||
from .runner import H8536Emulator
|
||||
from .uart import UartTiming
|
||||
|
||||
|
||||
CHECKSUM_SEED = 0x5A
|
||||
@@ -120,6 +121,8 @@ class RunContext:
|
||||
class FrameResult:
|
||||
input_frame: bytes
|
||||
checksum_ok: bool
|
||||
rx_injection: str
|
||||
uart_byte_cycles: int | None
|
||||
steps: int
|
||||
stopped_reason: str
|
||||
new_tx_bytes: bytes
|
||||
@@ -132,6 +135,7 @@ class FrameResult:
|
||||
def lines(self, index: int) -> list[str]:
|
||||
lines = [
|
||||
f"host_frame[{index}]={format_frame(self.input_frame)} checksum_ok={int(self.checksum_ok)}",
|
||||
f" rx_injection={self.rx_injection}",
|
||||
f" stopped={self.stopped_reason} steps={self.steps}",
|
||||
f" new_tx_bytes={format_frame(self.new_tx_bytes) if self.new_tx_bytes else 'none'}",
|
||||
]
|
||||
@@ -201,6 +205,8 @@ def run_rx_probe(
|
||||
boot_steps: int = 250_000,
|
||||
per_byte_steps: int = 5_000,
|
||||
post_frame_steps: int = 80_000,
|
||||
uart_timing: bool = False,
|
||||
uart_baud: int = 38_400,
|
||||
interval_steps: int = 512,
|
||||
frt1_ocia_steps: int | None = None,
|
||||
frt2_ocia_steps: int | None = None,
|
||||
@@ -236,6 +242,8 @@ def run_rx_probe(
|
||||
frame,
|
||||
per_byte_steps=per_byte_steps,
|
||||
post_frame_steps=post_frame_steps,
|
||||
uart_timing=uart_timing,
|
||||
uart_baud=uart_baud,
|
||||
stop_after_tx_frame=stop_after_tx_frame,
|
||||
)
|
||||
for frame in frames
|
||||
@@ -249,8 +257,10 @@ def build_arg_parser() -> argparse.ArgumentParser:
|
||||
parser.add_argument("--rom", type=Path, help="ROM image path; defaults to ROM/M27C512@DIP28_1.BIN when present")
|
||||
parser.add_argument("--preset", choices=("connect-lcd",), help="append a built-in host-frame set")
|
||||
parser.add_argument("--boot-steps", type=int, default=250_000, help="maximum steps to boot until SCI1 RXI is serviceable")
|
||||
parser.add_argument("--per-byte-steps", type=int, default=5_000, help="maximum steps after each injected RX byte")
|
||||
parser.add_argument("--per-byte-steps", type=int, default=5_000, help="polite mode byte-consume limit, or UART mode step limit between byte arrivals")
|
||||
parser.add_argument("--post-frame-steps", type=int, default=80_000, help="maximum steps after a full injected frame")
|
||||
parser.add_argument("--uart-timing", action="store_true", help="inject frame bytes at real 8N1 UART inter-byte timing instead of waiting for RDRF consumption")
|
||||
parser.add_argument("--uart-baud", type=parse_int, default=38_400, help="baud rate for --uart-timing; 38400 gives about 260 us per 8N1 byte")
|
||||
parser.add_argument("--keep-listening", action="store_true", help="use all post-frame steps instead of stopping at the first new TX frame")
|
||||
parser.add_argument("--interval-steps", type=int, default=512, help="rough step period for the scaffolded interval timer interrupt")
|
||||
parser.add_argument("--clock-hz", type=parse_int, default=10_000_000, help="CPU/phi clock in Hz for calibrated FRT timing")
|
||||
@@ -275,6 +285,8 @@ def main(argv: list[str] | None = None) -> int:
|
||||
boot_steps=args.boot_steps,
|
||||
per_byte_steps=args.per_byte_steps,
|
||||
post_frame_steps=args.post_frame_steps,
|
||||
uart_timing=args.uart_timing,
|
||||
uart_baud=args.uart_baud,
|
||||
interval_steps=args.interval_steps,
|
||||
frt1_ocia_steps=args.frt1_ocia_steps,
|
||||
frt2_ocia_steps=args.frt2_ocia_steps,
|
||||
@@ -300,6 +312,8 @@ def _run_frame(
|
||||
*,
|
||||
per_byte_steps: int,
|
||||
post_frame_steps: int,
|
||||
uart_timing: bool,
|
||||
uart_baud: int,
|
||||
stop_after_tx_frame: bool,
|
||||
) -> FrameResult:
|
||||
state_before = _state_snapshot(emulator)
|
||||
@@ -309,15 +323,29 @@ def _run_frame(
|
||||
context = RunContext()
|
||||
stopped_reason = "post_frame_steps"
|
||||
steps_total = 0
|
||||
timing = UartTiming(baud=uart_baud)
|
||||
|
||||
for offset, value in enumerate(frame):
|
||||
emulator.inject_sci1_rx_byte(value)
|
||||
steps, reason = _run_until(emulator, per_byte_steps, _rx_byte_consumed, context)
|
||||
steps_total += steps
|
||||
if reason != "predicate":
|
||||
stopped_reason = f"rx_byte_{offset}_{reason}"
|
||||
break
|
||||
if uart_timing:
|
||||
steps_total, stopped_reason = _inject_frame_uart_timed(
|
||||
emulator,
|
||||
frame,
|
||||
timing=timing,
|
||||
max_steps_per_gap=per_byte_steps,
|
||||
context=context,
|
||||
)
|
||||
injected_all_bytes = stopped_reason == "frame_injected_uart_timing"
|
||||
else:
|
||||
injected_all_bytes = True
|
||||
for offset, value in enumerate(frame):
|
||||
emulator.inject_sci1_rx_byte(value)
|
||||
steps, reason = _run_until(emulator, per_byte_steps, _rx_byte_consumed, context)
|
||||
steps_total += steps
|
||||
if reason != "predicate":
|
||||
stopped_reason = f"rx_byte_{offset}_{reason}"
|
||||
injected_all_bytes = False
|
||||
break
|
||||
|
||||
if injected_all_bytes:
|
||||
target_frame_count = tx_frame_start + 1
|
||||
|
||||
def post_predicate(inner: H8536Emulator) -> bool:
|
||||
@@ -332,6 +360,8 @@ def _run_frame(
|
||||
return FrameResult(
|
||||
input_frame=frame,
|
||||
checksum_ok=frame_checksum_ok(frame),
|
||||
rx_injection=timing.summary(emulator.clock_hz) if uart_timing else "polite_wait_for_rdrf_clear",
|
||||
uart_byte_cycles=timing.cycles_per_character(emulator.clock_hz) if uart_timing else None,
|
||||
steps=steps_total,
|
||||
stopped_reason=stopped_reason,
|
||||
new_tx_bytes=bytes(emulator.sci1.tx_bytes[tx_byte_start:]),
|
||||
@@ -343,6 +373,28 @@ def _run_frame(
|
||||
)
|
||||
|
||||
|
||||
def _inject_frame_uart_timed(
|
||||
emulator: H8536Emulator,
|
||||
frame: bytes,
|
||||
*,
|
||||
timing: UartTiming,
|
||||
max_steps_per_gap: int,
|
||||
context: RunContext,
|
||||
) -> tuple[int, str]:
|
||||
steps_total = 0
|
||||
start_cycles = emulator.cpu.cycles
|
||||
byte_cycles = timing.cycles_per_character(emulator.clock_hz)
|
||||
for offset, value in enumerate(frame):
|
||||
if offset:
|
||||
target_cycles = start_cycles + (offset * byte_cycles)
|
||||
steps, reason = _run_until_cycle(emulator, target_cycles, max_steps_per_gap, context)
|
||||
steps_total += steps
|
||||
if reason != "target_cycle":
|
||||
return steps_total, f"rx_byte_{offset}_{reason}"
|
||||
emulator.inject_sci1_rx_byte(value)
|
||||
return steps_total, "frame_injected_uart_timing"
|
||||
|
||||
|
||||
def _run_until(
|
||||
emulator: H8536Emulator,
|
||||
max_steps: int,
|
||||
@@ -362,6 +414,27 @@ def _run_until(
|
||||
return max_steps, "max_steps"
|
||||
|
||||
|
||||
def _run_until_cycle(
|
||||
emulator: H8536Emulator,
|
||||
target_cycles: int,
|
||||
max_steps: int,
|
||||
context: RunContext,
|
||||
) -> tuple[int, str]:
|
||||
for index in range(max(0, max_steps)):
|
||||
if emulator.cpu.cycles >= target_cycles:
|
||||
return index, "target_cycle"
|
||||
pc = emulator.cpu.pc
|
||||
context.record_pc(pc)
|
||||
try:
|
||||
emulator.step()
|
||||
except UnsupportedInstruction as exc:
|
||||
context.unsupported = str(exc)
|
||||
return index, "unsupported_instruction"
|
||||
if emulator.cpu.cycles >= target_cycles:
|
||||
return max(0, max_steps), "target_cycle"
|
||||
return max(0, max_steps), "max_steps"
|
||||
|
||||
|
||||
def _rx_ready(emulator: H8536Emulator) -> bool:
|
||||
if not (emulator.sci1.scr & SCI_SCR_RIE and emulator.sci1.scr & SCI_SCR_RE):
|
||||
return False
|
||||
@@ -477,4 +550,5 @@ __all__ = [
|
||||
"main",
|
||||
"parse_frame",
|
||||
"run_rx_probe",
|
||||
"UartTiming",
|
||||
]
|
||||
|
||||
@@ -105,6 +105,11 @@ class SCI1:
|
||||
self.ssr = (self.ssr & (writable_zero_flags & value)) | (value & ~writable_zero_flags)
|
||||
|
||||
def inject_rx(self, value: int) -> None:
|
||||
if self.ssr & SCI_SSR_ORER:
|
||||
return
|
||||
if self.ssr & SCI_SSR_RDRF:
|
||||
self.ssr |= SCI_SSR_ORER
|
||||
return
|
||||
self.rdr = value & 0xFF
|
||||
self.ssr |= SCI_SSR_RDRF
|
||||
|
||||
|
||||
35
h8536/emulator/uart.py
Normal file
35
h8536/emulator/uart.py
Normal file
@@ -0,0 +1,35 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class UartTiming:
|
||||
baud: int = 38_400
|
||||
data_bits: int = 8
|
||||
parity_bits: int = 0
|
||||
stop_bits: int = 1
|
||||
start_bits: int = 1
|
||||
|
||||
@property
|
||||
def bits_per_character(self) -> int:
|
||||
return self.start_bits + self.data_bits + self.parity_bits + self.stop_bits
|
||||
|
||||
def seconds_per_character(self) -> float:
|
||||
return self.bits_per_character / max(1, self.baud)
|
||||
|
||||
def micros_per_character(self) -> float:
|
||||
return 1_000_000.0 * self.seconds_per_character()
|
||||
|
||||
def cycles_per_character(self, clock_hz: int) -> int:
|
||||
return max(1, round(max(1, clock_hz) * self.seconds_per_character()))
|
||||
|
||||
def summary(self, clock_hz: int) -> str:
|
||||
return (
|
||||
f"uart_{self.data_bits}{'N' if self.parity_bits == 0 else 'P'}{self.stop_bits} "
|
||||
f"baud={self.baud} byte_us={self.micros_per_character():.3f} "
|
||||
f"byte_cycles={self.cycles_per_character(clock_hz)}"
|
||||
)
|
||||
|
||||
|
||||
__all__ = ["UartTiming"]
|
||||
Reference in New Issue
Block a user