Python Asyncio WebSocket类初始化错误排查及异步服务器需求

1. 报错原因：TypeError: init() missing 2 required positional arguments: 'client' and 'handler' #

这个错误通常是因为你错误地直接实例化了websockets库中的底层类（比如ServerProtocol），而没有使用官方推荐的websockets.serve()方法启动服务器。websockets的核心API是通过serve()函数注册连接处理回调，不需要手动初始化需要client和handler参数的类。

2. 整合pymodbus的非阻塞WebSocket服务器示例 #

下面是一个完整的异步实现，包含：

• 异步WebSocket服务器，处理客户端指令
• 异步Modbus TCP客户端，控制硬件设备
• 非阻塞的温度监控/设置逻辑

import asyncio
import websockets
from pymodbus.client import AsyncModbusTcpClient
from pymodbus.payload import BinaryPayloadBuilder, BinaryPayloadDecoder
from pymodbus.constants import Endian

# Modbus客户端实例（后续可封装为类）
modbus_client = None

async def init_modbus():
    """初始化Modbus TCP客户端连接"""
    global modbus_client
    modbus_client = AsyncModbusTcpClient("127.0.0.1", port=5020)  # 替换为你的硬件IP和端口
    await modbus_client.connect()
    print("Modbus客户端连接成功" if modbus_client.connected else "Modbus客户端连接失败")

async def set_hardware_temp(target_temp):
    """异步写入Modbus寄存器设置温度"""
    if not modbus_client or not modbus_client.connected:
        return False, "Modbus未连接"
    
    # 按硬件寄存器格式构建浮点数据（需根据实际设备调整字节序）
    builder = BinaryPayloadBuilder(byteorder=Endian.Big, wordorder=Endian.Big)
    builder.add_32bit_float(target_temp)
    payload = builder.build()
    
    # 写入起始地址0的2个寄存器（示例，需匹配硬件寄存器配置）
    result = await modbus_client.write_registers(address=0, values=payload, count=2, unit=1)
    if not result.isError():
        return True, f"温度已设置为{target_temp}℃"
    return False, f"设置失败: {str(result)}"

async def get_hardware_temp():
    """异步读取Modbus寄存器获取当前温度"""
    if not modbus_client or not modbus_client.connected:
        return None, "Modbus未连接"
    
    # 读取起始地址0的2个寄存器（示例，需匹配硬件寄存器配置）
    result = await modbus_client.read_holding_registers(address=0, count=2, unit=1)
    if not result.isError():
        decoder = BinaryPayloadDecoder.fromRegisters(result.registers, byteorder=Endian.Big, wordorder=Endian.Big)
        current_temp = decoder.decode_32bit_float()
        return round(current_temp, 2), "读取成功"
    return None, f"读取失败: {str(result)}"

async def handle_websocket(websocket):
    """WebSocket客户端连接处理函数"""
    print(f"新客户端接入: {websocket.remote_address}")
    try:
        async for msg in websocket:
            # 处理客户端指令，示例支持SET_TEMP:数值 和 GET_TEMP
            if msg.startswith("SET_TEMP:"):
                try:
                    temp = float(msg.split(":")[1])
                    success, response = await set_hardware_temp(temp)
                    await websocket.send(f"设置响应: {response}")
                except ValueError:
                    await websocket.send("错误: 温度格式无效，请输入数字")
            elif msg == "GET_TEMP":
                temp, response = await get_hardware_temp()
                await websocket.send(f"当前温度: {temp}℃" if temp is not None else f"读取失败: {response}")
            else:
                await websocket.send(f"未知指令: {msg}")
    except websockets.exceptions.ConnectionClosed:
        print(f"客户端断开连接: {websocket.remote_address}")

async def temp_monitor_loop():
    """非阻塞温度监控，每5秒读取一次"""
    while True:
        temp, msg = await get_hardware_temp()
        print(f"[监控] 当前温度: {temp}℃" if temp is not None else f"[监控] {msg}")
        await asyncio.sleep(5)  # 异步睡眠，不阻塞事件循环

async def main():
    # 初始化Modbus
    await init_modbus()
    
    # 启动WebSocket服务器
    async with websockets.serve(handle_websocket, "0.0.0.0", 8765):
        print("WebSocket服务器启动，监听端口8765")
        # 启动监控任务
        monitor_task = asyncio.create_task(temp_monitor_loop())
        # 持续运行直到手动终止
        await monitor_task

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("服务器已终止")

3. 项目优化建议 #

• 面向对象重构：把Modbus控制、WebSocket服务封装成独立类，比如ModbusHardwareController和AsyncWebSocketServer，替代全局变量，提升代码可维护性和复用性。
• 增强错误处理：添加Modbus自动重连逻辑，当连接断开时定时尝试重连；细化WebSocket异常捕获（如网络超时、客户端异常断开）。
• 日志替换打印：用Python标准库logging模块替代print，配置不同级别日志，方便调试和生产环境问题排查。
• 配置分离：把Modbus地址、端口、WebSocket端口、监控间隔等参数放到配置文件（如config.yaml）或环境变量中，避免硬编码。
• 标准化消息格式：改用JSON传递消息，比如客户端发送{"action": "set_temp", "value": 25}，服务器返回{"status": "success", "data": 25}，比纯文本指令更清晰易扩展。
• 添加身份验证：如果需要限制客户端访问，在WebSocket连接建立时添加token验证或账号密码校验。
• 编写测试用例：用pytest-asyncio编写单元测试，覆盖Modbus读写、WebSocket消息处理等核心逻辑，保障代码稳定性。

内容的提问来源于stack exchange，提问作者Gool