241 lines
8.5 KiB
Python
241 lines
8.5 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
ESP32云端报警上报功能集成测试脚本
|
||
验证ESP32能否成功上报报警到云端服务器
|
||
"""
|
||
|
||
import requests
|
||
import time
|
||
import json
|
||
from datetime import datetime
|
||
|
||
# 配置
|
||
ESP32_IP = "192.168.1.3"
|
||
ESP32_PORT = 80
|
||
CLOUD_SERVER = "http://115.190.167.176:36988"
|
||
DEVICE_KEY = "39dc753bd0cb19da5050a2f0edf932f851f5b21b90a6d06ecf9a28d81079a0a6"
|
||
|
||
def test_esp32_connection():
|
||
"""测试ESP32连接"""
|
||
print("🔍 测试ESP32连接...")
|
||
try:
|
||
response = requests.get(f"http://{ESP32_IP}:{ESP32_PORT}/api/status", timeout=5)
|
||
if response.status_code == 200:
|
||
print(f"✅ ESP32连接正常: {response.json()}")
|
||
return True
|
||
else:
|
||
print(f"❌ ESP32响应异常: {response.status_code}")
|
||
return False
|
||
except Exception as e:
|
||
print(f"❌ ESP32连接失败: {e}")
|
||
return False
|
||
|
||
def get_esp32_device_info():
|
||
"""获取ESP32设备信息"""
|
||
print("📱 获取ESP32设备信息...")
|
||
try:
|
||
response = requests.get(f"http://{ESP32_IP}:{ESP32_PORT}/api/device/info", timeout=5)
|
||
if response.status_code == 200:
|
||
device_info = response.json()
|
||
print(f"✅ 设备信息: {json.dumps(device_info, indent=2)}")
|
||
return device_info
|
||
else:
|
||
print(f"❌ 获取设备信息失败: {response.status_code}")
|
||
return None
|
||
except Exception as e:
|
||
print(f"❌ 获取设备信息异常: {e}")
|
||
return None
|
||
|
||
def get_esp32_alarm_history():
|
||
"""获取ESP32报警历史"""
|
||
print("📋 获取ESP32报警历史...")
|
||
try:
|
||
response = requests.get(f"http://{ESP32_IP}:{ESP32_PORT}/api/alarm/history", timeout=5)
|
||
if response.status_code == 200:
|
||
history = response.json()
|
||
print(f"✅ 报警历史: {json.dumps(history, indent=2)}")
|
||
return history
|
||
else:
|
||
print(f"❌ 获取报警历史失败: {response.status_code}")
|
||
return None
|
||
except Exception as e:
|
||
print(f"❌ 获取报警历史异常: {e}")
|
||
return None
|
||
|
||
def trigger_test_alarm():
|
||
"""触发测试报警（如果ESP32支持）"""
|
||
print("🚨 尝试触发测试报警...")
|
||
try:
|
||
# 尝试通过API触发测试报警
|
||
test_data = {
|
||
"action": "test_alarm",
|
||
"type": "burning",
|
||
"message": "集成测试报警",
|
||
"temperature": 45.5
|
||
}
|
||
response = requests.post(f"http://{ESP32_IP}:{ESP32_PORT}/api/test/alarm",
|
||
json=test_data, timeout=5)
|
||
if response.status_code == 200:
|
||
print("✅ 测试报警触发成功")
|
||
return True
|
||
else:
|
||
print(f"⚠️ 测试报警API不可用: {response.status_code}")
|
||
return False
|
||
except Exception as e:
|
||
print(f"⚠️ 测试报警触发失败: {e}")
|
||
return False
|
||
|
||
def check_cloud_server():
|
||
"""检查云端服务器状态"""
|
||
print("☁️ 检查云端服务器状态...")
|
||
try:
|
||
response = requests.get(f"{CLOUD_SERVER}/health", timeout=10)
|
||
if response.status_code == 200:
|
||
print("✅ 云端服务器正常")
|
||
return True
|
||
else:
|
||
print(f"❌ 云端服务器异常: {response.status_code}")
|
||
return False
|
||
except Exception as e:
|
||
print(f"❌ 云端服务器连接失败: {e}")
|
||
return False
|
||
|
||
def test_cloud_alarm_endpoint():
|
||
"""测试云端报警接收端点"""
|
||
print("🧪 测试云端报警接收端点...")
|
||
try:
|
||
# 模拟ESP32发送的报警数据
|
||
test_alarm = {
|
||
"physicalUid": "206EF1B3AD74",
|
||
"sourceEventId": f"206EF1B3AD74:{int(time.time() * 1000)}",
|
||
"title": "集成测试报警",
|
||
"level": "warning",
|
||
"message": "这是一个集成测试报警，用于验证ESP32云端上报功能",
|
||
"temp": 35.8,
|
||
"occurredAtMs": int(time.time() * 1000)
|
||
}
|
||
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"X-Device-Key": DEVICE_KEY
|
||
}
|
||
|
||
response = requests.post(f"{CLOUD_SERVER}/device-alarms",
|
||
json=test_alarm, headers=headers, timeout=10)
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
print(f"✅ 云端报警接收成功: {result}")
|
||
return True
|
||
else:
|
||
print(f"❌ 云端报警接收失败: {response.status_code}")
|
||
print(f"响应内容: {response.text}")
|
||
return False
|
||
except Exception as e:
|
||
print(f"❌ 云端报警测试异常: {e}")
|
||
return False
|
||
|
||
def monitor_esp32_logs():
|
||
"""监控ESP32日志输出（提示用户手动操作）"""
|
||
print("📺 ESP32日志监控指导:")
|
||
print("=" * 50)
|
||
print("请在另一个终端中运行以下命令监控ESP32日志:")
|
||
print(f"cd d:\\XinJiaPo\\Firefly_code\\smart-home\\firefly_esp32")
|
||
print(f"idf.py -p COM7 monitor")
|
||
print()
|
||
print("关键日志标识:")
|
||
print("- 🌐 CloudAlarmReporter: 初始化云端报警上报模块")
|
||
print("- ✅ CloudAlarmReporter: 云端报警上报模块初始化成功")
|
||
print("- ☁️ AlarmManager: 报警已加入云端上报队列")
|
||
print("- 📨 CloudAlarmReporter: 开始上报报警")
|
||
print("- 📡 CloudAlarmReporter: HTTP响应状态码: 200")
|
||
print("- ✅ CloudAlarmReporter: 报警上报成功")
|
||
print("=" * 50)
|
||
|
||
def wait_for_real_alarm():
|
||
"""等待真实报警触发"""
|
||
print("⏳ 等待真实报警触发...")
|
||
print("请通过以下方式触发报警:")
|
||
print("1. 用手接触热成像传感器（触发高温报警）")
|
||
print("2. 快速移动热源（触发异常升温报警）")
|
||
print("3. 等待系统自动检测")
|
||
print()
|
||
print("按 Enter 键继续检查报警历史...")
|
||
input()
|
||
|
||
def main():
|
||
"""主测试流程"""
|
||
print("🧪 ESP32云端报警上报功能集成测试")
|
||
print("=" * 60)
|
||
print(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print(f"ESP32地址: {ESP32_IP}:{ESP32_PORT}")
|
||
print(f"云端服务器: {CLOUD_SERVER}")
|
||
print("=" * 60)
|
||
|
||
# 测试步骤
|
||
test_results = {}
|
||
|
||
# 1. 测试ESP32连接
|
||
test_results['esp32_connection'] = test_esp32_connection()
|
||
|
||
# 2. 获取设备信息
|
||
device_info = get_esp32_device_info()
|
||
test_results['device_info'] = device_info is not None
|
||
|
||
# 3. 检查云端服务器
|
||
test_results['cloud_server'] = check_cloud_server()
|
||
|
||
# 4. 测试云端报警端点
|
||
test_results['cloud_endpoint'] = test_cloud_alarm_endpoint()
|
||
|
||
# 5. 获取当前报警历史
|
||
print("\n📋 获取当前报警历史作为基准...")
|
||
initial_history = get_esp32_alarm_history()
|
||
initial_count = len(initial_history.get('alarms', [])) if initial_history else 0
|
||
print(f"当前报警历史数量: {initial_count}")
|
||
|
||
# 6. 尝试触发测试报警
|
||
test_results['test_alarm'] = trigger_test_alarm()
|
||
|
||
# 7. 监控指导
|
||
monitor_esp32_logs()
|
||
|
||
# 8. 等待真实报警
|
||
wait_for_real_alarm()
|
||
|
||
# 9. 检查报警历史变化
|
||
print("\n📋 检查报警历史变化...")
|
||
final_history = get_esp32_alarm_history()
|
||
final_count = len(final_history.get('alarms', [])) if final_history else 0
|
||
|
||
if final_count > initial_count:
|
||
print(f"✅ 检测到新报警! 数量从 {initial_count} 增加到 {final_count}")
|
||
new_alarms = final_history.get('alarms', [])[:final_count - initial_count]
|
||
for alarm in new_alarms:
|
||
print(f" 📨 新报警: {alarm}")
|
||
else:
|
||
print(f"⚠️ 未检测到新报警 (当前: {final_count}, 之前: {initial_count})")
|
||
|
||
# 测试结果总结
|
||
print("\n🎯 测试结果总结:")
|
||
print("=" * 40)
|
||
for test_name, result in test_results.items():
|
||
status = "✅ 通过" if result else "❌ 失败"
|
||
print(f"{test_name}: {status}")
|
||
|
||
success_rate = sum(test_results.values()) / len(test_results) * 100
|
||
print(f"\n总体成功率: {success_rate:.1f}%")
|
||
|
||
if success_rate >= 80:
|
||
print("🎉 集成测试基本通过！")
|
||
else:
|
||
print("⚠️ 集成测试需要进一步调试")
|
||
|
||
print("\n💡 下一步验证:")
|
||
print("1. 检查云端数据库是否有新的alarm_records记录")
|
||
print("2. 在APP中查看报警历史是否同步")
|
||
print("3. 验证离线状态下APP是否能显示云端历史")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|