add tui and persistent targets
This commit is contained in:
495
scripts/dryer_tui.py
Normal file
495
scripts/dryer_tui.py
Normal file
@@ -0,0 +1,495 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Curses TUI for the filament dryer Arduino."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import curses
|
||||
import re
|
||||
import sys
|
||||
import threading
|
||||
from collections import deque
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from capture_csv import (
|
||||
decode_line,
|
||||
write_csv_row,
|
||||
)
|
||||
|
||||
PRESETS: list[tuple[str, float]] = [
|
||||
("PLA", 55),
|
||||
("PETG", 65),
|
||||
("ABS", 65),
|
||||
("ASA", 65),
|
||||
("HIPS", 65),
|
||||
("Nylon", 75),
|
||||
("PC", 75),
|
||||
("PVA", 45),
|
||||
("TPU/TPE", 55),
|
||||
("PP", 55),
|
||||
("Idle", 0),
|
||||
]
|
||||
|
||||
STATUS_RE = re.compile(
|
||||
r"target=(?P<target>\S+)\s+"
|
||||
r"cutoff=(?P<cutoff_temp>\S+)\s+"
|
||||
r"avg=(?P<avg>[\d.]+)C\s+"
|
||||
r"min=(?P<min>[\d.]+)C\s+"
|
||||
r"max=(?P<max>[\d.]+)C\s+"
|
||||
r"spread=(?P<spread>[\d.]+)C\s+"
|
||||
r"heatlim=(?P<heatlim>\d+)%\s+"
|
||||
r"heater=(?P<heater>[\d.]+)%\s+"
|
||||
r"fan=(?P<fan>\d+)(?P<fan_note>\([^)]*\))?\s+"
|
||||
r"cutoff=(?P<cutoff_active>\S+)\s+"
|
||||
r"failsafe=(?P<failsafe>\S+)\s+"
|
||||
r"mode=(?P<mode>\S+)\s+"
|
||||
r"sensors=\[(?P<sensors>.*)\]"
|
||||
)
|
||||
|
||||
SENSOR_RE = re.compile(r"ch(\d+):([\d.]+)C/(\d+)%|ch(\d+):ERR")
|
||||
|
||||
|
||||
@dataclass
|
||||
class DryerState:
|
||||
target: str = "—"
|
||||
cutoff: str = "—"
|
||||
avg: str = "—"
|
||||
min_temp: str = "—"
|
||||
max_temp: str = "—"
|
||||
spread: str = "—"
|
||||
heatlim: str = "—"
|
||||
heater: str = "—"
|
||||
fan: str = "—"
|
||||
fan_note: str = ""
|
||||
cutoff_active: str = "no"
|
||||
failsafe: str = "no"
|
||||
mode: str = "—"
|
||||
sensors: list[tuple[str, str, str]] = field(default_factory=list)
|
||||
messages: deque[str] = field(default_factory=lambda: deque(maxlen=12))
|
||||
csv_logging: bool = False
|
||||
csv_path: Path | None = None
|
||||
port: str = ""
|
||||
connected: bool = False
|
||||
|
||||
|
||||
def parse_status(line: str) -> dict | None:
|
||||
match = STATUS_RE.search(line)
|
||||
if not match:
|
||||
return None
|
||||
data = match.groupdict()
|
||||
sensors: list[tuple[str, str, str]] = []
|
||||
for part in data["sensors"].split(", "):
|
||||
part = part.strip()
|
||||
if not part:
|
||||
continue
|
||||
m = SENSOR_RE.match(part)
|
||||
if not m:
|
||||
continue
|
||||
if m.group(4):
|
||||
sensors.append((m.group(4), "ERR", ""))
|
||||
else:
|
||||
sensors.append((m.group(1), m.group(2), m.group(3)))
|
||||
data["sensor_list"] = sensors
|
||||
data["fan_note"] = data.get("fan_note") or ""
|
||||
return data
|
||||
|
||||
|
||||
def apply_status(state: DryerState, data: dict) -> None:
|
||||
target = data["target"]
|
||||
if target.startswith("idle"):
|
||||
state.target = "idle"
|
||||
elif target.endswith("C"):
|
||||
state.target = target[:-1]
|
||||
else:
|
||||
state.target = target
|
||||
cutoff = data["cutoff_temp"]
|
||||
if cutoff in ("n/aC", "n/a"):
|
||||
state.cutoff = "n/a"
|
||||
elif cutoff.endswith("C"):
|
||||
state.cutoff = cutoff[:-1]
|
||||
else:
|
||||
state.cutoff = cutoff
|
||||
state.avg = data["avg"]
|
||||
state.min_temp = data["min"]
|
||||
state.max_temp = data["max"]
|
||||
state.spread = data["spread"]
|
||||
state.heatlim = data["heatlim"]
|
||||
state.heater = data["heater"]
|
||||
state.fan = data["fan"]
|
||||
state.fan_note = data["fan_note"]
|
||||
state.cutoff_active = data["cutoff_active"]
|
||||
state.failsafe = data["failsafe"]
|
||||
state.mode = data["mode"]
|
||||
state.sensors = data["sensor_list"]
|
||||
|
||||
|
||||
class SerialWorker:
|
||||
def __init__(self, ser, state: DryerState, lock: threading.Lock):
|
||||
self.ser = ser
|
||||
self.state = state
|
||||
self.lock = lock
|
||||
self.stop = threading.Event()
|
||||
self._log_fh = None
|
||||
self._header_written = [False]
|
||||
self._thread: threading.Thread | None = None
|
||||
|
||||
def start(self) -> None:
|
||||
self._thread = threading.Thread(target=self._run, daemon=True)
|
||||
self._thread.start()
|
||||
with self.lock:
|
||||
self.state.connected = True
|
||||
self.state.port = getattr(self.ser, "port", "") or getattr(self.ser, "name", "")
|
||||
self.state.messages.append("Connected")
|
||||
|
||||
def close(self) -> None:
|
||||
self.stop.set()
|
||||
if self._thread is not None:
|
||||
self._thread.join(timeout=1.5)
|
||||
if self._log_fh is not None:
|
||||
self._log_fh.close()
|
||||
self._log_fh = None
|
||||
|
||||
def send(self, command: str) -> None:
|
||||
if self.ser is None:
|
||||
return
|
||||
self.ser.write((command.strip() + "\n").encode("utf-8"))
|
||||
self.ser.flush()
|
||||
|
||||
def set_csv_logging(self, enabled: bool, log_dir: Path) -> None:
|
||||
with self.lock:
|
||||
if enabled and not self.state.csv_logging:
|
||||
log_dir.mkdir(parents=True, exist_ok=True)
|
||||
path = log_dir / f"dryer_{datetime.now():%Y%m%d_%H%M%S}.csv"
|
||||
self._log_fh = path.open("w", encoding="utf-8")
|
||||
self._header_written = [False]
|
||||
self.state.csv_path = path
|
||||
self.state.csv_logging = True
|
||||
self.state.messages.append(f"CSV -> {path.name}")
|
||||
self.send("log on")
|
||||
elif not enabled and self.state.csv_logging:
|
||||
self.send("log off")
|
||||
self.state.csv_logging = False
|
||||
self.state.csv_path = None
|
||||
if self._log_fh is not None:
|
||||
self._log_fh.close()
|
||||
self._log_fh = None
|
||||
self.state.messages.append("CSV logging off")
|
||||
|
||||
def _note(self, line: str) -> None:
|
||||
with self.lock:
|
||||
self.state.messages.append(line)
|
||||
|
||||
def _run(self) -> None:
|
||||
assert self.ser is not None
|
||||
while not self.stop.is_set():
|
||||
try:
|
||||
raw = self.ser.readline()
|
||||
except Exception as exc:
|
||||
with self.lock:
|
||||
self.state.messages.append(f"Serial error: {exc}")
|
||||
break
|
||||
if not raw:
|
||||
continue
|
||||
line = decode_line(raw)
|
||||
if not line:
|
||||
continue
|
||||
|
||||
if line.startswith("csv,") or line.startswith("csv_hdr,"):
|
||||
if self._log_fh is not None:
|
||||
write_csv_row(self._log_fh, line, self._header_written)
|
||||
continue
|
||||
|
||||
parsed = parse_status(line)
|
||||
if parsed:
|
||||
with self.lock:
|
||||
apply_status(self.state, parsed)
|
||||
continue
|
||||
|
||||
if line.startswith("OK") or line.startswith("ERR") or line.startswith("WARN"):
|
||||
self._note(line)
|
||||
|
||||
|
||||
def _draw_box(win, y: int, x: int, h: int, w: int, title: str) -> None:
|
||||
if h < 2 or w < 4:
|
||||
return
|
||||
try:
|
||||
win.addstr(y, x, "+" + "-" * (w - 2) + "+")
|
||||
win.addstr(y, x + 2, f" {title} "[: max(0, w - 4)])
|
||||
for row in range(1, h - 1):
|
||||
win.addstr(y + row, x, "|" + " " * (w - 2) + "|")
|
||||
win.addstr(y + h - 1, x, "+" + "-" * (w - 2) + "+")
|
||||
except curses.error:
|
||||
pass
|
||||
|
||||
|
||||
def _safe_addstr(win, y: int, x: int, text: str, attr: int = 0) -> None:
|
||||
height, width = win.getmaxyx()
|
||||
if y < 0 or y >= height or x >= width:
|
||||
return
|
||||
win.addnstr(y, x, text, max(0, width - x - 1), attr)
|
||||
|
||||
|
||||
def _prompt(stdscr, label: str) -> str | None:
|
||||
stdscr.nodelay(False)
|
||||
stdscr.timeout(-1)
|
||||
curses.curs_set(1)
|
||||
height, width = stdscr.getmaxyx()
|
||||
prompt = f" {label}: "
|
||||
row = height - 1
|
||||
col = len(prompt)
|
||||
buf: list[str] = []
|
||||
|
||||
_safe_addstr(stdscr, row, 0, " " * max(0, width - 1))
|
||||
_safe_addstr(stdscr, row, 0, prompt)
|
||||
stdscr.move(row, col)
|
||||
stdscr.refresh()
|
||||
|
||||
try:
|
||||
while True:
|
||||
ch = stdscr.getch()
|
||||
if ch in (10, 13, curses.KEY_ENTER):
|
||||
break
|
||||
if ch in (27,): # Esc
|
||||
return None
|
||||
if ch in (curses.KEY_BACKSPACE, 127, 8):
|
||||
if buf:
|
||||
buf.pop()
|
||||
elif ch == curses.KEY_DC: # Delete — ignore
|
||||
pass
|
||||
elif 32 <= ch <= 126 and len(buf) < 24:
|
||||
buf.append(chr(ch))
|
||||
_safe_addstr(stdscr, row, col, (" " * 24))
|
||||
_safe_addstr(stdscr, row, col, "".join(buf))
|
||||
stdscr.move(row, col + len(buf))
|
||||
stdscr.refresh()
|
||||
finally:
|
||||
curses.curs_set(0)
|
||||
stdscr.nodelay(True)
|
||||
stdscr.timeout(200)
|
||||
|
||||
text = "".join(buf).strip()
|
||||
return text
|
||||
|
||||
|
||||
def _preset_menu(stdscr, worker: SerialWorker) -> None:
|
||||
height, width = stdscr.getmaxyx()
|
||||
menu_h = min(len(PRESETS) + 2, height - 4)
|
||||
menu_w = 28
|
||||
y0 = (height - menu_h) // 2
|
||||
x0 = (width - menu_w) // 2
|
||||
selected = 0
|
||||
|
||||
while True:
|
||||
stdscr.erase()
|
||||
_draw_box(stdscr, y0, x0, menu_h, menu_w, "Presets")
|
||||
for i, (name, temp) in enumerate(PRESETS):
|
||||
label = f" {name:<10} {temp:>5.0f} °C"
|
||||
attr = curses.A_REVERSE if i == selected else 0
|
||||
_safe_addstr(stdscr, y0 + 1 + i, x0 + 1, label.ljust(menu_w - 2), attr)
|
||||
_safe_addstr(stdscr, y0 + menu_h - 1, x0 + 2, "Enter select Esc cancel")
|
||||
stdscr.refresh()
|
||||
key = stdscr.getch()
|
||||
if key in (27, ord("q")):
|
||||
return
|
||||
if key in (curses.KEY_UP, ord("k")):
|
||||
selected = (selected - 1) % len(PRESETS)
|
||||
elif key in (curses.KEY_DOWN, ord("j")):
|
||||
selected = (selected + 1) % len(PRESETS)
|
||||
elif key in (10, 13, curses.KEY_ENTER):
|
||||
name, temp = PRESETS[selected]
|
||||
worker.send(f"target {temp:g}")
|
||||
worker._note(f"Preset {name} -> {temp:g} °C")
|
||||
return
|
||||
|
||||
|
||||
def _draw_dashboard(stdscr, state: DryerState) -> None:
|
||||
stdscr.erase()
|
||||
height, width = stdscr.getmaxyx()
|
||||
if height < 18 or width < 60:
|
||||
_safe_addstr(stdscr, 0, 0, "Terminal too small (need 60x18).")
|
||||
stdscr.refresh()
|
||||
return
|
||||
|
||||
title = f" Filament Dryer — {state.port} "
|
||||
_safe_addstr(stdscr, 0, 1, title, curses.A_BOLD)
|
||||
|
||||
row = 2
|
||||
_safe_addstr(stdscr, row, 2, f"Target: {state.target:>8} °C", curses.A_BOLD)
|
||||
_safe_addstr(stdscr, row, 28, f"Mode: {state.mode}")
|
||||
cutoff_attr = curses.A_BOLD | curses.color_pair(1) if state.cutoff_active == "YES" else 0
|
||||
_safe_addstr(stdscr, row, 48, f"Cutoff: {state.cutoff_active}", cutoff_attr)
|
||||
|
||||
row += 1
|
||||
_safe_addstr(stdscr, row, 2, f"Avg: {state.avg} °C Min: {state.min_temp} °C Max: {state.max_temp} °C Spread: {state.spread} °C")
|
||||
|
||||
row += 1
|
||||
fan_text = f"{state.fan} %{state.fan_note}"
|
||||
_safe_addstr(stdscr, row, 2, f"Heater: {state.heater} % Fan: {fan_text} Limit: {state.heatlim} %")
|
||||
|
||||
row += 2
|
||||
_draw_box(stdscr, row, 1, 5, width - 2, "Sensors")
|
||||
sensor_row = row + 1
|
||||
col = 3
|
||||
if state.sensors:
|
||||
for ch, temp, hum in state.sensors:
|
||||
if temp == "ERR":
|
||||
text = f"ch{ch}: ERR"
|
||||
else:
|
||||
text = f"ch{ch}: {temp} °C {hum} %"
|
||||
_safe_addstr(stdscr, sensor_row, col, text.ljust(22))
|
||||
col += 24
|
||||
if col + 22 >= width - 2:
|
||||
sensor_row += 1
|
||||
col = 3
|
||||
else:
|
||||
_safe_addstr(stdscr, sensor_row, 3, "Waiting for readings…")
|
||||
|
||||
row += 5
|
||||
help_y = height - 2
|
||||
log_h = help_y - row - 1
|
||||
if log_h < 4:
|
||||
_safe_addstr(stdscr, row, 2, "Terminal too small for message panel.")
|
||||
stdscr.refresh()
|
||||
return
|
||||
|
||||
_draw_box(stdscr, row, 1, log_h, width - 2, "Messages")
|
||||
csv_line = "on" if state.csv_logging else "off"
|
||||
if state.csv_path:
|
||||
csv_line += f" ({state.csv_path.name})"
|
||||
_safe_addstr(stdscr, row + 1, 3, f"CSV: {csv_line}")
|
||||
msg_row = row + 2
|
||||
for msg in list(state.messages)[-(log_h - 3) :]:
|
||||
_safe_addstr(stdscr, msg_row, 3, msg[: width - 6])
|
||||
msg_row += 1
|
||||
|
||||
help_y = height - 2
|
||||
_safe_addstr(
|
||||
stdscr,
|
||||
help_y,
|
||||
1,
|
||||
"0 idle | t target | p presets | f fan | l log | a autotune | : cmd | q quit",
|
||||
curses.A_DIM,
|
||||
)
|
||||
stdscr.refresh()
|
||||
|
||||
|
||||
def _curses_main(stdscr, ser, log_dir: Path, auto_log_on: bool) -> int:
|
||||
curses.curs_set(0)
|
||||
curses.start_color()
|
||||
curses.use_default_colors()
|
||||
curses.init_pair(1, curses.COLOR_RED, -1)
|
||||
stdscr.nodelay(True)
|
||||
stdscr.timeout(200)
|
||||
stdscr.clear()
|
||||
|
||||
state = DryerState()
|
||||
lock = threading.Lock()
|
||||
worker = SerialWorker(ser, state, lock)
|
||||
worker.start()
|
||||
worker.send("status")
|
||||
if auto_log_on:
|
||||
worker.set_csv_logging(True, log_dir)
|
||||
|
||||
try:
|
||||
while True:
|
||||
with lock:
|
||||
snapshot = DryerState(
|
||||
target=state.target,
|
||||
cutoff=state.cutoff,
|
||||
avg=state.avg,
|
||||
min_temp=state.min_temp,
|
||||
max_temp=state.max_temp,
|
||||
spread=state.spread,
|
||||
heatlim=state.heatlim,
|
||||
heater=state.heater,
|
||||
fan=state.fan,
|
||||
fan_note=state.fan_note,
|
||||
cutoff_active=state.cutoff_active,
|
||||
failsafe=state.failsafe,
|
||||
mode=state.mode,
|
||||
sensors=list(state.sensors),
|
||||
messages=deque(state.messages, maxlen=12),
|
||||
csv_logging=state.csv_logging,
|
||||
csv_path=state.csv_path,
|
||||
port=state.port,
|
||||
connected=state.connected,
|
||||
)
|
||||
_draw_dashboard(stdscr, snapshot)
|
||||
|
||||
key = stdscr.getch()
|
||||
if key == -1:
|
||||
continue
|
||||
if key in (ord("q"), ord("Q"), 27):
|
||||
break
|
||||
if key == ord("0"):
|
||||
worker.send("target 0")
|
||||
elif key == ord("t"):
|
||||
value = _prompt(stdscr, "Target °C (0 = idle)")
|
||||
if value is not None and value != "":
|
||||
worker.send(f"target {value}")
|
||||
elif key == ord("p"):
|
||||
stdscr.nodelay(False)
|
||||
stdscr.timeout(-1)
|
||||
_preset_menu(stdscr, worker)
|
||||
stdscr.nodelay(True)
|
||||
stdscr.timeout(200)
|
||||
elif key == ord("f"):
|
||||
worker.send("fan on")
|
||||
elif key == ord("F"):
|
||||
worker.send("fan off")
|
||||
elif key == ord("l"):
|
||||
with lock:
|
||||
enable = not state.csv_logging
|
||||
worker.set_csv_logging(enable, log_dir)
|
||||
elif key == ord("a"):
|
||||
value = _prompt(stdscr, "Autotune °C (Enter = 40)")
|
||||
if value is not None:
|
||||
cmd = "autotune" if value == "" else f"autotune {value}"
|
||||
worker.send(cmd)
|
||||
elif key == ord(":"):
|
||||
value = _prompt(stdscr, "Command")
|
||||
if value is not None and value != "":
|
||||
worker.send(value)
|
||||
elif key == ord("s"):
|
||||
worker.send("status")
|
||||
finally:
|
||||
worker.close()
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
def run_tui(port: str | None, baud: int, log_dir: Path, auto_log_on: bool = True) -> int:
|
||||
if not sys.stdin.isatty() or not sys.stdout.isatty():
|
||||
print("TUI requires an interactive terminal.", file=sys.stderr)
|
||||
print("Use: ssh -t user@host 'python3 scripts/capture_csv.py tui'", file=sys.stderr)
|
||||
print("Headless logging: python3 scripts/capture_csv.py log", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
from capture_csv import import_serial, open_serial, resolve_port
|
||||
|
||||
resolved = resolve_port(port)
|
||||
import_serial()
|
||||
try:
|
||||
ser = open_serial(resolved, baud)
|
||||
except Exception as exc:
|
||||
print(f"Cannot open serial port {resolved}: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
try:
|
||||
return curses.wrapper(
|
||||
lambda stdscr: _curses_main(stdscr, ser, log_dir, auto_log_on)
|
||||
)
|
||||
except curses.error as exc:
|
||||
print(f"TUI failed: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
except Exception as exc:
|
||||
print(f"Error: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
finally:
|
||||
if ser.is_open:
|
||||
ser.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(run_tui(None, 115200, Path("logs"), True))
|
||||
Reference in New Issue
Block a user