test_client.py

#!/usr/bin/env python3
"""
分布式空调控制器 - 测试客户端
用于测试TCP服务器的功能，兼容新的消息协议架构
"""

import socket
import json
import time
import random
import threading
import struct
from datetime import datetime
from uuid import uuid4

class MessageType:
    """消息类型枚举"""
    HEARTBEAT = 1
    QUERY = 2
    REPORT = 3
    COMMAND = 4
    RESPONSE = 5
    ERROR = 6
    UNKNOWN = 0

class MessageStatus:
    """消息状态码"""
    SUCCESS = 200
    BAD_REQUEST = 400
    NOT_FOUND = 404
    TIMEOUT = 408
    INTERNAL_ERROR = 500

class ProtocolPacketMaker:
    """协议包制作器 - Python版本"""
    
    @staticmethod
    def create_message_id():
        """创建消息ID"""
        return f"{int(time.time() * 1000)}_{random.randint(1000, 9999)}"
    
    @staticmethod
    def encode_with_length_header(json_data):
        """使用长度头编码数据包"""
        if isinstance(json_data, str):
            data = json_data.encode('utf-8')
        else:
            data = json.dumps(json_data).encode('utf-8')
        
        # 4字节长度头 + JSON数据
        length = len(data)
        header = struct.pack('!I', length)  # 网络字节序，无符号整数
        return header + data
    
    @staticmethod
    def decode_length_header(data):
        """解码长度头数据包"""
        if len(data) < 4:
            return None, data
        
        length = struct.unpack('!I', data[:4])[0]
        if len(data) < 4 + length:
            return None, data  # 数据不完整
        
        json_data = data[4:4+length].decode('utf-8')
        remaining = data[4+length:]
        return json_data, remaining

class TestClient:
    def __init__(self, server_ip="192.168.4.1", server_port=8848, mac_address=None):
        self.server_ip = server_ip
        self.server_port = server_port
        self.mac_address = mac_address or self.generate_mac_address()
        self.socket = None
        self.connected = False
        self.current_temp = round(random.uniform(20.0, 30.0), 1)
        self.target_temp = 25.0
        self.humidity = round(random.uniform(40.0, 80.0), 1)
        self.device_status = "normal"
        self.running = True
        self.receive_buffer = b''
        
    def generate_mac_address(self):
        """生成随机MAC地址"""
        mac = [random.randint(0, 255) for _ in range(6)]
        return ':'.join([f'{x:02x}' for x in mac]).upper()
        
    def connect(self):
        """连接到服务器"""
        try:
            self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.socket.connect((self.server_ip, self.server_port))
            self.connected = True
            print(f"[{datetime.now()}] 已连接到服务器 {self.server_ip}:{self.server_port}")
            print(f"[{datetime.now()}] 客户端MAC地址: {self.mac_address}")
            return True
        except Exception as e:
            print(f"[{datetime.now()}] 连接失败: {e}")
            return False
    
    def disconnect(self):
        """断开连接"""
        self.running = False
        if self.socket:
            self.socket.close()
            self.connected = False
            print(f"[{datetime.now()}] 已断开连接")
    
    def send_message(self, message_data):
        """发送消息"""
        if not self.connected:
            return False
        
        try:
            # 使用协议包编码
            packet_data = ProtocolPacketMaker.encode_with_length_header(message_data)
            self.socket.send(packet_data)
            return True
        except Exception as e:
            print(f"[{datetime.now()}] 发送失败: {e}")
            return False
    
    def create_heartbeat(self):
        """创建心跳消息"""
        return {
            "type": "heartbeat",
            "messageId": ProtocolPacketMaker.create_message_id(),
            "sourceMAC": self.mac_address,
            "targetMAC": "",  # 广播
            "timestamp": int(time.time() * 1000),
            "signalStrength": random.randint(-80, -30),
            "status": "online"
        }
    
    def create_temperature_report(self):
        """创建温度报告消息"""
        return {
            "type": "report",
            "messageId": ProtocolPacketMaker.create_message_id(),
            "sourceMAC": self.mac_address,
            "targetMAC": "",  # 广播给所有监听者
            "timestamp": int(time.time() * 1000),
            "temperature": self.current_temp,
            "humidity": self.humidity,
            "targetTemperature": self.target_temp,
            "deviceStatus": self.device_status
        }
    
    def create_query(self, query_type, target_mac=""):
        """创建查询消息"""
        return {
            "type": "query",
            "messageId": ProtocolPacketMaker.create_message_id(),
            "sourceMAC": self.mac_address,
            "targetMAC": target_mac,
            "timestamp": int(time.time() * 1000),
            "queryType": query_type
        }
    
    def create_response(self, request_id, status=MessageStatus.SUCCESS, message=""):
        """创建响应消息"""
        return {
            "type": "response",
            "messageId": ProtocolPacketMaker.create_message_id(),
            "sourceMAC": self.mac_address,
            "targetMAC": "",
            "timestamp": int(time.time() * 1000),
            "requestId": request_id,
            "status": status,
            "message": message
        }
    
    def send_heartbeat(self):
        """发送心跳包"""
        heartbeat = self.create_heartbeat()
        if self.send_message(heartbeat):
            print(f"[{datetime.now()}] 发送心跳包: 状态={heartbeat['status']}, 信号强度={heartbeat['signalStrength']}")
            return True
        return False
    
    def send_temperature_report(self):
        """发送温度报告"""
        report = self.create_temperature_report()
        if self.send_message(report):
            print(f"[{datetime.now()}] 发送温度报告: {self.current_temp}°C, 湿度={self.humidity}%, 目标温度={self.target_temp}°C")
            return True
        return False
    
    def send_query(self, query_type, target_mac=""):
        """发送查询请求"""
        query = self.create_query(query_type, target_mac)
        if self.send_message(query):
            print(f"[{datetime.now()}] 发送查询请求: 类型={query_type}, 目标={target_mac or '广播'}")
            return True
        return False
    
    def listen_for_messages(self):
        """监听服务器消息"""
        while self.running and self.connected:
            try:
                self.socket.settimeout(1.0)  # 1秒超时
                data = self.socket.recv(4096)
                if data:
                    self.receive_buffer += data
                    self.process_received_data()
            except socket.timeout:
                continue
            except Exception as e:
                if self.running:
                    print(f"[{datetime.now()}] 接收数据错误: {e}")
                break
    
    def process_received_data(self):
        """处理接收到的数据"""
        while len(self.receive_buffer) >= 4:
            try:
                # 尝试解码数据包
                json_str, remaining = ProtocolPacketMaker.decode_length_header(self.receive_buffer)
                if json_str is None:
                    break  # 数据不完整，等待更多数据
                
                self.receive_buffer = remaining
                
                # 解析JSON消息
                message = json.loads(json_str)
                self.handle_message(message)
                
            except Exception as e:
                print(f"[{datetime.now()}] 处理数据包错误: {e}")
                # 清空缓冲区避免持续错误
                self.receive_buffer = b''
                break
    
    def handle_message(self, message):
        """处理接收到的消息"""
        msg_type = message.get("type", "unknown")
        source_mac = message.get("sourceMAC", "unknown")
        target_mac = message.get("targetMAC", "")
        
        # 检查消息是否针对我们（空目标表示广播）
        if target_mac and target_mac != self.mac_address:
            return
        
        if msg_type == "heartbeat":
            self.handle_heartbeat(message)
        elif msg_type == "query":
            self.handle_query(message)
        elif msg_type == "report":
            self.handle_report(message)
        elif msg_type == "command":
            self.handle_command(message)
        elif msg_type == "response":
            self.handle_response(message)
        elif msg_type == "error":
            self.handle_error(message)
        else:
            print(f"[{datetime.now()}] 收到未知消息类型: {msg_type} 来自 {source_mac}")
    
    def handle_heartbeat(self, message):
        """处理心跳消息"""
        source = message.get("sourceMAC", "unknown")
        status = message.get("status", "unknown")
        signal = message.get("signalStrength", 0)
        print(f"[{datetime.now()}] 收到心跳包 来自 {source}: 状态={status}, 信号={signal}")
    
    def handle_query(self, message):
        """处理查询消息"""
        source = message.get("sourceMAC", "unknown")
        query_type = message.get("queryType", "unknown")
        message_id = message.get("messageId", "")
        
        print(f"[{datetime.now()}] 收到查询请求 来自 {source}: 类型={query_type}")
        
        # 根据查询类型回复
        response_message = ""
        status = MessageStatus.SUCCESS
        
        if query_type == "temperature":
            response_message = f"当前温度: {self.current_temp}°C, 湿度: {self.humidity}%"
        elif query_type == "status":
            response_message = f"设备状态: {self.device_status}"
        elif query_type == "config":
            response_message = f"目标温度: {self.target_temp}°C"
        else:
            response_message = f"未知查询类型: {query_type}"
            status = MessageStatus.BAD_REQUEST
        
        # 发送响应
        response = self.create_response(message_id, status, response_message)
        response["targetMAC"] = source  # 回复给查询者
        self.send_message(response)
        print(f"[{datetime.now()}] 发送查询响应: {response_message}")
    
    def handle_report(self, message):
        """处理温度报告消息"""
        source = message.get("sourceMAC", "unknown")
        temp = message.get("temperature", 0)
        humidity = message.get("humidity", 0)
        target_temp = message.get("targetTemperature", 0)
        status = message.get("deviceStatus", "unknown")
        
        print(f"[{datetime.now()}] 收到温度报告 来自 {source}: {temp}°C, 湿度={humidity}%, 目标={target_temp}°C, 状态={status}")
    
    def handle_command(self, message):
        """处理命令消息"""
        source = message.get("sourceMAC", "unknown")
        command = message.get("command", "unknown")
        message_id = message.get("messageId", "")
        priority = message.get("priority", 5)
        
        print(f"[{datetime.now()}] 收到控制命令 来自 {source}: {command} (优先级={priority})")
        
        # 执行命令
        response_message = ""
        status = MessageStatus.SUCCESS
        
        if command == "set_temperature":
            # 这里应该从parameters中获取目标温度，简化处理
            new_target = random.uniform(20.0, 28.0)
            self.target_temp = round(new_target, 1)
            response_message = f"目标温度已设置为 {self.target_temp}°C"
            print(f"[{datetime.now()}] 执行命令: 设置目标温度为 {self.target_temp}°C")
            
        elif command == "turn_on":
            self.device_status = "normal"
            response_message = "设备已开启"
            print(f"[{datetime.now()}] 执行命令: 设备开启")
            
        elif command == "turn_off":
            self.device_status = "offline"
            response_message = "设备已关闭"
            print(f"[{datetime.now()}] 执行命令: 设备关闭")
            
        elif command == "cool_down":
            self.target_temp = max(18.0, self.target_temp - 2.0)
            response_message = f"启动降温，目标温度调整为 {self.target_temp}°C"
            print(f"[{datetime.now()}] 执行命令: 降温到 {self.target_temp}°C")
            
        elif command == "warm_up":
            self.target_temp = min(32.0, self.target_temp + 2.0)
            response_message = f"启动升温，目标温度调整为 {self.target_temp}°C"
            print(f"[{datetime.now()}] 执行命令: 升温到 {self.target_temp}°C")
            
        elif command == "reset":
            self.target_temp = 25.0
            self.device_status = "normal"
            response_message = "设备已重置"
            print(f"[{datetime.now()}] 执行命令: 设备重置")
            
        else:
            response_message = f"未知命令: {command}"
            status = MessageStatus.BAD_REQUEST
        
        # 发送响应
        response = self.create_response(message_id, status, response_message)
        response["targetMAC"] = source  # 回复给命令发送者
        self.send_message(response)
        print(f"[{datetime.now()}] 发送命令响应: {response_message}")
    
    def handle_response(self, message):
        """处理响应消息"""
        source = message.get("sourceMAC", "unknown")
        request_id = message.get("requestId", "")
        status = message.get("status", 0)
        response_msg = message.get("message", "")
        
        print(f"[{datetime.now()}] 收到响应 来自 {source}: 状态={status}, 消息='{response_msg}' (请求ID: {request_id})")
    
    def handle_error(self, message):
        """处理错误消息"""
        source = message.get("sourceMAC", "unknown")
        error_code = message.get("errorCode", 0)
        error_msg = message.get("errorMessage", "")
        context = message.get("context", "")
        
        print(f"[{datetime.now()}] 收到错误消息 来自 {source}: 代码={error_code}, 消息='{error_msg}', 上下文='{context}'")
    
    def simulate_temperature_changes(self):
        """模拟温度变化"""
        while self.running and self.connected:
            try:
                # 模拟向目标温度调节
                temp_diff = self.target_temp - self.current_temp
                if abs(temp_diff) > 0.1:
                    # 模拟空调调节效果
                    adjustment = min(0.3, abs(temp_diff) * 0.1)
                    if temp_diff > 0:
                        self.current_temp += adjustment
                    else:
                        self.current_temp -= adjustment
                
                # 添加环境噪声
                noise = random.uniform(-0.2, 0.2)
                self.current_temp += noise
                
                # 保持在合理范围内
                self.current_temp = max(15.0, min(35.0, self.current_temp))
                self.current_temp = round(self.current_temp, 1)
                
                # 湿度也略微变化
                humidity_change = random.uniform(-2.0, 2.0)
                self.humidity += humidity_change
                self.humidity = max(30.0, min(90.0, self.humidity))
                self.humidity = round(self.humidity, 1)
                
                time.sleep(3)  # 每3秒更新一次
                
            except Exception as e:
                print(f"[{datetime.now()}] 温度模拟错误: {e}")
                break
    
    def run(self):
        """运行客户端"""
        if not self.connect():
            return
        
        # 发送初始心跳
        self.send_heartbeat()
        
        # 启动监听线程
        listen_thread = threading.Thread(target=self.listen_for_messages)
        listen_thread.daemon = True
        listen_thread.start()
        
        # 启动温度变化模拟线程
        temp_thread = threading.Thread(target=self.simulate_temperature_changes)
        temp_thread.daemon = True
        temp_thread.start()
        
        try:
            # 主循环：定期发送消息
            last_heartbeat = 0
            last_report = 0
            last_query = 0
            
            while self.running:
                current_time = time.time()
                
                if self.connected:
                    # 每30秒发送心跳
                    if current_time - last_heartbeat >= 30:
                        if not self.send_heartbeat():
                            break
                        last_heartbeat = current_time
                    
                    # 每10秒发送温度报告
                    if current_time - last_report >= 10:
                        if not self.send_temperature_report():
                            break
                        last_report = current_time
                    
                    # 每60秒发送一次查询（随机类型）
                    if current_time - last_query >= 60:
                        query_types = ["temperature", "status", "config"]
                        query_type = random.choice(query_types)
                        self.send_query(query_type)
                        last_query = current_time
                
                time.sleep(1)  # 主循环1秒检查一次
                
        except KeyboardInterrupt:
            print(f"\n[{datetime.now()}] 用户中断")
        finally:
            self.disconnect()

def main():
    print("=== 分布式空调控制器测试客户端 (新协议版本) ===")
    print("功能特性:")
    print("✓ 支持完整的消息协议 (心跳、查询、报告、命令、响应、错误)")
    print("✓ 使用长度头的TCP数据包格式")
    print("✓ 自动生成MAC地址和消息ID")
    print("✓ 模拟温度控制和环境变化")
    print("✓ 支持多客户端并发测试")
    print()
    print("使用说明:")
    print("1. 确保ESP32设备已启动并创建WiFi热点")
    print("2. 连接到热点 'ND_ESP_TEST' (密码: 1234567890)")
    print("3. 运行此脚本测试TCP通信")
    print("4. 按 Ctrl+C 停止客户端")
    print()
    
    # 创建多个客户端进行测试
    clients = []
    
    try:
        num_clients = int(input("输入要创建的客户端数量 (1-5, 默认1): ") or "1")
        num_clients = max(1, min(5, num_clients))
        
        print(f"\n创建 {num_clients} 个测试客户端...")
        
        # 创建客户端
        for i in range(num_clients):
            # 为每个客户端生成不同的初始温度
            client = TestClient()
            client.current_temp = round(random.uniform(20.0, 30.0), 1)
            client.target_temp = round(random.uniform(22.0, 26.0), 1)
            clients.append(client)
        
        # 启动所有客户端
        threads = []
        for i, client in enumerate(clients):
            thread = threading.Thread(target=client.run)
            thread.daemon = True
            thread.start()
            threads.append(thread)
            print(f"客户端 {i+1} 已启动 (MAC: {client.mac_address}, 初始温度: {client.current_temp}°C)")
            time.sleep(1)  # 错开启动时间
        
        print(f"\n所有客户端已启动，开始模拟分布式空调控制...")
        print("客户端将自动发送心跳、温度报告和查询请求")
        print("您可以通过Web界面或其他客户端发送控制命令进行测试")
        
        # 等待所有线程
        for thread in threads:
            thread.join()
            
    except KeyboardInterrupt:
        print("\n正在停止所有客户端...")
        for client in clients:
            client.disconnect()
    except Exception as e:
        print(f"错误: {e}")

if __name__ == "__main__":
    main()