152 lines
4.7 KiB
Python
Executable File
152 lines
4.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Capture filament dryer CSV lines from serial into a file.
|
|
|
|
Designed for a Raspberry Pi attached to the dryer Nano over USB. The Arduino
|
|
still owns sensors and control; this script only records the csv,* stream.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import sys
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
FALLBACK_HEADER = (
|
|
"wall_time,ms,target_c,avg_c,min_c,max_c,spread_c,heatlim_pct,heater_pct,"
|
|
"fan_pct,cutoff,failsafe,ch2_t,ch2_h,ch3_t,ch3_h,ch4_t,ch4_h,ch5_t,ch5_h"
|
|
)
|
|
|
|
|
|
def detect_serial_port() -> str | None:
|
|
by_id = Path("/dev/serial/by-id")
|
|
if by_id.is_dir():
|
|
patterns = ("*Arduino*", "*arduino*", "*2341*", "*1a86*", "*CH340*", "*ch340*")
|
|
for pattern in patterns:
|
|
matches = sorted(by_id.glob(pattern))
|
|
if matches:
|
|
return str(matches[0])
|
|
for candidate in ("/dev/ttyACM0", "/dev/ttyACM1", "/dev/ttyUSB0", "/dev/ttyUSB1"):
|
|
if Path(candidate).exists():
|
|
return candidate
|
|
return None
|
|
|
|
|
|
def enable_dryer_logging(ser, retries: int = 3) -> None:
|
|
for attempt in range(retries):
|
|
ser.reset_input_buffer()
|
|
ser.write(b"log on\n")
|
|
ser.flush()
|
|
deadline = time.monotonic() + 2.0
|
|
while time.monotonic() < deadline:
|
|
raw = ser.readline()
|
|
if not raw:
|
|
continue
|
|
line = raw.decode("utf-8", errors="replace").strip()
|
|
if line == "OK csv logging on" or line.startswith("csv_hdr,"):
|
|
return
|
|
if line.startswith("csv,"):
|
|
return
|
|
if line:
|
|
print(line)
|
|
time.sleep(0.5 * (attempt + 1))
|
|
print("WARN: did not see 'OK csv logging on' — continuing anyway", file=sys.stderr)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument(
|
|
"-p",
|
|
"--port",
|
|
help="Serial port (default: auto-detect on Pi, else /dev/ttyUSB0)",
|
|
)
|
|
parser.add_argument("-b", "--baud", type=int, default=115200)
|
|
parser.add_argument(
|
|
"-o",
|
|
"--output",
|
|
type=Path,
|
|
help="Output CSV file (default: logs/dryer_YYYYMMDD_HHMMSS.csv)",
|
|
)
|
|
parser.add_argument(
|
|
"--log-dir",
|
|
type=Path,
|
|
default=Path("logs"),
|
|
help="Directory for default timestamped log files",
|
|
)
|
|
parser.add_argument(
|
|
"--auto-log-on",
|
|
action=argparse.BooleanOptionalAction,
|
|
default=True,
|
|
help="Send 'log on' to the dryer after connect (default: on)",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
port = args.port
|
|
if port is None:
|
|
port = detect_serial_port()
|
|
if port is None:
|
|
port = "/dev/ttyUSB0"
|
|
print(
|
|
f"WARN: no serial device found, using {port}",
|
|
file=sys.stderr,
|
|
)
|
|
|
|
try:
|
|
import serial
|
|
except ImportError:
|
|
print("Install pyserial: pip install pyserial", file=sys.stderr)
|
|
return 1
|
|
|
|
out = args.output
|
|
if out is None:
|
|
out = args.log_dir / f"dryer_{datetime.now():%Y%m%d_%H%M%S}.csv"
|
|
out.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
print(f"Logging {port} -> {out}", file=sys.stderr)
|
|
if args.auto_log_on:
|
|
print("Will send 'log on' after connect", file=sys.stderr)
|
|
else:
|
|
print("Send 'log on' to the dryer if CSV rows are not appearing", file=sys.stderr)
|
|
|
|
header_written = False
|
|
with serial.Serial(port, args.baud, timeout=1) as ser, out.open("w", encoding="utf-8") as fh:
|
|
time.sleep(2.0) # allow Nano reset after USB open
|
|
if args.auto_log_on:
|
|
enable_dryer_logging(ser)
|
|
|
|
while True:
|
|
try:
|
|
raw = ser.readline()
|
|
except KeyboardInterrupt:
|
|
print("\nStopped.", file=sys.stderr)
|
|
return 0
|
|
|
|
if not raw:
|
|
continue
|
|
|
|
line = raw.decode("utf-8", errors="replace").strip()
|
|
if not line.startswith("csv_hdr,") and not line.startswith("csv,"):
|
|
if line:
|
|
print(line)
|
|
continue
|
|
|
|
if line.startswith("csv_hdr,"):
|
|
device_header = line[len("csv_hdr,") :]
|
|
fh.write("wall_time," + device_header + "\n")
|
|
header_written = True
|
|
fh.flush()
|
|
continue
|
|
|
|
if not header_written:
|
|
fh.write(FALLBACK_HEADER + "\n")
|
|
header_written = True
|
|
|
|
wall_time = datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
fh.write(wall_time + "," + line[len("csv,") :] + "\n")
|
|
fh.flush()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|