85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
"""Render the annotated debug trace (PLAN.md §8).
|
|
|
|
Consumes :class:`~iecpoc.device.IecRecord` objects and emits one human-readable
|
|
line per record. This is the primary PoC deliverable on the userspace side, and
|
|
it is fully decoupled from any timing: it just formats whatever the kernel
|
|
pushed up. ``--raw`` switches to a bare hex dump of the byte stream.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Iterable, Iterator
|
|
|
|
from . import decode, device
|
|
|
|
# bracketed state tag, padded so the content columns line up
|
|
_TAG_WIDTH = 8
|
|
|
|
|
|
def _line(tag: str, content: str) -> str:
|
|
return f"{f'[{tag}]':<{_TAG_WIDTH}} {content}"
|
|
|
|
|
|
class TraceFormatter:
|
|
"""Format an IEC record stream into the annotated trace.
|
|
|
|
Parameters
|
|
----------
|
|
my_address:
|
|
Our primary device address; used to annotate which commands target us.
|
|
raw:
|
|
When True, emit only a bare hex dump of command + data bytes and drop
|
|
the symbolic annotations and state events.
|
|
"""
|
|
|
|
def __init__(self, my_address: int = 4, raw: bool = False):
|
|
self.my_address = my_address
|
|
self.raw = raw
|
|
|
|
def format(self, rec: device.IecRecord) -> str | None:
|
|
"""Format one record. Returns ``None`` for records to be skipped."""
|
|
if self.raw:
|
|
# raw mode: just the byte value of command/data records
|
|
if rec.is_command or rec.is_data:
|
|
return f"{rec.value:02X}"
|
|
return None
|
|
|
|
if rec.is_event:
|
|
return self._format_event(rec)
|
|
if rec.is_command:
|
|
return self._format_command(rec)
|
|
if rec.is_data:
|
|
return self._format_data(rec)
|
|
return _line("?", f"unknown kind {rec.kind}")
|
|
|
|
def format_stream(self, records: Iterable[device.IecRecord]) -> Iterator[str]:
|
|
"""Format a stream, dropping skipped records."""
|
|
for rec in records:
|
|
line = self.format(rec)
|
|
if line is not None:
|
|
yield line
|
|
|
|
# -- per-kind formatters -------------------------------------------------
|
|
def _format_event(self, rec: device.IecRecord) -> str:
|
|
text = decode.describe_event(rec)
|
|
if rec.value == device.EV_IDLE:
|
|
tag = "IDLE"
|
|
elif rec.value == device.EV_RESET:
|
|
tag = "RESET"
|
|
else:
|
|
tag = "ATN"
|
|
return _line(tag, text)
|
|
|
|
def _format_command(self, rec: device.IecRecord) -> str:
|
|
cmd = decode.decode_command(rec.value)
|
|
content = f"${rec.value:02X} {cmd.describe()}"
|
|
if cmd.targets(self.my_address):
|
|
content = f"{content:<22} (addressed: ME)"
|
|
return _line("CMD", content)
|
|
|
|
def _format_data(self, rec: device.IecRecord) -> str:
|
|
content = decode.describe_data(rec.value)
|
|
if rec.eoi:
|
|
content = f"{content} <EOI>"
|
|
return _line("DATA", content)
|