"""
serial_capture.py – Capture Teensy ADC DMA USB streamer data into a numpy array.
The Teensy firmware streams raw binary uint16_t samples in 512-byte blocks.
This script:
1. Opens the port and sends Enter to start streaming.
2. Discards the text preamble lines sent before binary data begins.
3. Reads binary blocks into a growing numpy uint16 array.
4. Stops on Ctrl-C and reports basic stats.
Usage:
python serial_capture.py [PORT] [--max-samples N] [--out FILE.npy]
Requires: pip install pyserial numpy
"""
import argparse
import sys
import time
import matplotlib.pyplot as plt
import numpy as np
import serial
BLOCK_BYTES = 512 # must match BUF_BLOCK_SIZE in firmware
READ_CHUNK = 16 * BLOCK_BYTES # larger host read to reduce Python/USB overhead
SAMPLES_BLOCK = BLOCK_BYTES // 2 # uint16_t per block = 256
PREAMBLE_LINES = 2 # "Streaming…" + "Press 'q'…"
def flush_preamble(ser: "serial.Serial") -> None:
"""Read and discard the text lines the firmware sends before binary data."""
for _ in range(PREAMBLE_LINES):
line = ser.readline()
print("[preamble]", line.decode(errors="replace").rstrip())
def capture(port: str, max_samples: "int | None", out_path: "str | None") -> np.ndarray:
chunks: list[np.ndarray] = [] # dtype: uint16
total = 0
with serial.Serial(port, baudrate=115200, timeout=10.0) as ser:
# Disable DTR/RTS to avoid spurious resets.
ser.dtr = False
ser.rts = False
# Drain anything already in the OS buffer.
ser.reset_input_buffer()
# Wait for the Teensy's "Enter to" prompt before triggering, so we
# don't send Enter before the Teensy is ready (e.g. after a reset).
ser.timeout = 5.0
while True:
line = ser.readline().decode(errors="replace").strip()
if not line:
break # timed out
print("[boot]", line)
if "Enter to" in line:
break
# Send Enter to trigger the Teensy streaming start.
# Do not delay here: at >1 Msps, 100 ms can overflow the MCU ring buffer
# before the host starts draining binary data.
ser.write(b"\r\n")
ser.flush()
ser.timeout = 10.0
flush_preamble(ser)
print(f"Capturing from {port} (Ctrl-C to stop) …")
partial = bytearray() # buffer for partial blocks
timeout_count = 0
try:
while True:
raw = ser.read(READ_CHUNK)
if len(raw) == 0:
timeout_count += 1
if timeout_count >= 3:
print(f"\nNo data for {timeout_count} read attempts – stopping.")
break
time.sleep(0.01)
continue
timeout_count = 0 # reset on any data
partial.extend(raw)
# Extract complete blocks from partial buffer.
while len(partial) >= BLOCK_BYTES:
block = np.frombuffer(bytes(partial[:BLOCK_BYTES]), dtype=np.uint16)
chunks.append(block)
total += len(block)
del partial[:BLOCK_BYTES]
# Progress tick every 256 blocks.
if len(chunks) % 256 == 0:
print(f" {total:,} samples collected …", end="\r", flush=True)
if max_samples and total >= max_samples:
print(f"\nReached {max_samples:,} sample limit.")
raise KeyboardInterrupt # exit cleanly
except KeyboardInterrupt:
print(f"\nStopped by user.")
finally:
# Send 'q' to tell the Teensy to stop streaming.
ser.write(b"q")
ser.flush()
# Discard any trailing binary data until the stats sentinel.
ser.timeout = 2.0
while True:
line = ser.readline()
if not line:
break
decoded = line.decode(errors="replace").rstrip()
if decoded == "---STATS---":
break # found sentinel, stats follow
while True:
line = ser.readline()
if not line:
break
print("[teensy]", line.decode(errors="replace").rstrip())
if not chunks:
print("No data received.")
return np.empty(0, dtype=np.uint16)
samples = np.concatenate(chunks)
print(f"Captured {len(samples):,} samples "
f"({len(samples) * 2 / 1024:.1f} kB)")
print(f" min={samples.min()} max={samples.max()} "
f"mean={samples.mean():.1f} std={samples.std():.1f}")
if out_path:
np.save(out_path, samples)
print(f"Saved → {out_path}")
return samples
def main() -> None:
parser = argparse.ArgumentParser(description="Capture Teensy ADC stream to numpy array.")
parser.add_argument("port", nargs="?", default="COM3",
help="Serial port (default: COM3)")
parser.add_argument("--max-samples", type=int, default=None,
metavar="N", help="Stop after N samples")
parser.add_argument("--out", type=str, default=None,
metavar="FILE.npy", help="Save array to .npy file")
args = parser.parse_args()
samples = capture(args.port, args.max_samples, args.out)
if len(samples):
print(f"\nArray shape: {samples.shape}, dtype: {samples.dtype}")
# Values are raw 12-bit ADC counts in range [0, 4095].
fig, ax = plt.subplots(figsize=(12, 4))
ax.plot(samples, linewidth=0.5)
ax.set_xlabel("Sample index")
ax.set_ylabel("ADC count (uint16)")
ax.set_title(f"ADC capture – {len(samples):,} samples")
ax.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
if __name__ == "__main__":
main()