p9 bus emulation

h8536/emulator/peripherals/p9_bus.py

@@ -17,25 +17,63 @@ class P9StrobeEvent:
     bit7_output: bool
 
 
+@dataclass(frozen=True)
+class P9TraceEvent:
+    kind: str
+    ddr: int
+    dr: int
+    value: int | None = None
+    bit: int | None = None
+    source: str | None = None
+    success: bool | None = None
+    queue_depth: int | None = None
+
+    def line(self) -> str:
+        parts = [self.kind, f"ddr={self.ddr:02X}", f"dr={self.dr:02X}"]
+        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.success is not None:
+            parts.append(f"success={int(self.success)}")
+        if self.source is not None:
+            parts.append(f"source={self.source}")
+        if self.queue_depth is not None:
+            parts.append(f"queued={self.queue_depth}")
+        return " ".join(parts)
+
+
 class P9Bus:
     """Small model for the ROM's P9 bit-banged serial handshake."""
 
-    def __init__(self, ddr: int = 0x00, dr: int = 0x00, input_bits: Iterable[int] = ()) -> None:
+    def __init__(
+        self,
+        ddr: int = 0x00,
+        dr: int = 0x00,
+        input_bits: Iterable[int] = (),
+        wrapper_results: Iterable[bool] = (),
+    ) -> None:
         self.ddr = ddr & 0xFF
         self.dr_latch = dr & 0xFF
         self.input_bits: list[int] = [1 if bit else 0 for bit in input_bits]
         self.default_input_bit = 0
+        self.wrapper_results: list[bool] = [bool(result) for result in wrapper_results]
+        self.wrapper_sources: list[str] = ["initial"] * len(self.wrapper_results)
+        self.default_wrapper_success = False
         self.strobe_edges: list[P9StrobeEvent] = []
+        self.trace_events: list[P9TraceEvent] = []
         self.transmitted_bits: list[int] = []
         self.byte_candidates: list[int] = []
 
     def write_ddr(self, value: int) -> int:
         self.ddr = value & 0xFF
+        self.trace_events.append(P9TraceEvent("write_ddr", self.ddr, self.dr_latch, value=self.ddr))
         return self.ddr
 
     def write_dr(self, value: int) -> int:
         previous = self.dr_latch
         self.dr_latch = value & 0xFF
+        self.trace_events.append(P9TraceEvent("write_dr", self.ddr, self.dr_latch, value=self.dr_latch))
 
         previous_strobe = bool(previous & P9_STROBE_BIT)
         current_strobe = bool(self.dr_latch & P9_STROBE_BIT)
@@ -44,6 +82,7 @@ class P9Bus:
             data_bit = 1 if self.dr_latch & P9_ACK_BIT else 0
             bit7_output = bool(self.ddr & P9_ACK_BIT)
             self.strobe_edges.append(P9StrobeEvent(edge, self.ddr, self.dr_latch, data_bit, bit7_output))
+            self.trace_events.append(P9TraceEvent(f"strobe_{edge}", self.ddr, self.dr_latch, bit=data_bit))
             if edge == "rising" and bit7_output:
                 self._record_transmitted_bit(data_bit)
 
@@ -54,15 +93,20 @@ class P9Bus:
 
     def read_dr(self) -> int:
         value = self.dr_latch
+        input_bit = None
+        source = None
         if not (self.ddr & P9_ACK_BIT):
             if self.input_bits:
                 input_bit = self.input_bits.pop(0)
+                source = "queued_bit"
             else:
                 input_bit = self.default_input_bit
+                source = "default_bit"
             if input_bit:
                 value |= P9_ACK_BIT
             else:
                 value &= ~P9_ACK_BIT
+        self.trace_events.append(P9TraceEvent("read_dr", self.ddr, value & 0xFF, value=value, bit=input_bit, source=source))
         return value & 0xFF
 
     def queue_input_bits(self, bits: Iterable[int]) -> None:
@@ -71,10 +115,42 @@ class P9Bus:
     def set_default_input_bit(self, bit: int) -> None:
         self.default_input_bit = 1 if bit else 0
 
+    def queue_wrapper_results(self, results: Iterable[bool], source: str = "queued_wrapper") -> None:
+        for result in results:
+            self.wrapper_results.append(bool(result))
+            self.wrapper_sources.append(source)
+
+    def consume_wrapper_result(self) -> tuple[bool, str, int]:
+        if self.wrapper_results:
+            success = self.wrapper_results.pop(0)
+            source = self.wrapper_sources.pop(0) if self.wrapper_sources else "queued_wrapper"
+        else:
+            success = self.default_wrapper_success
+            source = "default_success" if success else "default_timeout"
+        queue_depth = len(self.wrapper_results)
+        self.trace_events.append(
+            P9TraceEvent(
+                "wrapper_result",
+                self.ddr,
+                self.dr_latch,
+                value=1 if success else 0,
+                success=success,
+                source=source,
+                queue_depth=queue_depth,
+            )
+        )
+        return success, source, queue_depth
+
+    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 _record_transmitted_bit(self, bit: int) -> None:
         self.transmitted_bits.append(bit)
+        self.trace_events.append(P9TraceEvent("tx_bit", self.ddr, self.dr_latch, bit=bit))
         if len(self.transmitted_bits) % 8 == 0:
             byte = 0
             for data_bit in self.transmitted_bits[-8:]:
                 byte = (byte << 1) | data_bit
             self.byte_candidates.append(byte)
+            self.trace_events.append(P9TraceEvent("tx_byte", self.ddr, self.dr_latch, value=byte))