187 lines
7.4 KiB
Python
187 lines
7.4 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
验证ESP32云端报警上报功能
|
||
通过模拟云端服务器来测试ESP32的上报功能
|
||
"""
|
||
|
||
import socket
|
||
import threading
|
||
import json
|
||
import time
|
||
from datetime import datetime
|
||
from http.server import HTTPServer, BaseHTTPRequestHandler
|
||
import urllib.parse
|
||
|
||
class CloudAlarmHandler(BaseHTTPRequestHandler):
|
||
"""模拟云端报警接收服务器"""
|
||
|
||
received_alarms = []
|
||
|
||
def do_POST(self):
|
||
"""处理POST请求"""
|
||
if self.path == '/device-alarms':
|
||
try:
|
||
# 读取请求体
|
||
content_length = int(self.headers.get('Content-Length', 0))
|
||
post_data = self.rfile.read(content_length)
|
||
|
||
# 解析JSON数据
|
||
alarm_data = json.loads(post_data.decode('utf-8'))
|
||
|
||
# 检查设备密钥
|
||
device_key = self.headers.get('X-Device-Key')
|
||
expected_key = "39dc753bd0cb19da5050a2f0edf932f851f5b21b90a6d06ecf9a28d81079a0a6"
|
||
|
||
print(f"\n📨 收到ESP32报警上报:")
|
||
print(f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print(f"设备密钥: {device_key}")
|
||
print(f"报警数据: {json.dumps(alarm_data, indent=2, ensure_ascii=False)}")
|
||
|
||
if device_key == expected_key:
|
||
# 保存报警数据
|
||
alarm_data['received_at'] = datetime.now().isoformat()
|
||
CloudAlarmHandler.received_alarms.append(alarm_data)
|
||
|
||
# 返回成功响应
|
||
response = {
|
||
"code": 200,
|
||
"message": "报警上报成功",
|
||
"data": {
|
||
"alarmId": len(CloudAlarmHandler.received_alarms),
|
||
"receivedAt": alarm_data['received_at']
|
||
}
|
||
}
|
||
|
||
self.send_response(200)
|
||
self.send_header('Content-Type', 'application/json')
|
||
self.end_headers()
|
||
self.wfile.write(json.dumps(response).encode('utf-8'))
|
||
|
||
print(f"✅ 报警接收成功! 总计收到 {len(CloudAlarmHandler.received_alarms)} 个报警")
|
||
else:
|
||
# 设备密钥错误
|
||
self.send_response(401)
|
||
self.send_header('Content-Type', 'application/json')
|
||
self.end_headers()
|
||
error_response = {"code": 401, "message": "设备密钥验证失败"}
|
||
self.wfile.write(json.dumps(error_response).encode('utf-8'))
|
||
print(f"❌ 设备密钥验证失败")
|
||
|
||
except Exception as e:
|
||
print(f"❌ 处理报警数据时出错: {e}")
|
||
self.send_response(500)
|
||
self.send_header('Content-Type', 'application/json')
|
||
self.end_headers()
|
||
error_response = {"code": 500, "message": f"服务器内部错误: {str(e)}"}
|
||
self.wfile.write(json.dumps(error_response).encode('utf-8'))
|
||
else:
|
||
# 其他路径返回404
|
||
self.send_response(404)
|
||
self.end_headers()
|
||
|
||
def do_GET(self):
|
||
"""处理GET请求"""
|
||
if self.path == '/health':
|
||
self.send_response(200)
|
||
self.send_header('Content-Type', 'application/json')
|
||
self.end_headers()
|
||
response = {"status": "ok", "message": "模拟云端服务器运行正常"}
|
||
self.wfile.write(json.dumps(response).encode('utf-8'))
|
||
else:
|
||
self.send_response(404)
|
||
self.end_headers()
|
||
|
||
def log_message(self, format, *args):
|
||
"""禁用默认日志输出"""
|
||
pass
|
||
|
||
def start_mock_cloud_server(port=36988):
|
||
"""启动模拟云端服务器"""
|
||
try:
|
||
server = HTTPServer(('0.0.0.0', port), CloudAlarmHandler)
|
||
print(f"🌐 模拟云端服务器启动成功")
|
||
print(f"监听地址: http://0.0.0.0:{port}")
|
||
print(f"报警接收端点: http://0.0.0.0:{port}/device-alarms")
|
||
print("=" * 50)
|
||
server.serve_forever()
|
||
except Exception as e:
|
||
print(f"❌ 启动模拟云端服务器失败: {e}")
|
||
|
||
def trigger_esp32_alarm():
|
||
"""尝试触发ESP32报警（通过接触传感器）"""
|
||
print("\n🔥 触发ESP32报警指导:")
|
||
print("=" * 40)
|
||
print("请执行以下操作之一来触发报警:")
|
||
print("1. 用手指接触热成像传感器表面")
|
||
print("2. 用打火机或热源靠近传感器")
|
||
print("3. 快速移动热源触发异常升温")
|
||
print("4. 等待系统自动检测")
|
||
print("=" * 40)
|
||
print("触发报警后，ESP32应该会自动上报到模拟云端服务器")
|
||
|
||
def monitor_received_alarms():
|
||
"""监控接收到的报警"""
|
||
print("\n📊 报警监控中...")
|
||
last_count = 0
|
||
|
||
while True:
|
||
current_count = len(CloudAlarmHandler.received_alarms)
|
||
if current_count > last_count:
|
||
print(f"\n🚨 新报警通知! 总计: {current_count}")
|
||
for i in range(last_count, current_count):
|
||
alarm = CloudAlarmHandler.received_alarms[i]
|
||
print(f"报警 #{i+1}:")
|
||
print(f" 设备ID: {alarm.get('physicalUid')}")
|
||
print(f" 标题: {alarm.get('title')}")
|
||
print(f" 级别: {alarm.get('level')}")
|
||
print(f" 温度: {alarm.get('temp')}°C")
|
||
print(f" 消息: {alarm.get('message')}")
|
||
print(f" 发生时间: {datetime.fromtimestamp(alarm.get('occurredAtMs', 0)/1000)}")
|
||
print(f" 接收时间: {alarm.get('received_at')}")
|
||
last_count = current_count
|
||
|
||
time.sleep(2) # 每2秒检查一次
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print("🧪 ESP32云端报警上报功能验证")
|
||
print("=" * 50)
|
||
print(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print("=" * 50)
|
||
|
||
# 启动模拟云端服务器（在后台线程中）
|
||
server_thread = threading.Thread(target=start_mock_cloud_server, daemon=True)
|
||
server_thread.start()
|
||
|
||
# 等待服务器启动
|
||
time.sleep(2)
|
||
|
||
# 显示触发指导
|
||
trigger_esp32_alarm()
|
||
|
||
# 开始监控报警
|
||
try:
|
||
monitor_received_alarms()
|
||
except KeyboardInterrupt:
|
||
print(f"\n\n🎯 测试结果总结:")
|
||
print("=" * 30)
|
||
print(f"总共接收到报警: {len(CloudAlarmHandler.received_alarms)} 个")
|
||
|
||
if CloudAlarmHandler.received_alarms:
|
||
print("✅ ESP32云端报警上报功能正常!")
|
||
print("\n📋 接收到的报警列表:")
|
||
for i, alarm in enumerate(CloudAlarmHandler.received_alarms):
|
||
print(f"{i+1}. {alarm.get('title')} - {alarm.get('level')} - {alarm.get('temp')}°C")
|
||
else:
|
||
print("⚠️ 未接收到任何报警")
|
||
print("可能的原因:")
|
||
print("1. ESP32未连接到网络")
|
||
print("2. CloudAlarmReporter模块未正确初始化")
|
||
print("3. 报警未被触发")
|
||
print("4. 网络配置问题")
|
||
|
||
print(f"\n💡 如果测试成功，说明ESP32能够正确上报报警到云端!")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|