67 lines
2.0 KiB
Python
67 lines
2.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Capture Trikke sensor CSV output until interrupted with Ctrl-C."""
|
|
|
|
import argparse
|
|
import signal
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import serial
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--port", default="/dev/cu.usbmodem1134101")
|
|
parser.add_argument("--baud", type=int, default=115200)
|
|
parser.add_argument("--output", type=Path)
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
output = args.output or Path("captures") / datetime.now().strftime(
|
|
"motion_%Y%m%d_%H%M%S.csv"
|
|
)
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
stop_requested = False
|
|
|
|
def request_stop(_signum: int, _frame: object) -> None:
|
|
nonlocal stop_requested
|
|
stop_requested = True
|
|
|
|
signal.signal(signal.SIGINT, request_stop)
|
|
signal.signal(signal.SIGTERM, request_stop)
|
|
|
|
sample_count = 0
|
|
print(f"Recording {args.port} to {output}; press Ctrl-C to stop", flush=True)
|
|
|
|
try:
|
|
with serial.Serial(args.port, args.baud, timeout=0.25) as sensor, output.open(
|
|
"w", encoding="utf-8", newline=""
|
|
) as capture:
|
|
while not stop_requested:
|
|
raw_line = sensor.readline()
|
|
if not raw_line:
|
|
continue
|
|
|
|
line = raw_line.decode("utf-8", errors="replace").rstrip("\r\n")
|
|
capture.write(line + "\n")
|
|
|
|
if line and line[0].isdigit() and line.count(",") == 13:
|
|
sample_count += 1
|
|
if sample_count % 500 == 0:
|
|
capture.flush()
|
|
print(f" {sample_count} samples captured", flush=True)
|
|
except serial.SerialException as exc:
|
|
print(f"Serial error: {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
print(f"Stopped after {sample_count} samples; saved {output}", flush=True)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|