1
0

command advance sweep

This commit is contained in:
Aiden
2026-05-26 15:21:52 +10:00
parent 74a2e2fd2c
commit a48fa0ed18
14 changed files with 821 additions and 78 deletions

View File

@@ -178,6 +178,8 @@ class ReplayConfig:
clock_hz: int = 10_000_000
uart_timing: bool = True
uart_baud: int = 38_400
uart_format: str = "8E1"
tx_wire_timing: bool = True
p9_fast_path: bool = True
p9_fast_input: int = 0xFF
p9_fast_optimistic_wrapper: bool = False
@@ -248,6 +250,9 @@ def run_bench_replay(log_path: Path, *, rom_path: Path | None = None, config: Re
p9_fast_default_wrapper_success=config.p9_fast_optimistic_wrapper,
p7_input=config.p7_input,
eeprom_seed=config.eeprom_seed,
sci1_tx_timing=UartTiming.from_format(config.uart_format, baud=config.uart_baud)
if config.tx_wire_timing
else None,
)
context = RunContext()
@@ -258,6 +263,7 @@ def run_bench_replay(log_path: Path, *, rom_path: Path | None = None, config: Re
f"rx_serviceable={int(_rx_ready(emulator))} "
f"sci1_priority={_sci1_priority(emulator)} interrupt_mask={_interrupt_mask(emulator)} "
f"clock_hz={emulator.clock_hz} "
f"uart_format={config.uart_format.upper()} tx_wire_timing={int(config.tx_wire_timing)} "
f"lcd_display={emulator.memory.lcd.display_text(lines=4)!r}"
)
@@ -270,7 +276,7 @@ def run_bench_replay(log_path: Path, *, rom_path: Path | None = None, config: Re
gap_frames = tuple(emulator.sci1.tx_frames[tx_frame_start_before_delay:])
tx_frame_start = len(emulator.sci1.tx_frames)
if config.uart_timing:
timing = UartTiming(baud=config.uart_baud)
timing = UartTiming.from_format(config.uart_format, baud=config.uart_baud)
steps_during_rx, inject_reason = _inject_frame_uart_timed(
emulator,
host.frame,
@@ -367,7 +373,9 @@ def build_arg_parser() -> argparse.ArgumentParser:
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("--uart-format", default=ReplayConfig.uart_format, help="UART character format for bench-style timing; real RCP link is 8E1")
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("--no-tx-wire-timing", action="store_true", help="use the legacy tiny TDRE delay instead of modeled UART TX character time")
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")
@@ -395,6 +403,8 @@ def main(argv: list[str] | None = None) -> int:
clock_hz=args.clock_hz,
uart_timing=not args.polite_rx,
uart_baud=args.uart_baud,
uart_format=args.uart_format,
tx_wire_timing=not args.no_tx_wire_timing,
p9_fast_path=not args.no_p9_fast_path,
p9_fast_input=args.p9_fast_input,
p9_fast_optimistic_wrapper=args.p9_fast_optimistic_wrapper,

View File

@@ -35,6 +35,7 @@ class MemoryAccess:
value: int
kind: str
region: str
pc: int | None = None
class MemoryMap:
@@ -48,6 +49,7 @@ class MemoryMap:
self.external: dict[int, int] = {}
self.port_inputs: dict[int, int] = {P7DR: p7_input & 0xFF}
self.access_log: list[MemoryAccess] = []
self.current_pc: int | None = None
self._set_register(SCI1_SMR, self.sci1.smr)
self._set_register(SCI1_BRR, self.sci1.brr)
@@ -185,7 +187,7 @@ class MemoryMap:
return ((latch & ddr) | (pins & ~ddr)) & 0xFF
def _log(self, kind: str, address: int, size: int, value: int) -> None:
self.access_log.append(MemoryAccess(address, size, value, kind, self.region(address).name))
self.access_log.append(MemoryAccess(address, size, value, kind, self.region(address).name, self.current_pc))
def describe_regions(regions: Iterable[MemoryRegion] = MEMORY_REGIONS) -> str:

View File

@@ -41,6 +41,7 @@ from .fast_paths import P9FastPath, P9FastPathConfig
from .memory import MemoryMap
from .sci import SCI1
from .timers import FrtOciaScheduler, FrtRegisters
from .uart import UartTiming
@dataclass
@@ -94,6 +95,7 @@ class H8536Emulator:
p9_fast_default_wrapper_success: bool = False,
p7_input: int = 0xFF,
eeprom_seed: str = "blank",
sci1_tx_timing: UartTiming | None = None,
) -> None:
if not rom_bytes:
raise ValueError("ROM image is empty")
@@ -103,6 +105,7 @@ class H8536Emulator:
self.memory = MemoryMap(rom_bytes, self.sci1, p7_input=p7_input)
if eeprom_seed == "factory":
self.memory.seed_factory_eeprom_and_shadow()
self.sci1.configure_tx_timing(sci1_tx_timing, clock_hz=clock_hz)
self.memory.p9_bus.default_wrapper_success = bool(p9_fast_default_wrapper_success)
self.p9_fast_path = p9_fast_path or P9FastPath(
P9FastPathConfig(enabled=p9_fast_path_enabled, default_input_byte=p9_fast_default_input_byte)
@@ -134,66 +137,70 @@ class H8536Emulator:
def step(self) -> str:
pc = self.cpu.pc
cycles_before = self.cpu.cycles
if self.p9_fast_path.try_handle(self):
self._tick_peripherals(self.cpu.cycles - cycles_before)
return f"{h16(pc)}: {'<p9-fast-path>':<17} P9 fast-path"
self.memory.current_pc = pc
try:
if self.p9_fast_path.try_handle(self):
self._tick_peripherals(self.cpu.cycles - cycles_before)
return f"{h16(pc)}: {'<p9-fast-path>':<17} P9 fast-path"
decoder = H8536Decoder(self.memory.rom, br=self.cpu.br)
ins = decoder.decode(pc)
if not ins.valid:
raise UnsupportedInstruction(pc, ins.raw, ins.text)
decoder = H8536Decoder(self.memory.rom, br=self.cpu.br)
ins = decoder.decode(pc)
if not ins.valid:
raise UnsupportedInstruction(pc, ins.raw, ins.text)
next_pc = (pc + ins.size) & 0xFFFF
raw = ins.raw
text = ins.text
next_pc = (pc + ins.size) & 0xFFFF
raw = ins.raw
text = ins.text
if raw[0] == 0x00:
pass
elif raw[0] == 0x02 and len(raw) == 2:
self._pop_register_mask(raw[1])
elif raw[0] == 0x11 and len(raw) >= 2:
next_pc = self._indirect_jump_call(raw, pc, next_pc)
elif raw[0] == 0x12 and len(raw) == 2:
self._push_register_mask(raw[1])
elif raw[0] in (0x01, 0x06, 0x07) and len(raw) == 3 and 0xB8 <= raw[1] <= 0xBF:
next_pc = self._scb(raw, pc, next_pc)
elif raw[0] in (0x04, 0x0C):
next_pc = self._execute_general(pc, next_pc)
elif 0x40 <= raw[0] <= 0x47 and len(raw) == 2:
reg = raw[0] & 0x07
self._cmp(self._reg_read(reg, 1), raw[1], 1)
elif 0x48 <= raw[0] <= 0x4F and len(raw) == 3:
reg = raw[0] & 0x07
self._cmp(self.cpu.regs[reg], int.from_bytes(raw[1:3], "big"), 2)
elif 0x50 <= raw[0] <= 0x57 and len(raw) == 2:
self._reg_write(raw[0] & 0x07, raw[1], 1)
self._set_logic_flags(raw[1], 1)
elif 0x58 <= raw[0] <= 0x5F and len(raw) == 3:
self.cpu.regs[raw[0] & 0x07] = int.from_bytes(raw[1:3], "big")
self._set_logic_flags(self.cpu.regs[raw[0] & 0x07], 2)
elif raw[0] in (0x0E, 0x1E, 0x18):
next_pc = self._direct_call(raw, next_pc)
elif raw[0] == 0x19:
next_pc = self._pop16()
elif raw[0] == 0x0A:
next_pc = self._return_from_interrupt()
elif raw[0] in (0x15, 0x1D) and len(raw) >= 4:
next_pc = self._execute_general(pc, next_pc)
elif raw[0] in range(0xA0, 0x100):
next_pc = self._execute_general(pc, next_pc)
elif raw[0] in range(0x20, 0x30) and len(raw) == 2:
next_pc = self._branch8(raw, pc, next_pc)
elif raw[0] in range(0x30, 0x40) and len(raw) == 3:
next_pc = self._branch16(raw, pc, next_pc)
else:
raise UnsupportedInstruction(pc, raw, text)
if raw[0] == 0x00:
pass
elif raw[0] == 0x02 and len(raw) == 2:
self._pop_register_mask(raw[1])
elif raw[0] == 0x11 and len(raw) >= 2:
next_pc = self._indirect_jump_call(raw, pc, next_pc)
elif raw[0] == 0x12 and len(raw) == 2:
self._push_register_mask(raw[1])
elif raw[0] in (0x01, 0x06, 0x07) and len(raw) == 3 and 0xB8 <= raw[1] <= 0xBF:
next_pc = self._scb(raw, pc, next_pc)
elif raw[0] in (0x04, 0x0C):
next_pc = self._execute_general(pc, next_pc)
elif 0x40 <= raw[0] <= 0x47 and len(raw) == 2:
reg = raw[0] & 0x07
self._cmp(self._reg_read(reg, 1), raw[1], 1)
elif 0x48 <= raw[0] <= 0x4F and len(raw) == 3:
reg = raw[0] & 0x07
self._cmp(self.cpu.regs[reg], int.from_bytes(raw[1:3], "big"), 2)
elif 0x50 <= raw[0] <= 0x57 and len(raw) == 2:
self._reg_write(raw[0] & 0x07, raw[1], 1)
self._set_logic_flags(raw[1], 1)
elif 0x58 <= raw[0] <= 0x5F and len(raw) == 3:
self.cpu.regs[raw[0] & 0x07] = int.from_bytes(raw[1:3], "big")
self._set_logic_flags(self.cpu.regs[raw[0] & 0x07], 2)
elif raw[0] in (0x0E, 0x1E, 0x18):
next_pc = self._direct_call(raw, next_pc)
elif raw[0] == 0x19:
next_pc = self._pop16()
elif raw[0] == 0x0A:
next_pc = self._return_from_interrupt()
elif raw[0] in (0x15, 0x1D) and len(raw) >= 4:
next_pc = self._execute_general(pc, next_pc)
elif raw[0] in range(0xA0, 0x100):
next_pc = self._execute_general(pc, next_pc)
elif raw[0] in range(0x20, 0x30) and len(raw) == 2:
next_pc = self._branch8(raw, pc, next_pc)
elif raw[0] in range(0x30, 0x40) and len(raw) == 3:
next_pc = self._branch16(raw, pc, next_pc)
else:
raise UnsupportedInstruction(pc, raw, text)
self.cpu.pc = next_pc
self.cpu.steps += 1
cycle_delta = self._rough_cycles(raw)
self.cpu.cycles += cycle_delta
self._tick_peripherals(cycle_delta)
return f"{h16(pc)}: {' '.join(f'{byte:02X}' for byte in raw):<17} {text}"
self.cpu.pc = next_pc
self.cpu.steps += 1
cycle_delta = self._rough_cycles(raw)
self.cpu.cycles += cycle_delta
self._tick_peripherals(cycle_delta)
return f"{h16(pc)}: {' '.join(f'{byte:02X}' for byte in raw):<17} {text}"
finally:
self.memory.current_pc = None
def run(self, max_steps: int, trace: bool = False, stop_on_heartbeat: bool = False) -> RunReport:
trace_lines: list[str] = []
@@ -434,7 +441,7 @@ class H8536Emulator:
return next_pc
def _tick_peripherals(self, cycle_delta: int) -> None:
self.sci1.tick()
self.sci1.tick(cycle_delta)
self._interval_counter += 1
if self.frt1_ocia_steps is None:
self.frt1_ocia.tick(self.memory, cycle_delta)

View File

@@ -163,6 +163,8 @@ class RxDivergenceConfig:
clock_hz: int = 10_000_000
uart_timing: bool = False
uart_baud: int = 38_400
uart_format: str = "8E1"
tx_wire_timing: bool = False
p7_input: int = 0xFF
p9_fast_path: bool = True
p9_fast_input: int = 0xFF
@@ -259,6 +261,9 @@ def run_rx_divergence(
p9_fast_default_wrapper_success=config.p9_fast_optimistic_wrapper,
p7_input=config.p7_input,
eeprom_seed=config.eeprom_seed,
sci1_tx_timing=UartTiming.from_format(config.uart_format, baud=config.uart_baud)
if config.tx_wire_timing
else None,
)
eeprom_load = config.eeprom_load
if eeprom_load is not None and eeprom_load.is_file():
@@ -271,7 +276,8 @@ def run_rx_divergence(
f"SCR={emulator.sci1.scr:02X} SSR={emulator.sci1.ssr:02X} "
f"rx_serviceable={int(_rx_ready(emulator))} "
f"sci1_priority={_sci1_priority(emulator)} interrupt_mask={_interrupt_mask(emulator)} "
f"clock_hz={emulator.clock_hz} p7_input={config.p7_input:#04x}"
f"clock_hz={emulator.clock_hz} p7_input={config.p7_input:#04x} "
f"uart_format={config.uart_format.upper()} tx_wire_timing={int(config.tx_wire_timing)}"
)
heartbeat_summary = None
@@ -303,6 +309,8 @@ def build_arg_parser() -> argparse.ArgumentParser:
parser.add_argument("--wait-heartbeat-steps", type=int, default=RxDivergenceConfig.wait_heartbeat_steps)
parser.add_argument("--uart-timing", action="store_true", help="inject bytes at UART character timing instead of waiting for RDRF clear")
parser.add_argument("--uart-baud", type=parse_int, default=RxDivergenceConfig.uart_baud)
parser.add_argument("--uart-format", default=RxDivergenceConfig.uart_format, help="UART character format for timing; real RCP link is 8E1")
parser.add_argument("--tx-wire-timing", action="store_true", help="delay SCI1 TDRE/TXI by one modeled UART character after each TDR write")
parser.add_argument("--post-frame-steps", type=int, default=RxDivergenceConfig.post_frame_steps)
parser.add_argument("--per-byte-steps", type=int, default=RxDivergenceConfig.per_byte_steps)
parser.add_argument("--clock-hz", type=parse_int, default=RxDivergenceConfig.clock_hz)
@@ -341,6 +349,8 @@ def main(argv: list[str] | None = None) -> int:
clock_hz=args.clock_hz,
uart_timing=args.uart_timing,
uart_baud=args.uart_baud,
uart_format=args.uart_format,
tx_wire_timing=args.tx_wire_timing,
p7_input=args.p7_input,
p9_fast_path=not args.no_p9_fast_path,
p9_fast_input=args.p9_fast_input,
@@ -363,7 +373,7 @@ def _trace_frame(emulator: H8536Emulator, frame: bytes, config: RxDivergenceConf
steps_total = 0
if config.uart_timing:
timing = UartTiming(baud=config.uart_baud)
timing = UartTiming.from_format(config.uart_format, baud=config.uart_baud)
steps_total, stopped_reason = _inject_frame_uart_timed(
emulator,
frame,

View File

@@ -24,6 +24,7 @@ from .uart import UartTiming
CHECKSUM_SEED = 0x5A
FRAME_LENGTH = 6
HEARTBEAT_FRAME = bytes.fromhex("0000000080DA")
CONNECT_LCD_FRAMES = (
bytes.fromhex("04000040001E"),
@@ -66,6 +67,12 @@ ACCESS_RANGES = (
(0xE400, 0xE401, "secondary_table_index_0000"),
(0xE800, 0xE801, "current_table_index_0000"),
(0xEC00, 0xEC01, "flag_table_index_0000"),
(0xE000, 0xE3FF, "primary_table_E000"),
(0xE400, 0xE7FF, "secondary_table_E400"),
(0xE800, 0xEBFF, "current_table_E800"),
(0xEC00, 0xEFFF, "flag_table_EC00"),
(0xF400, 0xF4FF, "eeprom_shadow_F400"),
(0xF7B0, 0xF82F, "persistent_record_ram"),
(0xF200, 0xF201, "lcd_ports"),
)
@@ -85,12 +92,30 @@ STATE_BYTES = {
0xFAA4: "rx_error_latch",
0xFAA5: "retry_or_gate_flags",
0xFAA6: "retry_counter",
0xEC02: "EC00_flag_index_0002",
0xEC04: "EC00_flag_index_0004",
0xEC12: "EC00_flag_index_0012",
0xEC13: "EC00_flag_index_0013",
0xEC15: "EC00_flag_index_0015",
0xEC82: "EC00_flag_index_0082",
}
STATE_WORDS = {
0xE000: "E000_index_0000_primary",
0xE004: "E000_index_0002_primary",
0xE008: "E000_index_0004_primary",
0xE024: "E000_index_0012_primary",
0xE026: "E000_index_0013_primary",
0xE02A: "E000_index_0015_primary",
0xE104: "E000_index_0082_primary",
0xE400: "E400_index_0000_secondary",
0xE800: "E800_index_0000_current",
0xE804: "E800_index_0002_current",
0xE808: "E800_index_0004_current",
0xE824: "E800_index_0012_current",
0xE826: "E800_index_0013_current",
0xE82A: "E800_index_0015_current",
0xE904: "E800_index_0082_current",
0xF860: "rx_frame_01",
0xF862: "rx_frame_23",
0xF864: "rx_frame_45",
@@ -206,8 +231,13 @@ def run_rx_probe(
boot_steps: int = 250_000,
per_byte_steps: int = 5_000,
post_frame_steps: int = 80_000,
post_frame_ms: int | None = None,
wait_heartbeats: int = 0,
wait_heartbeat_steps: int = 1_500_000,
uart_timing: bool = False,
uart_baud: int = 38_400,
uart_format: str = "8E1",
tx_wire_timing: bool = False,
interval_steps: int = 512,
frt1_ocia_steps: int | None = None,
frt2_ocia_steps: int | None = None,
@@ -232,6 +262,7 @@ def run_rx_probe(
p9_fast_default_wrapper_success=p9_fast_optimistic_wrapper,
p7_input=p7_input,
eeprom_seed=eeprom_seed,
sci1_tx_timing=UartTiming.from_format(uart_format, baud=uart_baud) if tx_wire_timing else None,
)
if eeprom_image is not None:
emulator.memory.load_eeprom_image(eeprom_image)
@@ -242,9 +273,24 @@ def run_rx_probe(
f"boot={boot_reason} steps={boot_steps_used} pc={h16(emulator.cpu.pc)} "
f"SCR={emulator.sci1.scr:02X} SSR={emulator.sci1.ssr:02X} "
f"rx_serviceable={int(_rx_ready(emulator))} "
f"clock_hz={emulator.clock_hz} "
f"clock_hz={emulator.clock_hz} uart_format={uart_format.upper()} "
f"tx_wire_timing={int(tx_wire_timing)} "
f"lcd_display={emulator.memory.lcd.display_text(lines=4)!r}"
)
if wait_heartbeats:
initial_heartbeats = sum(1 for frame in emulator.sci1.tx_frames if frame == HEARTBEAT_FRAME)
target_heartbeats = initial_heartbeats + wait_heartbeats
def heartbeat_predicate(inner: H8536Emulator) -> bool:
return sum(1 for frame in inner.sci1.tx_frames if frame == HEARTBEAT_FRAME) >= target_heartbeats
wait_context = RunContext()
wait_steps, wait_reason = _run_until(emulator, wait_heartbeat_steps, heartbeat_predicate, wait_context)
final_heartbeats = sum(1 for frame in emulator.sci1.tx_frames if frame == HEARTBEAT_FRAME)
boot_summary += (
f" wait_heartbeats={wait_heartbeats} wait_reason={wait_reason} "
f"wait_steps={wait_steps} heartbeat_count={final_heartbeats}"
)
results = [
_run_frame(
@@ -252,8 +298,10 @@ def run_rx_probe(
frame,
per_byte_steps=per_byte_steps,
post_frame_steps=post_frame_steps,
post_frame_ms=post_frame_ms,
uart_timing=uart_timing,
uart_baud=uart_baud,
uart_format=uart_format,
stop_after_tx_frame=stop_after_tx_frame,
)
for frame in frames
@@ -269,8 +317,13 @@ def build_arg_parser() -> argparse.ArgumentParser:
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="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("--post-frame-ms", type=int, help="run this many emulated milliseconds after each injected frame")
parser.add_argument("--wait-heartbeats", type=int, default=0, help="wait for this many heartbeat frames before injecting the first host frame")
parser.add_argument("--wait-heartbeat-steps", type=int, default=1_500_000, help="maximum steps while waiting for pre-injection heartbeat frames")
parser.add_argument("--uart-timing", action="store_true", help="inject frame bytes at real 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("--uart-format", default="8E1", help="UART character format for timed RX/TX modeling; real RCP link is 8E1")
parser.add_argument("--tx-wire-timing", action="store_true", help="delay SCI1 TDRE/TXI by one modeled UART character after each TDR write")
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")
@@ -303,8 +356,13 @@ 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,
post_frame_ms=args.post_frame_ms,
wait_heartbeats=args.wait_heartbeats,
wait_heartbeat_steps=args.wait_heartbeat_steps,
uart_timing=args.uart_timing,
uart_baud=args.uart_baud,
uart_format=args.uart_format,
tx_wire_timing=args.tx_wire_timing,
interval_steps=args.interval_steps,
frt1_ocia_steps=args.frt1_ocia_steps,
frt2_ocia_steps=args.frt2_ocia_steps,
@@ -327,6 +385,13 @@ def main(argv: list[str] | None = None) -> int:
for line in result.lines(index):
print(line)
print("total_tx_frames=" + " | ".join(format_frame(frame) for frame in emulator.sci1.tx_frames))
eeprom_writes = emulator.memory.p9_bus.x24164_bus.write_log_lines(limit=80)
if eeprom_writes:
print("eeprom_writes:")
for line in eeprom_writes:
print(f" {line}")
else:
print("eeprom_writes=none")
if args.eeprom_save:
args.eeprom_save.parent.mkdir(parents=True, exist_ok=True)
args.eeprom_save.write_bytes(emulator.memory.dump_eeprom_image())
@@ -352,8 +417,10 @@ def _run_frame(
*,
per_byte_steps: int,
post_frame_steps: int,
post_frame_ms: int | None,
uart_timing: bool,
uart_baud: int,
uart_format: str,
stop_after_tx_frame: bool,
) -> FrameResult:
state_before = _state_snapshot(emulator)
@@ -363,7 +430,7 @@ def _run_frame(
context = RunContext()
stopped_reason = "post_frame_steps"
steps_total = 0
timing = UartTiming(baud=uart_baud)
timing = UartTiming.from_format(uart_format, baud=uart_baud)
if uart_timing:
steps_total, stopped_reason = _inject_frame_uart_timed(
@@ -391,9 +458,13 @@ def _run_frame(
def post_predicate(inner: H8536Emulator) -> bool:
return stop_after_tx_frame and len(inner.sci1.tx_frames) >= target_frame_count
steps, reason = _run_until(emulator, post_frame_steps, post_predicate, context)
if post_frame_ms is not None:
steps, reason = _run_cycles_for_ms(emulator, post_frame_ms, context)
stopped_reason = reason
else:
steps, reason = _run_until(emulator, post_frame_steps, post_predicate, context)
stopped_reason = "tx_frame" if reason == "predicate" and stop_after_tx_frame else reason
steps_total += steps
stopped_reason = "tx_frame" if reason == "predicate" and stop_after_tx_frame else reason
log_end = len(emulator.memory.access_log)
state_after = _state_snapshot(emulator)
@@ -475,6 +546,22 @@ def _run_until_cycle(
return max(0, max_steps), "max_steps"
def _run_cycles_for_ms(emulator: H8536Emulator, delta_ms: int, context: RunContext) -> tuple[int, str]:
target_delta_cycles = int((max(0, delta_ms) * max(1, emulator.clock_hz)) / 1000)
target_cycles = emulator.cpu.cycles + target_delta_cycles
completed = 0
while emulator.cpu.cycles < target_cycles:
pc = emulator.cpu.pc
context.record_pc(pc)
try:
emulator.step()
except UnsupportedInstruction as exc:
context.unsupported = str(exc)
return completed, "unsupported_instruction"
completed += 1
return completed, f"post_frame_ms_{delta_ms}"
def _rx_ready(emulator: H8536Emulator) -> bool:
if not (emulator.sci1.scr & SCI_SCR_RIE and emulator.sci1.scr & SCI_SCR_RE):
return False
@@ -546,7 +633,8 @@ def _access_lines(accesses: list[MemoryAccess]) -> list[str]:
lines = []
for access in interesting[:80]:
label = _access_label(access.address)
lines.append(f"{access.kind:<5} {h16(access.address)} {access.value:02X} {label}")
pc = f" pc={h16(access.pc)}" if access.pc is not None else ""
lines.append(f"{access.kind:<5} {h16(access.address)} {access.value:02X} {label}{pc}")
if len(interesting) > 80:
lines.append(f"... {len(interesting) - 80} more interesting accesses")
return lines

View File

@@ -17,6 +17,7 @@ from .constants import (
SCI_SSR_RDRF,
SCI_SSR_TDRE,
)
from .uart import UartTiming
@dataclass
@@ -53,6 +54,8 @@ class SCI1:
_frame_buffer: bytearray = field(default_factory=bytearray)
tx_ready_delay: int = 0
tx_ready_ticks: int = 2
clock_hz: int = 10_000_000
tx_timing: UartTiming | None = None
_tx_ready_pending: bool = False
def read(self, address: int) -> int:
@@ -96,7 +99,7 @@ class SCI1:
if len(self._frame_buffer) == len(HEARTBEAT_FRAME):
self.tx_frames.append(bytes(self._frame_buffer))
self._frame_buffer.clear()
self.tx_ready_delay = max(0, self.tx_ready_ticks)
self.tx_ready_delay = self._tx_ready_delay()
self._tx_ready_pending = True
self.tx_events.append(SciTxEvent(SCI1_TDR, value, self.scr, self.ssr, emitted))
@@ -116,9 +119,22 @@ class SCI1:
def saw_heartbeat(self) -> bool:
return HEARTBEAT_FRAME in self.tx_frames
def tick(self) -> None:
def configure_tx_timing(self, timing: UartTiming | None, *, clock_hz: int | None = None) -> None:
self.tx_timing = timing
if clock_hz is not None:
self.clock_hz = max(1, clock_hz)
def tx_busy(self) -> bool:
return self._tx_ready_pending and self.tx_ready_delay > 0
def tick(self, cycles: int = 1) -> None:
if self._tx_ready_pending and self.tx_ready_delay:
self.tx_ready_delay -= 1
self.tx_ready_delay = max(0, self.tx_ready_delay - max(1, cycles))
if self._tx_ready_pending and self.tx_ready_delay == 0 and not (self.ssr & SCI_SSR_TDRE):
self.ssr |= SCI_SSR_TDRE
self._tx_ready_pending = False
def _tx_ready_delay(self) -> int:
if self.tx_timing is None:
return max(0, self.tx_ready_ticks)
return self.tx_timing.cycles_per_character(self.clock_hz)

View File

@@ -7,10 +7,20 @@ from dataclasses import dataclass
class UartTiming:
baud: int = 38_400
data_bits: int = 8
parity_bits: int = 0
parity: str = "N"
stop_bits: int = 1
start_bits: int = 1
def __post_init__(self) -> None:
parity = self.parity.upper()
if parity not in {"N", "E", "O"}:
raise ValueError("parity must be N, E, or O")
object.__setattr__(self, "parity", parity)
@property
def parity_bits(self) -> int:
return 0 if self.parity == "N" else 1
@property
def bits_per_character(self) -> int:
return self.start_bits + self.data_bits + self.parity_bits + self.stop_bits
@@ -26,10 +36,27 @@ class UartTiming:
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"uart_{self.data_bits}{self.parity}{self.stop_bits} "
f"baud={self.baud} byte_us={self.micros_per_character():.3f} "
f"byte_cycles={self.cycles_per_character(clock_hz)}"
)
@classmethod
def from_format(cls, text: str, *, baud: int = 38_400) -> "UartTiming":
normalized = text.strip().upper()
if len(normalized) != 3 or normalized[0] not in "78" or normalized[1] not in "NEO" or normalized[2] not in "12":
raise ValueError(f"unsupported UART format {text!r}; expected 8E1, 8N1, 8O1, etc.")
return cls(baud=baud, data_bits=int(normalized[0]), parity=normalized[1], stop_bits=int(normalized[2]))
@classmethod
def from_sci_smr(cls, smr: int, *, baud: int = 38_400) -> "UartTiming":
data_bits = 7 if smr & 0x40 else 8
if smr & 0x20:
parity = "O" if smr & 0x10 else "E"
else:
parity = "N"
stop_bits = 2 if smr & 0x08 else 1
return cls(baud=baud, data_bits=data_bits, parity=parity, stop_bits=stop_bits)
__all__ = ["UartTiming"]