non-volatile storage emulation
This commit is contained in:
Aiden
362
h8536/emulator/peripherals/x24164.py
Normal file
362
h8536/emulator/peripherals/x24164.py
Normal file
@@ -0,0 +1,362 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
|
||||
X24164_SIZE = 2048
|
||||
|
||||
|
||||
@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)
|
||||
|
||||
|
||||
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.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 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)
|
||||
device.write(offset, (value >> 8) & 0xFF)
|
||||
device.write((offset + 1) & (X24164_SIZE - 1), value & 0xFF)
|
||||
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 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 _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.selected.write(self.address, value)
|
||||
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 _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),
|
||||
]
|
||||
Reference in New Issue
Block a user