#!/usr/bin/env python3 """Filament dryer host tools — TUI dashboard and CSV logging. capture_csv.py TUI when run in a terminal (default) capture_csv.py tui same capture_csv.py log headless CSV capture (Pi systemd logger) """ 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 resolve_port(port: str | None) -> str: if port is not None: return port detected = detect_serial_port() if detected is not None: return detected fallback = "/dev/ttyUSB0" print(f"WARN: no serial device found, using {fallback}", file=sys.stderr) return fallback def import_serial(): try: import serial except ImportError: print("Install pyserial: pip install pyserial", file=sys.stderr) raise SystemExit(1) from None return serial def open_serial(port: str, baud: int): serial = import_serial() ser = serial.Serial() ser.port = port ser.baudrate = baud ser.timeout = 0.1 # DTR low: avoid resetting the Nano (CH340) on every reconnect. ser.dtr = False ser.rts = False ser.open() time.sleep(0.3) return ser def decode_line(raw: bytes) -> str: return raw.decode("utf-8", errors="replace").strip() 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 = decode_line(raw) 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 write_csv_row(fh, line: str, header_written: list[bool]) -> None: if line.startswith("csv_hdr,"): device_header = line[len("csv_hdr,") :] fh.write("wall_time," + device_header + "\n") header_written[0] = True fh.flush() return if not line.startswith("csv,"): return if not header_written[0]: fh.write(FALLBACK_HEADER + "\n") header_written[0] = True wall_time = datetime.now(timezone.utc).isoformat(timespec="seconds") fh.write(wall_time + "," + line[len("csv,") :] + "\n") fh.flush() def cmd_log(args: argparse.Namespace) -> int: port = resolve_port(args.port) 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) header_written = False with open_serial(port, args.baud) as ser, out.open("w", encoding="utf-8") as fh: 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 = decode_line(raw) 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() def cmd_tui(args: argparse.Namespace) -> int: from dryer_tui import run_tui return run_tui(args.port, args.baud, args.log_dir, args.auto_log_on) def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Filament dryer host tools (CSV logging and TUI).", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=( "Examples:\n" " %(prog)s tui\n" " %(prog)s log --log-dir logs\n" " %(prog)s --log-dir logs # legacy headless logger\n" ), ) parser.add_argument("-p", "--port", help="Serial port (default: auto-detect)") parser.add_argument("-b", "--baud", type=int, default=115200) parser.add_argument("-o", "--output", type=Path, help="Output CSV file (log mode)") parser.add_argument("--log-dir", type=Path, default=Path("logs"), help="CSV log directory") parser.add_argument( "--auto-log-on", action=argparse.BooleanOptionalAction, default=True, help="Send 'log on' after connect in log mode (default: on)", ) subparsers = parser.add_subparsers(dest="action") log_p = subparsers.add_parser("log", help="Headless CSV capture", add_help=False) log_p.add_argument("-p", "--port") log_p.add_argument("-b", "--baud", type=int, default=115200) log_p.add_argument("-o", "--output", type=Path) log_p.add_argument("--log-dir", type=Path, default=Path("logs")) log_p.add_argument("--auto-log-on", action=argparse.BooleanOptionalAction, default=True) tui_p = subparsers.add_parser("tui", help="Interactive curses dashboard") tui_p.add_argument("-p", "--port") tui_p.add_argument("-b", "--baud", type=int, default=115200) tui_p.add_argument("--log-dir", type=Path, default=Path("logs")) tui_p.add_argument( "--auto-log-on", action=argparse.BooleanOptionalAction, default=True, help="Start CSV logging when TUI opens (default: on)", ) return parser def main(argv: list[str] | None = None) -> int: argv = sys.argv[1:] if argv is None else list(argv) parser = build_parser() args = parser.parse_args(argv) if args.action == "tui": return cmd_tui(args) if args.action == "log": return cmd_log(args) # No subcommand: TUI in an interactive terminal, headless log otherwise (systemd). if sys.stdin.isatty() and sys.stdout.isatty(): return cmd_tui(args) return cmd_log(args) if __name__ == "__main__": raise SystemExit(main())