621 lines
20 KiB
Python
621 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
"""Curses TUI for the filament dryer Arduino."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import curses
|
|
import re
|
|
import sys
|
|
import threading
|
|
import time
|
|
from collections import deque
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from capture_csv import (
|
|
CsvSession,
|
|
decode_line,
|
|
)
|
|
|
|
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"htop=(?P<htop>\S+)\s+"
|
|
r"hblk=(?P<hblk>\S+)\s+"
|
|
r"ssr=(?P<ssr>on|off)\s+"
|
|
r"fan=(?P<fan>\d+/255\([^)]+\)(?:\([^)]+\))?(?:\s+TEST)?)\s+"
|
|
r"cutoff=(?P<cutoff_active>\S+)\s+"
|
|
r"failsafe=(?P<failsafe>\S+)\s+"
|
|
r"mode=(?P<mode>.+?)\s+sensors=\[(?P<sensors>.*)\]"
|
|
)
|
|
|
|
SENSOR_RE = re.compile(r"ch(\d+):([\d.]+)C/(\d+)%|ch(\d+):ERR")
|
|
|
|
AUTOTUNE_MODE_RE = re.compile(
|
|
r"autotune/(?P<phase>[\w-]+) (?P<elapsed>\d+)s (?P<cycles>\d+/\d+)cyc pre>=(?P<pre>\d+)C"
|
|
)
|
|
|
|
FANCHARS_MODE_RE = re.compile(
|
|
r"fanchars/(?P<phase>[\w-]+) (?P<elapsed>\d+)s run (?P<run>[\w]+)/(?P<runs>\d+) "
|
|
r"fan=(?P<testfan>\d+) heat=(?P<heat>\d+)%"
|
|
)
|
|
|
|
FANCHARS_PHASE_HELP: dict[str, str] = {
|
|
"precool": "Cooling chamber to 40 C avg before first fan test (fan at 100% now)",
|
|
"cool": "Cooling to 40 C avg before next fan test (fan at 100% now)",
|
|
"heat": "Heating to 60 C max corner at test fan speed",
|
|
"hold": "Holding at max — measuring temperature spread",
|
|
"refine": "Refine run — midpoint PWM between two best spreads",
|
|
}
|
|
|
|
|
|
def fan_pct_from_pwm(pwm: int) -> int:
|
|
return (pwm * 100) // 255
|
|
|
|
|
|
def format_fan_display(fan_raw: str) -> str:
|
|
match = re.match(r"(\d+)/255\((\d+)%\)(.*)$", fan_raw.strip())
|
|
if not match:
|
|
return fan_raw
|
|
suffix = match.group(3).strip()
|
|
pct = match.group(2)
|
|
if suffix:
|
|
return f"{pct}% {suffix}"
|
|
return f"{pct}%"
|
|
|
|
|
|
def format_mode_line(mode: str, avg: str = "") -> tuple[str, str]:
|
|
"""Return (mode summary, activity detail) for the dashboard."""
|
|
match = AUTOTUNE_MODE_RE.match(mode)
|
|
if match:
|
|
d = match.groupdict()
|
|
summary = (
|
|
f"Autotune {d['phase']}: {d['elapsed']}s, "
|
|
f"{d['cycles']} cycles, preheat avg >= {d['pre']} C"
|
|
)
|
|
return summary, "Relay tuning heat PI — heater bang-bang around setpoint"
|
|
|
|
match = FANCHARS_MODE_RE.match(mode)
|
|
if match:
|
|
d = match.groupdict()
|
|
phase = d["phase"]
|
|
test_pct = fan_pct_from_pwm(int(d["testfan"]))
|
|
run = d["run"]
|
|
runs = d["runs"]
|
|
if run == "pre":
|
|
run_text = f"preparing (before 1/{runs})"
|
|
elif run.startswith("n"):
|
|
run_text = f"before {run[1:]}/{runs} ({test_pct}% fan next)"
|
|
elif run == "refine":
|
|
run_text = f"refine ({test_pct}% fan)"
|
|
else:
|
|
run_text = f"{run}/{runs} ({test_pct}% fan)"
|
|
summary = f"Fan chars {phase}: {d['elapsed']}s — {run_text}"
|
|
detail = FANCHARS_PHASE_HELP.get(phase, "")
|
|
if phase in ("precool", "cool") and avg not in ("—", ""):
|
|
try:
|
|
detail += f" — avg {avg} C"
|
|
except ValueError:
|
|
pass
|
|
return summary, detail
|
|
|
|
if mode in ("manual", "regulating"):
|
|
return f"Mode: {mode}", "Normal temperature control"
|
|
return f"Mode: {mode}", ""
|
|
|
|
|
|
@dataclass
|
|
class DryerState:
|
|
target: str = "—"
|
|
cutoff: str = "—"
|
|
avg: str = "—"
|
|
min_temp: str = "—"
|
|
max_temp: str = "—"
|
|
spread: str = "—"
|
|
heatlim: str = "—"
|
|
heater: str = "—"
|
|
htop: str = "—"
|
|
hblk: str = "—"
|
|
ssr: str = "—"
|
|
fan: str = "—"
|
|
fan_note: str = ""
|
|
cutoff_active: str = "no"
|
|
failsafe: str = "no"
|
|
mode: str = "—"
|
|
activity: str = ""
|
|
sensors: list[tuple[str, str, str]] = field(default_factory=list)
|
|
messages: deque[str] = field(default_factory=lambda: deque(maxlen=24))
|
|
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
|
|
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.htop = data["htop"].removesuffix("C") if data["htop"].endswith("C") else data["htop"]
|
|
state.hblk = data["hblk"]
|
|
state.ssr = data["ssr"]
|
|
fan_raw = data["fan"]
|
|
state.fan = format_fan_display(fan_raw)
|
|
state.fan_note = ""
|
|
if "(off)" in fan_raw or "(cooldown)" in fan_raw or "(off<40C)" in fan_raw:
|
|
state.fan_note = fan_raw[fan_raw.find("(") :] if "(" in fan_raw else ""
|
|
elif "(manual)" in fan_raw or "(stir)" in fan_raw:
|
|
state.fan_note = fan_raw[fan_raw.find("(") :] if "(" in fan_raw else ""
|
|
elif "(fanchars-" in fan_raw:
|
|
state.fan_note = fan_raw[fan_raw.find("(fanchars-") :]
|
|
state.cutoff_active = data["cutoff_active"]
|
|
state.failsafe = data["failsafe"]
|
|
state.mode = data["mode"]
|
|
summary, activity = format_mode_line(data["mode"], data["avg"])
|
|
state.mode = summary
|
|
state.activity = activity
|
|
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._csv: CsvSession | None = None
|
|
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._csv is not None:
|
|
self._csv.close()
|
|
self._csv = 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._csv = CsvSession(path)
|
|
self.state.csv_path = path
|
|
self.state.csv_logging = True
|
|
self.state.messages.append(f"CSV -> {path.name} (on status)")
|
|
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._csv is not None:
|
|
rows = self._csv.row_count
|
|
self._csv.close()
|
|
self._csv = None
|
|
self.state.messages.append(f"CSV logging off ({rows} rows)")
|
|
else:
|
|
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("fanchars:"):
|
|
self._note(line)
|
|
continue
|
|
|
|
if line.startswith("csv,") or line.startswith("csv_hdr,"):
|
|
continue
|
|
|
|
parsed = parse_status(line)
|
|
if parsed:
|
|
with self.lock:
|
|
apply_status(self.state, parsed)
|
|
if self._csv is not None:
|
|
self._csv.write_status(parsed)
|
|
continue
|
|
|
|
if line.startswith("target="):
|
|
self._note("WARN: could not parse status line")
|
|
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 < 20 or width < 60:
|
|
_safe_addstr(stdscr, 0, 0, "Terminal too small (need 60x20).")
|
|
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:>6} C Limit: {state.cutoff} C", curses.A_BOLD)
|
|
cutoff_attr = curses.A_BOLD | curses.color_pair(1) if state.cutoff_active == "YES" else 0
|
|
_safe_addstr(stdscr, row, 36, f"Trip: {state.cutoff_active}", cutoff_attr)
|
|
|
|
row += 1
|
|
mode_text = state.mode
|
|
_safe_addstr(stdscr, row, 2, f"{mode_text[: max(0, width - 18)]} FS: {state.failsafe}")
|
|
|
|
row += 1
|
|
if state.activity:
|
|
_safe_addstr(stdscr, row, 2, state.activity[: max(0, width - 4)], curses.A_DIM)
|
|
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 = state.fan
|
|
if state.fan_note and "(fanchars-" not in state.fan_note:
|
|
fan_text = f"{state.fan} {state.fan_note}"
|
|
_safe_addstr(
|
|
stdscr,
|
|
row,
|
|
2,
|
|
f"Heater: {state.heater} % SSR: {state.ssr} Fan: {fan_text} Limit: {state.heatlim} %",
|
|
)
|
|
|
|
row += 1
|
|
_safe_addstr(stdscr, row, 2, f"Heat stop: {state.htop} C Block: {state.hblk}")
|
|
|
|
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 | c fanchars | : 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")
|
|
time.sleep(0.4)
|
|
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,
|
|
htop=state.htop,
|
|
hblk=state.hblk,
|
|
ssr=state.ssr,
|
|
fan=state.fan,
|
|
fan_note=state.fan_note,
|
|
cutoff_active=state.cutoff_active,
|
|
failsafe=state.failsafe,
|
|
mode=state.mode,
|
|
activity=state.activity,
|
|
sensors=list(state.sensors),
|
|
messages=deque(state.messages, maxlen=24),
|
|
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("c"):
|
|
worker.send("fanchars")
|
|
worker._note("Started fanchars — 30/100/60/80% then refine if needed")
|
|
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))
|