357 lines
11 KiB
Python
Executable File
357 lines
11 KiB
Python
Executable File
#!/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 re
|
|
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"
|
|
)
|
|
|
|
SENSOR_CHANNELS = [2, 3, 4, 5]
|
|
|
|
# If no bytes at all arrive for this long, treat the device as hung (e.g. an
|
|
# I2C bus lockup freezing the Arduino) rather than looping forever in silence.
|
|
DEFAULT_STALL_TIMEOUT_S = 20.0
|
|
|
|
_FAN_PCT_RE = re.compile(r"\((\d+)%\)")
|
|
_FAN_PWM_RE = re.compile(r"^(\d+)/")
|
|
|
|
|
|
def fan_pct_from_status(fan: str) -> int:
|
|
match = _FAN_PCT_RE.search(fan)
|
|
if match:
|
|
return int(match.group(1))
|
|
match = _FAN_PWM_RE.match(fan)
|
|
if match:
|
|
return (int(match.group(1)) * 100) // 255
|
|
return 0
|
|
|
|
|
|
def target_c_from_status(target: str) -> str:
|
|
if target.startswith("idle"):
|
|
return "0.0"
|
|
if target.endswith("C"):
|
|
return target[:-1]
|
|
return target
|
|
|
|
|
|
def build_csv_payload_from_status(data: dict, ms: int | None = None) -> str:
|
|
if ms is None:
|
|
ms = int(time.time() * 1000)
|
|
|
|
sensors = {ch: (temp, hum) for ch, temp, hum in data.get("sensor_list", [])}
|
|
parts = [
|
|
str(ms),
|
|
target_c_from_status(data["target"]),
|
|
data["avg"],
|
|
data["min"],
|
|
data["max"],
|
|
data["spread"],
|
|
data["heatlim"],
|
|
data["heater"],
|
|
str(fan_pct_from_status(data.get("fan", "0"))),
|
|
"1" if data.get("cutoff_active") == "YES" else "0",
|
|
"1" if data.get("failsafe") == "YES" else "0",
|
|
]
|
|
for ch in SENSOR_CHANNELS:
|
|
if str(ch) in sensors:
|
|
temp, hum = sensors[str(ch)]
|
|
if temp == "ERR":
|
|
parts.extend(["", ""])
|
|
else:
|
|
parts.extend([temp, hum])
|
|
else:
|
|
parts.extend(["", ""])
|
|
return ",".join(parts)
|
|
|
|
|
|
class CsvSession:
|
|
"""Deferred CSV writer — no empty file until the first row lands."""
|
|
|
|
def __init__(self, path: Path):
|
|
self.path = path
|
|
self._fh = None
|
|
self._header_written = False
|
|
self.row_count = 0
|
|
|
|
def _ensure_open(self) -> None:
|
|
if self._fh is None:
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._fh = self.path.open("w", encoding="utf-8")
|
|
|
|
def _write_header(self) -> None:
|
|
if not self._header_written:
|
|
self._ensure_open()
|
|
assert self._fh is not None
|
|
self._fh.write(FALLBACK_HEADER + "\n")
|
|
self._header_written = True
|
|
|
|
def write_device_line(self, line: str) -> None:
|
|
if line.startswith("csv_hdr,"):
|
|
self._ensure_open()
|
|
assert self._fh is not None
|
|
device_header = line[len("csv_hdr,") :]
|
|
self._fh.write("wall_time," + device_header + "\n")
|
|
self._header_written = True
|
|
self._fh.flush()
|
|
return
|
|
if not line.startswith("csv,"):
|
|
return
|
|
self._write_header()
|
|
assert self._fh is not None
|
|
wall_time = datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
self._fh.write(wall_time + "," + line[len("csv,") :] + "\n")
|
|
self._fh.flush()
|
|
self.row_count += 1
|
|
|
|
def write_status(self, data: dict) -> None:
|
|
self._write_header()
|
|
assert self._fh is not None
|
|
wall_time = datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
payload = build_csv_payload_from_status(data)
|
|
self._fh.write(wall_time + "," + payload + "\n")
|
|
self._fh.flush()
|
|
self.row_count += 1
|
|
|
|
def close(self) -> None:
|
|
if self._fh is not None:
|
|
self._fh.close()
|
|
self._fh = None
|
|
if self.row_count == 0 and self.path.exists():
|
|
try:
|
|
self.path.unlink()
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
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 log_notice(message: str) -> None:
|
|
stamp = datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
print(f"{stamp} {message}", file=sys.stderr)
|
|
|
|
|
|
def cmd_log(args: argparse.Namespace) -> int:
|
|
from dryer_tui import parse_status
|
|
|
|
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"
|
|
|
|
stall_timeout = args.stall_timeout
|
|
print(f"Logging {port} -> {out}", file=sys.stderr)
|
|
if args.auto_log_on:
|
|
print("Will send 'log on' after connect", file=sys.stderr)
|
|
|
|
session = CsvSession(out)
|
|
with open_serial(port, args.baud) as ser:
|
|
if args.auto_log_on:
|
|
enable_dryer_logging(ser)
|
|
|
|
last_activity = time.monotonic()
|
|
while True:
|
|
try:
|
|
raw = ser.readline()
|
|
except KeyboardInterrupt:
|
|
print(f"\nStopped ({session.row_count} rows).", file=sys.stderr)
|
|
session.close()
|
|
return 0
|
|
except Exception as exc:
|
|
log_notice(
|
|
f"ERROR: serial read failed ({exc}) — closing after "
|
|
f"{session.row_count} rows"
|
|
)
|
|
session.close()
|
|
return 1
|
|
|
|
if not raw:
|
|
if time.monotonic() - last_activity >= stall_timeout:
|
|
log_notice(
|
|
f"WARN: no data from {port} for {stall_timeout:.0f}s — "
|
|
f"device likely hung (e.g. I2C bus lockup on the Arduino) "
|
|
f"— closing after {session.row_count} rows"
|
|
)
|
|
session.close()
|
|
return 1
|
|
continue
|
|
|
|
last_activity = time.monotonic()
|
|
line = decode_line(raw)
|
|
parsed = parse_status(line)
|
|
if parsed:
|
|
session.write_status(parsed)
|
|
print(line)
|
|
continue
|
|
|
|
if line.startswith("csv_hdr,") or line.startswith("csv,"):
|
|
session.write_device_line(line)
|
|
continue
|
|
|
|
if line:
|
|
print(line)
|
|
return 0
|
|
|
|
|
|
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)",
|
|
)
|
|
parser.add_argument(
|
|
"--stall-timeout",
|
|
type=float,
|
|
default=DEFAULT_STALL_TIMEOUT_S,
|
|
help=(
|
|
"log mode: seconds without any data before treating the device as "
|
|
f"hung and exiting (default: {DEFAULT_STALL_TIMEOUT_S:.0f})"
|
|
),
|
|
)
|
|
|
|
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)
|
|
log_p.add_argument("--stall-timeout", type=float, default=DEFAULT_STALL_TIMEOUT_S)
|
|
|
|
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())
|