fd16fa8f90
Live sniff/sim stream trace frames from a background thread. bpython replaces sys.stdout with a capture object whose write() does a greenlet switch back to the main thread — which raises off-thread; the text buffers but only repaints on the next keypress, so streamed frames 'pile up until you touch a key'. New trace/output.py emit() buffers via the repl's on_write and fires bpython's threadsafe repaint trigger; a plain write everywhere else, with a safe fallback on any internals mismatch. Route TraceFormatter.print(), SimSession's sim-ended status, and SniffSession's state message through emit(). Flip the trace-line newline convention from leading to trailing so each emitted line self-terminates (format_sniff_line + test updated to match). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
248 lines
8.7 KiB
Python
248 lines
8.7 KiB
Python
"""Trace formatting — colored, decoded, column-wrapped trace output."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import signal
|
|
import sys
|
|
|
|
from .decode_iso15 import decode_15693
|
|
from .output import emit
|
|
|
|
# ---- ANSI colors ----
|
|
_C_CYAN = "\033[36m"
|
|
_C_YELLOW = "\033[33m"
|
|
_C_MAGENTA = "\033[35m"
|
|
_C_GREEN = "\033[32m"
|
|
_C_DIM = "\033[2m"
|
|
_C_CRC = "\033[1;37m" # bold white — CRC bytes
|
|
_C_RESET = "\033[0m"
|
|
|
|
_MODE_TAGS = {
|
|
"sim": ("[Sim]", _C_MAGENTA),
|
|
"reader": ("[Rdr]", _C_CYAN),
|
|
"sniff": ("[Snf]", ""),
|
|
}
|
|
|
|
|
|
class TraceFormatter:
|
|
"""Colored, decoded, column-wrapped trace output.
|
|
|
|
Args:
|
|
mode: "sim", "reader", or "sniff"
|
|
decoder: Callable[[int, bytes], str | None] for annotation, or None
|
|
width: override terminal width (None = auto-detect)
|
|
is_tty: override TTY detection (None = auto-detect)
|
|
crc_len: number of CRC bytes to split from payload end (0 = none)
|
|
"""
|
|
|
|
def __init__(self, mode: str = "sim", decoder=None,
|
|
width: int | None = None, is_tty: bool | None = None,
|
|
crc_len: int = 0):
|
|
self._mode = mode
|
|
self._decoder = decoder
|
|
self._crc_len = crc_len
|
|
self._width = width or self._detect_width()
|
|
self._is_tty = is_tty if is_tty is not None else sys.stdout.isatty()
|
|
self._line_count = 0
|
|
|
|
# Try to register SIGWINCH for dynamic resize
|
|
if width is None:
|
|
try:
|
|
signal.signal(signal.SIGWINCH, self._on_resize)
|
|
except (OSError, ValueError):
|
|
pass # not main thread or not Unix
|
|
|
|
def _detect_width(self) -> int:
|
|
try:
|
|
return os.get_terminal_size().columns
|
|
except (OSError, ValueError):
|
|
return 80
|
|
|
|
def _on_resize(self, signum, frame):
|
|
self._width = self._detect_width()
|
|
|
|
def _color(self, code: str, text: str) -> str:
|
|
if not self._is_tty or not code:
|
|
return text
|
|
return f"{code}{text}{_C_RESET}"
|
|
|
|
def decode(self, direction: int, payload: bytes, crc_fail: bool = False) -> str | None:
|
|
"""Return the plain-text decoded annotation for a frame — no ANSI, no hex.
|
|
|
|
The same annotation ``format()`` renders (e.g. "INVENTORY", "READ BLOCK 4"),
|
|
exposed as structured data for scripting/callbacks.
|
|
"""
|
|
if self._crc_len > 0 and len(payload) > self._crc_len:
|
|
data = payload[:-self._crc_len]
|
|
else:
|
|
data = payload
|
|
annotation = self._decoder(direction, data) if self._decoder is not None else None
|
|
if crc_fail:
|
|
annotation = f"{annotation} BAD CRC" if annotation else "BAD CRC"
|
|
return annotation
|
|
|
|
def format(self, direction: int, payload: bytes, crc_fail: bool = False) -> str:
|
|
"""Format a trace line with colors, decoding, and wrapping."""
|
|
self._line_count += 1
|
|
if self._line_count % 10 == 0:
|
|
try:
|
|
self._width = self._detect_width()
|
|
except Exception:
|
|
pass
|
|
|
|
# Mode tag
|
|
mode_tag, mode_color = _MODE_TAGS.get(self._mode, ("[???]", ""))
|
|
|
|
# Direction
|
|
if direction == 0:
|
|
arrow = "Reader \u2192 Tag:"
|
|
dir_color = _C_CYAN
|
|
is_ours = (self._mode == "reader")
|
|
else:
|
|
arrow = "Tag \u2192 Reader:"
|
|
dir_color = _C_YELLOW
|
|
is_ours = (self._mode == "sim")
|
|
|
|
# Build prefix
|
|
prefix_plain = f"{mode_tag} {arrow} "
|
|
prefix_len = len(prefix_plain)
|
|
|
|
if is_ours:
|
|
dir_color = _C_DIM + dir_color
|
|
prefix_colored = self._color(mode_color, mode_tag) + " " + self._color(dir_color, arrow) + " "
|
|
|
|
# Split payload into data and CRC
|
|
if self._crc_len > 0 and len(payload) > self._crc_len:
|
|
data = payload[:-self._crc_len]
|
|
crc = payload[-self._crc_len:]
|
|
else:
|
|
data = payload
|
|
crc = b""
|
|
|
|
data_hex = " ".join(f"{b:02X}" for b in data)
|
|
crc_hex = " ".join(f"{b:02X}" for b in crc)
|
|
full_hex = f"{data_hex} {crc_hex}" if crc_hex else data_hex
|
|
|
|
# Decode annotation
|
|
annotation = None
|
|
if self._decoder is not None:
|
|
annotation = self._decoder(direction, data)
|
|
if crc_fail:
|
|
crc_note = "BAD CRC"
|
|
annotation = f"{annotation} {crc_note}" if annotation else crc_note
|
|
ann_color = (_C_DIM + dir_color) if is_ours else dir_color
|
|
if crc_fail:
|
|
ann_color = "\033[31m"
|
|
|
|
# Layout
|
|
avail = self._width - prefix_len
|
|
if annotation:
|
|
one_line = f"{full_hex} {annotation}"
|
|
else:
|
|
one_line = full_hex
|
|
|
|
if len(one_line) <= avail:
|
|
hex_colored = self._color(_C_DIM, data_hex)
|
|
if crc_hex:
|
|
hex_colored += " " + self._color(_C_CRC, crc_hex)
|
|
if annotation:
|
|
ann_colored = self._color(ann_color, annotation)
|
|
gap = max(avail - len(full_hex) - len(annotation), 2)
|
|
line = f"{prefix_colored}{hex_colored}{' ' * gap}{ann_colored}"
|
|
else:
|
|
line = f"{prefix_colored}{hex_colored}"
|
|
return f"{line}\n"
|
|
|
|
# Multi-line
|
|
pad = " " * prefix_len
|
|
hex_lines = self._wrap_hex(full_hex, avail)
|
|
parts = []
|
|
for i, hl in enumerate(hex_lines):
|
|
colored_hl = self._color_hex_line(hl, data_hex, crc_hex)
|
|
if i == 0:
|
|
parts.append(f"{prefix_colored}{colored_hl}")
|
|
else:
|
|
parts.append(f"{pad}{colored_hl}")
|
|
if annotation:
|
|
ann_lines = self._wrap_annotation(annotation, avail)
|
|
for al in ann_lines:
|
|
ann_pad = max(avail - len(al), 0)
|
|
parts.append(f"{pad}{' ' * ann_pad}{self._color(ann_color, al)}")
|
|
|
|
return "\n".join(parts) + "\n"
|
|
|
|
def _color_hex_line(self, line: str, data_hex: str, crc_hex: str) -> str:
|
|
if not crc_hex:
|
|
return self._color(_C_DIM, line)
|
|
crc_tokens = crc_hex.split(" ")
|
|
line_tokens = line.split(" ")
|
|
crc_start = None
|
|
for i in range(len(line_tokens)):
|
|
if line_tokens[i:] == crc_tokens[-len(line_tokens) + i:]:
|
|
crc_start = i
|
|
break
|
|
remaining = line_tokens[i:]
|
|
if crc_hex.startswith(" ".join(remaining)) or " ".join(remaining) == crc_hex:
|
|
crc_start = i
|
|
break
|
|
if crc_start is not None and crc_start < len(line_tokens):
|
|
data_part = " ".join(line_tokens[:crc_start])
|
|
crc_part = " ".join(line_tokens[crc_start:])
|
|
result = ""
|
|
if data_part:
|
|
result += self._color(_C_DIM, data_part) + " "
|
|
result += self._color(_C_CRC, crc_part)
|
|
return result
|
|
return self._color(_C_DIM, line)
|
|
|
|
def _wrap_annotation(self, annotation: str, avail: int) -> list[str]:
|
|
if len(annotation) <= avail:
|
|
return [annotation]
|
|
if " | " in annotation:
|
|
parts = annotation.split(" | ")
|
|
lines = []
|
|
current = parts[0]
|
|
for part in parts[1:]:
|
|
candidate = f"{current} | {part}"
|
|
if len(candidate) <= avail:
|
|
current = candidate
|
|
else:
|
|
lines.append(current)
|
|
current = part
|
|
lines.append(current)
|
|
return lines
|
|
return [annotation[:avail - 3] + "..."]
|
|
|
|
def _wrap_hex(self, hex_str: str, avail: int) -> list[str]:
|
|
tokens = hex_str.split(" ")
|
|
lines: list[str] = []
|
|
current = ""
|
|
for tok in tokens:
|
|
candidate = f"{current} {tok}" if current else tok
|
|
if len(candidate) <= avail:
|
|
current = candidate
|
|
else:
|
|
if current:
|
|
lines.append(current)
|
|
current = tok
|
|
if current:
|
|
lines.append(current)
|
|
return lines or [""]
|
|
|
|
def print(self, direction: int, payload: bytes, crc_fail: bool = False) -> None:
|
|
"""Format and print a trace line, live even under bpython."""
|
|
emit(self.format(direction, payload, crc_fail=crc_fail))
|
|
|
|
|
|
def format_sniff_line(entry: dict, width: int = 120, is_tty: bool = True) -> str:
|
|
"""Format a single tracelog entry as a colored sniff line.
|
|
|
|
Convenience wrapper around TraceFormatter for sniff mode with 15693 decoder.
|
|
"""
|
|
fmt = TraceFormatter(mode="sniff", decoder=decode_15693,
|
|
width=width, is_tty=is_tty, crc_len=2)
|
|
direction = 1 if entry["is_response"] else 0
|
|
result = fmt.format(direction, entry["data"])
|
|
# Strip the trailing newline that TraceFormatter adds
|
|
return result.rstrip("\n")
|