下面是一个完整的 Python 脚本，用于**定时向多个串口设备发送指令**，每个串口对应的指令从一个 **配置文件（JSON 格式）** 中读取。脚本使用 `pyserial` 进行串口通信，`schedule` 实现定时任务，`threading` 支持多串口并发操作。 --- ### ✅ 功能说明： - 从 JSON 配置文件中读取串口信息：端口名、波特率、发送周期（秒）、要发送的指令。 - 每个串口独立运行在自己的线程中。 - 定时按周期发送指令。 - 错误处理：串口不存在或断开时自动重连并输出日志。 --- ### 🔧 所需依赖库： ```bash pip install pyserial schedule ``` --- ### 📁 配置文件 `config.json` 示例： ```json [ { "port": "COM3", "baudrate": 9600, "interval": 5, "command": "GET_STATUS\n" }, { "port": "COM4", "baudrate": 115200, "interval": 10, "command": "RESET_DEVICE\n" } ] ``` > 注意：根据你的系统修改串口号（Windows 是 `COMx`，Linux 是 `/dev/ttyUSB0` 等） --- ### 🐍 Python 脚本：`serial_scheduler.py` ```python import json import serial import time import threading import schedule from functools import partial def send_command(port, baudrate, command): """向指定串口发送指令""" try: with serial.Serial(port, baudrate, timeout=1) as ser: # 编码字符串为 bytes data = command.encode('utf-8') ser.write(data) print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Sent to {port}: {command.strip()}") except serial.SerialException as e: print(f"[ERROR] Serial error on {port}: {e}") except Exception as e: print(f"[ERROR] Unexpected error with {port}: {e}") def start_serial_job(port, baudrate, interval, command): """启动一个定时任务，定期向该串口发送指令""" job_func = partial(send_command, port, baudrate, command) schedule.every(interval).seconds.do(job_func) # 该线程专属的调度循环 def run_schedule(): while True: schedule.run_pending() time.sleep(1) # 启动调度线程 thread = threading.Thread(target=run_schedule, daemon=True) thread.start() return thread def load_config_and_start(filename): """加载配置文件并启动所有串口任务""" try: with open(filename, 'r', encoding='utf-8') as f: configs = json.load(f) except FileNotFoundError: print(f"Error: Config file '{filename}' not found.") return except json.JSONDecodeError as e: print(f"Error: Invalid JSON in config file: {e}") return threads = [] for cfg in configs: port = cfg.get("port") baudrate = cfg.get("baudrate", 9600) interval = cfg.get("interval", 5) command = cfg.get("command", "\n") if not port: print(f"Warning: Missing 'port' in config entry: {cfg}") continue print(f"Setting up {port} @ {baudrate} bps, every {interval}s -> {command.strip()}") t = start_serial_job(port, baudrate, interval, command) threads.append(t) return threads if __name__ == "__main__": config_file = "config.json" threads = load_config_and_start(config_file) if not threads: print("No valid configuration loaded. Exiting.") exit(1) print("All serial tasks started. Press Ctrl+C to exit.") try: while True: time.sleep(1) except KeyboardInterrupt: print("\nShutting down...") ``` --- ### 💡 代码解释： - `send_command`: 封装了串口打开 → 发送数据 → 自动关闭逻辑，避免资源泄漏。 - `start_serial_job`: 使用 `partial` 固定参数后注册到 `schedule`，并在独立线程中运行调度器。 - 多线程设计：每个串口共享主线程中的调度器可能冲突，因此采用 **每个串口不单独开线程调度**，而是所有任务共用一个调度器（本例中所有任务共用同一个调度循环）。 - `daemon=True`: 线程随主程序退出而结束。 - `load_config_and_start`: 加载 JSON 并验证字段，启动各个任务。 --- ### ⚠️ 注意事项： 1. 确保串口名称正确，且当前用户有访问权限（Linux 下常需加入 `dialout` 组）。 2. 指令末尾是否加 `\n` 或 `\r\n` 取决于设备协议，请根据实际情况调整。 3. 如果需要接收响应，可在 `send_command` 中添加 `ser.readline()` 并打印。 --- ### ✅ 扩展建议： - 添加日志记录到文件。 - 增加串口自动重连机制（当前每次发送都重新打开关闭，已具备一定容错性）。 - 使用 `argparse` 支持命令行传入配置文件路径。 --- ###