144 lines
5.0 KiB
Python
144 lines
5.0 KiB
Python
"""Userspace view of the ``/dev/iec0`` record stream.
|
|
|
|
The kernel module (``kernel/iec_listener.c``) pushes fixed-size ``struct
|
|
iec_record`` entries through a character device. This module mirrors that wire
|
|
format so the whole userspace side can be exercised on a host by replaying a
|
|
captured record file -- no Pi or kernel required (PLAN.md Phase 0).
|
|
|
|
Wire format (must match ``struct iec_record`` in ``kernel/iec_listener.h``)::
|
|
|
|
struct iec_record {
|
|
__u8 kind; // KIND_*
|
|
__u8 value; // the byte, or an EV_* code when kind == KIND_EVENT
|
|
__u8 flags; // FLAG_* bitmask
|
|
__u8 _pad;
|
|
__u64 ts_ns; // kernel monotonic timestamp (ktime_get_ns)
|
|
} __packed; // 12 bytes, little-endian on the Pi (ARM LE)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import struct
|
|
from dataclasses import dataclass
|
|
from typing import BinaryIO, Iterable, Iterator
|
|
|
|
# --- record kinds -----------------------------------------------------------
|
|
KIND_COMMAND = 0 # a command byte received during the ATN phase
|
|
KIND_DATA = 1 # a data byte received in the data phase
|
|
KIND_EVENT = 2 # a state-machine event; ``value`` is an EV_* code
|
|
|
|
# --- record flags (bitmask) -------------------------------------------------
|
|
FLAG_EOI = 0x01 # this byte was the last in the stream (EOI signalled)
|
|
FLAG_ADDRESSED = 0x02 # this transfer / transition concerns *our* address
|
|
|
|
# --- event codes (value field when kind == KIND_EVENT) ----------------------
|
|
EV_IDLE = 0 # entered IDLE, waiting for ATN
|
|
EV_ATN_ASSERTED = 1 # ATN went low from IDLE -> we pulled DATA low (ack)
|
|
EV_ATN_COMMAND = 2 # ATN re-asserted mid-stream -> back to command phase
|
|
EV_ATN_RELEASED = 3 # ATN released; FLAG_ADDRESSED set => we become listener
|
|
EV_RESET = 4 # RESET asserted -> bus released, state cleared
|
|
|
|
# struct layout: kind, value, flags, 1 pad byte, u64 timestamp (little-endian)
|
|
RECORD_FMT = "<BBBxQ"
|
|
RECORD_SIZE = struct.calcsize(RECORD_FMT)
|
|
assert RECORD_SIZE == 12, RECORD_SIZE
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class IecRecord:
|
|
"""One decoded ``struct iec_record``."""
|
|
|
|
kind: int
|
|
value: int
|
|
flags: int
|
|
ts_ns: int
|
|
|
|
# -- convenience predicates ---------------------------------------------
|
|
@property
|
|
def is_command(self) -> bool:
|
|
return self.kind == KIND_COMMAND
|
|
|
|
@property
|
|
def is_data(self) -> bool:
|
|
return self.kind == KIND_DATA
|
|
|
|
@property
|
|
def is_event(self) -> bool:
|
|
return self.kind == KIND_EVENT
|
|
|
|
@property
|
|
def eoi(self) -> bool:
|
|
return bool(self.flags & FLAG_EOI)
|
|
|
|
@property
|
|
def addressed(self) -> bool:
|
|
return bool(self.flags & FLAG_ADDRESSED)
|
|
|
|
# -- (de)serialisation ---------------------------------------------------
|
|
@classmethod
|
|
def from_bytes(cls, raw: bytes) -> "IecRecord":
|
|
if len(raw) != RECORD_SIZE:
|
|
raise ValueError(f"expected {RECORD_SIZE} bytes, got {len(raw)}")
|
|
kind, value, flags, ts_ns = struct.unpack(RECORD_FMT, raw)
|
|
return cls(kind=kind, value=value, flags=flags, ts_ns=ts_ns)
|
|
|
|
def to_bytes(self) -> bytes:
|
|
return struct.pack(RECORD_FMT, self.kind, self.value, self.flags, self.ts_ns)
|
|
|
|
|
|
def iter_records(stream: BinaryIO) -> Iterator[IecRecord]:
|
|
"""Yield :class:`IecRecord` objects from a binary stream.
|
|
|
|
Reads exactly ``RECORD_SIZE`` bytes at a time, which matches the kernel
|
|
char device returning one record per ``read()``. Stops cleanly at EOF; a
|
|
trailing partial record raises ``ValueError`` (a corrupt capture).
|
|
"""
|
|
while True:
|
|
raw = stream.read(RECORD_SIZE)
|
|
if not raw:
|
|
return
|
|
if len(raw) < RECORD_SIZE:
|
|
raise ValueError(
|
|
f"truncated record: {len(raw)} of {RECORD_SIZE} bytes at EOF"
|
|
)
|
|
yield IecRecord.from_bytes(raw)
|
|
|
|
|
|
def write_records(stream: BinaryIO, records: Iterable[IecRecord]) -> None:
|
|
"""Serialise records back to a binary stream (used to build test captures)."""
|
|
for rec in records:
|
|
stream.write(rec.to_bytes())
|
|
|
|
|
|
class IecDevice:
|
|
"""Open and iterate the live kernel character device (or a replay file).
|
|
|
|
On the Pi this wraps ``/dev/iec0``. On a host pass a captured file to
|
|
``path`` to replay it through the exact same code path.
|
|
"""
|
|
|
|
def __init__(self, path: str = "/dev/iec0"):
|
|
self.path = path
|
|
self._fh: BinaryIO | None = None
|
|
|
|
def open(self) -> "IecDevice":
|
|
# buffering=0 so each read() maps to one kernel record on the real device.
|
|
self._fh = open(self.path, "rb", buffering=0)
|
|
return self
|
|
|
|
def __enter__(self) -> "IecDevice":
|
|
return self.open()
|
|
|
|
def __exit__(self, *exc) -> None:
|
|
self.close()
|
|
|
|
def close(self) -> None:
|
|
if self._fh is not None:
|
|
self._fh.close()
|
|
self._fh = None
|
|
|
|
def __iter__(self) -> Iterator[IecRecord]:
|
|
if self._fh is None:
|
|
raise RuntimeError("device not open; call open() or use as a context manager")
|
|
return iter_records(self._fh)
|