EC20模组AT指令实战：用Python脚本自动化测试4G模组联网状态

EC20模组AT指令实战用Python脚本自动化测试4G模组联网状态在物联网设备从原型走向量产的过程中有一个环节常常让开发者感到繁琐又无法回避如何高效、批量地验证成百上千个4G通信模组的联网状态与基础功能手动连接串口、逐条输入AT指令、肉眼观察返回结果这种工作方式不仅效率低下而且极易出错尤其是在需要反复测试信号强度、网络注册、SIM卡状态等关键指标时。对于负责产线测试、质量保障或现场部署维护的工程师而言一套稳定、可复用的自动化测试工具其价值不言而喻。本文将聚焦于移远通信的EC20系列4G CAT4模组但核心思路适用于绝大多数基于AT指令交互的通信模块。我们不讨论硬件供电、引脚焊接这些基础硬件操作而是直接切入软件调试的核心——如何用Python构建一个健壮、灵活的AT指令自动化测试框架。这个框架的目标是连接串口、发送指令、解析响应、判断状态、记录日志最终生成一份清晰的可读报告将工程师从重复劳动中解放出来把精力投入到更复杂的业务逻辑和问题排查中。1. 自动化测试框架的设计哲学与核心组件在动手写代码之前我们需要先想清楚一个好的自动化测试脚本应该具备哪些特质。它绝不仅仅是几条serial.write()命令的堆砌。一个工业级的测试工具需要兼顾可靠性、可维护性、可扩展性和易用性。可靠性是底线。脚本必须能稳定处理各种异常情况串口突然断开、模组响应超时、返回非预期数据、网络环境瞬时波动等。一个脆弱的脚本可能在实验室运行良好一到产线或现场就频频崩溃失去信任。可维护性意味着代码结构清晰关键参数如串口配置、超时时间、判断阈值集中管理而非散落在各个函数里。当测试用例需要增减或者AT指令集因固件升级而变化时我们能快速定位并修改而不是在成百上千行代码中大海捞针。可扩展性则要求我们的设计易于添加新的测试项。今天可能只测信号强度(ATCSQ)和网络注册(ATCREG?)明天可能就需要加入查询国际移动设备识别码(ATCGSN)、检查运营商(ATCOPS?)等功能。框架应该支持以“插件”或“配置”的方式轻松纳入新测试。易用性关注使用体验。脚本应该提供清晰的命令行参数支持配置文件并能输出结构化的日志和测试报告方便非开发人员如测试工程师查看结果。基于这些考量一个典型的自动化测试框架核心组件包括串口通信管理器负责打开、配置、关闭串口并提供基础的读写方法。AT指令会话处理器封装发送指令、等待响应、解析结果的核心逻辑内置超时和重试机制。测试用例集合以结构化的方式定义每一个要执行的测试项如“检查SIM卡”、“查询信号”包括指令、预期响应格式、成功/失败的条件判断函数。结果收集与报告生成器在执行过程中收集每个测试用例的结果成功、失败、超时、数据最终生成文本、JSON或HTML格式的报告。日志系统记录详细的调试信息便于在测试失败时进行问题追踪。接下来我们将从环境搭建开始一步步实现这些组件。2. 环境准备与Python生态工具链工欲善其事必先利其器。我们的自动化脚本将完全基于Python实现这得益于Python在串口通信和自动化测试领域丰富的库支持。首先你需要一个Python环境推荐3.7及以上版本。核心依赖库是pyserial它是Python事实上的串口通信标准库。# 使用pip安装pyserial pip install pyserial如果你计划生成更美观的测试报告可以额外安装pandas用于数据处理或Jinja2用于HTML报告模板渲染。pip install pandas Jinja2在硬件连接方面确保你的EC20模组已经正确上电并通过USB转TTL串口线或直接通过开发板的UART接口连接到电脑。你需要知道串口号在Windows上是COM3、COM4等在Linux/macOS上是/dev/ttyUSB0、/dev/ttyACM0等以及模组默认的波特率EC20通常是115200。注意在Linux系统下你可能需要将当前用户添加到dialout组以获得串口访问权限sudo usermod -a -G dialout $USER操作后需要重新登录生效。为了后续代码的清晰我们建议创建一个项目目录并规划如下结构ec20_auto_tester/ ├── config.yaml # 配置文件存放串口参数、测试项等 ├── test_runner.py # 主程序入口 ├── core/ │ ├── __init__.py │ ├── serial_manager.py # 串口管理类 │ ├── at_command.py # AT指令会话处理类 │ └── test_cases.py # 测试用例定义 ├── utils/ │ ├── __init__.py │ ├── logger.py # 日志配置 │ └── report_generator.py # 报告生成 └── results/ └── (测试报告和日志将输出至此)这种模块化的结构让代码职责分明便于后续维护和扩展。3. 构建健壮的串口通信与AT指令会话层这是整个框架的基石。我们不能简单地使用pyserial的初级读写而需要在其之上封装一层具备容错能力的会话管理。首先在core/serial_manager.py中我们创建一个串口管理器类。import serial import time from typing import Optional, Tuple class SerialManager: def __init__(self, port: str, baudrate: int 115200, timeout: float 1.0): 初始化串口管理器。 :param port: 串口设备路径如 COM3 或 /dev/ttyUSB0 :param baudrate: 波特率默认115200 :param timeout: 读超时时间秒 self.port port self.baudrate baudrate self.timeout timeout self.serial_conn: Optional[serial.Serial] None def open(self) - bool: 尝试打开串口连接 try: self.serial_conn serial.Serial( portself.port, baudrateself.baudrate, bytesizeserial.EIGHTBITS, parityserial.PARITY_NONE, stopbitsserial.STOPBITS_ONE, timeoutself.timeout ) # 给模组一点时间初始化清空可能的缓冲区残留数据 time.sleep(0.5) if self.serial_conn.is_open: self.serial_conn.reset_input_buffer() self.serial_conn.reset_output_buffer() return True except (serial.SerialException, OSError) as e: print(f无法打开串口 {self.port}: {e}) return False def close(self): 关闭串口连接 if self.serial_conn and self.serial_conn.is_open: self.serial_conn.close() def send_and_receive(self, command: str, wait_time: float 0.5) - Tuple[bool, list]: 发送命令并读取响应。 :param command: 要发送的AT指令无需额外添加回车换行 :param wait_time: 发送后等待响应的基础时间秒 :return: (是否成功, 响应行列表) if not self.serial_conn or not self.serial_conn.is_open: return False, [串口未打开] # 确保指令以回车结尾 if not command.endswith(\r): command \r try: self.serial_conn.write(command.encode(utf-8, errorsignore)) time.sleep(wait_time) # 等待模组处理 response_lines [] # 循环读取直到超时或读到明确的结束标志如OK或ERROR while True: line self.serial_conn.readline() if not line: # 超时无更多数据 break try: decoded_line line.decode(utf-8, errorsignore).strip() except UnicodeDecodeError: decoded_line str(line).strip() # 备用方案 if decoded_line: response_lines.append(decoded_line) # 如果读到OK或ERROR可以提前结束但继续读完缓冲区 if decoded_line in (OK, ERROR): # 再短暂读取一次确保没有紧随其后的数据 time.sleep(0.05) continue except (serial.SerialException, OSError) as e: return False, [f串口通信错误: {e}] return True, response_lines这个SerialManager提供了基础的打开、关闭和收发功能。但针对AT指令我们需要更高级的抽象。接下来在core/at_command.py中创建ATSession类。import re import time from .serial_manager import SerialManager class ATSession: def __init__(self, serial_manager: SerialManager, max_retries: int 2): self.sm serial_manager self.max_retries max_retries self._echo_enabled None # 初始未知 def execute(self, command: str, expected_okTrue, pattern: str None, timeout_per_retry: float 3.0) - dict: 执行一条AT指令并解析结果。 :param command: AT指令如 ATCSQ :param expected_ok: 是否期望以OK作为成功响应 :param pattern: 用于从响应中提取数据的正则表达式模式 :param timeout_per_retry: 单次尝试的超时时间 :return: 包含执行结果的字典 result { success: False, command: command, raw_response: [], data: None, error: None, retries: 0 } for attempt in range(self.max_retries 1): result[retries] attempt success, lines self.sm.send_and_receive(command, wait_time0.5) if not success: result[error] 串口通信失败 time.sleep(0.5) # 失败后稍作等待再重试 continue result[raw_response] lines # 处理常见的AT响应 if not lines: result[error] 无响应 continue # 检查是否有ERROR if ERROR in lines: result[error] f模组返回ERROR: {lines} # 如果是语法错误等重试无意义直接退出 if any(err in .join(lines) for err in [CME ERROR, CMS ERROR]): break continue # 如果期望OK则检查OK if expected_ok and OK not in lines: result[error] 响应中未找到OK continue # 如果提供了正则表达式模式尝试提取数据 if pattern: for line in lines: match re.search(pattern, line) if match: result[data] match.groups() if len(match.groups()) 1 else match.group(1) break if pattern and result[data] is None: result[error] f未匹配到模式: {pattern} continue # 所有检查通过 result[success] True result[error] None break # 成功则跳出重试循环 return resultATSession.execute方法是核心它封装了重试逻辑、错误判断和数据提取。使用正则表达式 (pattern参数) 可以非常灵活地从形如CSQ: 24,99这样的响应中提取出信号强度值24和误码率99。4. 定义测试用例与实现关键状态查询有了强大的会话层我们就可以定义具体的测试用例了。在core/test_cases.py中我们将测试项组织成结构化的列表或字典。每个测试项至少包含名称、要执行的AT指令、用于解析响应的正则表达式、以及一个判断测试是否通过的函数。首先我们定义一些EC20模组常用的状态查询指令及其解析方式测试项目AT指令成功响应示例解析目标与正则表达式说明通信功能测试ATOK无基础AT指令响应检查串口通信是否正常。信号强度与质量ATCSQCSQ: 24,99r\CSQ:\s*(\d),(\d)提取信号强度(0-31, 99为未知)和误码率(0-7, 99为未知)。网络注册状态ATCREG?CREG: 0,1r\CREG:\s*\d,(\d)提取第二个参数1-已注册本地网5-已注册漫游网其他值表示未注册或搜索中。SIM卡状态ATCPIN?CPIN: READYr\CPIN:\s*(.)检查返回是否为READY其他如SIM PIN、SIM PUK表示需要输入密码。运营商信息ATCOPS?COPS: 0,0,China Mobiler\COPS:.*?([^]*)提取运营商名称。模块型号与版本ATIQuectelbrEC20CEFAR02A07M4G无读取多行响应验证模块身份。国际移动设备识别码ATCGSN864839040123456r(\d{15})提取15位IMEI号。基于上表我们可以编写具体的测试用例函数。以信号强度测试为例def test_signal_strength(session: ATSession) - dict: 测试信号强度。 返回的字典包含标准化的测试结果。 test_name 信号强度查询 command ATCSQ pattern r\CSQ:\s*(\d),(\d) result session.execute(command, expected_okTrue, patternpattern) test_result { name: test_name, command: command, passed: result[success], data: {}, message: } if result[success] and result[data]: rssi, ber result[data] rssi_int int(rssi) # 将RSSI转换为dBm近似值EC20常用换算 if rssi_int 0: dBm -113 dBm 或更差 elif rssi_int 1: dBm -111 dBm elif 2 rssi_int 30: dBm f-{109 - 2 * (rssi_int - 2)} dBm # 近似公式 elif rssi_int 31: dBm -51 dBm 或更好 else: # 99 dBm 未知或不可用 test_result[data][rssi] rssi_int test_result[data][ber] int(ber) test_result[data][dBm_approx] dBm test_result[message] f信号强度: {rssi} (约{dBm}), 误码率: {ber} # 判断是否通过通常信号强度大于2即大于-109dBm认为可用 test_result[passed] test_result[passed] and (rssi_int 2 and rssi_int ! 99) if not test_result[passed]: test_result[message] - 信号强度不足 else: test_result[message] f查询失败: {result.get(error, 未知错误)} return test_result类似地我们可以编写test_network_registration,test_sim_status等函数。为了让主程序能方便地调用所有测试我们可以创建一个测试套件# 在 test_cases.py 中定义一个测试套件列表 TEST_SUITE [ {func: test_communication, name: 基础通信测试}, {func: test_sim_status, name: SIM卡状态检查}, {func: test_signal_strength, name: 信号强度与质量}, {func: test_network_registration, name: 网络注册状态}, {func: test_operator_info, name: 运营商信息}, {func: test_module_identity, name: 模块身份识别}, ]这种设计允许我们通过修改这个列表来轻松调整测试执行的顺序和内容。5. 整合与实战从单次测试到批量自动化现在我们将所有组件整合到主程序test_runner.py中。这个脚本应该能够接受命令行参数如串口号加载配置按顺序执行测试套件并生成报告。#!/usr/bin/env python3 import sys import time import json from datetime import datetime from core.serial_manager import SerialManager from core.at_command import ATSession from core.test_cases import TEST_SUITE from utils.logger import setup_logger from utils.report_generator import generate_html_report def main(port: str): 主测试流程 # 1. 初始化日志 log_file fresults/test_log_{datetime.now().strftime(%Y%m%d_%H%M%S)}.log logger setup_logger(log_file) logger.info(f开始EC20模组自动化测试串口: {port}) # 2. 初始化串口和AT会话 sm SerialManager(portport, baudrate115200, timeout2) if not sm.open(): logger.error(串口打开失败程序退出。) sys.exit(1) session ATSession(sm, max_retries2) overall_results { start_time: datetime.now().isoformat(), port: port, total_tests: len(TEST_SUITE), passed_tests: 0, failed_tests: 0, test_details: [] } # 3. 顺序执行测试套件 for test_item in TEST_SUITE: test_func test_item[func] test_name test_item[name] logger.info(f执行测试: {test_name}) try: test_result test_func(session) # 调用具体的测试函数 test_result[timestamp] datetime.now().isoformat() if test_result[passed]: overall_results[passed_tests] 1 logger.info(f [通过] {test_result[message]}) else: overall_results[failed_tests] 1 logger.warning(f [失败] {test_result[message]}) overall_results[test_details].append(test_result) except Exception as e: logger.error(f执行测试 {test_name} 时发生异常: {e}, exc_infoTrue) error_result { name: test_name, passed: False, message: f测试执行异常: {e}, timestamp: datetime.now().isoformat() } overall_results[test_details].append(error_result) overall_results[failed_tests] 1 # 测试项之间短暂间隔避免模组过载 time.sleep(0.3) # 4. 生成报告 overall_results[end_time] datetime.now().isoformat() overall_results[duration] (datetime.fromisoformat(overall_results[end_time]) - datetime.fromisoformat(overall_results[start_time])).total_seconds() # 保存JSON格式的详细结果 json_report_path fresults/test_report_{datetime.now().strftime(%Y%m%d_%H%M%S)}.json with open(json_report_path, w, encodingutf-8) as f: json.dump(overall_results, f, ensure_asciiFalse, indent2) logger.info(f详细测试报告已保存至: {json_report_path}) # 生成更易读的HTML报告如果实现了该功能 html_report_path fresults/test_report_{datetime.now().strftime(%Y%m%d_%H%M%S)}.html try: generate_html_report(overall_results, html_report_path) logger.info(fHTML测试报告已生成: {html_report_path}) except Exception as e: logger.warning(f生成HTML报告失败: {e}) # 5. 清理与总结 sm.close() final_status 通过 if overall_results[failed_tests] 0 else 失败 logger.info(f测试完成。总计 {overall_results[total_tests]} 项通过 {overall_results[passed_tests]} 项失败 {overall_results[failed_tests]} 项。整体状态: {final_status}) print(f\n 测试总结 ) print(f串口: {port}) print(f状态: {final_status}) print(f通过率: {overall_results[passed_tests]}/{overall_results[total_tests]}) print(f报告文件: {json_report_path}) if os.path.exists(html_report_path): print(fHTML报告: {html_report_path}) if __name__ __main__: # 可以从命令行参数或配置文件读取串口号 if len(sys.argv) 1: port sys.argv[1] else: # 默认值或从config.yaml读取 port COM3 # Windows示例 # port /dev/ttyUSB0 # Linux示例 main(port)这个主程序流程清晰初始化、执行测试、收集结果、生成报告、清理资源。它已经具备了处理单个模组测试的能力。迈向批量自动化在实际产线测试中我们往往需要循环测试多个串口对应多个模组。这时可以编写一个外层脚本读取一个设备列表包含串口号、设备ID等然后为每个设备启动一个测试进程或线程并将所有结果汇总到一个总报告中。这涉及到多进程/线程管理和数据同步是另一个层次的优化但核心的测试逻辑完全复用。6. 异常处理与调试技巧让脚本更可靠在自动化测试中最耗时的往往不是编写成功的流程而是处理各种边界情况和异常。我们的框架已经内置了基础的重试和错误判断但在实际使用中你可能会遇到一些棘手的问题。问题一模组响应不稳定偶尔超时或无响应。应对策略除了增加重试次数 (max_retries)还可以在重试之间加入递增的等待时间指数退避。同时在发送下一条关键指令前如查询网络注册可以先发送一个简单的AT指令来确认模组是否处于可响应状态。问题二AT指令返回的数据格式与文档略有差异。应对策略在解析响应时使用更宽松的正则表达式并增加更多的日志输出。例如解析CSQ时模式r\CSQ:\s*(\d).*比r\CSQ:\s*(\d),(\d)更宽容能应对某些固件版本可能省略误码率的情况。同时将原始的响应行详细记录到日志文件中便于后续分析。问题三需要测试模组的特定业务功能如TCP连接、HTTP GET等。应对策略我们的框架很容易扩展。只需在test_cases.py中编写新的测试函数。例如测试TCP连接def test_tcp_connection(session: ATSession, server_ip: str, port: int) - dict: 测试TCP连接功能 # 1. 设置APN (如果需要) # session.execute(ATCGDCONT1,IP,your_apn) # 2. 激活PDP上下文 # session.execute(ATQIACT1) # 3. 建立TCP连接 result session.execute(fATQIOPEN1,0,TCP,{server_ip},{port}) # ... 解析结果检查是否返回 CONNECT OK # 4. 发送测试数据 # 5. 关闭连接 # 返回结构化的测试结果 pass调试时的一个实用技巧在开发初期可以启用一个“调试模式”让脚本将所有发送和接收到的原始字节包括不可打印字符都打印到控制台或日志中。这能帮助你确认回车换行符(\r\n)是否正确以及模组返回的确切内容。# 在SerialManager的send_and_receive方法中添加调试输出 def send_and_receive(self, command: str, wait_time: float 0.5, debugFalse): # ... if debug: print(f[TX] {repr(command)}) self.serial_conn.write(command.encode(utf-8)) # ... while True: line self.serial_conn.readline() if debug and line: print(f[RX] {repr(line)}) # ...最后记得将你的测试脚本和配置纳入版本控制如Git。每次固件升级或测试需求变更时你可以清晰地追踪修改历史并轻松回滚到稳定版本。自动化测试的价值正是在这一次次迭代和问题解决中得以体现它最终成为你物联网产品开发流程中一个可靠、高效的守门员。