alex/arduino-filament-dryer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix tui

Browse Source
This commit is contained in:
Alex
2026-07-05 20:15:29 +02:00
parent 960831529e
commit 02e51717e8
4 changed files with 192 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

164
scripts/capture_csv.py
View File

@@ -9,6 +9,7 @@
from __future__ import annotations
import argparse
import re
import sys
import time
from datetime import datetime, timezone
@@ -19,6 +20,118 @@ FALLBACK_HEADER = (
"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]
_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")
@@ -93,36 +206,20 @@ def enable_dryer_logging(ser, retries: int = 3) -> None:
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:
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"
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:
session = CsvSession(out)
with open_serial(port, args.baud) as ser:
if args.auto_log_on:
enable_dryer_logging(ser)
@@ -130,32 +227,27 @@ def cmd_log(args: argparse.Namespace) -> int:
try:
raw = ser.readline()
except KeyboardInterrupt:
print("\nStopped.", file=sys.stderr)
print(f"\nStopped ({session.row_count} rows).", file=sys.stderr)
session.close()
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)
parsed = parse_status(line)
if parsed:
session.write_status(parsed)
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()
if line.startswith("csv_hdr,") or line.startswith("csv,"):
session.write_device_line(line)
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()
if line:
print(line)
return 0
def cmd_tui(args: argparse.Namespace) -> int: