初始化
This commit is contained in:
290
ir_receive.py
Normal file
290
ir_receive.py
Normal file
@@ -0,0 +1,290 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
红外收发板 — 接收测试
|
||||
=====================
|
||||
USB转TTL 连接红外收发板,接收并显示红外信号数据。
|
||||
|
||||
用法:
|
||||
python ir_receive.py # 自动查找串口, 9600 波特率
|
||||
python ir_receive.py --ports # 仅扫描串口
|
||||
python ir_receive.py -p COM5 # 指定端口
|
||||
python ir_receive.py -p COM5 -b 115200 # 指定端口和波特率
|
||||
python ir_receive.py -p COM5 --hex-only # 仅显示十六进制
|
||||
|
||||
依赖: pip install pyserial
|
||||
"""
|
||||
|
||||
import sys
|
||||
import time
|
||||
import argparse
|
||||
import serial
|
||||
import serial.tools.list_ports
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════
|
||||
# 常见红外板串口波特率 (按优先级排列)
|
||||
# ═══════════════════════════════════════════════════════════════
|
||||
COMMON_BAUDRATES = [9600, 115200, 57600, 19200, 38400, 4800, 2400]
|
||||
|
||||
# CH34x / 常见 USB-to-TTL 芯片 VID:PID
|
||||
USB_TTL_CHIPS = {
|
||||
(0x1A86, 0x55D3): "CH343",
|
||||
(0x1A86, 0x55D4): "CH344",
|
||||
(0x1A86, 0x7523): "CH340",
|
||||
(0x1A86, 0x5523): "CH341",
|
||||
(0x1A86, 0x55D5): "CH342",
|
||||
(0x0403, 0x6001): "FT232",
|
||||
(0x067B, 0x2303): "PL2303",
|
||||
(0x10C4, 0xEA60): "CP2102",
|
||||
}
|
||||
|
||||
USB_TTL_KEYWORDS = [
|
||||
"ch343", "ch340", "ch341", "ch342", "ch344",
|
||||
"usb-enhanced-serial", "usb serial", "usb-serial",
|
||||
"ft232", "ftdi", "cp210", "pl2303", "wch",
|
||||
]
|
||||
|
||||
|
||||
def scan_ports():
|
||||
"""扫描所有串口,标注 USB-to-TTL 设备"""
|
||||
ports = serial.tools.list_ports.comports()
|
||||
results = []
|
||||
|
||||
print("=" * 70)
|
||||
print(f"{'端口':<8} {'芯片':<10} {'VID:PID':<12} {'描述'}")
|
||||
print("-" * 70)
|
||||
|
||||
for port in ports:
|
||||
vid_pid = (port.vid, port.pid) if port.vid and port.pid else None
|
||||
chip = USB_TTL_CHIPS.get(vid_pid, "")
|
||||
if not chip:
|
||||
combined = f"{port.description or ''} {port.hwid or ''}".lower()
|
||||
for kw in USB_TTL_KEYWORDS:
|
||||
if kw in combined:
|
||||
chip = "USB-TTL"
|
||||
break
|
||||
|
||||
info = {
|
||||
"device": port.device,
|
||||
"description": port.description or "",
|
||||
"vid": port.vid,
|
||||
"pid": port.pid,
|
||||
"chip": chip or None,
|
||||
}
|
||||
results.append(info)
|
||||
|
||||
vid_str = f"{port.vid:04X}:{port.pid:04X}" if vid_pid else "N/A"
|
||||
chip_str = chip or "-"
|
||||
print(f"{port.device:<8} {chip_str:<10} {vid_str:<12} {port.description or ''}")
|
||||
|
||||
print("=" * 70)
|
||||
return results
|
||||
|
||||
|
||||
def find_usb_ttl_port():
|
||||
"""自动查找 USB-to-TTL 串口"""
|
||||
ports = serial.tools.list_ports.comports()
|
||||
candidates = []
|
||||
|
||||
for port in ports:
|
||||
combined = f"{port.description or ''} {port.hwid or ''}".lower()
|
||||
if any(kw in combined for kw in USB_TTL_KEYWORDS):
|
||||
candidates.append(port)
|
||||
|
||||
if candidates:
|
||||
p = candidates[0]
|
||||
print(f"→ 自动选择: {p.device} — {p.description}")
|
||||
return p.device
|
||||
|
||||
# 回退: 返回第一个可用端口
|
||||
all_ports = list(ports)
|
||||
if all_ports:
|
||||
p = all_ports[0]
|
||||
print(f"→ 回退使用: {p.device} — {p.description}")
|
||||
return p.device
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def try_connect(port: str, baudrate: int, timeout: float = 1.0) -> serial.Serial | None:
|
||||
"""尝试以指定波特率打开串口"""
|
||||
try:
|
||||
ser = serial.Serial(
|
||||
port=port,
|
||||
baudrate=baudrate,
|
||||
bytesize=serial.EIGHTBITS,
|
||||
parity=serial.PARITY_NONE,
|
||||
stopbits=serial.STOPBITS_ONE,
|
||||
timeout=timeout,
|
||||
)
|
||||
return ser
|
||||
except serial.SerialException:
|
||||
return None
|
||||
|
||||
|
||||
def format_hex(data: bytes) -> str:
|
||||
"""格式化十六进制显示"""
|
||||
return data.hex(" ").upper()
|
||||
|
||||
|
||||
def format_binary(data: bytes) -> str:
|
||||
"""格式化二进制显示"""
|
||||
return " ".join(f"{b:08b}" for b in data)
|
||||
|
||||
|
||||
def run_receiver(port: str, baudrate: int, hex_only: bool = False):
|
||||
"""
|
||||
主接收循环 — 持续读取串口数据并显示。
|
||||
按 Ctrl+C 退出。
|
||||
"""
|
||||
print(f"\n打开串口 {port} @ {baudrate} baud...")
|
||||
ser = try_connect(port, baudrate, timeout=0.5)
|
||||
if ser is None:
|
||||
print(f"✗ 无法打开 {port}")
|
||||
return
|
||||
|
||||
print(f"✓ 已连接。等待红外信号... (按 Ctrl+C 退出)\n")
|
||||
print("-" * 70)
|
||||
|
||||
buffer = bytearray()
|
||||
last_print_time = time.time()
|
||||
IDLE_PRINT_THRESHOLD = 0.15 # 空闲超过此秒数则打印之前累积的数据
|
||||
|
||||
try:
|
||||
while True:
|
||||
try:
|
||||
waiting = ser.in_waiting
|
||||
except (serial.SerialException, OSError) as e:
|
||||
print(f"\n✗ 串口断开: {e}")
|
||||
break
|
||||
|
||||
if waiting > 0:
|
||||
try:
|
||||
chunk = ser.read(waiting)
|
||||
except serial.SerialException:
|
||||
break
|
||||
buffer.extend(chunk)
|
||||
last_print_time = time.time()
|
||||
else:
|
||||
# 空闲时: 如果缓冲区有数据且空闲足够久,打印
|
||||
if buffer and (time.time() - last_print_time) > IDLE_PRINT_THRESHOLD:
|
||||
data = bytes(buffer)
|
||||
timestamp = time.strftime("%H:%M:%S")
|
||||
|
||||
if hex_only:
|
||||
print(f"[{timestamp}] ({len(data)}B) {format_hex(data)}")
|
||||
else:
|
||||
print(f"[{timestamp}] 收到 {len(data)} 字节")
|
||||
print(f" HEX: {format_hex(data)}")
|
||||
print(f" BIN: {format_binary(data)}")
|
||||
# 尝试 ASCII 解码
|
||||
try:
|
||||
ascii_str = data.decode("ascii")
|
||||
if ascii_str.isprintable():
|
||||
print(f" ASCII: {ascii_str.strip()}")
|
||||
except UnicodeDecodeError:
|
||||
pass
|
||||
print()
|
||||
|
||||
buffer.clear()
|
||||
|
||||
time.sleep(0.02)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\n\n用户中断")
|
||||
|
||||
finally:
|
||||
ser.close()
|
||||
print(f"✓ 已关闭 {port}")
|
||||
|
||||
|
||||
def auto_baudrate_test(port: str):
|
||||
"""
|
||||
波特率自动探测 — 提示用户按红外遥控器,
|
||||
在不同波特率下监听,看哪个能收到有效数据。
|
||||
"""
|
||||
print("\n" + "=" * 70)
|
||||
print(" 波特率自动探测")
|
||||
print("=" * 70)
|
||||
print("请持续按红外遥控器任意键...\n")
|
||||
|
||||
for baud in COMMON_BAUDRATES:
|
||||
print(f" 测试 {baud:>6} baud ... ", end="", flush=True)
|
||||
ser = try_connect(port, baud, timeout=0.3)
|
||||
if ser is None:
|
||||
print("无法打开")
|
||||
continue
|
||||
|
||||
# 等待一小段时间收集数据
|
||||
received = bytearray()
|
||||
deadline = time.time() + 2.0
|
||||
while time.time() < deadline:
|
||||
try:
|
||||
n = ser.in_waiting
|
||||
if n > 0:
|
||||
received.extend(ser.read(n))
|
||||
except serial.SerialException:
|
||||
break
|
||||
time.sleep(0.05)
|
||||
|
||||
ser.close()
|
||||
|
||||
if received:
|
||||
print(f"收到 {len(received)} 字节! → {format_hex(bytes(received))}")
|
||||
else:
|
||||
print("无数据")
|
||||
|
||||
print("\n提示: 有数据的波特率即为正确波特率。")
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="红外收发板 — 接收测试",
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
epilog="""
|
||||
示例:
|
||||
%(prog)s 自动查找串口, 9600bps, 持续接收
|
||||
%(prog)s --ports 仅扫描串口
|
||||
%(prog)s -p COM5 指定端口接收
|
||||
%(prog)s -p COM5 -b 115200 指定端口和波特率
|
||||
%(prog)s -p COM5 --auto-baud 自动探测波特率
|
||||
%(prog)s -p COM5 --hex-only 仅显示十六进制
|
||||
""",
|
||||
)
|
||||
parser.add_argument("-p", "--port", default=None, help="串口 (如 COM5)")
|
||||
parser.add_argument("-b", "--baudrate", type=int, default=9600, help="波特率 (默认 9600)")
|
||||
parser.add_argument("--ports", action="store_true", help="仅扫描串口")
|
||||
parser.add_argument("--auto-baud", action="store_true", help="自动探测波特率")
|
||||
parser.add_argument("--hex-only", action="store_true", help="仅显示十六进制")
|
||||
args = parser.parse_args()
|
||||
|
||||
# 仅扫描
|
||||
if args.ports:
|
||||
scan_ports()
|
||||
return
|
||||
|
||||
# 确定端口
|
||||
port = args.port
|
||||
if port is None:
|
||||
print("未指定端口,自动扫描 USB-to-TTL 设备...\n")
|
||||
scan_ports()
|
||||
port = find_usb_ttl_port()
|
||||
if port is None:
|
||||
print("\n✗ 未找到任何串口。请检查:")
|
||||
print(" 1. USB-to-TTL 是否已插入")
|
||||
print(" 2. 设备管理器中是否出现 COM 口")
|
||||
print(" 3. 驱动是否已安装")
|
||||
return
|
||||
print()
|
||||
|
||||
# 自动探测波特率
|
||||
if args.auto_baud:
|
||||
auto_baudrate_test(port)
|
||||
return
|
||||
|
||||
# 持续接收
|
||||
run_receiver(port, args.baudrate, hex_only=args.hex_only)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user