148 lines
4.2 KiB
Python
148 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Cycle fan speeds on the dryer for wiring / PWM verification.
|
|
|
|
Uses the firmware `fan test <pwm>` command (heater stays off). Sends `target 0`
|
|
first so the thermal loop is idle.
|
|
|
|
Example:
|
|
./fan_test.py
|
|
./fan_test.py --pct 30 50 100 --interval 3 --loop
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import re
|
|
import sys
|
|
import time
|
|
|
|
from capture_csv import decode_line, open_serial, resolve_port
|
|
|
|
OK_RE = re.compile(r"^OK ")
|
|
ERR_RE = re.compile(r"^ERR ")
|
|
FAN_STATUS_RE = re.compile(r"fan=(\d+)/255\((\d+)%\)")
|
|
|
|
|
|
def pct_to_pwm(pct: int) -> int:
|
|
if pct < 0 or pct > 100:
|
|
raise ValueError(f"fan percent must be 0-100, got {pct}")
|
|
return round(pct * 255 / 100)
|
|
|
|
|
|
def send_command(ser, command: str, timeout: float = 2.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 OK_RE.match(line) or ERR_RE.match(line):
|
|
break
|
|
return lines
|
|
|
|
|
|
def drain_status(ser, duration: float) -> str | None:
|
|
"""Read serial for `duration` seconds; return last status fan field if seen."""
|
|
fan_field: str | None = None
|
|
deadline = time.monotonic() + duration
|
|
while time.monotonic() < deadline:
|
|
raw = ser.readline()
|
|
if not raw:
|
|
continue
|
|
line = decode_line(raw)
|
|
if not line or line.startswith("csv"):
|
|
continue
|
|
match = FAN_STATUS_RE.search(line)
|
|
if match:
|
|
fan_field = f"{match.group(1)}/255 ({match.group(2)}%)"
|
|
elif line.startswith("target="):
|
|
print(f" status: {line}", flush=True)
|
|
return fan_field
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Cycle fan PWM to verify fan control")
|
|
parser.add_argument(
|
|
"-p",
|
|
"--port",
|
|
help="Serial port (default: auto-detect)",
|
|
)
|
|
parser.add_argument("-b", "--baud", type=int, default=115200)
|
|
parser.add_argument(
|
|
"--pct",
|
|
type=int,
|
|
nargs="+",
|
|
default=[0, 30, 100, 200, 255],
|
|
metavar="PCT",
|
|
help="Fan speeds in percent (default: 0 30 100 200 255)",
|
|
)
|
|
parser.add_argument(
|
|
"--interval",
|
|
type=float,
|
|
default=5.0,
|
|
metavar="SEC",
|
|
help="Seconds to hold each step (default: 5)",
|
|
)
|
|
parser.add_argument(
|
|
"--loop",
|
|
action="store_true",
|
|
help="Repeat the sequence until Ctrl+C",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
port = resolve_port(args.port)
|
|
sequence = [(pct, pct_to_pwm(pct)) for pct in args.pct]
|
|
|
|
print(f"Port: {port}", file=sys.stderr)
|
|
print(
|
|
f"Sequence: {' -> '.join(str(p) + '%' for p, _ in sequence)} "
|
|
f"every {args.interval:g}s (heater off)",
|
|
file=sys.stderr,
|
|
)
|
|
print("Ctrl+C to stop\n", file=sys.stderr)
|
|
|
|
interrupted = False
|
|
with open_serial(port, args.baud) as ser:
|
|
ser.reset_input_buffer()
|
|
|
|
lines = send_command(ser, "target 0")
|
|
for line in lines:
|
|
print(line, flush=True)
|
|
if any(ERR_RE.match(line) for line in lines):
|
|
return 1
|
|
|
|
try:
|
|
while True:
|
|
for pct, pwm in sequence:
|
|
print(f">>> fan test {pwm} ({pct}%)", flush=True)
|
|
lines = send_command(ser, f"fan test {pwm}")
|
|
for line in lines:
|
|
print(f" {line}", flush=True)
|
|
if any(ERR_RE.match(line) for line in lines):
|
|
return 1
|
|
|
|
reported = drain_status(ser, args.interval)
|
|
if reported:
|
|
print(f" reported fan={reported}", flush=True)
|
|
|
|
if not args.loop:
|
|
break
|
|
except KeyboardInterrupt:
|
|
interrupted = True
|
|
print("\nInterrupted", file=sys.stderr)
|
|
finally:
|
|
print(">>> fan test 0 (stop)", flush=True)
|
|
send_command(ser, "fan test 0")
|
|
|
|
return 130 if interrupted else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|