From: 孙春阳 Date: Fri, 10 Oct 2025 01:15:50 +0000 (+0800) Subject: 添加 ppp 串口测试功能,检验串口读的性能,目前看来中断模式效果最好。但是1M的情况下,还是会有丢数据的情况出现。 X-Git-Tag: V1.0.9_20251128_194606_LN~276 X-Git-Url: http://git.snicker.com.cn/?a=commitdiff_plain;h=3a8b80c3c4cac3f4cc46c4aa97c5118581f0711f;p=mirrors%2FFD230.git 添加 ppp 串口测试功能,检验串口读的性能,目前看来中断模式效果最好。但是1M的情况下,还是会有丢数据的情况出现。 --- diff --git a/code/sdk/drivers/chips/brandy/porting/uart/uart_porting.c b/code/sdk/drivers/chips/brandy/porting/uart/uart_porting.c index 35e02eca2..7399f2b8b 100644 --- a/code/sdk/drivers/chips/brandy/porting/uart/uart_porting.c +++ b/code/sdk/drivers/chips/brandy/porting/uart/uart_porting.c @@ -41,13 +41,10 @@ #define UART_RX_BUFFER_SIZE 1 #else -#if defined(SUPPORT_PPP_SERVICE) -#define UART_RX_BUFFER_SIZE 2048 -// #warning "UART_RX_BUFFER_SIZE is 2048" -#else -#define UART_RX_BUFFER_SIZE 56 -// #warning "UART_RX_BUFFER_SIZE is 56" -#endif +/** + * UART_RX_BUFFER SIZE 似乎只能是 56 更多没用 + */ +#define UART_RX_BUFFER_SIZE 56 #endif diff --git a/code/sdk/zhihui/ppp/doc/serial_test/SERIAL_TEST_USAGE.md b/code/sdk/zhihui/ppp/doc/serial_test/SERIAL_TEST_USAGE.md new file mode 100644 index 000000000..14ec67aea --- /dev/null +++ b/code/sdk/zhihui/ppp/doc/serial_test/SERIAL_TEST_USAGE.md @@ -0,0 +1,335 @@ +# 串口测试工具使用说明 + +## 主要功能 + +### ✅ 优雅的测试停止机制 + +**新特性:先停发送,延时后再停接收** + +测试结束时的执行顺序: +1. 🛑 停止发送线程 +2. ⏱️ 等待1秒(让最后的数据有时间传输) +3. 🛑 停止接收线程 +4. 📊 打印最终统计 + +这样可以确保最后发送的数据包能够被完整接收,避免误报丢包。 + +### 示例输出 + +```bash +$ python3 serial_test.py -t /dev/ttyUSB0 -r /dev/ttyUSB1 -d 10 -i 100 -s 64 -b 1000000 + +====================================================================== + 串口测试程序 +====================================================================== +发送串口: /dev/ttyUSB0 +接收串口: /dev/ttyUSB1 +波特率: 1000000 +测试时长: 10 秒 +发送间隔: 100 毫秒 +包大小: 64 字节 +====================================================================== + +打开发送串口: /dev/ttyUSB0 @ 1000000 +打开接收串口: /dev/ttyUSB1 @ 1000000 +串口打开成功! + +发送线程启动 (间隔: 100.00ms, 包大小: 64字节) +接收线程启动 + +测试开始,运行时长: 10 秒 +按 Ctrl+C 可以提前终止测试 + +====================================================================== +统计时间: 2025-10-09 14:23:45 +运行时长: 10.05 秒 +---------------------------------------------------------------------- +发送: 100 包, 6,400 字节, 636.82 字节/秒 +接收: 100 包, 6,400 字节, 636.80 字节/秒 +错误: 0 次 (数据不连续: 0) +丢包率: 0.00% +====================================================================== + +正在停止发送... # ← 先停发送 +发送线程已停止 +等待1秒以接收剩余数据... # ← 延时1秒 +正在停止接收... # ← 再停接收 +接收线程已停止 + +###################################################################### +# 最终统计信息 # +###################################################################### + +====================================================================== +统计时间: 2025-10-09 14:23:56 +运行时长: 11.08 秒 +---------------------------------------------------------------------- +发送: 100 包, 6,400 字节, 577.62 字节/秒 +接收: 100 包, 6,400 字节, 577.60 字节/秒 +错误: 0 次 (数据不连续: 0) +丢包率: 0.00% +====================================================================== + +发送串口已关闭 +接收串口已关闭 +日志文件已关闭 +``` + +## 使用场景 + +### 1. 基本测试 + +```bash +python3 serial_test.py \ + -t /dev/ttyUSB0 \ + -r /dev/ttyUSB1 \ + -d 60 \ + -i 100 \ + -s 64 \ + -b 1000000 +``` + +### 2. 保存日志文件 + +```bash +python3 serial_test.py \ + -t /dev/ttyUSB0 \ + -r /dev/ttyUSB1 \ + -d 60 \ + -i 100 \ + -s 64 \ + -b 1000000 \ + --tx-log tx_data.log \ + --rx-log rx_data.log +``` + +### 3. 测试完成后对比 + +```bash +# 运行测试 +python3 serial_test.py \ + -t /dev/ttyUSB0 -r /dev/ttyUSB1 \ + -d 30 -i 50 -s 128 -b 1000000 \ + --tx-log tx.log --rx-log rx.log + +# 对比日志 +python3 compare_logs.py tx.log rx.log +``` + +## 优雅停止的优势 + +### 问题场景:没有延时时 +``` +时间轴: + 0.0s: 发送最后一个包 + 0.0s: 同时停止发送和接收线程 + 结果: 最后一个包可能还在传输中就停止接收 ❌ +``` + +### 改进后:有延时 +``` +时间轴: + 0.0s: 发送最后一个包 + 0.0s: 停止发送线程 + 0.1s: 数据在串口传输中... + 0.2s: 接收线程接收到最后的数据 + 1.0s: 停止接收线程 + 结果: 所有数据都被正确接收 ✅ +``` + +## 参数说明 + +| 参数 | 简写 | 必需 | 说明 | 示例 | +|------|------|------|------|------| +| `--tx-port` | `-t` | ✅ | 发送串口设备 | `/dev/ttyUSB0` | +| `--rx-port` | `-r` | ✅ | 接收串口设备 | `/dev/ttyUSB1` | +| `--duration` | `-d` | ✅ | 测试时长(秒) | `60` | +| `--interval` | `-i` | ✅ | 发送间隔(毫秒) | `100` | +| `--size` | `-s` | ✅ | 每包数据大小(字节) | `64` | +| `--baudrate` | `-b` | ❌ | 波特率(默认115200) | `1000000` | +| `--tx-log` | | ❌ | 发送日志文件 | `tx_data.log` | +| `--rx-log` | | ❌ | 接收日志文件 | `rx_data.log` | + +## 实时统计 + +测试运行时会: +- ✅ 每10秒自动打印统计信息 +- ✅ 发现数据不连续时立即打印错误 +- ✅ 测试结束时打印最终统计 +- ✅ 可以按 `Ctrl+C` 提前终止测试(同样会优雅停止) + +## 统计信息说明 + +``` +发送: 100 包, 6,400 字节, 636.82 字节/秒 + └─ 包数 └─ 总字节 └─ 平均速率 + +接收: 100 包, 6,400 字节, 636.80 字节/秒 + └─ 包数 └─ 总字节 └─ 平均速率 + +错误: 0 次 (数据不连续: 0) + └─ 总错误 └─ 不连续错误 + +丢包率: 0.00% + └─ (发送字节 - 接收字节) / 发送字节 * 100 +``` + +## 日志文件格式 + +生成的日志文件每行一个字节(十六进制大写): + +**tx_data.log** +``` +00 +01 +02 +03 +... +FE +FF +00 ← 循环 +``` + +**rx_data.log** +``` +00 +01 +02 +03 +... +``` + +## 常见问题 + +### Q1: 为什么要延时1秒? + +**A:** 串口数据传输需要时间: +- 1Mbps波特率下,64字节数据传输约需 **0.5ms** +- 但系统调度、缓冲区处理可能需要 **几十到几百毫秒** +- 1秒的延时足够覆盖绝大多数场景 + +### Q2: 可以调整延时时间吗? + +**A:** 当前是硬编码1秒。如需调整,可以修改 `serial_test.py` 第274行: +```python +time.sleep(1) # 改为其他值,如 0.5 或 2 +``` + +### Q3: 高速传输时会丢数据吗? + +**A:** 取决于: +- 串口驱动缓冲区大小(通常4KB~64KB) +- 发送间隔和包大小 +- 系统负载 + +建议: +- 高速传输时适当增加发送间隔 +- 使用日志文件保存数据,测试完成后对比 + +### Q4: 为什么最后接收可能比发送多? + +**A:** 这是不可能的。如果出现这种情况,可能是: +- Bug(请报告) +- 统计计数错误 + +## 测试技巧 + +### 1. 压力测试 +```bash +# 高频小包(每10ms发送64字节) +python3 serial_test.py -t /dev/ttyUSB0 -r /dev/ttyUSB1 -d 300 -i 10 -s 64 -b 1000000 + +# 低频大包(每500ms发送2048字节) +python3 serial_test.py -t /dev/ttyUSB0 -r /dev/ttyUSB1 -d 300 -i 500 -s 2048 -b 1000000 +``` + +### 2. 稳定性测试 +```bash +# 长时间运行(1小时) +python3 serial_test.py -t /dev/ttyUSB0 -r /dev/ttyUSB1 -d 3600 -i 100 -s 128 -b 1000000 +``` + +### 3. 回环测试 +```bash +# 设备端开启 PPP_DO_SERIAL_TEST +# PC端使用同一串口收发 +python3 serial_test.py \ + -t /dev/ttyUSB0 \ + -r /dev/ttyUSB0 \ + -d 60 -i 100 -s 64 -b 1000000 \ + --tx-log tx.log --rx-log rx.log +``` + +## 依赖安装 + +```bash +# Ubuntu/Debian +sudo apt-get install python3-serial + +# 或使用 pip +pip3 install pyserial +``` + +## 权限问题 + +Linux下可能需要串口权限: +```bash +# 方法1:临时授权 +sudo chmod 666 /dev/ttyUSB0 + +# 方法2:将用户加入dialout组(推荐) +sudo usermod -a -G dialout $USER +# 重新登录后生效 +``` + +## 组合使用 + +完整的测试流程: + +```bash +#!/bin/bash +# 串口完整测试脚本 + +DATE=$(date +%Y%m%d_%H%M%S) +TX_LOG="tx_${DATE}.log" +RX_LOG="rx_${DATE}.log" +RESULT="result_${DATE}.txt" + +echo "开始串口测试..." + +# 1. 运行测试 +python3 serial_test.py \ + -t /dev/ttyUSB0 \ + -r /dev/ttyUSB1 \ + -d 60 \ + -i 100 \ + -s 64 \ + -b 1000000 \ + --tx-log "$TX_LOG" \ + --rx-log "$RX_LOG" \ + | tee test_output.txt + +# 2. 对比日志 +echo -e "\n\n对比收发数据..." +python3 compare_logs.py "$TX_LOG" "$RX_LOG" | tee "$RESULT" + +# 3. 检查结果 +if grep -q "✅ 完美匹配" "$RESULT"; then + echo -e "\n✅ 测试通过!" + exit 0 +else + echo -e "\n❌ 测试失败!" + exit 1 +fi +``` + +## 总结 + +优雅停止机制确保: +- ✅ 最后的数据包不会丢失 +- ✅ 统计信息更准确 +- ✅ 测试结果更可靠 +- ✅ 适用于各种波特率和数据量 + +是串口测试的重要改进!🎯 + diff --git a/code/sdk/zhihui/ppp/doc/serial_test/compare_logs.py b/code/sdk/zhihui/ppp/doc/serial_test/compare_logs.py new file mode 100644 index 000000000..903e13b7f --- /dev/null +++ b/code/sdk/zhihui/ppp/doc/serial_test/compare_logs.py @@ -0,0 +1,205 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +串口日志对比工具 +用于对比发送日志和接收日志,查找差异 +""" + +import sys +import argparse + +def load_hex_log(filename): + """加载十六进制日志文件""" + data = [] + try: + with open(filename, 'r') as f: + for line_num, line in enumerate(f, 1): + line = line.strip() + if line: + try: + # 解析十六进制值 + byte_val = int(line, 16) + data.append((line_num, byte_val)) + except ValueError: + print(f"警告: {filename}:{line_num} 无效的十六进制值: {line}") + return data + except FileNotFoundError: + print(f"错误: 文件不存在: {filename}") + return None + except Exception as e: + print(f"错误: 读取文件失败: {e}") + return None + +def find_missing_ranges(tx_data, rx_data): + """ + 找出发送了但没收到的数据范围 + 使用LCS(最长公共子序列)思想来对齐数据 + """ + tx_len = len(tx_data) + rx_len = len(rx_data) + + # 创建接收数据的字典,方便快速查找 + rx_dict = {rx_data[i][1]: i for i in range(rx_len)} + + # 简单策略:逐个对比,找出丢失的字节 + missing_ranges = [] + current_range_start = None + current_range_data = [] + + rx_index = 0 + + for tx_index in range(tx_len): + tx_line, tx_byte = tx_data[tx_index] + + # 检查这个字节是否在接收数据中(从当前位置开始) + found = False + if rx_index < rx_len: + rx_line, rx_byte = rx_data[rx_index] + if tx_byte == rx_byte: + # 匹配,移动到下一个接收位置 + found = True + rx_index += 1 + + # 如果之前有未收到的范围,现在结束它 + if current_range_start is not None: + missing_ranges.append({ + 'start': current_range_start, + 'end': tx_index - 1, + 'count': len(current_range_data), + 'data': current_range_data + }) + current_range_start = None + current_range_data = [] + + if not found: + # 这个字节没有收到 + if current_range_start is None: + current_range_start = tx_index + current_range_data.append((tx_line, tx_byte)) + + # 处理最后一个范围 + if current_range_start is not None: + missing_ranges.append({ + 'start': current_range_start, + 'end': tx_len - 1, + 'count': len(current_range_data), + 'data': current_range_data + }) + + return missing_ranges + +def compare_logs(tx_data, rx_data, max_errors=100): + """对比发送和接收数据""" + tx_len = len(tx_data) + rx_len = len(rx_data) + + print("="*70) + print(" "*25 + "日志对比结果") + print("="*70) + print(f"发送字节数: {tx_len}") + print(f"接收字节数: {rx_len}") + print(f"丢失字节数: {tx_len - rx_len}") + + if tx_len > 0: + loss_rate = (tx_len - rx_len) / tx_len * 100 + print(f"丢包率: {loss_rate:.2f}%") + print("-"*70) + + # 找出丢失的数据范围 + missing_ranges = find_missing_ranges(tx_data, rx_data) + + if not missing_ranges: + print("\n✅ 完美匹配!所有数据都收到了。") + print("="*70) + return True + + # 显示丢失的数据范围 + print(f"\n发现 {len(missing_ranges)} 个丢失区域:\n") + + total_missing = 0 + shown_ranges = 0 + + for i, range_info in enumerate(missing_ranges): + if shown_ranges >= max_errors: + remaining = len(missing_ranges) - shown_ranges + print(f"\n... (还有 {remaining} 个丢失区域未显示)") + break + + start_pos = range_info['start'] + end_pos = range_info['end'] + count = range_info['count'] + data = range_info['data'] + + total_missing += count + shown_ranges += 1 + + if count == 1: + # 单个字节丢失 + line, byte_val = data[0] + print(f"❌ 位置 {start_pos}: 丢失 1 字节 = 0x{byte_val:02X} (发送行{line})") + else: + # 连续多个字节丢失 + print(f"❌ 位置 {start_pos}~{end_pos}: 丢失 {count} 字节") + + # 显示前8个和后8个字节(如果数据较多) + if count <= 16: + # 数据不多,全部显示 + hex_str = ' '.join([f"{byte_val:02X}" for _, byte_val in data]) + print(f" 数据: {hex_str}") + else: + # 数据太多,只显示首尾 + head = ' '.join([f"{byte_val:02X}" for _, byte_val in data[:8]]) + tail = ' '.join([f"{byte_val:02X}" for _, byte_val in data[-8:]]) + print(f" 数据: {head} ... {tail}") + + # 添加空行分隔不同的范围 + if i < len(missing_ranges) - 1: + print() + + print("-"*70) + print(f"总计丢失: {total_missing} 字节") + + if tx_len != rx_len: + print(f"长度差异: {tx_len - rx_len} 字节") + + print("\n❌ 发现数据丢失!") + print("="*70) + + return False + +def main(): + parser = argparse.ArgumentParser( + description='串口日志对比工具 - 对比发送和接收的十六进制日志文件', + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +示例: + %(prog)s tx.log rx.log + %(prog)s tx.log rx.log --max-errors 50 + """ + ) + + parser.add_argument('tx_log', help='发送日志文件') + parser.add_argument('rx_log', help='接收日志文件') + parser.add_argument('--max-errors', type=int, default=100, + help='最多显示的错误数量 (默认: 100)') + + args = parser.parse_args() + + print(f"加载发送日志: {args.tx_log}") + tx_data = load_hex_log(args.tx_log) + if tx_data is None: + return 1 + + print(f"加载接收日志: {args.rx_log}") + rx_data = load_hex_log(args.rx_log) + if rx_data is None: + return 1 + + print() + success = compare_logs(tx_data, rx_data, args.max_errors) + + return 0 if success else 1 + +if __name__ == '__main__': + sys.exit(main()) + diff --git a/code/sdk/zhihui/ppp/doc/serial_test/serial_test.py b/code/sdk/zhihui/ppp/doc/serial_test/serial_test.py new file mode 100644 index 000000000..3af7a48fa --- /dev/null +++ b/code/sdk/zhihui/ppp/doc/serial_test/serial_test.py @@ -0,0 +1,374 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +串口测试程序 +功能: +- 支持双串口发送和接收测试 +- 数据按照0x00~0xFF循环发送 +- 检测接收数据连续性 +- 统计错误和吞吐量 +""" + +import serial +import threading +import time +import argparse +import sys +from datetime import datetime + +class SerialTestStats: + """统计信息类""" + def __init__(self, tx_log_file=None, rx_log_file=None): + self.tx_bytes = 0 # 发送字节数 + self.rx_bytes = 0 # 接收字节数 + self.tx_packets = 0 # 发送包数 + self.rx_packets = 0 # 接收包数 + self.error_count = 0 # 错误计数 + self.discontinuity_count = 0 # 不连续计数 + self.last_expected_byte = None # 期望的下一个字节 + self.lock = threading.Lock() + self.start_time = time.time() + + # 日志文件 + self.tx_log_file = None + self.rx_log_file = None + + # 打开日志文件 + if tx_log_file: + try: + self.tx_log_file = open(tx_log_file, 'w') + print(f"发送数据日志: {tx_log_file}") + except Exception as e: + print(f"警告: 无法创建发送日志文件: {e}") + + if rx_log_file: + try: + self.rx_log_file = open(rx_log_file, 'w') + print(f"接收数据日志: {rx_log_file}") + except Exception as e: + print(f"警告: 无法创建接收日志文件: {e}") + + def update_tx(self, data): + """更新发送统计并记录到日志""" + with self.lock: + self.tx_bytes += len(data) + self.tx_packets += 1 + + # 记录到日志文件(每个字节一行) + if self.tx_log_file: + for byte in data: + self.tx_log_file.write(f"{byte:02X}\n") + self.tx_log_file.flush() # 实时刷新 + + def update_rx(self, data): + """更新接收统计并检查连续性""" + with self.lock: + self.rx_bytes += len(data) + self.rx_packets += 1 + + # 记录到日志文件(每个字节一行) + if self.rx_log_file: + for byte in data: + self.rx_log_file.write(f"{byte:02X}\n") + self.rx_log_file.flush() # 实时刷新 + + # 检查数据连续性 + for byte in data: + if self.last_expected_byte is None: + # 第一个字节,初始化 + self.last_expected_byte = (byte + 1) % 256 + else: + if byte != self.last_expected_byte: + self.error_count += 1 + self.discontinuity_count += 1 + print(f"\n[ERROR] 数据不连续! 期望: 0x{self.last_expected_byte:02X}, 实际: 0x{byte:02X}") + self.last_expected_byte = (byte + 1) % 256 + + def close(self): + """关闭日志文件""" + if self.tx_log_file: + self.tx_log_file.close() + self.tx_log_file = None + if self.rx_log_file: + self.rx_log_file.close() + self.rx_log_file = None + + def print_stats(self): + """打印统计信息""" + with self.lock: + elapsed = time.time() - self.start_time + tx_rate = self.tx_bytes / elapsed if elapsed > 0 else 0 + rx_rate = self.rx_bytes / elapsed if elapsed > 0 else 0 + + print("\n" + "="*70) + print(f"统计时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") + print(f"运行时长: {elapsed:.2f} 秒") + print("-"*70) + print(f"发送: {self.tx_packets:,} 包, {self.tx_bytes:,} 字节, {tx_rate:.2f} 字节/秒") + print(f"接收: {self.rx_packets:,} 包, {self.rx_bytes:,} 字节, {rx_rate:.2f} 字节/秒") + print(f"错误: {self.error_count:,} 次 (数据不连续: {self.discontinuity_count:,})") + if self.tx_bytes > 0: + loss_rate = (self.tx_bytes - self.rx_bytes) / self.tx_bytes * 100 + print(f"丢包率: {loss_rate:.2f}%") + print("="*70 + "\n") + +class SerialTester: + """串口测试器""" + def __init__(self, tx_port, rx_port, duration, interval_ms, packet_size, baudrate=115200, + tx_log_file=None, rx_log_file=None): + self.tx_port = tx_port + self.rx_port = rx_port + self.duration = duration + self.interval = interval_ms / 1000.0 # 转换为秒 + self.packet_size = packet_size + self.baudrate = baudrate + self.stats = SerialTestStats(tx_log_file, rx_log_file) + self.running = False + self.tx_running = False # 发送线程运行标志 + self.rx_running = False # 接收线程运行标志 + self.tx_serial = None + self.rx_serial = None + self.current_byte = 0 # 当前要发送的字节值 + + def open_ports(self): + """打开串口""" + try: + print(f"打开发送串口: {self.tx_port} @ {self.baudrate}") + self.tx_serial = serial.Serial( + port=self.tx_port, + baudrate=self.baudrate, + bytesize=serial.EIGHTBITS, + parity=serial.PARITY_NONE, + stopbits=serial.STOPBITS_ONE, + timeout=1 + ) + + print(f"打开接收串口: {self.rx_port} @ {self.baudrate}") + self.rx_serial = serial.Serial( + port=self.rx_port, + baudrate=self.baudrate, + bytesize=serial.EIGHTBITS, + parity=serial.PARITY_NONE, + stopbits=serial.STOPBITS_ONE, + timeout=1 + ) + + print("串口打开成功!\n") + return True + + except serial.SerialException as e: + print(f"打开串口失败: {e}") + return False + + def close_ports(self): + """关闭串口和日志文件""" + if self.tx_serial and self.tx_serial.is_open: + self.tx_serial.close() + print("发送串口已关闭") + + if self.rx_serial and self.rx_serial.is_open: + self.rx_serial.close() + print("接收串口已关闭") + + # 关闭日志文件 + self.stats.close() + print("日志文件已关闭") + + def generate_packet(self): + """生成测试数据包""" + packet = bytearray() + for i in range(self.packet_size): + packet.append(self.current_byte) + self.current_byte = (self.current_byte + 1) % 256 + return bytes(packet) + + def tx_thread(self): + """发送线程""" + print(f"发送线程启动 (间隔: {self.interval*1000:.2f}ms, 包大小: {self.packet_size}字节)") + + while self.tx_running: + try: + # 生成并发送数据 + packet = self.generate_packet() + self.tx_serial.write(packet) + self.stats.update_tx(packet) + + # 等待指定间隔 + time.sleep(self.interval) + + except serial.SerialException as e: + print(f"\n[ERROR] 发送错误: {e}") + break + except Exception as e: + print(f"\n[ERROR] 发送异常: {e}") + break + + print("发送线程退出") + + def rx_thread(self): + """接收线程""" + print(f"接收线程启动") + + while self.rx_running: + try: + # 检查是否有数据可读 + if self.rx_serial.in_waiting > 0: + data = self.rx_serial.read(self.rx_serial.in_waiting) + self.stats.update_rx(data) + else: + time.sleep(0.01) # 短暂休眠避免CPU占用过高 + + except serial.SerialException as e: + print(f"\n[ERROR] 接收错误: {e}") + break + except Exception as e: + print(f"\n[ERROR] 接收异常: {e}") + break + + print("接收线程退出") + + def stats_thread(self): + """统计打印线程""" + while self.running: + time.sleep(10) + if self.running: + self.stats.print_stats() + + def run(self): + """运行测试""" + if not self.open_ports(): + return False + + # 启动所有线程 + self.running = True + self.tx_running = True + self.rx_running = True + + # 创建线程 + tx_t = threading.Thread(target=self.tx_thread, name="TX-Thread") + rx_t = threading.Thread(target=self.rx_thread, name="RX-Thread") + stats_t = threading.Thread(target=self.stats_thread, name="Stats-Thread") + + # 启动线程 + tx_t.start() + rx_t.start() + stats_t.start() + + print(f"\n测试开始,运行时长: {self.duration} 秒") + print("按 Ctrl+C 可以提前终止测试\n") + + try: + # 等待指定时长 + time.sleep(self.duration) + except KeyboardInterrupt: + print("\n\n收到中断信号,正在停止测试...") + + # 优雅地停止测试:先停发送,等1秒后再停接收 + print("\n正在停止发送...") + self.tx_running = False + tx_t.join(timeout=5) + print("发送线程已停止") + + # 等待1秒,让接收线程接收最后的数据 + print("等待1秒以接收剩余数据...") + time.sleep(1) + + print("正在停止接收...") + self.rx_running = False + rx_t.join(timeout=5) + print("接收线程已停止") + + # 停止统计线程 + self.running = False + stats_t.join(timeout=2) + + # 打印最终统计 + print("\n" + "#"*70) + print("#" + " "*25 + "最终统计信息" + " "*25 + "#") + print("#"*70) + self.stats.print_stats() + + # 关闭串口 + self.close_ports() + + return True + +def main(): + parser = argparse.ArgumentParser( + description='串口测试程序 - 支持双串口收发测试', + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +示例: + %(prog)s -t /dev/ttyUSB0 -r /dev/ttyUSB1 -d 60 -i 100 -s 64 + %(prog)s --tx-port COM1 --rx-port COM2 --duration 120 --interval 50 --size 128 --baudrate 9600 + %(prog)s -t /dev/ttyUSB0 -r /dev/ttyUSB1 -d 60 -i 100 -s 64 --tx-log tx.log --rx-log rx.log + """ + ) + + parser.add_argument('-t', '--tx-port', required=True, + help='发送串口 (例如: /dev/ttyUSB0 或 COM1)') + parser.add_argument('-r', '--rx-port', required=True, + help='接收串口 (例如: /dev/ttyUSB1 或 COM2)') + parser.add_argument('-d', '--duration', type=int, required=True, + help='测试时长 (秒)') + parser.add_argument('-i', '--interval', type=int, required=True, + help='发送间隔 (毫秒)') + parser.add_argument('-s', '--size', type=int, required=True, + help='每包数据大小 (字节)') + parser.add_argument('-b', '--baudrate', type=int, default=115200, + help='波特率 (默认: 115200)') + parser.add_argument('--tx-log', type=str, default=None, + help='发送数据日志文件路径 (十六进制格式,一字节一行)') + parser.add_argument('--rx-log', type=str, default=None, + help='接收数据日志文件路径 (十六进制格式,一字节一行)') + + args = parser.parse_args() + + # 参数验证 + if args.duration <= 0: + print("错误: 测试时长必须大于0") + return 1 + + if args.interval <= 0: + print("错误: 发送间隔必须大于0") + return 1 + + if args.size <= 0 or args.size > 4096: + print("错误: 数据大小必须在1-4096字节之间") + return 1 + + # 打印配置信息 + print("="*70) + print(" "*25 + "串口测试程序" + " "*25) + print("="*70) + print(f"发送串口: {args.tx_port}") + print(f"接收串口: {args.rx_port}") + print(f"波特率: {args.baudrate}") + print(f"测试时长: {args.duration} 秒") + print(f"发送间隔: {args.interval} 毫秒") + print(f"包大小: {args.size} 字节") + if args.tx_log: + print(f"发送日志: {args.tx_log}") + if args.rx_log: + print(f"接收日志: {args.rx_log}") + print("="*70 + "\n") + + # 创建测试器并运行 + tester = SerialTester( + tx_port=args.tx_port, + rx_port=args.rx_port, + duration=args.duration, + interval_ms=args.interval, + packet_size=args.size, + baudrate=args.baudrate, + tx_log_file=args.tx_log, + rx_log_file=args.rx_log + ) + + success = tester.run() + + return 0 if success else 1 + +if __name__ == '__main__': + sys.exit(main()) + diff --git a/code/sdk/zhihui/ppp/example/ppp_at_cmd.c b/code/sdk/zhihui/ppp/example/ppp_at_cmd.c index 7cf942584..5851e4327 100644 --- a/code/sdk/zhihui/ppp/example/ppp_at_cmd.c +++ b/code/sdk/zhihui/ppp/example/ppp_at_cmd.c @@ -1318,13 +1318,13 @@ static errcode_t at_ppp_get_signal_info(uint8_t *para, uint32_t paraLen) (void)para; (void)paraLen; - PRINT("[PPP_INFO]PPP SIGNAL INFO: Getting network signal information...\r\n"); + PRINT("[PPP_INFO]PPP SIGNAL INFO IN ALL MODE: Getting network signal information...\r\n"); - errcode_t ret = ppp_get_network_signal_info(); + errcode_t ret = ppp_api_update_netinfo_in_all_mode(); if (ret == ERRCODE_SUCC) { - PRINT("[PPP_INFO]PPP SIGNAL INFO: Network signal information retrieved successfully\r\n"); + PRINT("[PPP_INFO]PPP SIGNAL INFO IN ALL MODE: Network signal information retrieved successfully\r\n"); } else { - PRINT("[PPP_ERR]PPP SIGNAL INFO: Failed to get network signal information, error: %d\r\n", ret); + PRINT("[PPP_ERR]PPP SIGNAL INFO IN ALL MODE: Failed to get network signal information, error: %d\r\n", ret); } return ret; @@ -1782,9 +1782,9 @@ static errcode_t at_ppp_tcp_performance_test(uint8_t *para, uint32_t paraLen) { if (!para || paraLen == 0) { PRINT("[PPP_ERR]PPP TCP PERF: Missing parameters\r\n"); - PRINT("[PPP_INFO]Usage: AT^PPPTCPPERF=,,,