ouch finger

scripts/fan_characterize.py

@@ -0,0 +1,187 @@
+#!/usr/bin/env python3
+"""Run fan characterize sweep and capture fc,... serial log lines to CSV.
+
+Each profile heats from ~35 C avg to max corner (default 60 C) at a fixed fan
+PWM, measures spread during a hold, cools, then repeats for the next speed.
+Firmware prints the best fan at the end; use fanchars save on the device.
+
+Example:
+  ./fan_characterize.py
+  ./fan_characterize.py -o logs/fanchars.csv
+"""
+
+from __future__ import annotations
+
+import argparse
+import re
+import sys
+import time
+from collections import defaultdict
+from datetime import datetime, timezone
+from pathlib import Path
+
+from capture_csv import decode_line, open_serial, resolve_port
+
+FC_RE = re.compile(
+    r"^fc,(?P<ms>\d+),(?P<phase>\w+),(?P<run>\d+/\d+),"
+    r"(?P<fan>\d+),(?P<heater>\d+),"
+    r"(?P<avg>[\d.]+),(?P<min>[\d.]+),(?P<max>[\d.]+),(?P<spread>[\d.]+)"
+    r"(?:,(?P<temps>.*))?$"
+)
+DONE_RE = re.compile(r"^fanchars: done")
+FAIL_RE = re.compile(r"^fanchars: abort")
+RUN_SUMMARY_RE = re.compile(r"^fanchars: f=(?P<fan>\d+) spr=(?P<mean>[\d.]+)")
+BEST_RE = re.compile(r"^  best (?P<fan>\d+) spr=(?P<mean>[\d.]+)")
+
+HEADER = (
+    "wall_time,ms,phase,run,fan_pwm,fan_pct,heater_pct,avg_c,min_c,max_c,spread_c,"
+    "ch2_t,ch3_t,ch4_t,ch5_t"
+)
+
+
+def fan_pct(pwm: int) -> int:
+    return (pwm * 100) // 255
+
+
+def send_command(ser, command: str, timeout: float = 3.0) -> list[str]:
+    ser.write((command.strip() + "\n").encode("utf-8"))
+    ser.flush()
+    lines: list[str] = []
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        raw = ser.readline()
+        if not raw:
+            continue
+        line = decode_line(raw)
+        if not line:
+            continue
+        lines.append(line)
+        if line.startswith("OK ") or line.startswith("ERR "):
+            break
+    return lines
+
+
+def parse_fc_line(line: str) -> dict | None:
+    match = FC_RE.match(line)
+    if not match:
+        return None
+    data = match.groupdict()
+    temps = data.pop("temps") or ""
+    channels = (temps.split(",") + ["", "", "", ""])[:4]
+    data["ch2_t"], data["ch3_t"], data["ch4_t"], data["ch5_t"] = channels
+    data["fan_pwm"] = data.pop("fan")
+    data["heater_pct"] = data.pop("heater")
+    return data
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Capture fan characterize sweep")
+    parser.add_argument("-p", "--port", help="Serial port (default: auto-detect)")
+    parser.add_argument("-b", "--baud", type=int, default=115200)
+    parser.add_argument("--max", type=float, default=60.0, help="Max corner temp (C, default 60)")
+    parser.add_argument(
+        "-o",
+        "--output",
+        type=Path,
+        help="Output CSV (default: logs/fanchars_YYYYMMDD_HHMMSS.csv)",
+    )
+    parser.add_argument(
+        "--timeout",
+        type=float,
+        default=5 * 3600,
+        help="Abort if not done within this many seconds (default: 5 h)",
+    )
+    args = parser.parse_args()
+
+    out = args.output
+    if out is None:
+        out = Path("logs") / f"fanchars_{datetime.now():%Y%m%d_%H%M%S}.csv"
+    out.parent.mkdir(parents=True, exist_ok=True)
+
+    cmd = f"fanchars {args.max:g}" if args.max != 60.0 else "fanchars"
+    print(f"Command: {cmd}", file=sys.stderr)
+    print(f"Output: {out}", file=sys.stderr)
+    print("Expect heat/cool cycles per fan speed. Ctrl+C sends fanchars stop.", file=sys.stderr)
+
+    port = resolve_port(args.port)
+    run_means: dict[int, float] = {}
+    best_line = ""
+
+    try:
+        with open_serial(port, args.baud) as ser:
+            time.sleep(0.3)
+            while ser.in_waiting:
+                decode_line(ser.readline())
+
+            lines = send_command(ser, cmd, timeout=5.0)
+            for line in lines:
+                print(line, file=sys.stderr)
+                if line.startswith("ERR "):
+                    return 1
+
+            send_command(ser, "log on", timeout=2.0)
+
+            with out.open("w", encoding="utf-8") as fh:
+                fh.write(HEADER + "\n")
+                deadline = time.monotonic() + args.timeout
+                while time.monotonic() < deadline:
+                    raw = ser.readline()
+                    if not raw:
+                        continue
+                    line = decode_line(raw)
+                    if not line:
+                        continue
+
+                    row = parse_fc_line(line)
+                    if row:
+                        wall = datetime.now(timezone.utc).isoformat()
+                        fh.write(
+                            f"{wall},{row['ms']},{row['phase']},{row['run']},"
+                            f"{row['fan_pwm']},{fan_pct(int(row['fan_pwm']))},"
+                            f"{row['heater_pct']},{row['avg']},{row['min']},{row['max']},"
+                            f"{row['spread']},{row['ch2_t']},{row['ch3_t']},"
+                            f"{row['ch4_t']},{row['ch5_t']}\n"
+                        )
+                        fh.flush()
+
+                    match = RUN_SUMMARY_RE.match(line)
+                    if match:
+                        run_means[int(match.group("fan"))] = float(match.group("mean"))
+                        print(line, file=sys.stderr)
+
+                    if BEST_RE.search(line):
+                        best_line = line.strip()
+
+                    if DONE_RE.search(line) or FAIL_RE.search(line):
+                        print(line, file=sys.stderr)
+                        break
+                else:
+                    print("Timeout waiting for fanchars to finish", file=sys.stderr)
+                    return 1
+
+            if best_line:
+                print(best_line, file=sys.stderr)
+                print("Run: fanchars save  (on device) to store stir PWM", file=sys.stderr)
+            elif run_means:
+                best_fan = min(run_means, key=run_means.get)
+                print(
+                    f"Best from logs: fan {best_fan} ({fan_pct(best_fan)}%) "
+                    f"mean spread {run_means[best_fan]:.2f} C",
+                    file=sys.stderr,
+                )
+
+    except KeyboardInterrupt:
+        print("\nStopping…", file=sys.stderr)
+        try:
+            with open_serial(port, args.baud) as ser:
+                send_command(ser, "fanchars stop")
+        except OSError:
+            pass
+        return 130
+
+    print(f"Wrote {out}", file=sys.stderr)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())