add fanchars
This commit is contained in:
@@ -7,6 +7,7 @@ import curses
|
||||
import re
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
from collections import deque
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
@@ -43,7 +44,7 @@ STATUS_RE = re.compile(
|
||||
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\(\d+%\)(?:\([^)]+\))?(?:\s+TEST)?)\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>.*)\]"
|
||||
@@ -55,27 +56,73 @@ AUTOTUNE_MODE_RE = re.compile(
|
||||
r"autotune/(?P<phase>[\w-]+) (?P<elapsed>\d+)s (?P<cycles>\d+/\d+)cyc pre>=(?P<pre>\d+)C"
|
||||
)
|
||||
|
||||
STEPRESP_MODE_RE = re.compile(
|
||||
r"stepresp/(?P<phase>[\w-]+) (?P<elapsed>\d+)s step (?P<step>\d+/\d+) heat=(?P<heat>\d+)%"
|
||||
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 format_mode_line(mode: str) -> str:
|
||||
|
||||
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()
|
||||
return (
|
||||
f"Autotune {d['phase']}: {d['elapsed']}s elapsed, "
|
||||
summary = (
|
||||
f"Autotune {d['phase']}: {d['elapsed']}s, "
|
||||
f"{d['cycles']} cycles, preheat avg >= {d['pre']} C"
|
||||
)
|
||||
match = STEPRESP_MODE_RE.match(mode)
|
||||
return summary, "Relay tuning heat PI — heater bang-bang around setpoint"
|
||||
|
||||
match = FANCHARS_MODE_RE.match(mode)
|
||||
if match:
|
||||
d = match.groupdict()
|
||||
return (
|
||||
f"Step response {d['phase']}: {d['elapsed']}s, "
|
||||
f"step {d['step']}, heater {d['heat']}%"
|
||||
)
|
||||
return f"Mode: {mode}"
|
||||
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
|
||||
@@ -96,8 +143,9 @@ class DryerState:
|
||||
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=12))
|
||||
messages: deque[str] = field(default_factory=lambda: deque(maxlen=24))
|
||||
csv_logging: bool = False
|
||||
csv_path: Path | None = None
|
||||
port: str = ""
|
||||
@@ -150,13 +198,18 @@ def apply_status(state: DryerState, data: dict) -> None:
|
||||
state.hblk = data["hblk"]
|
||||
state.ssr = data["ssr"]
|
||||
fan_raw = data["fan"]
|
||||
state.fan = fan_raw
|
||||
state.fan = format_fan_display(fan_raw)
|
||||
state.fan_note = ""
|
||||
if fan_raw.endswith("(off)") or fan_raw.endswith("(cooldown)") or fan_raw.endswith("(cmd-off)") or " TEST" in fan_raw:
|
||||
if "(off)" in fan_raw or "(cooldown)" in fan_raw or "(cmd-off)" 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"]
|
||||
|
||||
|
||||
@@ -232,6 +285,10 @@ class SerialWorker:
|
||||
if not line:
|
||||
continue
|
||||
|
||||
if line.startswith("fanchars:"):
|
||||
self._note(line)
|
||||
continue
|
||||
|
||||
if line.startswith("csv,") or line.startswith("csv_hdr,"):
|
||||
continue
|
||||
|
||||
@@ -347,8 +404,8 @@ def _preset_menu(stdscr, worker: SerialWorker) -> None:
|
||||
def _draw_dashboard(stdscr, state: DryerState) -> None:
|
||||
stdscr.erase()
|
||||
height, width = stdscr.getmaxyx()
|
||||
if height < 19 or width < 60:
|
||||
_safe_addstr(stdscr, 0, 0, "Terminal too small (need 60x19).")
|
||||
if height < 20 or width < 60:
|
||||
_safe_addstr(stdscr, 0, 0, "Terminal too small (need 60x20).")
|
||||
stdscr.refresh()
|
||||
return
|
||||
|
||||
@@ -361,14 +418,19 @@ def _draw_dashboard(stdscr, state: DryerState) -> None:
|
||||
_safe_addstr(stdscr, row, 36, f"Trip: {state.cutoff_active}", cutoff_attr)
|
||||
|
||||
row += 1
|
||||
mode_text = format_mode_line(state.mode)
|
||||
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 == "" else state.fan
|
||||
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,
|
||||
@@ -420,7 +482,7 @@ def _draw_dashboard(stdscr, state: DryerState) -> None:
|
||||
stdscr,
|
||||
help_y,
|
||||
1,
|
||||
"0 idle | t target | p presets | f fan | l log | a autotune | r stepresp | : cmd | q quit",
|
||||
"0 idle | t target | p presets | f fan | l log | a autotune | c fanchars | : cmd | q quit",
|
||||
curses.A_DIM,
|
||||
)
|
||||
stdscr.refresh()
|
||||
@@ -440,6 +502,8 @@ def _curses_main(stdscr, ser, log_dir: Path, auto_log_on: bool) -> int:
|
||||
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)
|
||||
|
||||
@@ -463,8 +527,9 @@ def _curses_main(stdscr, ser, log_dir: Path, auto_log_on: bool) -> int:
|
||||
cutoff_active=state.cutoff_active,
|
||||
failsafe=state.failsafe,
|
||||
mode=state.mode,
|
||||
activity=state.activity,
|
||||
sensors=list(state.sensors),
|
||||
messages=deque(state.messages, maxlen=12),
|
||||
messages=deque(state.messages, maxlen=24),
|
||||
csv_logging=state.csv_logging,
|
||||
csv_path=state.csv_path,
|
||||
port=state.port,
|
||||
@@ -502,21 +567,9 @@ def _curses_main(stdscr, ser, log_dir: Path, auto_log_on: bool) -> int:
|
||||
if value is not None:
|
||||
cmd = "autotune" if value == "" else f"autotune {value}"
|
||||
worker.send(cmd)
|
||||
elif key == ord("r"):
|
||||
temp = _prompt(stdscr, "Stepresp temp °C (Enter = 45)")
|
||||
if temp is None:
|
||||
continue
|
||||
heater = _prompt(stdscr, "Heater % (Enter = 35)")
|
||||
if heater is None:
|
||||
continue
|
||||
if temp == "" and heater == "":
|
||||
worker.send("stepresp")
|
||||
elif heater == "":
|
||||
worker.send(f"stepresp {temp}")
|
||||
elif temp == "":
|
||||
worker.send(f"stepresp 45 {heater}")
|
||||
else:
|
||||
worker.send(f"stepresp {temp} {heater}")
|
||||
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 != "":
|
||||
|
||||
Reference in New Issue
Block a user