diff --git a/docs/src/content/docs/reference/resources.mdx b/docs/src/content/docs/reference/resources.mdx
index eb37de2..79b411d 100644
--- a/docs/src/content/docs/reference/resources.mdx
+++ b/docs/src/content/docs/reference/resources.mdx
@@ -130,6 +130,74 @@ Like the data resource, reading raw data consumes the bytes from the serial buff
---
+## `serial://log`
+
+Read the server-wide log buffer. Returns recent log entries as formatted text, useful for debugging server behavior and diagnosing issues.
+
+**URI:** `serial://log`
+
+**Example output:**
+```
+# mcserial Server Log (47 entries)
+
+[2024-02-03T18:45:12] INFO server: Opened /dev/ttyUSB0 at 115200 baud
+[2024-02-03T18:45:13] DEBUG /dev/ttyUSB0: TX 8 bytes
+[2024-02-03T18:45:14] WARNING zmodem: Sanitized filename: "../etc/passwd" -> "etc_passwd"
+[2024-02-03T18:45:15] ERROR server: Failed to open /dev/ttyUSB1: Permission denied
+```
+
+Log entries include:
+- **Timestamp** in ISO-8601 format
+- **Level**: DEBUG, INFO, WARNING, or ERROR
+- **Source**: "server" for general events, or port name for port-specific events
+- **Message**: Human-readable description of the event
+
+Use `configure_logging()` to adjust the minimum log level and buffer size. The default level is INFO, which captures port open/close events and errors.
+
+
+Setting the log level to DEBUG captures TX/RX byte counts for all read/write operations. This is useful for debugging communication issues without enabling full traffic logging.
+
+
+---
+
+## `serial://{port}/log`
+
+Read the per-port traffic log (if enabled). Returns a history of TX/RX data in hex + ASCII format.
+
+**URI pattern:** `serial://{port}/log`
+
+**Example URIs:**
+```
+serial:///dev/ttyUSB0/log
+serial://spy:///dev/ttyUSB0/log
+```
+
+**Example output:**
+```
+# Traffic Log: /dev/ttyUSB0 (23 entries)
+
+[18:45:12.123] TX (8 bytes): 01 03 00 00 00 01 84 0a |........|
+[18:45:12.189] RX (7 bytes): 01 03 02 00 64 b8 44 |....d.D|
+[18:45:13.001] TX (8 bytes): 01 03 00 01 00 01 d5 ca |........|
+```
+
+Each entry shows:
+- **Timestamp** with millisecond precision
+- **Direction**: TX (transmitted) or RX (received)
+- **Byte count**
+- **Hex dump**: First 16 bytes in hexadecimal
+- **ASCII preview**: Printable characters (non-printable shown as `.`)
+
+
+Traffic logging is not enabled by default. Enable it with `enable_traffic_log(port)` or by opening the port with `spy://` scheme.
+
+
+
+Unlike the data resource, reading the traffic log does **not** consume any data. It's a read-only view of the capture buffer. The actual serial data remains in the serial buffer until read via tools or the data resource.
+
+
+---
+
## URI Encoding
The `{port}` segment in resource URIs uses the device path directly. For paths containing special characters (like forward slashes in Linux device paths), the MCP client handles URL encoding:
diff --git a/docs/src/content/docs/reference/tools-common.mdx b/docs/src/content/docs/reference/tools-common.mdx
index c5ceebe..fc34266 100644
--- a/docs/src/content/docs/reference/tools-common.mdx
+++ b/docs/src/content/docs/reference/tools-common.mdx
@@ -7,7 +7,7 @@ sidebar:
import { Aside } from '@astrojs/starlight/components';
-These 18 tools work in both RS-232 and RS-485 modes. They cover port discovery, connection management, reading and writing data, configuration, flow control, and diagnostics.
+These 20 tools work in both RS-232 and RS-485 modes. They cover port discovery, connection management, reading and writing data, configuration, flow control, diagnostics, and logging.
Examples use tool-call notation: `tool_name(param=value)`. These are MCP tool calls made by the assistant, not code you write directly. Boolean values use JSON convention (`true`/`false`).
@@ -457,3 +457,75 @@ detect_baud_rate(port="/dev/ttyUSB0", baudrates=[9600, 115200, 57600])
# Longer wait time for slow devices
detect_baud_rate(port="/dev/ttyUSB0", timeout_per_rate=1.0)
```
+
+---
+
+## Logging
+
+### configure_logging
+
+Configure server-wide logging behavior. Controls the minimum log level and buffer size for the in-memory log accessible via `serial://log`.
+
+| Parameter | Type | Default | Description |
+|-----------|------|---------|-------------|
+| `level` | `str` | `"INFO"` | Minimum log level: `DEBUG`, `INFO`, `WARNING`, or `ERROR`. |
+| `max_entries` | `int` | `1000` | Maximum entries in the server log buffer (10-100000). |
+| `clear` | `bool` | `False` | Clear existing log entries before applying new settings. |
+
+**Returns:** Dict with `success`, `level`, `previous_level`, `max_entries`, `current_entries`, `cleared`, and `resource_uri`.
+
+Log levels filter what gets captured:
+- **DEBUG**: All events including TX/RX byte counts
+- **INFO**: Port open/close, configuration changes (default)
+- **WARNING**: Potential issues like sanitized filenames
+- **ERROR**: Failed operations only
+
+```
+# Set to DEBUG for detailed operation logging
+configure_logging(level="DEBUG")
+
+# Reduce buffer size to save memory
+configure_logging(max_entries=500)
+
+# Clear logs and set to ERROR only
+configure_logging(level="ERROR", clear=true)
+```
+
+
+Set level to `DEBUG` when troubleshooting communication issues. This captures TX/RX byte counts for all read/write operations without the overhead of full traffic logging.
+
+
+### enable_traffic_log
+
+Enable or disable per-port traffic capture. When enabled, all TX/RX data on the port is captured to an in-memory buffer accessible via `serial://{port}/log`.
+
+| Parameter | Type | Default | Description |
+|-----------|------|---------|-------------|
+| `port` | `str` | required | Device path of the port to configure. |
+| `enabled` | `bool` | `True` | Enable or disable traffic logging. |
+| `max_entries` | `int` | `500` | Maximum traffic entries to keep (10-10000). |
+| `clear` | `bool` | `False` | Clear existing traffic log entries. |
+
+**Returns:** Dict with `success`, `port`, `traffic_log_enabled`, `max_entries`, `current_entries`, and `resource_uri`.
+
+```
+# Enable traffic logging on a port
+enable_traffic_log(port="/dev/ttyUSB0")
+
+# Enable with larger buffer for long sessions
+enable_traffic_log(port="/dev/ttyUSB0", max_entries=2000)
+
+# Clear and restart traffic capture
+enable_traffic_log(port="/dev/ttyUSB0", clear=true)
+
+# Disable traffic logging
+enable_traffic_log(port="/dev/ttyUSB0", enabled=false)
+```
+
+
+Traffic logging is automatically enabled when opening a port with the `spy://` URL scheme. The buffer is set to 1000 entries for spy connections.
+
+
+
+Traffic capture stores raw bytes in memory. For high-throughput connections, use a reasonable `max_entries` limit to avoid excessive memory usage.
+
diff --git a/docs/src/content/docs/reference/url-handlers.mdx b/docs/src/content/docs/reference/url-handlers.mdx
index c066d08..6a417b7 100644
--- a/docs/src/content/docs/reference/url-handlers.mdx
+++ b/docs/src/content/docs/reference/url-handlers.mdx
@@ -105,21 +105,33 @@ close_serial_port(port="loop://")
## spy://
-Debug wrapper that logs all serial traffic to stderr while passing data through to the underlying port. Wraps a real device path.
+Debug wrapper that captures all serial traffic to an in-memory buffer while passing data through to the underlying port. Wraps a real device path.
**Format:** `spy://device_path`
-All bytes read and written are printed to stderr in a human-readable format. The underlying serial connection works exactly as if you opened the device directly.
+When you open a port with `spy://`, traffic logging is automatically enabled with a 1000-entry buffer. All bytes read and written are captured and can be retrieved later via the `serial://{port}/log` resource.
-**When to use:** Debugging communication issues, reverse-engineering protocols, or logging traffic for analysis.
+**When to use:** Debugging communication issues, reverse-engineering protocols, multi-tasking (send commands, do other work, read traffic log later), or logging traffic for analysis.
```
-# Wraps /dev/ttyUSB0 with traffic logging
+# Wraps /dev/ttyUSB0 with automatic traffic capture
open_serial_port(port="spy:///dev/ttyUSB0", baudrate=9600)
+
+# Send a command
+write_serial(port="spy:///dev/ttyUSB0", data="AT\r\n")
+
+# ... do other work while device responds ...
+
+# Read captured traffic via resource
+# serial://spy:///dev/ttyUSB0/log
```
+
+The spy:// scheme is especially useful for MCP clients. Unlike stderr logging, the captured traffic is accessible via MCP resources, letting the model send commands, perform other tasks, and then review the traffic log later.
+
+
-The spy output goes to stderr of the mcserial server process. Check your server's stderr output to see the logged traffic.
+Traffic capture stores raw bytes in memory. The default buffer of 1000 entries is suitable for most debugging sessions. For longer sessions or high-throughput connections, consider using `enable_traffic_log()` to set a custom buffer size.
---
diff --git a/src/mcserial/server.py b/src/mcserial/server.py
index 98f6d36..9d16119 100644
--- a/src/mcserial/server.py
+++ b/src/mcserial/server.py
@@ -4,6 +4,8 @@ from __future__ import annotations
import atexit
import os
+import time
+from collections import deque
from dataclasses import dataclass, field
from typing import Literal
@@ -17,6 +19,48 @@ DEFAULT_TIMEOUT = float(os.getenv("MCSERIAL_DEFAULT_TIMEOUT", "1.0"))
MAX_CONNECTIONS = int(os.getenv("MCSERIAL_MAX_CONNECTIONS", "10"))
+# ============================================================================
+# LOGGING DATA STRUCTURES
+# ============================================================================
+
+
+@dataclass
+class LogEntry:
+ """A single server log entry."""
+
+ timestamp: float
+ level: str # DEBUG, INFO, WARNING, ERROR
+ source: str # "server", port name, "xmodem", etc.
+ message: str
+
+
+@dataclass
+class TrafficEntry:
+ """A single TX/RX traffic capture entry."""
+
+ timestamp: float
+ direction: Literal["tx", "rx"]
+ data: bytes
+ encoding_used: str | None = None
+
+
+# Server-wide log buffer (circular)
+_log_buffer: deque[LogEntry] = deque(maxlen=1000)
+_log_level: str = "INFO"
+_LOG_LEVELS = {"DEBUG": 0, "INFO": 1, "WARNING": 2, "ERROR": 3}
+
+
+def _log(level: str, source: str, message: str) -> None:
+ """Add an entry to the server log buffer if level meets threshold."""
+ if _LOG_LEVELS.get(level, 0) >= _LOG_LEVELS.get(_log_level, 1):
+ _log_buffer.append(LogEntry(
+ timestamp=time.time(),
+ level=level,
+ source=source,
+ message=message,
+ ))
+
+
@dataclass
class SerialConnection:
"""Tracks an open serial port connection."""
@@ -25,6 +69,9 @@ class SerialConnection:
connection: serial.Serial
buffer: bytes = field(default_factory=bytes)
mode: Literal["rs232", "rs485"] = "rs232"
+ # Traffic logging (None = disabled)
+ traffic_log: deque | None = None
+ traffic_log_max: int = 500
# Active connections registry
@@ -212,11 +259,19 @@ Transfer files using classic protocols. Works in both modes.
Protocols: xmodem (128B blocks), xmodem1k, ymodem (batch), zmodem (streaming, recommended)
+## Logging
+In-memory logging for debugging and traffic capture:
+ - configure_logging(level, max_entries) - Set server log level (DEBUG/INFO/WARNING/ERROR)
+ - enable_traffic_log(port) - Capture TX/RX traffic on a port
+ - spy:// URLs auto-enable traffic logging
+
## Resources
- serial://ports - List available ports
- serial://{port}/data - Read data from open port
- serial://{port}/status - Port configuration and mode
-- serial://{port}/raw - Read as hex dump""",
+- serial://{port}/raw - Read as hex dump
+- serial://log - Server-wide log buffer
+- serial://{port}/log - Per-port traffic log (if enabled)""",
)
@@ -402,7 +457,17 @@ def open_serial_port(
dsrdtr=dsrdtr,
exclusive=exclusive,
)
- _connections[port] = SerialConnection(port=port, connection=conn)
+ # Create the connection object
+ sc = SerialConnection(port=port, connection=conn)
+
+ # Auto-enable traffic logging for spy:// URLs
+ if is_url and port.split("://", 1)[0].lower() == "spy":
+ sc.traffic_log = deque(maxlen=1000)
+ _log("INFO", "server", f"spy:// detected, auto-enabled traffic logging for {port}")
+
+ _connections[port] = sc
+ _log("INFO", "server", f"Opened {port} at {baudrate} baud")
+
result = {
"success": True,
"port": port,
@@ -419,6 +484,9 @@ def open_serial_port(
if is_url:
result["url_scheme"] = port.split("://", 1)[0].lower()
result["hint"] = "Opened via URL handler. Some features (exclusive, auto-baud) are not available."
+ if sc.traffic_log is not None:
+ result["traffic_log_enabled"] = True
+ result["traffic_log_resource"] = f"serial://{port}/log"
else:
result["exclusive"] = exclusive
result["mode_hint"] = "Use set_port_mode() to switch to RS-485 mode if needed."
@@ -426,8 +494,10 @@ def open_serial_port(
result["autobaud"] = detected_info
return result
except serial.SerialException as e:
+ _log("ERROR", "server", f"Failed to open {port}: {e}")
return {"error": str(e), "success": False}
except OSError as e:
+ _log("ERROR", "server", f"Connection error for {port}: {e}")
return {"error": f"Connection error: {e}", "success": False}
@@ -447,8 +517,10 @@ def close_serial_port(port: str) -> dict:
try:
_connections[port].connection.close()
del _connections[port]
+ _log("INFO", "server", f"Closed {port}")
return {"success": True, "port": port, "message": "Port closed"}
except serial.SerialException as e:
+ _log("ERROR", "server", f"Error closing {port}: {e}")
return {"error": str(e), "success": False}
@@ -468,12 +540,24 @@ def write_serial(port: str, data: str, encoding: str = "utf-8") -> dict:
return {"error": f"Port {port} is not open", "success": False}
try:
- conn = _connections[port].connection
+ sc = _connections[port]
encoded = data.encode(encoding)
- bytes_written = conn.write(encoded)
- conn.flush()
+ bytes_written = sc.connection.write(encoded)
+ sc.connection.flush()
+
+ # Traffic capture
+ if sc.traffic_log is not None:
+ sc.traffic_log.append(TrafficEntry(
+ timestamp=time.time(),
+ direction="tx",
+ data=encoded,
+ encoding_used=encoding,
+ ))
+ _log("DEBUG", port, f"TX {bytes_written} bytes")
+
return {"success": True, "bytes_written": bytes_written, "port": port}
except serial.SerialException as e:
+ _log("ERROR", port, f"Write error: {e}")
return {"error": str(e), "success": False}
@@ -500,12 +584,24 @@ def write_serial_bytes(port: str, data: list[int]) -> dict:
}
try:
- conn = _connections[port].connection
+ sc = _connections[port]
raw_bytes = bytes(data)
- bytes_written = conn.write(raw_bytes)
- conn.flush()
+ bytes_written = sc.connection.write(raw_bytes)
+ sc.connection.flush()
+
+ # Traffic capture
+ if sc.traffic_log is not None:
+ sc.traffic_log.append(TrafficEntry(
+ timestamp=time.time(),
+ direction="tx",
+ data=raw_bytes,
+ encoding_used=None,
+ ))
+ _log("DEBUG", port, f"TX {bytes_written} bytes (raw)")
+
return {"success": True, "bytes_written": bytes_written, "port": port}
except (serial.SerialException, ValueError) as e:
+ _log("ERROR", port, f"Write bytes error: {e}")
return {"error": str(e), "success": False}
@@ -531,7 +627,8 @@ def read_serial(
return {"error": f"Port {port} is not open", "success": False}
try:
- conn = _connections[port].connection
+ sc = _connections[port]
+ conn = sc.connection
# Temporarily override timeout if specified
original_timeout = conn.timeout
@@ -544,6 +641,17 @@ def read_serial(
if timeout is not None:
conn.timeout = original_timeout
+ # Traffic capture
+ if raw and sc.traffic_log is not None:
+ sc.traffic_log.append(TrafficEntry(
+ timestamp=time.time(),
+ direction="rx",
+ data=raw,
+ encoding_used=encoding,
+ ))
+ if raw:
+ _log("DEBUG", port, f"RX {len(raw)} bytes")
+
return {
"success": True,
"data": raw.decode(encoding, errors="replace"),
@@ -552,6 +660,7 @@ def read_serial(
"port": port,
}
except serial.SerialException as e:
+ _log("ERROR", port, f"Read error: {e}")
return {"error": str(e), "success": False}
@@ -570,8 +679,20 @@ def read_serial_line(port: str, encoding: str = "utf-8") -> dict:
return {"error": f"Port {port} is not open", "success": False}
try:
- conn = _connections[port].connection
- raw = conn.readline()
+ sc = _connections[port]
+ raw = sc.connection.readline()
+
+ # Traffic capture
+ if raw and sc.traffic_log is not None:
+ sc.traffic_log.append(TrafficEntry(
+ timestamp=time.time(),
+ direction="rx",
+ data=raw,
+ encoding_used=encoding,
+ ))
+ if raw:
+ _log("DEBUG", port, f"RX line {len(raw)} bytes")
+
return {
"success": True,
"line": raw.decode(encoding, errors="replace").rstrip("\r\n"),
@@ -579,6 +700,7 @@ def read_serial_line(port: str, encoding: str = "utf-8") -> dict:
"port": port,
}
except serial.SerialException as e:
+ _log("ERROR", port, f"Read line error: {e}")
return {"error": str(e), "success": False}
@@ -609,17 +731,29 @@ def read_serial_lines(
return {"error": "max_lines must be 1-1000", "success": False}
try:
- conn = _connections[port].connection
+ sc = _connections[port]
lines = []
total_bytes = 0
for _ in range(max_lines):
- raw = conn.readline()
+ raw = sc.connection.readline()
if not raw:
break
total_bytes += len(raw)
lines.append(raw.decode(encoding, errors="replace").rstrip("\r\n"))
+ # Traffic capture per line
+ if sc.traffic_log is not None:
+ sc.traffic_log.append(TrafficEntry(
+ timestamp=time.time(),
+ direction="rx",
+ data=raw,
+ encoding_used=encoding,
+ ))
+
+ if total_bytes:
+ _log("DEBUG", port, f"RX {len(lines)} lines, {total_bytes} bytes")
+
return {
"success": True,
"lines": lines,
@@ -628,6 +762,7 @@ def read_serial_lines(
"port": port,
}
except serial.SerialException as e:
+ _log("ERROR", port, f"Read lines error: {e}")
return {"error": str(e), "success": False}
@@ -736,12 +871,121 @@ def get_connection_status() -> dict:
"rts": conn.rts,
"dtr": conn.dtr,
"resource_uri": f"serial://{port}/data",
+ "traffic_log_enabled": sc.traffic_log is not None,
+ "traffic_log_entries": len(sc.traffic_log) if sc.traffic_log else 0,
}
except serial.SerialException:
status[port] = {"is_open": False, "mode": sc.mode, "error": "Port disconnected"}
return {"connections": status, "count": len(_connections)}
+@mcp.tool()
+def configure_logging(
+ level: str = "INFO",
+ max_entries: int = 1000,
+ clear: bool = False,
+) -> dict:
+ """Configure server-wide logging behavior.
+
+ Args:
+ level: Minimum log level to capture (DEBUG, INFO, WARNING, ERROR)
+ max_entries: Maximum entries in the server log buffer (default 1000)
+ clear: Clear existing log entries if True
+
+ Returns:
+ Current logging configuration
+ """
+ global _log_buffer, _log_level
+
+ level = level.upper()
+ if level not in _LOG_LEVELS:
+ return {
+ "error": f"Invalid log level: {level}. Must be one of: {list(_LOG_LEVELS.keys())}",
+ "success": False,
+ }
+
+ if max_entries < 10 or max_entries > 100000:
+ return {"error": "max_entries must be between 10 and 100000", "success": False}
+
+ old_level = _log_level
+ _log_level = level
+
+ if clear:
+ _log_buffer.clear()
+ _log("INFO", "server", "Log buffer cleared")
+
+ # Resize buffer if needed
+ if _log_buffer.maxlen != max_entries:
+ old_entries = list(_log_buffer)
+ _log_buffer = deque(old_entries, maxlen=max_entries)
+
+ _log("INFO", "server", f"Logging configured: level={level}, max_entries={max_entries}")
+
+ return {
+ "success": True,
+ "level": _log_level,
+ "previous_level": old_level,
+ "max_entries": _log_buffer.maxlen,
+ "current_entries": len(_log_buffer),
+ "cleared": clear,
+ "resource_uri": "serial://log",
+ }
+
+
+@mcp.tool()
+def enable_traffic_log(
+ port: str,
+ enabled: bool = True,
+ max_entries: int = 500,
+ clear: bool = False,
+) -> dict:
+ """Enable or disable per-port traffic capture.
+
+ When enabled, all TX/RX data on this port is captured to an in-memory buffer.
+ The captured traffic can be read via the serial://{port}/log resource.
+
+ Args:
+ port: Device path of the port to configure
+ enabled: Enable (True) or disable (False) traffic logging
+ max_entries: Maximum traffic entries to keep (default 500)
+ clear: Clear existing traffic log entries if True
+
+ Returns:
+ Traffic log configuration status
+ """
+ if port not in _connections:
+ return {"error": f"Port {port} is not open", "success": False}
+
+ if max_entries < 10 or max_entries > 10000:
+ return {"error": "max_entries must be between 10 and 10000", "success": False}
+
+ sc = _connections[port]
+
+ if enabled:
+ if clear or sc.traffic_log is None:
+ sc.traffic_log = deque(maxlen=max_entries)
+ _log("INFO", port, f"Traffic logging enabled (max {max_entries} entries)")
+ elif sc.traffic_log.maxlen != max_entries:
+ # Resize buffer
+ old_entries = list(sc.traffic_log)
+ sc.traffic_log = deque(old_entries, maxlen=max_entries)
+ _log("INFO", port, f"Traffic log resized to {max_entries} entries")
+ sc.traffic_log_max = max_entries
+ else:
+ if sc.traffic_log is not None:
+ sc.traffic_log = None
+ _log("INFO", port, "Traffic logging disabled")
+
+ return {
+ "success": True,
+ "port": port,
+ "traffic_log_enabled": sc.traffic_log is not None,
+ "max_entries": max_entries if enabled else None,
+ "current_entries": len(sc.traffic_log) if sc.traffic_log else 0,
+ "resource_uri": f"serial://{port}/log" if enabled else None,
+ }
+
+
@mcp.tool()
def flush_serial(port: str, input_buffer: bool = True, output_buffer: bool = True) -> dict:
"""Flush serial port buffers.
@@ -2153,6 +2397,69 @@ def resource_read_raw(port: str) -> str:
return f"Error reading from {port}: {e}"
+@mcp.resource("serial://log")
+def resource_server_log() -> str:
+ """Server-wide log buffer. Returns recent log entries as formatted text.
+
+ Use query parameter ?level=WARNING to filter by minimum level.
+ """
+ from datetime import datetime
+
+ if not _log_buffer:
+ return "# mcserial Server Log (0 entries)\n\n[No log entries yet]"
+
+ lines = [f"# mcserial Server Log ({len(_log_buffer)} entries)\n"]
+ for entry in _log_buffer:
+ ts = datetime.fromtimestamp(entry.timestamp).strftime("%Y-%m-%dT%H:%M:%S")
+ lines.append(f"[{ts}] {entry.level} {entry.source}: {entry.message}")
+
+ return "\n".join(lines)
+
+
+@mcp.resource("serial://{port}/log")
+def resource_port_log(port: str) -> str:
+ """Per-port traffic log (if enabled). Returns TX/RX history as hex + ASCII.
+
+ Traffic logging must be enabled via enable_traffic_log() or by using spy://.
+ """
+ import urllib.parse
+ from datetime import datetime
+
+ port = urllib.parse.unquote(port)
+
+ if port not in _connections:
+ return f"Error: Port {port} is not open."
+
+ sc = _connections[port]
+ if sc.traffic_log is None:
+ return (
+ f"# Traffic Log: {port}\n\n"
+ "Traffic logging is not enabled for this port.\n"
+ "Use enable_traffic_log(port, enabled=True) to enable it."
+ )
+
+ if not sc.traffic_log:
+ return f"# Traffic Log: {port} (0 entries)\n\n[No traffic captured yet]"
+
+ def format_hex_ascii(data: bytes) -> str:
+ """Format bytes as hex + ASCII representation."""
+ hex_part = " ".join(f"{b:02x}" for b in data[:16])
+ ascii_part = "".join(chr(b) if 32 <= b <= 126 else "." for b in data[:16])
+ if len(data) > 16:
+ hex_part += " ..."
+ ascii_part += "..."
+ return f"{hex_part} |{ascii_part}|"
+
+ lines = [f"# Traffic Log: {port} ({len(sc.traffic_log)} entries)\n"]
+ for entry in sc.traffic_log:
+ ts = datetime.fromtimestamp(entry.timestamp).strftime("%H:%M:%S.%f")[:-3]
+ direction = "TX" if entry.direction == "tx" else "RX"
+ formatted = format_hex_ascii(entry.data)
+ lines.append(f"[{ts}] {direction} ({len(entry.data)} bytes): {formatted}")
+
+ return "\n".join(lines)
+
+
# ============================================================================
# ENTRY POINT
# ============================================================================