feat(protocol): enhance raw capture functionality and documentation updates

- Update `s3_bridge.py` to default raw capture file paths to "auto" for timestamped naming.
- Modify `gui_bridge.py` to pre-check raw capture options and streamline path handling.
- Extend `ach_server.py` to save both incoming and outgoing raw bytes for analysis.
- Revise `CHANGELOG.md` and `instantel_protocol_reference.md` to reflect changes in recording mode handling and compliance data encoding.
This commit is contained in:
2026-04-21 16:07:24 -04:00
parent b3dcfe7239
commit 4331215e23
6 changed files with 207 additions and 63 deletions
+37 -15
View File
@@ -35,6 +35,7 @@ Output per session
device_info.json — serial number, firmware version, calibration date, etc.
events.json — all events: timestamp, PPV per channel, peaks, metadata
raw_rx_<ts>.bin — raw bytes from the device (S3 side) for Analyzer
raw_tx_<ts>.bin — raw bytes we sent to the device (BW side) for Analyzer
session_<ts>.log — detailed protocol log
What to look for
@@ -172,16 +173,24 @@ class AchSession:
transport = SocketTransport(self.sock, peer=self.peer)
# Collect raw bytes in memory until startup succeeds, then flush to disk.
raw_buf: list[bytes] = []
_orig_read = transport.read
raw_rx_buf: list[bytes] = [] # device → us (S3 side)
raw_tx_buf: list[bytes] = [] # us → device (BW side)
_orig_read = transport.read
_orig_write = transport.write
def tapped_read(n: int) -> bytes:
data = _orig_read(n)
if data:
raw_buf.append(data)
raw_rx_buf.append(data)
return data
transport.read = tapped_read # type: ignore[method-assign]
def tapped_write(data: bytes) -> None:
_orig_write(data)
if data:
raw_tx_buf.append(data)
transport.read = tapped_read # type: ignore[method-assign]
transport.write = tapped_write # type: ignore[method-assign]
serial: Optional[str] = None
@@ -201,23 +210,35 @@ class AchSession:
# Startup succeeded — this is a real unit. Create session dir now.
session_dir = self.output_dir / f"ach_inbound_{ts}"
session_dir.mkdir(parents=True, exist_ok=True)
log_path = session_dir / f"session_{ts}.log"
raw_path = session_dir / f"raw_rx_{ts}.bin"
log_path = session_dir / f"session_{ts}.log"
raw_rx_path = session_dir / f"raw_rx_{ts}.bin" # device → us (S3 side)
raw_tx_path = session_dir / f"raw_tx_{ts}.bin" # us → device (BW side)
# Flush buffered raw bytes to file and switch to direct file writes.
raw_fh = open(raw_path, "wb")
for chunk in raw_buf:
raw_fh.write(chunk)
raw_buf.clear()
# Flush buffered bytes to files and switch to direct file writes.
raw_rx_fh = open(raw_rx_path, "wb")
raw_tx_fh = open(raw_tx_path, "wb")
for chunk in raw_rx_buf:
raw_rx_fh.write(chunk)
for chunk in raw_tx_buf:
raw_tx_fh.write(chunk)
raw_rx_buf.clear()
raw_tx_buf.clear()
def tapped_read_file(n: int) -> bytes:
data = _orig_read(n)
if data:
raw_fh.write(data)
raw_fh.flush()
raw_rx_fh.write(data)
raw_rx_fh.flush()
return data
transport.read = tapped_read_file # type: ignore[method-assign]
def tapped_write_file(data: bytes) -> None:
_orig_write(data)
if data:
raw_tx_fh.write(data)
raw_tx_fh.flush()
transport.read = tapped_read_file # type: ignore[method-assign]
transport.write = tapped_write_file # type: ignore[method-assign]
# Wire up file handler now that the session dir exists.
fh = logging.FileHandler(log_path, encoding="utf-8")
@@ -530,7 +551,8 @@ class AchSession:
log.warning(" [WARN] Failed to restart monitoring: %s", exc)
finally:
raw_fh.close()
raw_rx_fh.close()
raw_tx_fh.close()
client.close() # closes transport / socket cleanly
root_logger.removeHandler(fh)
fh.close()