479 lines
19 KiB
Python
479 lines
19 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
|
|
|
|
X24164_SIZE = 2048
|
|
X24164_LOGICAL_SIZE = 4096
|
|
X24164_FACTORY_DEFAULT_BASE = 0xC964
|
|
X24164_FACTORY_DEFAULT_BYTES = 0x0100
|
|
X24164_LOGICAL_PAGE_SIZE = 0x0100
|
|
X24164_LOGICAL_PAGE_COUNT = 0x10
|
|
|
|
|
|
@dataclass
|
|
class X24164Device:
|
|
"""Small Xicor X24164 serial EEPROM model.
|
|
|
|
The ROM uses two control-byte families on the P91/P97 two-wire bus:
|
|
H'A0/H'A1 for the low logical half and H'E0/H'E1 for the high logical half.
|
|
X24164 has unusual device-select encoding compared with later 24Cxx parts,
|
|
so the emulator stores the accepted high-nibble control family directly.
|
|
"""
|
|
|
|
name: str
|
|
control_base: int
|
|
data: bytearray = field(default_factory=lambda: bytearray([0xFF] * X24164_SIZE))
|
|
|
|
def __post_init__(self) -> None:
|
|
self.control_base &= 0xF0
|
|
if len(self.data) < X24164_SIZE:
|
|
self.data.extend([0xFF] * (X24164_SIZE - len(self.data)))
|
|
elif len(self.data) > X24164_SIZE:
|
|
del self.data[X24164_SIZE:]
|
|
|
|
def matches_control(self, value: int) -> bool:
|
|
return (value & 0xF0) == self.control_base
|
|
|
|
def offset_from_control(self, value: int, word_address: int) -> int:
|
|
high_address = (value >> 1) & 0x07
|
|
return ((high_address << 8) | (word_address & 0xFF)) & (X24164_SIZE - 1)
|
|
|
|
def read(self, offset: int) -> int:
|
|
return self.data[offset & (X24164_SIZE - 1)]
|
|
|
|
def write(self, offset: int, value: int) -> None:
|
|
self.data[offset & (X24164_SIZE - 1)] = value & 0xFF
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class X24164TraceEvent:
|
|
kind: str
|
|
device: str | None = None
|
|
value: int | None = None
|
|
address: int | None = None
|
|
bit: int | None = None
|
|
ack: bool | None = None
|
|
message: str | None = None
|
|
|
|
def line(self) -> str:
|
|
parts = [self.kind]
|
|
if self.device is not None:
|
|
parts.append(f"device={self.device}")
|
|
if self.address is not None:
|
|
parts.append(f"addr={self.address:03X}")
|
|
if self.value is not None:
|
|
parts.append(f"value={self.value:02X}")
|
|
if self.bit is not None:
|
|
parts.append(f"bit={self.bit}")
|
|
if self.ack is not None:
|
|
parts.append(f"ack={int(self.ack)}")
|
|
if self.message is not None:
|
|
parts.append(self.message)
|
|
return " ".join(parts)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class X24164WriteEvent:
|
|
logical_address: int
|
|
device: str
|
|
device_offset: int
|
|
old_value: int
|
|
new_value: int
|
|
source: str
|
|
|
|
def line(self) -> str:
|
|
return (
|
|
f"addr={self.logical_address & 0x0FFF:03X} device={self.device} "
|
|
f"offset={self.device_offset & (X24164_SIZE - 1):03X} "
|
|
f"{self.old_value & 0xFF:02X}->{self.new_value & 0xFF:02X} source={self.source}"
|
|
)
|
|
|
|
|
|
class X24164Bus:
|
|
"""Bit-level two-wire bus model for X24164 EEPROMs."""
|
|
|
|
def __init__(self, devices: list[X24164Device] | None = None) -> None:
|
|
self.devices = devices if devices is not None else default_x24164_devices()
|
|
self.trace_events: list[X24164TraceEvent] = []
|
|
self.write_events: list[X24164WriteEvent] = []
|
|
self.active = False
|
|
self.phase = "idle"
|
|
self.selected: X24164Device | None = None
|
|
self.control_byte = 0
|
|
self.address = 0
|
|
self._rx_bits: list[int] = []
|
|
self._ack_pending: bool | None = None
|
|
self._ack_armed_on_current_clock = False
|
|
self._read_byte = 0xFF
|
|
self._read_bit_index = 0
|
|
self._read_prepared_on_current_clock = False
|
|
self._awaiting_master_ack = False
|
|
|
|
def observe(
|
|
self,
|
|
*,
|
|
previous_scl: bool,
|
|
previous_master_sda: bool,
|
|
current_scl: bool,
|
|
current_master_sda: bool,
|
|
master_sda_output: bool,
|
|
) -> None:
|
|
if previous_scl and current_scl and previous_master_sda != current_master_sda:
|
|
if previous_master_sda and not current_master_sda:
|
|
self.start()
|
|
elif not previous_master_sda and current_master_sda:
|
|
self.stop()
|
|
|
|
if not previous_scl and current_scl:
|
|
self._scl_rising(current_master_sda, master_sda_output)
|
|
elif previous_scl and not current_scl:
|
|
self._scl_falling()
|
|
|
|
def start(self) -> None:
|
|
self.active = True
|
|
self.phase = "control"
|
|
self.selected = None
|
|
self._rx_bits.clear()
|
|
self._ack_pending = None
|
|
self._ack_armed_on_current_clock = False
|
|
self._read_prepared_on_current_clock = False
|
|
self._awaiting_master_ack = False
|
|
self.trace_events.append(X24164TraceEvent("x24164_start"))
|
|
|
|
def stop(self) -> None:
|
|
if self.active:
|
|
self.trace_events.append(X24164TraceEvent("x24164_stop", device=self._selected_name()))
|
|
self.active = False
|
|
self.phase = "idle"
|
|
self.selected = None
|
|
self._rx_bits.clear()
|
|
self._ack_pending = None
|
|
self._ack_armed_on_current_clock = False
|
|
self._read_prepared_on_current_clock = False
|
|
self._awaiting_master_ack = False
|
|
|
|
def sda_bit(self) -> int | None:
|
|
if not self.active:
|
|
return None
|
|
if self._ack_pending is not None:
|
|
return 0 if self._ack_pending else 1
|
|
if self.phase == "read_data" and not self._awaiting_master_ack:
|
|
return (self._read_byte >> (7 - self._read_bit_index)) & 1
|
|
return 1
|
|
|
|
def fast_start(self) -> None:
|
|
self.start()
|
|
|
|
def fast_stop(self) -> None:
|
|
self.stop()
|
|
|
|
def fast_write_byte(self, value: int) -> bool:
|
|
self._accept_byte(value & 0xFF)
|
|
ack = bool(self._ack_pending)
|
|
self.trace_events.append(
|
|
X24164TraceEvent("x24164_fast_write_byte", self._selected_name(), value=value & 0xFF, ack=ack)
|
|
)
|
|
self._ack_pending = None
|
|
self._ack_armed_on_current_clock = False
|
|
if self.phase == "read_data" and self.selected is not None:
|
|
self._prepare_read_byte()
|
|
return ack
|
|
|
|
def fast_read_byte(self) -> int | None:
|
|
if not self.active or self.phase != "read_data" or self.selected is None:
|
|
return None
|
|
value = self.selected.read(self.address)
|
|
self.trace_events.append(
|
|
X24164TraceEvent("x24164_fast_read_byte", self.selected.name, value=value, address=self.address)
|
|
)
|
|
self.address = (self.address + 1) & (X24164_SIZE - 1)
|
|
self._prepare_read_byte()
|
|
return value
|
|
|
|
def fast_master_ack(self, ack: bool = True) -> None:
|
|
if not self.active:
|
|
return
|
|
self.trace_events.append(X24164TraceEvent("x24164_fast_master_ack", self._selected_name(), ack=ack))
|
|
if ack and self.phase == "read_data" and self.selected is not None:
|
|
self._prepare_read_byte()
|
|
elif not ack:
|
|
self.phase = "idle"
|
|
|
|
def read_linear_word(self, address: int) -> tuple[bool, int]:
|
|
device = self._device_for_linear_address(address)
|
|
if device is None:
|
|
self.trace_events.append(
|
|
X24164TraceEvent("x24164_linear_read_miss", address=address & 0x0FFF, message="no_mapped_device")
|
|
)
|
|
return False, 0xFFFF
|
|
offset = address & (X24164_SIZE - 1)
|
|
high = device.read(offset)
|
|
low = device.read((offset + 1) & (X24164_SIZE - 1))
|
|
value = (high << 8) | low
|
|
self.trace_events.append(
|
|
X24164TraceEvent("x24164_linear_read_word", device.name, value=high, address=offset, message=f"word={value:04X}")
|
|
)
|
|
return True, value
|
|
|
|
def read_linear_byte(self, address: int) -> tuple[bool, int]:
|
|
device = self._device_for_linear_address(address)
|
|
if device is None:
|
|
self.trace_events.append(
|
|
X24164TraceEvent("x24164_linear_read_miss", address=address & 0x0FFF, message="no_mapped_device")
|
|
)
|
|
return False, 0xFF
|
|
offset = address & (X24164_SIZE - 1)
|
|
value = device.read(offset)
|
|
self.trace_events.append(X24164TraceEvent("x24164_linear_read_byte", device.name, value=value, address=offset))
|
|
return True, value
|
|
|
|
def write_linear_word(self, address: int, value: int) -> bool:
|
|
device = self._device_for_linear_address(address)
|
|
if device is None:
|
|
self.trace_events.append(
|
|
X24164TraceEvent("x24164_linear_write_miss", address=address & 0x0FFF, message="no_mapped_device")
|
|
)
|
|
return False
|
|
offset = address & (X24164_SIZE - 1)
|
|
self._write_device_byte(device, offset, (value >> 8) & 0xFF, source="linear_word")
|
|
self._write_device_byte(device, (offset + 1) & (X24164_SIZE - 1), value & 0xFF, source="linear_word")
|
|
self.trace_events.append(
|
|
X24164TraceEvent(
|
|
"x24164_linear_write_word",
|
|
device.name,
|
|
value=(value >> 8) & 0xFF,
|
|
address=offset,
|
|
message=f"word={value & 0xFFFF:04X}",
|
|
)
|
|
)
|
|
return True
|
|
|
|
def write_linear_byte(self, address: int, value: int, *, source: str = "linear_byte") -> bool:
|
|
device = self._device_for_linear_address(address)
|
|
if device is None:
|
|
self.trace_events.append(
|
|
X24164TraceEvent("x24164_linear_write_miss", address=address & 0x0FFF, message="no_mapped_device")
|
|
)
|
|
return False
|
|
offset = address & (X24164_SIZE - 1)
|
|
self._write_device_byte(device, offset, value, source=source)
|
|
self.trace_events.append(X24164TraceEvent("x24164_linear_write_byte", device.name, value=value & 0xFF, address=offset))
|
|
return True
|
|
|
|
def dump_linear(self) -> bytes:
|
|
data = bytearray()
|
|
for address in range(X24164_LOGICAL_SIZE):
|
|
device = self._device_for_linear_address(address)
|
|
if device is None:
|
|
data.append(0xFF)
|
|
else:
|
|
data.append(device.read(address & (X24164_SIZE - 1)))
|
|
return bytes(data)
|
|
|
|
def load_linear(self, data: bytes | bytearray, *, fill: int = 0xFF) -> None:
|
|
if len(data) > X24164_LOGICAL_SIZE:
|
|
raise ValueError(f"EEPROM image is too large: {len(data)} > {X24164_LOGICAL_SIZE}")
|
|
padded = bytearray([fill & 0xFF] * X24164_LOGICAL_SIZE)
|
|
padded[: len(data)] = data
|
|
for address, value in enumerate(padded):
|
|
device = self._device_for_linear_address(address)
|
|
if device is not None:
|
|
device.write(address & (X24164_SIZE - 1), value)
|
|
self.clear_write_log()
|
|
|
|
def clear_write_log(self) -> None:
|
|
self.write_events.clear()
|
|
|
|
def seed_factory_defaults_from_rom(self, rom_bytes: bytes) -> None:
|
|
for offset, word in factory_default_words_from_rom(rom_bytes):
|
|
for page in range(X24164_LOGICAL_PAGE_COUNT):
|
|
self.write_linear_word((page * X24164_LOGICAL_PAGE_SIZE) + offset, word)
|
|
|
|
for page in range(1, X24164_LOGICAL_PAGE_COUNT):
|
|
base = page * X24164_LOGICAL_PAGE_SIZE
|
|
for offset in range(0, 8, 2):
|
|
self.write_linear_word(base + offset, 0x2020)
|
|
self.clear_write_log()
|
|
|
|
def trace_lines(self, limit: int | None = None) -> list[str]:
|
|
events = self.trace_events if limit is None else self.trace_events[-limit:]
|
|
return [event.line() for event in events]
|
|
|
|
def write_log_lines(self, limit: int | None = None) -> list[str]:
|
|
events = self.write_events if limit is None else self.write_events[-limit:]
|
|
return [event.line() for event in events]
|
|
|
|
def _scl_rising(self, master_sda: bool, master_sda_output: bool) -> None:
|
|
if not self.active:
|
|
return
|
|
if self._ack_pending is not None:
|
|
self.trace_events.append(X24164TraceEvent("x24164_ack_bit", self._selected_name(), ack=self._ack_pending))
|
|
return
|
|
if self.phase == "read_data":
|
|
if self._awaiting_master_ack:
|
|
ack = not master_sda if master_sda_output else False
|
|
self.trace_events.append(X24164TraceEvent("x24164_master_ack", self._selected_name(), ack=ack))
|
|
self._awaiting_master_ack = False
|
|
if ack and self.selected is not None:
|
|
self._prepare_read_byte(skip_current_falling=True)
|
|
else:
|
|
self.phase = "idle"
|
|
return
|
|
if master_sda_output:
|
|
self._rx_bits.append(1 if master_sda else 0)
|
|
self.trace_events.append(X24164TraceEvent("x24164_rx_bit", self._selected_name(), bit=self._rx_bits[-1]))
|
|
if len(self._rx_bits) == 8:
|
|
value = 0
|
|
for bit in self._rx_bits:
|
|
value = (value << 1) | bit
|
|
self._rx_bits.clear()
|
|
self._accept_byte(value)
|
|
|
|
def _scl_falling(self) -> None:
|
|
if not self.active:
|
|
return
|
|
if self._ack_pending is not None:
|
|
if self._ack_armed_on_current_clock:
|
|
self._ack_armed_on_current_clock = False
|
|
return
|
|
self._ack_pending = None
|
|
if self.phase == "read_data" and self.selected is not None:
|
|
self._prepare_read_byte()
|
|
return
|
|
if self.phase == "read_data" and not self._awaiting_master_ack:
|
|
if self._read_prepared_on_current_clock:
|
|
self._read_prepared_on_current_clock = False
|
|
return
|
|
if self._read_bit_index < 7:
|
|
self._read_bit_index += 1
|
|
else:
|
|
self.trace_events.append(
|
|
X24164TraceEvent("x24164_tx_byte_done", self._selected_name(), value=self._read_byte)
|
|
)
|
|
if self.selected is not None:
|
|
self.address = (self.address + 1) & (X24164_SIZE - 1)
|
|
self._awaiting_master_ack = True
|
|
|
|
def _accept_byte(self, value: int) -> None:
|
|
value &= 0xFF
|
|
if self.phase == "control":
|
|
self.control_byte = value
|
|
self.selected = self._device_for_control(value)
|
|
read_mode = bool(value & 1)
|
|
self._ack_pending = self.selected is not None
|
|
self._ack_armed_on_current_clock = True
|
|
self.trace_events.append(
|
|
X24164TraceEvent(
|
|
"x24164_control",
|
|
self._selected_name(),
|
|
value=value,
|
|
ack=self._ack_pending,
|
|
message="read" if read_mode else "write",
|
|
)
|
|
)
|
|
if self.selected is None:
|
|
self.phase = "ignore"
|
|
elif read_mode:
|
|
self.phase = "read_data"
|
|
else:
|
|
self.phase = "word_address"
|
|
return
|
|
if self.phase == "word_address":
|
|
if self.selected is None:
|
|
self._ack_pending = False
|
|
self._ack_armed_on_current_clock = True
|
|
self.phase = "ignore"
|
|
return
|
|
self.address = self.selected.offset_from_control(self.control_byte, value)
|
|
self._ack_pending = True
|
|
self._ack_armed_on_current_clock = True
|
|
self.phase = "write_data"
|
|
self.trace_events.append(
|
|
X24164TraceEvent("x24164_word_address", self.selected.name, value=value, address=self.address, ack=True)
|
|
)
|
|
return
|
|
if self.phase == "write_data":
|
|
if self.selected is None:
|
|
self._ack_pending = False
|
|
self._ack_armed_on_current_clock = True
|
|
self.phase = "ignore"
|
|
return
|
|
self._write_device_byte(self.selected, self.address, value, source="bit_banged")
|
|
self.trace_events.append(
|
|
X24164TraceEvent("x24164_write_data", self.selected.name, value=value, address=self.address, ack=True)
|
|
)
|
|
self.address = (self.address + 1) & (X24164_SIZE - 1)
|
|
self._ack_pending = True
|
|
self._ack_armed_on_current_clock = True
|
|
return
|
|
self._ack_pending = False
|
|
self._ack_armed_on_current_clock = True
|
|
self.trace_events.append(X24164TraceEvent("x24164_ignored_byte", self._selected_name(), value=value, ack=False))
|
|
|
|
def _prepare_read_byte(self, *, skip_current_falling: bool = False) -> None:
|
|
if self.selected is None:
|
|
self._read_byte = 0xFF
|
|
return
|
|
self._read_byte = self.selected.read(self.address)
|
|
self._read_bit_index = 0
|
|
self._read_prepared_on_current_clock = skip_current_falling
|
|
self._awaiting_master_ack = False
|
|
self.trace_events.append(
|
|
X24164TraceEvent("x24164_prepare_read", self.selected.name, value=self._read_byte, address=self.address)
|
|
)
|
|
|
|
def _device_for_control(self, value: int) -> X24164Device | None:
|
|
for device in self.devices:
|
|
if device.matches_control(value):
|
|
return device
|
|
return None
|
|
|
|
def _device_for_linear_address(self, address: int) -> X24164Device | None:
|
|
bank = (address >> 11) & 1
|
|
wanted_base = 0xA0 if bank == 0 else 0xE0
|
|
for device in self.devices:
|
|
if device.control_base == wanted_base:
|
|
return device
|
|
return None
|
|
|
|
def _linear_base_for_device(self, device: X24164Device) -> int:
|
|
return 0x0800 if device.control_base == 0xE0 else 0x0000
|
|
|
|
def _write_device_byte(self, device: X24164Device, offset: int, value: int, *, source: str) -> None:
|
|
offset &= X24164_SIZE - 1
|
|
old_value = device.read(offset)
|
|
device.write(offset, value)
|
|
self.write_events.append(
|
|
X24164WriteEvent(
|
|
logical_address=(self._linear_base_for_device(device) + offset) & 0x0FFF,
|
|
device=device.name,
|
|
device_offset=offset,
|
|
old_value=old_value,
|
|
new_value=value & 0xFF,
|
|
source=source,
|
|
)
|
|
)
|
|
|
|
def _selected_name(self) -> str | None:
|
|
return self.selected.name if self.selected is not None else None
|
|
|
|
|
|
def default_x24164_devices() -> list[X24164Device]:
|
|
return [
|
|
X24164Device("x24164_a0_lower_2k", 0xA0),
|
|
X24164Device("x24164_e0_upper_2k", 0xE0),
|
|
]
|
|
|
|
|
|
def factory_default_words_from_rom(rom_bytes: bytes) -> list[tuple[int, int]]:
|
|
end = X24164_FACTORY_DEFAULT_BASE + X24164_FACTORY_DEFAULT_BYTES
|
|
if len(rom_bytes) < end:
|
|
raise ValueError(f"ROM is too small for X24164 factory default table at {X24164_FACTORY_DEFAULT_BASE:04X}")
|
|
words: list[tuple[int, int]] = []
|
|
for offset in range(0, X24164_FACTORY_DEFAULT_BYTES, 2):
|
|
address = X24164_FACTORY_DEFAULT_BASE + offset
|
|
word = (rom_bytes[address] << 8) | rom_bytes[address + 1]
|
|
words.append((offset, word))
|
|
return words
|