DASP trace: color send/receive, decode header fields, hex+ASCII dump

Sent lines are yellow and received lines green when stdout is a console;
pipes and files stay plain. Each message lists the header fields it
carries by name and dumps header and payload separately, 16 bytes a row.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-18 22:38:52 +03:00
co-authored by Claude Opus 5.5
parent 0dddea53f8
commit 6dd01b13b9
3 changed files with 50 additions and 11 deletions
+1 -1
View File
@@ -8,7 +8,7 @@ so later rebinding would only ever be seen by the module that did it.
import configparser
import os
VERSION = "0.0.0.181"
VERSION = "0.0.0.182"
PROPERTIES_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "editor.properties")
DEFAULT_GRID_SIZE = 10
DEFAULT_ADD_TIMESTAMP = True
+46 -8
View File
@@ -92,17 +92,55 @@ def _trace(direction, msg, datagram=None):
import datetime
ts = datetime.datetime.now().strftime("%H:%M:%S.%f")[:-3]
name = _MSG_TYPE_NAMES.get(msg.msg_type, f"?{msg.msg_type}")
seq = f" seq={msg.seq_num}" if msg.seq_num >= 0 else ""
ack = f" ack={msg.ack}" if msg.ack >= 0 else ""
plen = len(msg.payload) if msg.payload else 0
hexdump = ""
arrow = {"S": ">> SEND", "R": "<< RECV"}.get(direction, direction)
payload = msg.payload or b""
size = f" ({len(datagram)} bytes)" if datagram else ""
lines = [f"[DASP] {ts} {arrow} {name:<12} {_trace_fields(msg)}{size}"]
if datagram:
hexdump = " [" + datagram.hex(" ") + "]"
line = f"[DASP] {ts} {direction} {name}{seq}{ack} payload={plen}{hexdump}"
hdr_len = len(datagram) - len(payload)
lines.extend(_hexdump(datagram[:hdr_len], f"header {hdr_len}"))
if payload:
lines.extend(_hexdump(payload, f"payload {len(payload)}"))
color = _TRACE_COLORS.get(direction)
if color and _console_colors():
line = f"{color}{line}\033[0m"
_trace_out(line)
lines = [f"{color}{ln}\033[0m" for ln in lines]
_trace_out("\n".join(lines))
def _trace_fields(msg):
"""The message's header fields as name=value pairs, only those it carries."""
f = []
if msg.session_id >= 0: f.append(f"session=0x{msg.session_id:04x}")
if msg.seq_num >= 0: f.append(f"seq={msg.seq_num}")
if msg.ack >= 0: f.append(f"ack={msg.ack}")
if msg.ack_more: f.append(f"ackMore={msg.ack_more.hex()}")
if msg.version >= 0: f.append(f"version=0x{msg.version:04x}")
if msg.remote_id >= 0: f.append(f"remoteId=0x{msg.remote_id:04x}")
if msg.username: f.append(f"user={msg.username}")
if msg.digest_algorithm: f.append(f"digestAlg={msg.digest_algorithm}")
if msg.nonce: f.append(f"nonce=<{len(msg.nonce)} bytes>")
if msg.digest: f.append(f"digest=<{len(msg.digest)} bytes>")
if msg.ideal_max >= 0: f.append(f"idealMax={msg.ideal_max}")
if msg.abs_max >= 0: f.append(f"absMax={msg.abs_max}")
if msg.receive_max >= 0: f.append(f"receiveMax={msg.receive_max}")
if msg.receive_timeout >= 0: f.append(f"receiveTimeout={msg.receive_timeout // 1000}s")
if msg.platform_id: f.append(f"platform={msg.platform_id}")
if msg.error_code >= 0:
hint = _ERROR_HINTS.get(msg.error_code, "")
f.append(f"error=0x{msg.error_code:02x}{hint}")
return " ".join(f)
def _hexdump(data, label):
"""Hex and ASCII, 16 bytes a row, the first row headed by label."""
out = []
for off in range(0, len(data), 16):
chunk = data[off:off + 16]
hexs = chunk.hex(" ").ljust(47)
text = "".join(chr(b) if 32 <= b < 127 else "." for b in chunk)
head = label if off == 0 else ""
out.append(f" {head:<11} {off:04x} {hexs} |{text}|")
return out
# Sent lines yellow, received lines green, so the two directions read apart at a glance.
+3 -2
View File
@@ -5,10 +5,11 @@ add_timestamp = True
box_width = 250
left_pane_width = 230
trace_dasp = True
lazy_loading = False
lazy_loading = True
show_slot_numbers = False
[Device]
host = localhost
host = 192.168.0.26
port = 1876
user = admin