317 lines
12 KiB
Python
317 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
红外板 — 收发联合诊断
|
||
====================
|
||
一边发送候选协议帧，一边同时读取串口返回，
|
||
看板子收到数据后有没有任何回应（ACK/NAK/回显等）。
|
||
|
||
同时抓取原始遥控信号的完整原始字节（不合并，带时间戳），
|
||
检查是否有被 IDLE_TIMEOUT 截断的帧头/帧尾。
|
||
|
||
用法:
|
||
python ir_diag.py -p COM8 # 诊断模式1: 重新抓遥控信号
|
||
python ir_diag.py -p COM8 --echo # 诊断模式2: 发送+监听回应
|
||
python ir_diag.py -p COM8 --boot # 诊断模式3: 捕获上电启动信息
|
||
"""
|
||
|
||
import sys
|
||
import time
|
||
import argparse
|
||
import serial
|
||
import serial.tools.list_ports
|
||
|
||
BAUDRATE = 9600
|
||
|
||
# 已知数据
|
||
CUSTOMER_HI = 0x3B
|
||
CUSTOMER_LO = 0xC4
|
||
KEY_CODES = {0x0D: "上", 0x15: "下", 0x10: "左", 0x12: "右"}
|
||
|
||
|
||
def find_port():
|
||
"""自动找 USB-to-TTL"""
|
||
ports = serial.tools.list_ports.comports()
|
||
keywords = ["ch343", "ch340", "ch341", "usb-enhanced-serial",
|
||
"usb serial", "ft232", "cp210", "pl2303", "wch"]
|
||
for p in ports:
|
||
combined = f"{p.description or ''} {p.hwid or ''}".lower()
|
||
if any(kw in combined for kw in keywords):
|
||
return p.device
|
||
all_p = list(ports)
|
||
return all_p[0].device if all_p else None
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════
|
||
# 模式1: 详细捕获原始遥控信号 (逐字节, 不合并)
|
||
# ═══════════════════════════════════════════════════════════════
|
||
|
||
def capture_remote_detail(port: str):
|
||
"""
|
||
逐字节捕获，每个字节都带时间戳。
|
||
可能发现: 帧头(如 AA 55)、长度字节、校验和等之前被忽略的字节。
|
||
"""
|
||
print(f"打开 {port} @ {BAUDRATE} baud")
|
||
print("请按遥控器按键... (Ctrl+C 退出)\n")
|
||
|
||
ser = serial.Serial(port, baudrate=BAUDRATE, timeout=0.1)
|
||
print(f"{'时间':<12} {'字节':<6} {'HEX':<8} {'BIN':<12} {'ASCII'}")
|
||
print("-" * 65)
|
||
|
||
last_time = time.time()
|
||
|
||
try:
|
||
while True:
|
||
try:
|
||
n = ser.in_waiting
|
||
except serial.SerialException:
|
||
break
|
||
|
||
if n > 0:
|
||
now = time.time()
|
||
gap = now - last_time
|
||
data = ser.read(n)
|
||
|
||
for i, b in enumerate(data):
|
||
ascii_ch = chr(b) if 32 <= b < 127 else "."
|
||
ts = time.strftime("%H:%M:%S") + f".{int((now % 1) * 1000):03d}"
|
||
|
||
if i == 0 and gap > 0.3:
|
||
# 新的红外信号 — 画分隔线
|
||
print(f"\n{'─' * 65}")
|
||
print(f" 新信号 ↑ 距上次 {gap:.1f}s")
|
||
print(f"{ts:<12} [{i+1}/{n}] {b:02X} {b:08b} {ascii_ch}")
|
||
|
||
last_time = now
|
||
else:
|
||
time.sleep(0.01)
|
||
|
||
except KeyboardInterrupt:
|
||
pass
|
||
finally:
|
||
ser.close()
|
||
|
||
print("\n✓ 诊断完成。检查上面的字节序列:")
|
||
print(" 1. 每次按键是否只有 3 个字节?")
|
||
print(" 2. 前后有没有额外的帧头/帧尾字节?")
|
||
print(" 3. 字节间有没有明显的停顿(说明是分帧发送)?")
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════
|
||
# 模式2: 发送并监听回应
|
||
# ═══════════════════════════════════════════════════════════════
|
||
|
||
# 候选发送协议 — 带"编号"以便快速定位有效协议
|
||
CANDIDATES = [
|
||
# (名称, 帧bytes)
|
||
("raw_3bytes", bytes([0x3B, 0xC4, 0x0D])),
|
||
("raw_3bytes_rev", bytes([0x0D, 0xC4, 0x3B])),
|
||
("raw_plus_CR", bytes([0x3B, 0xC4, 0x0D, 0x0D])),
|
||
("raw_plus_CRLF", bytes([0x3B, 0xC4, 0x0D, 0x0D, 0x0A])),
|
||
("NEC_std", bytes([0x3B, 0xC4, 0x0D, 0xF2])),
|
||
("NEC_addr_inv", bytes([0x3B, 0xC4 ^ 0xFF, 0x0D, 0x0D ^ 0xFF])),
|
||
("A5_frame", bytes([0xA5, 0x03, 0x3B, 0xC4, 0x0D])),
|
||
("A5_frame_sum", bytes([0xA5, 0x03, 0x3B, 0xC4, 0x0D, 0xE9])),
|
||
("AA55_frame", bytes([0xAA, 0x55, 0x3B, 0xC4, 0x0D])),
|
||
("AA55_frame_sum", bytes([0xAA, 0x55, 0x3B, 0xC4, 0x0D, 0xA2])),
|
||
("FF_prefix", bytes([0xFF, 0xFF, 0x3B, 0xC4, 0x0D])),
|
||
("FE_FD_frame", bytes([0xFE, 0x3B, 0xC4, 0x0D, 0xFD])),
|
||
("A0_prefix", bytes([0xA0, 0x3B, 0xC4, 0x0D, 0x0A])),
|
||
("01_prefix", bytes([0x01, 0x3B, 0xC4, 0x0D])),
|
||
("AT_send_crlf", b"AT+SEND=3BC40D\r\n"),
|
||
("text_hex_crlf", b"3BC40D\r\n"),
|
||
("text_spaced_crlf", b"3B C4 0D\r\n"),
|
||
("text_send_crlf", b"SEND 3BC40D\r\n"),
|
||
("text_at_crlf", b"AT+IR=3BC40D\r\n"),
|
||
("IR_SEND_crlf", b"IR_SEND=3BC40D\r\n"),
|
||
]
|
||
|
||
|
||
def send_and_listen(port: str):
|
||
"""
|
||
发送每个候选帧，然后立即监听串口回应。
|
||
使用非阻塞读取: 发完等100ms，把期间收到的所有字节打印出来。
|
||
"""
|
||
print(f"打开 {port} @ {BAUDRATE} baud")
|
||
print("收发联合诊断 — 每个协议发送后监听板子是否有回应\n")
|
||
|
||
ser = serial.Serial(port, baudrate=BAUDRATE, timeout=0.1)
|
||
|
||
for i, (name, frame) in enumerate(CANDIDATES):
|
||
# 清空接收缓冲
|
||
ser.reset_input_buffer()
|
||
ser.reset_output_buffer()
|
||
|
||
# 发送
|
||
try:
|
||
ser.write(frame)
|
||
ser.flush()
|
||
except serial.SerialException as e:
|
||
print(f" [{i+1:>2}] {name}: ✗ 发送失败 — {e}")
|
||
continue
|
||
|
||
# 等待回应 (给板子足够的处理时间)
|
||
time.sleep(0.15)
|
||
|
||
# 读取所有可用数据
|
||
response = bytearray()
|
||
while True:
|
||
n = ser.in_waiting
|
||
if n == 0:
|
||
break
|
||
response.extend(ser.read(n))
|
||
time.sleep(0.02)
|
||
|
||
# 打印结果
|
||
resp_str = ""
|
||
if response:
|
||
resp_str = f"← 板子回应 {len(response)}B: {response.hex(' ').upper()}"
|
||
try:
|
||
ascii_str = response.decode("ascii", errors="replace").strip()
|
||
if ascii_str:
|
||
resp_str += f" [{ascii_str}]"
|
||
except Exception:
|
||
pass
|
||
|
||
print(f" [{i+1:>2}] {name}")
|
||
print(f" 发送: {frame.hex(' ').upper()}")
|
||
if resp_str:
|
||
print(f" {resp_str}")
|
||
else:
|
||
print(f" (板子无回应)")
|
||
|
||
time.sleep(0.05)
|
||
|
||
ser.close()
|
||
|
||
print("\n" + "=" * 65)
|
||
print("分析要点:")
|
||
print(" 1. 有回应的 → 可能是板子返回了 ACK/NAK/回显，说明协议接近")
|
||
print(" 2. 回应 = 发送内容 → 回显模式，说明板子期望特定格式")
|
||
print(" 3. 回应包含 'OK'/'ERROR' 等 → 文本协议")
|
||
print(" 4. 全无回应 → 要么协议全错，要么板子是单向的(不回显)")
|
||
print("\n提示: 同时也用手机摄像头看红外LED是否发光")
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════
|
||
# 模式3: 上电启动捕获
|
||
# ═══════════════════════════════════════════════════════════════
|
||
|
||
def capture_boot(port: str):
|
||
"""
|
||
打开串口后立即开始接收，看板子有没有上电启动信息。
|
||
也尝试切换 DTR (部分板子用 DTR 供电/复位)。
|
||
"""
|
||
print("上电诊断 — 捕获板子启动信息\n")
|
||
print("提示: 拔掉 USB-to-TTL 再插入，然后按 Enter 继续...")
|
||
input()
|
||
|
||
# 先用 DTR 低电平"复位"
|
||
try:
|
||
ser = serial.Serial(port, baudrate=BAUDRATE, timeout=0.1)
|
||
ser.dtr = False
|
||
time.sleep(0.5)
|
||
ser.dtr = True
|
||
print(f"已打开 {port} (DTR 复位完成)")
|
||
except serial.SerialException as e:
|
||
print(f"✗ 无法打开 {port}: {e}")
|
||
return
|
||
|
||
print("等待板子启动输出... (10s)\n")
|
||
|
||
buffer = bytearray()
|
||
start = time.time()
|
||
|
||
try:
|
||
while time.time() - start < 10:
|
||
n = ser.in_waiting
|
||
if n > 0:
|
||
data = ser.read(n)
|
||
buffer.extend(data)
|
||
# 有数据就打印
|
||
ts = time.strftime("%H:%M:%S")
|
||
print(f" [{ts}] {data.hex(' ').upper()}", end="")
|
||
try:
|
||
ascii_str = data.decode("ascii", errors="replace").strip()
|
||
if ascii_str:
|
||
print(f" [{ascii_str}]")
|
||
else:
|
||
print()
|
||
except Exception:
|
||
print()
|
||
start = time.time() # 有新数据就重置计时
|
||
else:
|
||
time.sleep(0.05)
|
||
except KeyboardInterrupt:
|
||
pass
|
||
finally:
|
||
ser.close()
|
||
|
||
if buffer:
|
||
print(f"\n启动阶段共收到 {len(buffer)} 字节:")
|
||
print(f" HEX: {buffer.hex(' ').upper()}")
|
||
try:
|
||
text = buffer.decode("ascii", errors="replace")
|
||
if text.strip():
|
||
print(f" ASCII: {text.strip()}")
|
||
except Exception:
|
||
pass
|
||
else:
|
||
print("\n无启动输出 — 板子不发送启动信息")
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════
|
||
# 综合诊断: 以上全部
|
||
# ═══════════════════════════════════════════════════════════════
|
||
|
||
def run_all(port: str):
|
||
"""运行全部诊断"""
|
||
print("=" * 65)
|
||
print(" 阶段1: 上电启动捕获")
|
||
print("=" * 65)
|
||
capture_boot(port)
|
||
|
||
print("\n" + "=" * 65)
|
||
print(" 阶段2: 详细遥控信号捕获")
|
||
print("=" * 65)
|
||
capture_remote_detail(port)
|
||
|
||
print("\n" + "=" * 65)
|
||
print(" 阶段3: 发送+监听回应")
|
||
print("=" * 65)
|
||
send_and_listen(port)
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(
|
||
description="红外板 — 收发联合诊断",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
)
|
||
parser.add_argument("-p", "--port", default=None, help="串口")
|
||
parser.add_argument("--boot", action="store_true", help="捕获上电启动信息")
|
||
parser.add_argument("--echo", action="store_true", help="发送候选协议+监听回应")
|
||
parser.add_argument("--capture", action="store_true", help="详细捕获遥控信号")
|
||
parser.add_argument("--all", action="store_true", help="运行全部诊断")
|
||
args = parser.parse_args()
|
||
|
||
port = args.port or find_port()
|
||
if not port:
|
||
print("✗ 未找到串口。请用 -p 指定。")
|
||
return
|
||
print(f"串口: {port}")
|
||
|
||
if args.all:
|
||
run_all(port)
|
||
elif args.boot:
|
||
capture_boot(port)
|
||
elif args.echo:
|
||
send_and_listen(port)
|
||
elif args.capture:
|
||
capture_remote_detail(port)
|
||
else:
|
||
# 默认: 全部运行
|
||
run_all(port)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|