测试树莓派串口,是ttyAMA0 还是ttyS0
·

打开程序后,会有这个:

简易程序判断ttyAMA0
import os
import select
import time
DEV = "/dev/ttyAMA0"
fd = os.open(
DEV,
os.O_RDWR |
os.O_NOCTTY |
os.O_NONBLOCK
)
while 1:
os.write(fd, b"\x55")
readable = select.select(
[fd], [], [], 0.2
)[0]
ok = readable and os.read(fd, 1)
if ok:
print(DEV)
else:
print("disconnect")
time.sleep(0.2)
入门之判断是否ttyAMA0回环
import os
import select
import time
d = "/dev/ttyAMA0"
if not os.path.exists(d):
print("without")
raise SystemExit
fd = os.open(
d,
os.O_RDWR |
os.O_NOCTTY |
os.O_NONBLOCK
)
while 1:
os.write(fd, b"\x55")
r = select.select(
[fd], [], [], 0.2
)[0]
if r and os.read(fd, 1):
print(d)
else:
print("disconnect")
time.sleep(0.2)
注释,行号对应上程序:
1. `import os`:导入 `os`,用于检查文件是否存在、打开设备、读写等
2. `import select`:导入 `select`,用于等待“这个 fd 是否可读”
3. `import time`:导入 `time`,用于 `sleep` 控制打印频率
4. (空行):分隔代码块,方便阅读
5. `d = "/dev/ttyAMA0"`:要测试的串口设备路径
6. (空行):分隔
7. `if not os.path.exists(d):`:如果系统里没有这个设备节点(文件)
8. ` print("without")`:打印 `without` 表示“没有这个串口”
9. ` raise SystemExit`:退出程序(默认退出码 0;需要非0可写 `SystemExit(1)`)
10. (空行):分隔
11. `fd = os.open(`:开始打开设备,返回文件描述符 `fd`
12. ` d,`:第一个参数是设备路径
13. ` os.O_RDWR |`:以读写方式打开
14. ` os.O_NOCTTY |`:不把这个串口当作控制终端
15. ` os.O_NONBLOCK`:非阻塞打开/读写(读不到数据会立刻返回)
16. `)`:结束 `os.open` 调用
17. (空行):分隔
18. `while 1:`:无限循环持续测试
19. ` os.write(fd, b"\x55")`:向串口写 1 个字节 `0x55`(用于回环测试)
20. (空行):分隔
21. ` r = select.select(`:调用 `select` 等待 fd 是否可读
22. ` [fd], [], [], 0.2`:读列表是 `[fd]`,写/异常列表空,超时 0.2 秒
23. ` )[0]`:取返回结果的第一个元素(“可读 fd 列表”)赋给 `r`
24. (空行):分隔
25. ` if r and os.read(fd, 1):`:如果 0.2 秒内变得可读,并且能读到 1 字节(非空)
26. ` print(d)`:打印设备名(表示回环收发成功)
27. ` else:`:否则(没变可读或读不到字节)
28. ` print("disconnect")`:打印 `disconnect`(表示不通/没回环数据)
29. (空行):分隔
30. ` time.sleep(0.2)`:睡眠 0.2 秒,限制循环速度,避免刷屏和占满 CPU
中文定位大白话测试程序
import os
import termios
import select
import time
import sys
要试的串口 = ("/dev/ttyS0", "/dev/ttyAMA0")
def 打开这个串口(串口名字):
串口编号 = os.open(串口名字, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
串口设置 = termios.tcgetattr(串口编号)
串口设置[4] = termios.B115200
串口设置[5] = termios.B115200
串口设置[2] = termios.CS8 | termios.CREAD | termios.CLOCAL
串口设置[3] = 0
串口设置[6][termios.VMIN] = 0
串口设置[6][termios.VTIME] = 0
termios.tcsetattr(串口编号, termios.TCSANOW, 串口设置)
return 串口编号
for 串口名字 in 要试的串口:
try:
串口编号 = 打开这个串口(串口名字)
except OSError:
print(f"\033[91m{串口名字} 打不开\033[0m")
continue
try:
while True:
os.write(串口编号, b"\x55")
有没有收到数据 = select.select([串口编号], [], [], 0.2)[0]
if 有没有收到数据:
收到的数据 = os.read(串口编号, 4096)
if 收到的数据:
print(f"\033[92m{串口名字} 回环成功\033[0m")
else:
print(f"\033[91m{串口名字} 没收到数据\033[0m")
else:
print(f"\033[91m{串口名字} 回环失败\033[0m")
time.sleep(0.2)
finally:
os.close(串口编号)
print("\033[91m两个串口都打不开\033[0m")
sys.exit(1)
UI 界面程序测试
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs
import json
import time
import psutil
import socket
import glob
import os
import threading
try:
import serial
except ImportError:
serial = None
HOST = "0.0.0.0"
PORT = 8080
LOCK = threading.Lock()
LAST_RESULTS = []
HTML = r"""<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>Raspberry Pi 串口扫描与测试</title>
<style>
body{font-family:Arial,"Microsoft YaHei";margin:20px;background:#f5f5f5;color:#222}
.box{background:#fff;border-radius:10px;padding:16px;margin-bottom:16px;box-shadow:0 2px 8px rgba(0,0,0,.08)}
button{padding:8px 14px;border:0;border-radius:6px;background:#1677ff;color:#fff;cursor:pointer}
button:hover{opacity:.9}
input{padding:8px;border:1px solid #ccc;border-radius:6px}
table{width:100%;border-collapse:collapse;background:#fff}
th,td{border-bottom:1px solid #eee;padding:10px;text-align:left;font-size:14px}
.ok{color:#0a8f3d;font-weight:bold}
.bad{color:#c62828;font-weight:bold}
pre{white-space:pre-wrap;word-break:break-all;background:#111;color:#0f0;padding:12px;border-radius:8px}
.small{font-size:13px;color:#666}
.url{font-size:18px;font-weight:bold;color:#1677ff}
</style>
</head>
<body>
<div class="box">
<h2>树莓派串口扫描与测试</h2>
<div>访问地址:<span class="url" id="open_url">加载中...</span></div>
<div class="small" id="iface_info">加载中...</div>
<br>
波特率:
<input id="baud" value="115200" style="width:120px">
测试内容:
<input id="payload" value="UART_LOOP_TEST_12345" style="width:260px">
<button onclick="loadPorts()">刷新串口</button>
<button onclick="testAll()">测试全部</button>
</div>
<div class="box">
<h3>串口列表</h3>
<table>
<thead>
<tr>
<th>设备</th>
<th>存在</th>
<th>操作</th>
</tr>
</thead>
<tbody id="ports_body"></tbody>
</table>
</div>
<div class="box">
<h3>测试结果</h3>
<pre id="log">等待操作...</pre>
</div>
<script>
function esc(s){
return String(s ?? "");
}
async function loadPorts(){
const r = await fetch('/api/ports?_=' + Date.now());
const d = await r.json();
document.getElementById('open_url').textContent = d.open_url;
document.getElementById('iface_info').textContent =
'当前接口: ' + d.best_if + ' IP: ' + d.best_ip + ' 端口: ' + d.port;
const body = document.getElementById('ports_body');
body.innerHTML = '';
if(!d.serial_ready){
body.innerHTML = '<tr><td colspan="3">未安装 pyserial,请先安装:sudo apt install -y python3-serial</td></tr>';
return;
}
if(!d.ports.length){
body.innerHTML = '<tr><td colspan="3">未发现串口设备</td></tr>';
return;
}
d.ports.forEach(p => {
const tr = document.createElement('tr');
tr.innerHTML = `
<td>${esc(p)}</td>
<td>是</td>
<td><button onclick="testOne('${esc(p)}')">测试</button></td>
`;
body.appendChild(tr);
});
}
async function testOne(port){
const baud = document.getElementById('baud').value.trim() || '115200';
const payload = encodeURIComponent(document.getElementById('payload').value);
document.getElementById('log').textContent = '正在测试 ' + port + ' ...';
const r = await fetch('/api/test?port=' + encodeURIComponent(port) + '&baud=' + baud + '&payload=' + payload + '&_=' + Date.now());
const d = await r.json();
document.getElementById('log').textContent = JSON.stringify(d, null, 2);
}
async function testAll(){
const baud = document.getElementById('baud').value.trim() || '115200';
const payload = encodeURIComponent(document.getElementById('payload').value);
document.getElementById('log').textContent = '正在测试全部串口 ...';
const r = await fetch('/api/test_all?baud=' + baud + '&payload=' + payload + '&_=' + Date.now());
const d = await r.json();
document.getElementById('log').textContent = JSON.stringify(d, null, 2);
}
loadPorts();
</script>
</body>
</html>
"""
def get_ipv4_map():
result = {}
addrs = psutil.net_if_addrs()
for ifname, addr_list in addrs.items():
for addr in addr_list:
if addr.family == socket.AF_INET and not addr.address.startswith("127."):
result[ifname] = addr.address
return result
def get_best_ip():
ip_map = get_ipv4_map()
for name in ("eth0", "wlan0", "en0"):
if name in ip_map:
return name, ip_map[name]
if ip_map:
k = list(ip_map.keys())[0]
return k, ip_map[k]
return "unknown", "0.0.0.0"
def list_serial_ports():
patterns = [
"/dev/serial0",
"/dev/ttyAMA*",
"/dev/ttyS*",
"/dev/ttyUSB*",
"/dev/ttyACM*",
]
ports = []
seen = set()
for pattern in patterns:
for p in glob.glob(pattern):
if p not in seen and os.path.exists(p):
seen.add(p)
ports.append(p)
ports.sort()
return ports
def read_back(ser, timeout=1.0):
end = time.time() + timeout
data = b""
while time.time() < end:
n = ser.in_waiting if hasattr(ser, "in_waiting") else 0
if n:
data += ser.read(n)
else:
chunk = ser.read(256)
if chunk:
data += chunk
else:
time.sleep(0.05)
return data
def test_serial_port(port, baudrate=115200, payload="UART_LOOP_TEST_12345"):
if serial is None:
return {
"port": port,
"ok": False,
"error": "pyserial not installed"
}
tx = payload.encode("utf-8", errors="ignore")
start = time.time()
try:
with LOCK:
ser = serial.Serial(
port=port,
baudrate=int(baudrate),
timeout=0.2,
write_timeout=0.5,
bytesize=serial.EIGHTBITS,
parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE,
xonxoff=False,
rtscts=False,
dsrdtr=False
)
try:
ser.reset_input_buffer()
ser.reset_output_buffer()
except Exception:
pass
ser.write(tx)
ser.flush()
time.sleep(0.15)
rx = read_back(ser, timeout=0.8)
ser.close()
ok = tx in rx or rx == tx
return {
"port": port,
"baudrate": int(baudrate),
"tx_text": payload,
"tx_hex": tx.hex(" "),
"rx_text": rx.decode("utf-8", errors="replace"),
"rx_hex": rx.hex(" "),
"ok": ok,
"msg": "loopback ok" if ok else "no echo data",
"elapsed_ms": int((time.time() - start) * 1000)
}
except Exception as e:
return {
"port": port,
"baudrate": int(baudrate),
"tx_text": payload,
"tx_hex": tx.hex(" "),
"rx_text": "",
"rx_hex": "",
"ok": False,
"msg": "open/test failed",
"error": str(e),
"elapsed_ms": int((time.time() - start) * 1000)
}
class Handler(BaseHTTPRequestHandler):
def log_message(self, format, *args):
pass
def send_json(self, obj, code=200):
data = json.dumps(obj, ensure_ascii=False, indent=2).encode("utf-8")
self.send_response(code)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)
def send_html(self, text, code=200):
data = text.encode("utf-8")
self.send_response(code)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)
def do_GET(self):
global LAST_RESULTS
parsed = urlparse(self.path)
path = parsed.path
q = parse_qs(parsed.query)
if path == "/":
self.send_html(HTML)
return
if path == "/api/ports":
best_if, best_ip = get_best_ip()
ports = list_serial_ports()
self.send_json({
"host": HOST,
"port": PORT,
"best_if": best_if,
"best_ip": best_ip,
"ip_map": get_ipv4_map(),
"open_url": f"http://{best_ip}:{PORT}",
"ports": ports,
"serial_ready": serial is not None,
"ts": int(time.time())
})
return
if path == "/api/test":
port = q.get("port", [""])[0].strip()
baud = q.get("baud", ["115200"])[0].strip()
payload = q.get("payload", ["UART_LOOP_TEST_12345"])[0]
if not port:
self.send_json({"ok": False, "error": "missing port"}, 400)
return
result = test_serial_port(port, baud, payload)
LAST_RESULTS = [result]
self.send_json(result)
return
if path == "/api/test_all":
baud = q.get("baud", ["115200"])[0].strip()
payload = q.get("payload", ["UART_LOOP_TEST_12345"])[0]
ports = list_serial_ports()
results = [test_serial_port(p, baud, payload) for p in ports]
LAST_RESULTS = results
self.send_json({
"count": len(results),
"results": results
})
return
if path == "/api/last":
self.send_json({"results": LAST_RESULTS})
return
self.send_json({"ok": False, "error": "not found"}, 404)
def main():
print("========================================")
print(" Raspberry Pi Serial Web Tester")
print("========================================")
best_if, best_ip = get_best_ip()
print("Open in browser:")
print(f" http://{best_ip}:{PORT}")
print("All IPv4:")
for k, v in get_ipv4_map().items():
print(f" {k}: {v}")
print("Listening on 0.0.0.0:%d" % PORT)
print("Need loopback: short TX <-> RX on target UART")
print("========================================")
httpd = ThreadingHTTPServer((HOST, PORT), Handler)
httpd.serve_forever()
if __name__ == "__main__":
main()
ttyAMA0 还是ttyS0 初确认
import os, termios, select, time, sys # 导入所需模块
PORTS = ("/dev/ttyAMA0", "/dev/ttyS0") # 依次测试的串口
def open_uart(dev): # 打开并配置串口
fd = os.open(dev, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK) # 非阻塞方式打开
a = termios.tcgetattr(fd) # 获取当前串口参数
a[4] = a[5] = termios.B115200 # 设置输入/输出波特率 115200
# 设置 8N1 + 允许接收 + 本地模式(不依赖硬件控制线)
a[2] = (termios.CS8 | termios.CREAD | termios.CLOCAL) # 组合 cflag
a[3] = 0 # raw 模式(尽量少处理)
a[6][termios.VMIN] = 0 # 读取最小字节数 0
a[6][termios.VTIME] = 0 # 读取超时 0(立即返回)
termios.tcsetattr(fd, termios.TCSANOW, a) # 立即应用设置
return fd # 返回 fd
for dev in PORTS: # 遍历串口
try:
fd = open_uart(dev) # 尝试打开
except OSError:
continue # 打不开就换下一个
try:
while 1: # 一直测试并滚动打印
os.write(fd, b"\x55") # 往 TXD 发 1 字节
ok = select.select([fd], [], [], 0.2)[0] and os.read(fd, 4096) # RXD 有数据就算回环
print(f"\033[92m{dev}\033[0m" if ok else "\033[91mfail\033[0m") # 成功打印设备名,否则 fail
time.sleep(0.2) # 控制打印频率
finally:
os.close(fd) # 退出时关闭 fd
print("\033[91mFAIL\033[0m") # 两个口都打不开才会到这里
sys.exit(1) # 非 0 退出码

确认有无对应的ttyAMA0 或 ttyS0 串口,确认有无回环测试中
import os, termios, select # os: 文件/设备操作,termios: 串口配置,select: 轮询等待
PORTS = ("/dev/ttyAMA0", "/dev/ttyS0") # 需要检测的串口设备列表
GREEN = "\033[92m" # 绿色 ANSI 颜色码(正常)
YELLOW = "\033[93m" # 黄色 ANSI 颜色码(异常但存在)
RED = "\033[91m" # 红色 ANSI 颜色码(设备不存在)
END = "\033[0m" # 颜色结束标记,恢复默认颜色
def open_uart(dev): # 打开并初始化串口函数
fd = os.open(dev, # 打开串口设备
os.O_RDWR # 读写模式
| os.O_NOCTTY # 不将该设备作为控制终端
| os.O_NONBLOCK) # 非阻塞方式
a = termios.tcgetattr(fd) # 读取当前串口属性
a[4] = a[5] = termios.B115200 # 设置输入和输出波特率为 115200
a[2] = (termios.CS8 # 8 位数据位
| termios.CREAD # 使能接收
| termios.CLOCAL) # 本地模式,不依赖调制解调器信号
a[3] = 0 # 关闭所有本地模式处理,接近 raw 模式
a[6][termios.VMIN] = 0 # 非阻塞读,最少读取 0 字节
a[6][termios.VTIME] = 0 # 读取超时为 0,立即返回
termios.tcsetattr(fd, # 将修改后的参数写回串口
termios.TCSANOW, # 立即生效
a)
return fd # 返回串口文件描述符
for dev in PORTS: # 依次检测每一个串口
if not os.path.exists(dev): # 如果系统中不存在该串口节点
print(f"{RED}{dev} without{END}")# 红色打印:无该串口
continue # 跳过,检测下一个
try:
fd = open_uart(dev) # 尝试打开并初始化串口
except OSError: # 打开失败(权限/占用/异常)
print(f"{YELLOW}{dev} disconnect{END}") # 黄色打印:存在但不可用
continue # 跳过,检测下一个
os.write(fd, b"\x55") # 向串口发送 0x55,用于回环测试
ok = ( # 判断是否收发成功
select.select([fd], [], [], 0.2)[0] # 0.2 秒内是否有可读数据
and os.read(fd, 1) # 读取 1 字节,能读到即认为成功
)
print( # 根据结果打印状态
f"{GREEN}{dev}{END}" # 绿色:串口存在且收发正常
if ok
else f"{YELLOW}{dev} disconnect{END}" # 黄色:存在但不通
)
os.close(fd) # 关闭串口文件描述符,释放资源

串口波特率范围测试
#!/usr/bin/env python3
"""
树莓派串口波特率测试程序
测试 ttyUSB0 和 ttyUSB1 之间的通信
"""
import serial
import time
import threading
import sys
class SerialTester:
def __init__(self):
self.serial_send = None
self.serial_recv = None
self.test_message = "Hello, this is a test message for serial communication!"
self.received_data = ""
self.test_complete = False
self.successful_rates = []
def setup_serials(self, baudrate):
"""设置发送和接收串口"""
try:
# 发送端 - ttyUSB0
self.serial_send = serial.Serial(
port='/dev/ttyUSB0',
baudrate=baudrate,
bytesize=serial.EIGHTBITS,
parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE,
timeout=1
)
# 接收端 - ttyUSB1
self.serial_recv = serial.Serial(
port='/dev/ttyUSB1',
baudrate=baudrate,
bytesize=serial.EIGHTBITS,
parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE,
timeout=1
)
# 清空缓冲区
self.serial_send.reset_input_buffer()
self.serial_send.reset_output_buffer()
self.serial_recv.reset_input_buffer()
self.serial_recv.reset_output_buffer()
return True
except Exception as e:
print(f"设置串口失败 (波特率 {baudrate}): {e}")
return False
def receiver_thread(self):
"""接收线程"""
while not self.test_complete:
if self.serial_recv and self.serial_recv.in_waiting > 0:
try:
data = self.serial_recv.read(self.serial_recv.in_waiting)
self.received_data += data.decode('utf-8', errors='ignore')
except Exception as e:
print(f"接收数据错误: {e}")
time.sleep(0.01)
def test_baudrate(self, baudrate, test_duration=2):
"""测试特定波特率"""
print(f"\n正在测试波特率: {baudrate}")
# 设置串口
if not self.setup_serials(baudrate):
return False
# 重置测试状态
self.received_data = ""
self.test_complete = False
# 启动接收线程
recv_thread = threading.Thread(target=self.receiver_thread)
recv_thread.daemon = True
recv_thread.start()
# 发送测试数据
total_bytes_sent = 0
start_time = time.time()
try:
while time.time() - start_time < test_duration:
# 发送测试消息
message = f"[{baudrate}]{self.test_message}\n"
bytes_sent = self.serial_send.write(message.encode('utf-8'))
self.serial_send.flush()
total_bytes_sent += bytes_sent
time.sleep(0.01) # 短暂延迟
# 结束测试
self.test_complete = True
time.sleep(0.5) # 等待最后的数据接收完成
# 计算统计信息
total_time = time.time() - start_time
actual_baudrate = (total_bytes_sent * 10) / total_time # 估算实际波特率
# 检查数据完整性
expected_pattern = f"[{baudrate}]{self.test_message}"
pattern_count = self.received_data.count(expected_pattern)
# 计算错误率
total_patterns_sent = total_bytes_sent // len(message)
error_rate = 0 if total_patterns_sent == 0 else (total_patterns_sent - pattern_count) / total_patterns_sent
print(f" 发送字节数: {total_bytes_sent}")
print(f" 实际波特率: {actual_baudrate:.0f}")
print(f" 数据包成功率: {pattern_count}/{total_patterns_sent}")
print(f" 错误率: {error_rate*100:.2f}%")
# 判断测试是否成功 (错误率低于5%)
success = error_rate < 0.05 and pattern_count > 0
if success:
self.successful_rates.append(baudrate)
print(f" ✓ 波特率 {baudrate} 测试成功")
else:
print(f" ✗ 波特率 {baudrate} 测试失败")
return success
except Exception as e:
print(f" 测试过程中出错: {e}")
return False
finally:
# 关闭串口
if self.serial_send:
self.serial_send.close()
if self.serial_recv:
self.serial_recv.close()
def run_comprehensive_test(self):
"""运行全面的波特率测试"""
print("=" * 60)
print("树莓派串口波特率测试程序")
print("测试 ttyUSB0 -> ttyUSB1 的通信")
print("=" * 60)
# 常见的波特率列表
baudrates = [
300, 600, 1200, 2400, 4800, 9600, 19200, 38400, 57600,
115200, 230400, 460800, 500000, 576000, 921600,
1000000, 1152000, 1500000, 2000000, 2500000, 3000000
]
print(f"将要测试的波特率: {baudrates}")
print("\n开始测试...")
# 测试每个波特率
for baudrate in baudrates:
try:
self.test_baudrate(baudrate)
except KeyboardInterrupt:
print("\n用户中断测试")
break
except Exception as e:
print(f"测试波特率 {baudrate} 时发生未知错误: {e}")
# 显示测试结果
self.display_results()
def display_results(self):
"""显示最终测试结果"""
print("\n" + "=" * 60)
print("测试完成!")
print("=" * 60)
if self.successful_rates:
max_baudrate = max(self.successful_rates)
print(f"成功测试的波特率: {self.successful_rates}")
print(f"最高可用的波特率: {max_baudrate}")
else:
print("没有成功的波特率测试!")
print("=" * 60)
def check_serial_ports():
"""检查串口设备是否存在"""
import os
ports = ['/dev/ttyUSB0', '/dev/ttyUSB1']
print("检查串口设备...")
for port in ports:
if os.path.exists(port):
print(f" ✓ {port} 存在")
else:
print(f" ✗ {port} 不存在")
return False
return True
def main():
"""主函数"""
# 检查串口设备
if not check_serial_ports():
print("\n错误: 请确保 ttyUSB0 和 ttyUSB1 设备已连接")
print("连接方法: 使用USB转TTL模块,交叉连接RX/TX")
print("ttyUSB0 TX -> ttyUSB1 RX")
print("ttyUSB0 RX -> ttyUSB1 TX")
print("ttyUSB0 GND -> ttyUSB1 GND")
sys.exit(1)
# 运行测试
tester = SerialTester()
try:
tester.run_comprehensive_test()
except KeyboardInterrupt:
print("\n测试被用户中断")
except Exception as e:
print(f"测试过程中发生错误: {e}")
if __name__ == "__main__":
main()

常见问题
FAQ 注意分串口
更多推荐







所有评论(0)