from __future__ import annotations import argparse from dataclasses import dataclass, field from pathlib import Path from typing import Iterable from .decoder import H8536Decoder from .formatting import h8, h16 from .memory import MEMORY_REGIONS, MemoryRegion, region_for from .rom import DecodeError, Rom from .vectors import read_vectors_min SCI1_SMR = 0xFED8 SCI1_BRR = 0xFED9 SCI1_SCR = 0xFEDA SCI1_TDR = 0xFEDB SCI1_SSR = 0xFEDC SCI1_RDR = 0xFEDD P9DDR = 0xFEFE P9DR = 0xFEFF IPRA = 0xFF00 IPRE = 0xFF04 WDT_TCSR_R = 0xFEEC SCI_SCR_TIE = 0x80 SCI_SCR_RIE = 0x40 SCI_SCR_TE = 0x20 SCI_SCR_RE = 0x10 SCI_SSR_TDRE = 0x80 SCI_SSR_RDRF = 0x40 SCI_SSR_ORER = 0x20 SCI_SSR_FER = 0x10 SCI_SSR_PER = 0x08 ON_CHIP_RAM_START = 0xF680 ON_CHIP_RAM_END = 0xFE7F REGISTER_FIELD_START = 0xFE80 REGISTER_FIELD_END = 0xFFFF RAMCR = 0xFF11 HEARTBEAT_FRAME = bytes([0x00, 0x00, 0x00, 0x00, 0x80, 0xDA]) VECTOR_INTERVAL_TIMER = 0x0042 VECTOR_SCI1_TXI = 0x0084 class EmulatorError(Exception): pass class UnsupportedInstruction(EmulatorError): def __init__(self, pc: int, raw: bytes, text: str) -> None: raw_text = " ".join(f"{byte:02X}" for byte in raw) super().__init__(f"unsupported instruction at {h16(pc)}: {raw_text} {text}".rstrip()) self.pc = pc self.raw = raw self.text = text @dataclass class SciTxEvent: address: int value: int scr: int ssr: int emitted: bool @dataclass class SCI1: """Small SCI1 model for the H8/536 serial path. Manual anchors: - RDR/TDR/SCR/SSR live at H'FEDD/H'FEDB/H'FEDA/H'FEDC for SCI1. - SCR bit 7 TIE, bit 6 RIE, bit 5 TE, bit 4 RE. - SSR bit 7 TDRE, bit 6 RDRF, bits 5..3 ORER/FER/PER. - Software normally writes TDR after TDRE=1, then clears SSR.TDRE. """ smr: int = 0x00 brr: int = 0xFF scr: int = 0x0C tdr: int = 0xFF ssr: int = 0x87 rdr: int = 0x00 tx_bytes: list[int] = field(default_factory=list) tx_events: list[SciTxEvent] = field(default_factory=list) tx_frames: list[bytes] = field(default_factory=list) _frame_buffer: bytearray = field(default_factory=bytearray) tx_ready_delay: int = 0 def read(self, address: int) -> int: if address == SCI1_SMR: return self.smr if address == SCI1_BRR: return self.brr if address == SCI1_SCR: return self.scr if address == SCI1_TDR: return self.tdr if address == SCI1_SSR: return self.ssr if address == SCI1_RDR: return self.rdr raise KeyError(address) def write(self, address: int, value: int) -> None: value &= 0xFF if address == SCI1_SMR: self.smr = value elif address == SCI1_BRR: self.brr = value elif address == SCI1_SCR: self.scr = value elif address == SCI1_TDR: self.tdr = value self._write_tdr(value) elif address == SCI1_SSR: # The real SSR is R/(W)*: writable zeroes clear latched flags after # the required read sequence. This scaffold applies zero bits # directly so ROM BCLR/BSET style accesses can be modeled. self.ssr = value elif address == SCI1_RDR: self.rdr = value else: raise KeyError(address) def _write_tdr(self, value: int) -> None: emitted = bool(self.scr & SCI_SCR_TE) if emitted: self.tx_bytes.append(value) self._frame_buffer.append(value) if len(self._frame_buffer) == len(HEARTBEAT_FRAME): self.tx_frames.append(bytes(self._frame_buffer)) self._frame_buffer.clear() self.ssr |= SCI_SSR_TDRE self.tx_ready_delay = 2 self.tx_events.append(SciTxEvent(SCI1_TDR, value, self.scr, self.ssr, emitted)) def inject_rx(self, value: int) -> None: self.rdr = value & 0xFF self.ssr |= SCI_SSR_RDRF def saw_heartbeat(self) -> bool: return HEARTBEAT_FRAME in self.tx_frames def tick(self) -> None: if self.tx_ready_delay: self.tx_ready_delay -= 1 if self.tx_ready_delay == 0: self.ssr |= SCI_SSR_TDRE @dataclass class MemoryAccess: address: int size: int value: int kind: str region: str class MemoryMap: def __init__(self, rom_bytes: bytes, sci1: SCI1 | None = None) -> None: self.rom = Rom(rom_bytes, base=0) self.sci1 = sci1 if sci1 is not None else SCI1() self.ram = bytearray(ON_CHIP_RAM_END - ON_CHIP_RAM_START + 1) self.registers = bytearray(REGISTER_FIELD_END - REGISTER_FIELD_START + 1) self.external: dict[int, int] = {} self.access_log: list[MemoryAccess] = [] self._set_register(SCI1_SMR, self.sci1.smr) self._set_register(SCI1_BRR, self.sci1.brr) self._set_register(SCI1_SCR, self.sci1.scr) self._set_register(SCI1_TDR, self.sci1.tdr) self._set_register(SCI1_SSR, self.sci1.ssr) self._set_register(SCI1_RDR, self.sci1.rdr) def region(self, address: int) -> MemoryRegion: return region_for(address & 0xFFFF) def read8(self, address: int) -> int: address &= 0xFFFF if address in (SCI1_SMR, SCI1_BRR, SCI1_SCR, SCI1_TDR, SCI1_SSR, SCI1_RDR): value = self.sci1.read(address) self._set_register(address, value) elif address in self.external: value = self.external[address] elif address in (0xF200, 0xF201): # LCD E-clock/status space. Default to ready/zero so boot can pass # busy-flag polling until a fuller external bus model exists. value = 0x00 elif ON_CHIP_RAM_START <= address <= ON_CHIP_RAM_END: value = self.ram[address - ON_CHIP_RAM_START] elif REGISTER_FIELD_START <= address <= REGISTER_FIELD_END: value = self.registers[address - REGISTER_FIELD_START] if address == P9DR and not (self.registers[P9DDR - REGISTER_FIELD_START] & 0x80): # P97 is used as an input during the serial panel/camera-side # bit-bang handshake. With no external device modeled, hold the # input low so the firmware sees an idle/acknowledged bus rather # than reading back its previous output latch forever. value &= 0x7F elif self.rom.contains(address): value = self.rom.u8(address) else: value = self.external.get(address, 0xFF) self._log("read", address, 1, value) return value def read16(self, address: int) -> int: high = self.read8(address) low = self.read8((address + 1) & 0xFFFF) return (high << 8) | low def write8(self, address: int, value: int) -> None: address &= 0xFFFF value &= 0xFF if address in (SCI1_SMR, SCI1_BRR, SCI1_SCR, SCI1_TDR, SCI1_SSR, SCI1_RDR): self.sci1.write(address, value) self._set_register(address, self.sci1.read(address)) elif ON_CHIP_RAM_START <= address <= ON_CHIP_RAM_END: self.ram[address - ON_CHIP_RAM_START] = value elif REGISTER_FIELD_START <= address <= REGISTER_FIELD_END: self._set_register(address, value) elif self.rom.contains(address): # The ROM image spans the whole address space, but the H8/536 map # can place external RAM/peripherals in these ranges. Keep writes # as external overrides while leaving instruction fetch immutable. self.external[address] = value else: self.external[address] = value self._log("write", address, 1, value) def write16(self, address: int, value: int) -> None: self.write8(address, (value >> 8) & 0xFF) self.write8((address + 1) & 0xFFFF, value & 0xFF) def _set_register(self, address: int, value: int) -> None: self.registers[address - REGISTER_FIELD_START] = value & 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)) @dataclass class CPUState: pc: int = 0 sr: int = 0 br: int = 0 regs: list[int] = field(default_factory=lambda: [0] * 8) cycles: int = 0 steps: int = 0 z: bool = False c: bool = False n: bool = False v: bool = False interrupt_depth: int = 0 @dataclass class RunReport: steps: int cycles: int pc: int stopped_reason: str tx_bytes: bytes tx_frames: list[bytes] heartbeat_seen: bool unsupported: str | None = None trace: list[str] = field(default_factory=list) def summary_lines(self) -> list[str]: lines = [ f"steps={self.steps}", f"cycles={self.cycles}", f"pc={h16(self.pc)}", f"stopped={self.stopped_reason}", "tx_bytes=" + self.tx_bytes.hex(" ").upper(), "tx_frames=" + ", ".join(frame.hex(" ").upper() for frame in self.tx_frames), f"heartbeat_seen={self.heartbeat_seen}", ] if self.unsupported: lines.append(f"unsupported={self.unsupported}") lines.append("next_todo=implement the stopped opcode, then add interrupt scheduling for SCI1 TXI and interval/watchdog timer overflow") return lines class H8536Emulator: def __init__(self, rom_bytes: bytes, *, interval_steps: int = 2048) -> None: if not rom_bytes: raise ValueError("ROM image is empty") self.sci1 = SCI1() self.memory = MemoryMap(rom_bytes, self.sci1) self.cpu = CPUState() self.vectors = read_vectors_min(self.memory.rom) self.interval_steps = max(1, interval_steps) self._interval_counter = 0 self.reset() def reset(self) -> None: self.cpu = CPUState(pc=self.reset_vector()) def reset_vector(self) -> int: if self.memory.rom.contains(0, 2): return self.memory.rom.u16(0) raise DecodeError("ROM does not contain a reset vector at H'0000") def step(self) -> str: pc = self.cpu.pc 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 if raw[0] == 0x00: pass elif raw[0] == 0x02 and len(raw) == 2: self._pop_register_mask(raw[1]) 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) 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 self.cpu.cycles += self._rough_cycles(raw) self._tick_peripherals() return f"{h16(pc)}: {' '.join(f'{byte:02X}' for byte in raw):<17} {text}" def run(self, max_steps: int, trace: bool = False, stop_on_heartbeat: bool = False) -> RunReport: trace_lines: list[str] = [] stopped_reason = "max_steps" unsupported: str | None = None for _ in range(max_steps): try: line = self.step() except UnsupportedInstruction as exc: stopped_reason = "unsupported_instruction" unsupported = str(exc) break if trace: trace_lines.append(line) if stop_on_heartbeat and self.sci1.saw_heartbeat(): stopped_reason = "heartbeat" break return RunReport( steps=self.cpu.steps, cycles=self.cpu.cycles, pc=self.cpu.pc, stopped_reason=stopped_reason, tx_bytes=bytes(self.sci1.tx_bytes), tx_frames=list(self.sci1.tx_frames), heartbeat_seen=self.sci1.saw_heartbeat(), unsupported=unsupported, trace=trace_lines, ) def _execute_general(self, pc: int, next_pc: int) -> int: ea = self._decode_ea(pc) op = self.memory.rom.u8(pc + ea["length"]) size = int(ea["size"]) raw = self.memory.rom.slice(pc, self._general_length(pc, ea, op)) if op == 0x00: ext = self.memory.rom.u8(pc + int(ea["length"]) + 1) ext_base = ext & 0xF8 reg = ext & 0x07 if ext_base == 0x80: value = self._read_ea(ea, 1) self._reg_write(reg, value, 1) self._set_logic_flags(value, 1) elif ext_base == 0x90: self._write_ea(ea, self._reg_read(reg, 1), 1) else: raise UnsupportedInstruction(pc, raw, H8536Decoder(self.memory.rom, br=self.cpu.br).decode(pc).text) return next_pc if op in (0x06, 0x07): value = raw[-1] if op == 0x06 else int.from_bytes(raw[-2:], "big") self._write_ea(ea, value, 1 if op == 0x06 else 2) self._set_logic_flags(value, 1 if op == 0x06 else 2) return next_pc base = op & 0xF8 rd = op & 0x07 if ea["mode"] == "imm" and base in (0x48, 0x58, 0x68): value = self._read_ea(ea, size) if base == 0x48: self.cpu.sr |= value elif base == 0x58: self.cpu.sr &= value else: self.cpu.sr ^= value elif base == 0x80: value = self._read_ea(ea, size) self._reg_write(rd, value, size) self._set_logic_flags(value, size) elif base == 0x90: self._write_ea(ea, self._reg_read(rd, size), size) elif base == 0x20: lhs = self._reg_read(rd, size) rhs = self._read_ea(ea, size) result = lhs + rhs self._reg_write(rd, result, size) self._set_add_flags(lhs, rhs, result, size) elif base == 0x30: lhs = self._reg_read(rd, size) rhs = self._read_ea(ea, size) result = lhs - rhs self._reg_write(rd, result, size) self._set_sub_flags(lhs, rhs, result, size) elif base == 0x50: result = self._reg_read(rd, size) & self._read_ea(ea, size) self._reg_write(rd, result, size) self._set_logic_flags(result, size) elif base == 0x40: result = self._reg_read(rd, size) | self._read_ea(ea, size) self._reg_write(rd, result, size) self._set_logic_flags(result, size) elif base == 0x60: result = self._reg_read(rd, size) ^ self._read_ea(ea, size) self._reg_write(rd, result, size) self._set_logic_flags(result, size) elif base == 0x70: self._cmp(self._reg_read(rd, size), self._read_ea(ea, size), size) elif op in (0x08, 0x09, 0x0C, 0x0D): delta = {0x08: 1, 0x09: 2, 0x0C: -1, 0x0D: -2}[op] old = self._read_ea(ea, size) result = old + delta self._write_ea(ea, result, size) if delta >= 0: self._set_add_flags(old, delta, result, size) else: self._set_sub_flags(old, -delta, result, size) elif op == 0x13: self._write_ea(ea, 0, size) self._set_logic_flags(0, size) elif op == 0x16: self._set_logic_flags(self._read_ea(ea, size), size) elif op == 0x10: value = self._read_ea(ea, size) result = ((value & 0xFF) << 8) | ((value >> 8) & 0xFF) self._write_ea(ea, result, 2) self._set_logic_flags(result, 2) elif op == 0x12: value = self._read_ea(ea, size) & 0xFF self._write_ea(ea, value, 2 if size == 2 else 1) self._set_logic_flags(value, size) elif op == 0x1A: value = self._read_ea(ea, size) result = (value << 1) & _mask(size) self.cpu.c = bool(value & _sign_bit(size)) self._write_ea(ea, result, size) self._set_logic_flags(result, size) elif op == 0x1B: value = self._read_ea(ea, size) result = value >> 1 self.cpu.c = bool(value & 1) self._write_ea(ea, result, size) self._set_logic_flags(result, size) elif 0xC0 <= op <= 0xFF: bit = op & 0x0F self._bit_operation(ea, size, op & 0xF0, bit) elif base in (0x48, 0x58, 0x68, 0x78): bit = self._reg_read(rd, 1) & 0x0F self._bit_operation(ea, size, base + 0x80, bit) elif base == 0x88: value = self._read_ea(ea, size) if size == 2 and rd == 0: self.cpu.sr = value & 0xFFFF elif size == 1 and rd == 1: self.cpu.br = value & 0xFF else: raise UnsupportedInstruction(pc, raw, H8536Decoder(self.memory.rom, br=self.cpu.br).decode(pc).text) elif base == 0x98: if size == 2 and rd == 0: self._write_ea(ea, self.cpu.sr, 2) elif size == 1 and rd == 1: self._write_ea(ea, self.cpu.br, 1) else: raise UnsupportedInstruction(pc, raw, H8536Decoder(self.memory.rom, br=self.cpu.br).decode(pc).text) elif op in (0x04, 0x05): compare_size = 1 if op == 0x04 else 2 immediate = raw[-1] if op == 0x04 else int.from_bytes(raw[-2:], "big") self._cmp(self._read_ea(ea, compare_size), immediate, compare_size) else: raise UnsupportedInstruction(pc, raw, H8536Decoder(self.memory.rom, br=self.cpu.br).decode(pc).text) return next_pc def _branch8(self, raw: bytes, pc: int, next_pc: int) -> int: cond = raw[0] & 0x0F disp = _s8(raw[1]) target = (pc + 2 + disp) & 0xFFFF if self._branch_condition(cond): return target return next_pc def _branch16(self, raw: bytes, pc: int, next_pc: int) -> int: cond = raw[0] & 0x0F disp = _s16(int.from_bytes(raw[1:3], "big")) target = (pc + 3 + disp) & 0xFFFF if self._branch_condition(cond): return target return next_pc def _rough_cycles(self, raw: bytes) -> int: if raw[0] in (0x15, 0x1D): return 9 if raw[0] in (0x0E, 0x1E, 0x18): return 14 if raw[0] in (0x19, 0x0A): return 12 if raw[0] in range(0x20, 0x40): return 8 if (raw[0] & 0x0F) == 0 else 4 return 3 def _direct_call(self, raw: bytes, next_pc: int) -> int: self._push16(next_pc) if raw[0] == 0x0E: return (next_pc + _s8(raw[1])) & 0xFFFF if raw[0] == 0x1E: return (next_pc + _s16(int.from_bytes(raw[1:3], "big"))) & 0xFFFF return int.from_bytes(raw[1:3], "big") def _scb(self, raw: bytes, pc: int, next_pc: int) -> int: reg = raw[1] & 0x07 condition = {0x01: False, 0x06: not self.cpu.z, 0x07: self.cpu.z}[raw[0]] if condition: return next_pc value = (self.cpu.regs[reg] - 1) & 0xFFFF self.cpu.regs[reg] = value self.cpu.z = value == 0 if value != 0: return (pc + 3 + _s8(raw[2])) & 0xFFFF return next_pc def _tick_peripherals(self) -> None: self.sci1.tick() self._interval_counter += 1 self._service_pending_interrupt() def _service_pending_interrupt(self) -> None: if self.cpu.interrupt_depth: return candidates: list[tuple[int, int, str]] = [] if self.sci1.scr & SCI_SCR_TIE and self.sci1.ssr & SCI_SSR_TDRE: target = self._vector_target(VECTOR_SCI1_TXI) if target is not None: candidates.append((self._sci1_priority(), target, "sci1_txi")) if self._interval_counter >= self.interval_steps: target = self._vector_target(VECTOR_INTERVAL_TIMER) if target is not None: candidates.append((self._interval_priority(), target, "interval_timer")) if not candidates: return priority, target, source = max(candidates, key=lambda item: item[0]) if priority <= self._interrupt_mask(): return if source == "interval_timer": self._interval_counter = 0 self._enter_interrupt(target) def _enter_interrupt(self, target: int) -> None: self._push16(self.cpu.sr) self._push16(self.cpu.pc) self.cpu.pc = target & 0xFFFF self.cpu.interrupt_depth += 1 def _return_from_interrupt(self) -> int: pc = self._pop16() self.cpu.sr = self._pop16() if self.cpu.interrupt_depth: self.cpu.interrupt_depth -= 1 return pc def _vector_target(self, vector_address: int) -> int | None: entry = self.vectors.get(vector_address) return entry[1] if entry else None def _interrupt_mask(self) -> int: return (self.cpu.sr >> 8) & 0x07 def _sci1_priority(self) -> int: return (self.memory.read8(IPRE) >> 4) & 0x07 def _interval_priority(self) -> int: return (self.memory.read8(IPRA) >> 4) & 0x07 def _push16(self, value: int) -> None: sp = (self.cpu.regs[7] - 2) & 0xFFFF self.cpu.regs[7] = sp self.memory.write16(sp, value) def _pop16(self) -> int: sp = self.cpu.regs[7] & 0xFFFF value = self.memory.read16(sp) self.cpu.regs[7] = (sp + 2) & 0xFFFF return value def _push_register_mask(self, mask: int) -> None: for reg in range(8): if mask & (1 << reg): self._push16(self.cpu.regs[reg]) def _pop_register_mask(self, mask: int) -> None: for reg in reversed(range(8)): if mask & (1 << reg): self.cpu.regs[reg] = self._pop16() def _decode_ea(self, pc: int) -> dict[str, int | str | None]: first = self.memory.rom.u8(pc) if 0xA0 <= first <= 0xAF: return {"mode": "reg", "reg": first & 0x07, "size": 2 if first & 0x08 else 1, "length": 1, "value": None} if 0xB0 <= first <= 0xBF: return {"mode": "predec", "reg": first & 0x07, "size": 2 if first & 0x08 else 1, "length": 1, "value": None} if 0xC0 <= first <= 0xCF: return {"mode": "postinc", "reg": first & 0x07, "size": 2 if first & 0x08 else 1, "length": 1, "value": None} if 0xD0 <= first <= 0xDF: return {"mode": "indirect", "reg": first & 0x07, "size": 2 if first & 0x08 else 1, "length": 1, "value": None} if 0xE0 <= first <= 0xEF: return {"mode": "disp", "reg": first & 0x07, "size": 2 if first & 0x08 else 1, "length": 2, "value": _s8(self.memory.rom.u8(pc + 1))} if 0xF0 <= first <= 0xFF: return {"mode": "disp", "reg": first & 0x07, "size": 2 if first & 0x08 else 1, "length": 3, "value": _s16(self.memory.rom.u16(pc + 1))} if first in (0x04, 0x0C): size = 2 if first == 0x0C else 1 value = self.memory.rom.u16(pc + 1) if size == 2 else self.memory.rom.u8(pc + 1) return {"mode": "imm", "reg": None, "size": size, "length": 3 if size == 2 else 2, "value": value} if first in (0x15, 0x1D): return {"mode": "abs16", "reg": None, "size": 2 if first == 0x1D else 1, "length": 3, "value": self.memory.rom.u16(pc + 1)} raise UnsupportedInstruction(pc, self.memory.rom.slice(pc, 1), H8536Decoder(self.memory.rom, br=self.cpu.br).decode(pc).text) def _general_length(self, pc: int, ea: dict[str, int | str | None], op: int) -> int: length = int(ea["length"]) + 1 if op == 0x00: return length + 1 if op == 0x06: return length + 1 if op in (0x07, 0x05): return length + 2 if op == 0x04: return length + 1 return length def _ea_address(self, ea: dict[str, int | str | None], size: int, *, write: bool = False) -> int: mode = ea["mode"] reg = ea.get("reg") if mode == "abs16": return int(ea["value"]) & 0xFFFF if mode == "indirect": return self.cpu.regs[int(reg)] & 0xFFFF if mode == "disp": return (self.cpu.regs[int(reg)] + int(ea["value"])) & 0xFFFF if mode == "predec": self.cpu.regs[int(reg)] = (self.cpu.regs[int(reg)] - size) & 0xFFFF return self.cpu.regs[int(reg)] if mode == "postinc": address = self.cpu.regs[int(reg)] & 0xFFFF if not write: self.cpu.regs[int(reg)] = (self.cpu.regs[int(reg)] + size) & 0xFFFF return address raise EmulatorError(f"EA mode {mode} does not have an address") def _read_ea(self, ea: dict[str, int | str | None], size: int) -> int: mode = ea["mode"] if mode == "imm": return int(ea["value"]) & _mask(size) if mode == "reg": return self._reg_read(int(ea["reg"]), size) address = self._ea_address(ea, size) return self.memory.read16(address) if size == 2 else self.memory.read8(address) def _write_ea(self, ea: dict[str, int | str | None], value: int, size: int) -> None: if ea["mode"] == "reg": self._reg_write(int(ea["reg"]), value, size) return address = self._ea_address(ea, size, write=True) if size == 2: self.memory.write16(address, value) else: self.memory.write8(address, value) def _reg_read(self, reg: int, size: int) -> int: value = self.cpu.regs[reg] & 0xFFFF return value if size == 2 else value & 0xFF def _reg_write(self, reg: int, value: int, size: int) -> None: if size == 2: self.cpu.regs[reg] = value & 0xFFFF else: self.cpu.regs[reg] = (self.cpu.regs[reg] & 0xFF00) | (value & 0xFF) def _cmp(self, lhs: int, rhs: int, size: int) -> None: self._set_sub_flags(lhs, rhs, lhs - rhs, size) def _set_logic_flags(self, value: int, size: int) -> None: value &= _mask(size) self.cpu.z = value == 0 self.cpu.n = bool(value & _sign_bit(size)) self.cpu.v = False def _set_add_flags(self, lhs: int, rhs: int, result: int, size: int) -> None: mask = _mask(size) sign = _sign_bit(size) result &= mask lhs &= mask rhs &= mask self.cpu.z = result == 0 self.cpu.n = bool(result & sign) self.cpu.c = lhs + rhs > mask self.cpu.v = bool((~(lhs ^ rhs) & (lhs ^ result) & sign) != 0) def _set_sub_flags(self, lhs: int, rhs: int, result: int, size: int) -> None: mask = _mask(size) sign = _sign_bit(size) result &= mask lhs &= mask rhs &= mask self.cpu.z = result == 0 self.cpu.n = bool(result & sign) self.cpu.c = lhs < rhs self.cpu.v = bool(((lhs ^ rhs) & (lhs ^ result) & sign) != 0) def _bit_operation(self, ea: dict[str, int | str | None], size: int, op_base: int, bit: int) -> None: value = self._read_ea(ea, size) bit &= (15 if size == 2 else 7) mask = 1 << bit self.cpu.z = not bool(value & mask) if op_base == 0xC0: self._write_ea(ea, value | mask, size) elif op_base == 0xD0: self._write_ea(ea, value & ~mask, size) elif op_base == 0xE0: self._write_ea(ea, value ^ mask, size) def _branch_condition(self, cond: int) -> bool: if cond == 0x0: return True if cond == 0x1: return False if cond == 0x2: return not self.cpu.c and not self.cpu.z if cond == 0x3: return self.cpu.c or self.cpu.z if cond == 0x4: return not self.cpu.c if cond == 0x5: return self.cpu.c if cond == 0x6: return not self.cpu.z if cond == 0x7: return self.cpu.z if cond == 0x8: return not self.cpu.v if cond == 0x9: return self.cpu.v if cond == 0xA: return not self.cpu.n if cond == 0xB: return self.cpu.n if cond == 0xC: return self.cpu.n == self.cpu.v if cond == 0xD: return self.cpu.n != self.cpu.v if cond == 0xE: return not self.cpu.z and self.cpu.n == self.cpu.v return self.cpu.z or self.cpu.n != self.cpu.v def _s8(value: int) -> int: return value - 0x100 if value & 0x80 else value def _s16(value: int) -> int: return value - 0x10000 if value & 0x8000 else value def _mask(size: int) -> int: return 0xFFFF if size == 2 else 0xFF def _sign_bit(size: int) -> int: return 0x8000 if size == 2 else 0x80 def discover_rom_path(root: Path) -> Path | None: candidates = [ root / "ROM" / "M27C512@DIP28_1.BIN", root / "rom.bin", ] candidates.extend(sorted((root / "ROM").glob("*.BIN")) if (root / "ROM").exists() else []) candidates.extend(sorted((root / "ROM").glob("*.bin")) if (root / "ROM").exists() else []) for candidate in candidates: if candidate.is_file(): return candidate return None def load_rom(path: Path | None = None, root: Path | None = None) -> tuple[bytes, Path]: root = root if root is not None else Path.cwd() rom_path = path if path is not None else discover_rom_path(root) if rom_path is None: raise FileNotFoundError( "could not discover ROM bytes; pass --rom PATH, expected ROM/M27C512@DIP28_1.BIN or another ROM/*.BIN" ) return rom_path.read_bytes(), rom_path def describe_regions(regions: Iterable[MemoryRegion] = MEMORY_REGIONS) -> str: return "\n".join(f"{h16(region.start)}-{h16(region.end)} {region.name} {region.kind}" for region in regions) def build_arg_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Minimal H8/536 emulation harness scaffold") parser.add_argument("--rom", type=Path, help="ROM image path; defaults to ROM/M27C512@DIP28_1.BIN when present") parser.add_argument("--max-steps", type=int, default=64, help="maximum CPU steps to execute") parser.add_argument("--trace", action="store_true", help="print decoded/executed instruction trace") parser.add_argument("--stop-on-heartbeat", action="store_true", help="stop only when 00 00 00 00 80 DA is emitted through SCI1 TDR") parser.add_argument("--memory-map", action="store_true", help="print the scaffolded memory map before running") parser.add_argument("--interval-steps", type=int, default=2048, help="rough step period for the scaffolded timer interrupt") return parser def main(argv: list[str] | None = None) -> int: args = build_arg_parser().parse_args(argv) try: rom_bytes, rom_path = load_rom(args.rom) except FileNotFoundError as exc: print(str(exc)) return 2 emulator = H8536Emulator(rom_bytes, interval_steps=args.interval_steps) print(f"rom={rom_path}") print(f"reset_vector={h16(emulator.reset_vector())}") if args.memory_map: print(describe_regions()) report = emulator.run(args.max_steps, trace=args.trace, stop_on_heartbeat=args.stop_on_heartbeat) if args.trace: for line in report.trace: print(line) for line in report.summary_lines(): print(line) if not report.heartbeat_seen: print("heartbeat_status=not reached; no heartbeat is reported unless bytes are emitted via SCI1_TDR") return 0