add: log also saved in binary. updated the reference material.
This commit is contained in:
@@ -1,14 +1,15 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
s3_bridge.py — S3 <-> Blastware serial bridge with frame-aware session logging
|
||||
Version: v0.4.0
|
||||
Version: v0.3.0
|
||||
|
||||
Key features:
|
||||
- Low CPU: avoids per-byte console printing
|
||||
- Forwards bytes immediately (true bridge)
|
||||
- Frame-aware logging: buffers per direction until ETX (0x03), then logs full frame on one line
|
||||
- Also logs plain ASCII bursts (e.g., "Operating System") cleanly
|
||||
- Session log file created on start, closed on Ctrl+C
|
||||
- Dual log output: hex text log (.log) AND raw binary log (.bin) written simultaneously
|
||||
- Session log files created on start, closed on Ctrl+C
|
||||
|
||||
Usage examples:
|
||||
python s3_bridge.py
|
||||
@@ -30,7 +31,7 @@ from typing import Optional
|
||||
import serial
|
||||
|
||||
|
||||
VERSION = "v0.4.0"
|
||||
VERSION = "v0.3.0"
|
||||
|
||||
|
||||
def now_ts() -> str:
|
||||
@@ -57,21 +58,30 @@ def looks_like_text(b: bytes) -> bool:
|
||||
|
||||
|
||||
class SessionLogger:
|
||||
def __init__(self, path: str):
|
||||
def __init__(self, path: str, bin_path: str):
|
||||
self.path = path
|
||||
self.bin_path = bin_path
|
||||
self._fh = open(path, "a", buffering=1, encoding="utf-8", errors="replace")
|
||||
self._bin_fh = open(bin_path, "ab", buffering=0)
|
||||
self._lock = threading.Lock()
|
||||
|
||||
def log_line(self, line: str) -> None:
|
||||
with self._lock:
|
||||
self._fh.write(line + "\n")
|
||||
|
||||
def log_raw(self, data: bytes) -> None:
|
||||
"""Write raw bytes directly to the binary log."""
|
||||
with self._lock:
|
||||
self._bin_fh.write(data)
|
||||
|
||||
def close(self) -> None:
|
||||
with self._lock:
|
||||
try:
|
||||
self._fh.flush()
|
||||
self._bin_fh.flush()
|
||||
finally:
|
||||
self._fh.close()
|
||||
self._bin_fh.close()
|
||||
|
||||
|
||||
class FrameAssembler:
|
||||
@@ -168,6 +178,7 @@ def forward_loop(
|
||||
for frame in frames:
|
||||
# Some devices send leading STX separately; we still log as-is.
|
||||
logger.log_line(f"[{now_ts()}] [{name}] {bytes_to_hex(frame)}")
|
||||
logger.log_raw(frame)
|
||||
|
||||
# If we have non-ETX data that looks like text, flush it as TEXT
|
||||
text = assembler.drain_as_text_if_any()
|
||||
@@ -177,6 +188,7 @@ def forward_loop(
|
||||
except Exception:
|
||||
s = repr(text)
|
||||
logger.log_line(f"[{now_ts()}] [{name}] [TEXT] {s}")
|
||||
logger.log_raw(text)
|
||||
|
||||
# minimal console heartbeat (cheap)
|
||||
if not quiet and status_every_s > 0:
|
||||
@@ -211,10 +223,12 @@ def main() -> int:
|
||||
print(f"Connected: {args.bw} <-> {args.s3}")
|
||||
|
||||
os.makedirs(args.logdir, exist_ok=True)
|
||||
log_name = _dt.datetime.now().strftime("s3_session_%Y%m%d_%H%M%S.log")
|
||||
log_path = os.path.join(args.logdir, log_name)
|
||||
logger = SessionLogger(log_path)
|
||||
print(f"[LOG] Writing session log to {log_path}")
|
||||
ts = _dt.datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
log_path = os.path.join(args.logdir, f"s3_session_{ts}.log")
|
||||
bin_path = os.path.join(args.logdir, f"s3_session_{ts}.bin")
|
||||
logger = SessionLogger(log_path, bin_path)
|
||||
print(f"[LOG] Writing hex log to {log_path}")
|
||||
print(f"[LOG] Writing binary log to {bin_path}")
|
||||
logger.log_line(f"[{now_ts()}] [INFO] s3_bridge {VERSION} start")
|
||||
logger.log_line(f"[{now_ts()}] [INFO] BW={args.bw} S3={args.s3} baud={args.baud}")
|
||||
|
||||
|
||||
Reference in New Issue
Block a user