258 lines
8.1 KiB
Python
258 lines
8.1 KiB
Python
#!/usr/bin/env python3
|
||
import argparse
|
||
import csv
|
||
import os
|
||
import time
|
||
from datetime import datetime
|
||
|
||
from gpiozero import OutputDevice
|
||
|
||
from rigol_phase import DEV as RIGOL_DEV, measure_phase
|
||
|
||
|
||
# 这里的“step”按全步计（不考虑细分）。细分后会自动把全步换算为脉冲数。
|
||
BASE_STEPS_PER_MM = 200 # 200 full-steps = 1mm
|
||
MM_MIN = 0
|
||
MM_MAX = 300
|
||
|
||
|
||
class GPIOStepper:
|
||
def __init__(
|
||
self,
|
||
*,
|
||
dir_pin: int,
|
||
step_pin: int,
|
||
microsteps: int = 8,
|
||
) -> None:
|
||
if microsteps <= 0:
|
||
raise ValueError("microsteps must be a positive integer")
|
||
self.microsteps = int(microsteps)
|
||
|
||
# gpiozero 默认使用 BCM 编号
|
||
self._dir = OutputDevice(dir_pin)
|
||
self._step = OutputDevice(step_pin)
|
||
self._step.off()
|
||
|
||
def close(self) -> None:
|
||
try:
|
||
self._step.off()
|
||
except Exception:
|
||
pass
|
||
try:
|
||
self._dir.off()
|
||
except Exception:
|
||
pass
|
||
|
||
def ping(self) -> str:
|
||
return "GPIO"
|
||
|
||
def set_dir(self, inc: bool) -> None:
|
||
value = 1 if inc else 0
|
||
self._dir.value = value
|
||
|
||
def move_steps(self, full_steps: int, freq_hz: int) -> None:
|
||
if full_steps < 0:
|
||
raise ValueError("steps must be >= 0")
|
||
if freq_hz <= 0:
|
||
raise ValueError("freq_hz must be > 0")
|
||
|
||
pulses = int(full_steps) * self.microsteps
|
||
if pulses <= 0:
|
||
return
|
||
|
||
half_period = 0.5 / float(freq_hz)
|
||
for _ in range(pulses):
|
||
self._step.on()
|
||
time.sleep(half_period)
|
||
self._step.off()
|
||
time.sleep(half_period)
|
||
|
||
|
||
def _default_csv_path() -> str:
|
||
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
return os.path.join(os.getcwd(), f"scan_{ts}.csv")
|
||
|
||
|
||
def _ensure_csv_header(path: str) -> None:
|
||
exists = os.path.exists(path) and os.path.getsize(path) > 0
|
||
if exists:
|
||
return
|
||
|
||
with open(path, "w", newline="", encoding="utf-8") as f:
|
||
w = csv.writer(f)
|
||
w.writerow(
|
||
[
|
||
"ts",
|
||
"mm",
|
||
"f0_hz",
|
||
"phi1_rad",
|
||
"phi2_rad",
|
||
"dphi_rad",
|
||
"dphi_deg",
|
||
"amp1_pp_adc",
|
||
"amp2_pp_adc",
|
||
"rigol_idn",
|
||
"points_mode",
|
||
"n",
|
||
]
|
||
)
|
||
|
||
|
||
def _append_row(path: str, *, mm: float, r: dict) -> None:
|
||
with open(path, "a", newline="", encoding="utf-8") as f:
|
||
w = csv.writer(f)
|
||
w.writerow(
|
||
[
|
||
r.get("ts"),
|
||
mm,
|
||
r.get("f0_hz"),
|
||
r.get("phi1_rad"),
|
||
r.get("phi2_rad"),
|
||
r.get("dphi_rad"),
|
||
r.get("dphi_deg"),
|
||
r.get("amp1_pp_adc"),
|
||
r.get("amp2_pp_adc"),
|
||
r.get("idn"),
|
||
r.get("points_mode"),
|
||
r.get("n"),
|
||
]
|
||
)
|
||
f.flush()
|
||
|
||
|
||
def main() -> None:
|
||
ap = argparse.ArgumentParser(description="电机每1mm移动一次，并读取Rigol示波器数据写入CSV")
|
||
ap.add_argument("--mm", type=float, required=True, help="启动时装置当前位置(mm)，范围0-300")
|
||
ap.add_argument("--end-mm", type=float, default=300.0, help="扫描终点(mm)，默认300")
|
||
ap.add_argument(
|
||
"--step",
|
||
type=int,
|
||
default=40,
|
||
help="每步移动(全步 steps)，默认40（200 step = 1mm；会按细分自动换算为脉冲）",
|
||
)
|
||
|
||
ap.add_argument("--microsteps", type=int, default=8, help="驱动器细分倍数，默认8")
|
||
ap.add_argument("--dir-pin", type=int, default=22, help="DIR 引脚(BCM编号)，默认22")
|
||
ap.add_argument("--step-pin", type=int, default=27, help="STEP 引脚(BCM编号)，默认27")
|
||
|
||
ap.add_argument("--move-freq", type=int, default=1200, help="MOVE命令使用的频率Hz，默认500")
|
||
ap.add_argument("--settle-s", type=float, default=0.05, help="每次移动后等待再测量(秒)，默认0.05")
|
||
|
||
ap.add_argument("--rigol-dev", default=RIGOL_DEV, help="Rigol设备路径(默认/dev/usbtmc0)")
|
||
ap.add_argument("--rigol-timeout", type=float, default=3.0)
|
||
ap.add_argument("--rigol-mode", default="NORM", choices=["NORM", "RAW", "MAX"], help="WAV:POIN:MODE")
|
||
|
||
ap.add_argument(
|
||
"--live",
|
||
action="store_true",
|
||
help="开启 Matplotlib 实时图形监视器（相位差/峰峰值/频率 + 进度条）",
|
||
)
|
||
|
||
ap.add_argument("--csv", default="", help="输出CSV路径，默认当前目录scan_时间戳.csv")
|
||
args = ap.parse_args()
|
||
|
||
if args.step <= 0:
|
||
raise SystemExit("--step 必须是正整数")
|
||
if args.microsteps <= 0:
|
||
raise SystemExit("--microsteps 必须是正整数")
|
||
|
||
if not (MM_MIN <= args.mm <= MM_MAX):
|
||
raise SystemExit("--mm 超出范围 0-300")
|
||
if not (MM_MIN <= args.end_mm <= MM_MAX):
|
||
raise SystemExit("--end-mm 超出范围 0-300")
|
||
|
||
# 用“步数”作为内部位置单位，避免浮点累积误差。
|
||
# 这里的 steps 是“全步”；实际输出到 STEP 引脚时会乘以 microsteps。
|
||
start_steps = int(round(args.mm * BASE_STEPS_PER_MM))
|
||
end_steps = int(round(args.end_mm * BASE_STEPS_PER_MM))
|
||
step_steps = int(args.step)
|
||
|
||
csv_path = args.csv or _default_csv_path()
|
||
_ensure_csv_header(csv_path)
|
||
|
||
motor = GPIOStepper(
|
||
dir_pin=args.dir_pin,
|
||
step_pin=args.step_pin,
|
||
microsteps=args.microsteps,
|
||
)
|
||
|
||
live = None
|
||
if args.live:
|
||
try:
|
||
from live_monitor import LiveMonitor
|
||
|
||
start_mm = start_steps / BASE_STEPS_PER_MM
|
||
end_mm = end_steps / BASE_STEPS_PER_MM
|
||
live = LiveMonitor(start_mm=start_mm, end_mm=end_mm, csv_path=csv_path)
|
||
except Exception as e:
|
||
print(f"[!] Live monitor disabled: {e}")
|
||
live = None
|
||
|
||
def live_add(mm: float, meas: dict) -> None:
|
||
nonlocal live
|
||
if live is None:
|
||
return
|
||
try:
|
||
live.add_point(mm=mm, meas=meas)
|
||
except Exception as e:
|
||
print(f"[!] Live monitor error (disabled): {e}")
|
||
try:
|
||
live.close()
|
||
except Exception:
|
||
pass
|
||
live = None
|
||
|
||
try:
|
||
r = motor.ping()
|
||
print(
|
||
f"[+] Motor GPIO: DIR={args.dir_pin}, STEP={args.step_pin}, microsteps={args.microsteps} (PING -> {r})"
|
||
)
|
||
|
||
# 先测一次（包含一次性IDN）
|
||
start_mm = start_steps / BASE_STEPS_PER_MM
|
||
print(f"[+] Measure @ {start_mm:.3f}mm")
|
||
meas = measure_phase(dev=args.rigol_dev, timeout=args.rigol_timeout, points_mode=args.rigol_mode, fetch_idn=True)
|
||
_append_row(csv_path, mm=start_mm, r=meas)
|
||
live_add(start_mm, meas)
|
||
print(f"[+] CSV: {csv_path}")
|
||
|
||
cur_steps = start_steps
|
||
if cur_steps == end_steps:
|
||
print("[+] 已在终点，无需移动")
|
||
return
|
||
|
||
inc = end_steps > cur_steps
|
||
direction_str = "+" if inc else "-"
|
||
step_mm = step_steps / BASE_STEPS_PER_MM
|
||
end_mm = end_steps / BASE_STEPS_PER_MM
|
||
print(f"[+] Scan: {start_mm:.3f} -> {end_mm:.3f} (step {direction_str}{step_steps} steps = {step_mm:.3f}mm)")
|
||
|
||
while cur_steps != end_steps:
|
||
motor.set_dir(inc=inc)
|
||
|
||
remaining = (end_steps - cur_steps) if inc else (cur_steps - end_steps)
|
||
move = step_steps if step_steps <= remaining else remaining
|
||
motor.move_steps(move, args.move_freq)
|
||
time.sleep(args.settle_s)
|
||
|
||
cur_steps = cur_steps + (move if inc else -move)
|
||
cur_mm = cur_steps / BASE_STEPS_PER_MM
|
||
print(f"[+] Measure @ {cur_mm:.3f}mm")
|
||
meas = measure_phase(dev=args.rigol_dev, timeout=args.rigol_timeout, points_mode=args.rigol_mode, fetch_idn=False)
|
||
_append_row(csv_path, mm=cur_mm, r=meas)
|
||
live_add(cur_mm, meas)
|
||
|
||
except KeyboardInterrupt:
|
||
print("\n[!] Interrupted")
|
||
finally:
|
||
motor.close()
|
||
if live is not None:
|
||
try:
|
||
live.close()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|