150 lines
5.7 KiB
Python
150 lines
5.7 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
测试ESP32云端报警上报到生产环境
|
||
验证ESP32能否成功上报到 http://115.190.167.176:36988/device-alarms
|
||
"""
|
||
|
||
import requests
|
||
import json
|
||
import time
|
||
from datetime import datetime
|
||
|
||
def test_production_cloud_server():
|
||
"""测试生产环境云端服务器"""
|
||
print("☁️ 测试生产环境云端服务器...")
|
||
|
||
try:
|
||
# 测试健康检查端点
|
||
response = requests.get("http://115.190.167.176:36988/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_production_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": "39dc753bd0cb19da5050a2f0edf932f851f5b21b90a6d06ecf9a28d81079a0a6"
|
||
}
|
||
|
||
response = requests.post("http://115.190.167.176:36988/device-alarms",
|
||
json=test_alarm, headers=headers, timeout=15)
|
||
|
||
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_alarm_history():
|
||
"""监控ESP32报警历史变化"""
|
||
print("\n📋 监控ESP32报警历史变化...")
|
||
|
||
try:
|
||
response = requests.get("http://192.168.1.3:80/api/alarm/history", timeout=5)
|
||
if response.status_code == 200:
|
||
history = response.json()
|
||
alarms = history.get('alarms', [])
|
||
print(f"📊 当前ESP32报警历史: {len(alarms)} 个")
|
||
|
||
if alarms:
|
||
latest = alarms[0] # 最新的报警
|
||
print(f"最新报警: {latest.get('title')} - {latest.get('time')} - {latest.get('temp')}°C")
|
||
|
||
return len(alarms)
|
||
else:
|
||
print("❌ 获取ESP32报警历史失败")
|
||
return 0
|
||
except Exception as e:
|
||
print(f"❌ 获取ESP32报警历史异常: {e}")
|
||
return 0
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print("🚀 ESP32生产环境云端报警上报测试")
|
||
print("=" * 60)
|
||
print(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print(f"生产服务器: http://115.190.167.176:36988/device-alarms")
|
||
print(f"ESP32地址: http://192.168.1.3:80")
|
||
print("=" * 60)
|
||
|
||
# 1. 测试生产环境云端服务器
|
||
cloud_ok = test_production_cloud_server()
|
||
|
||
# 2. 测试生产环境报警端点
|
||
endpoint_ok = test_production_alarm_endpoint()
|
||
|
||
# 3. 获取ESP32当前报警历史
|
||
initial_count = monitor_esp32_alarm_history()
|
||
|
||
print(f"\n🎯 测试结果:")
|
||
print("=" * 30)
|
||
print(f"生产环境服务器: {'✅ 正常' if cloud_ok else '❌ 异常'}")
|
||
print(f"报警接收端点: {'✅ 正常' if endpoint_ok else '❌ 异常'}")
|
||
print(f"ESP32报警历史: {initial_count} 个")
|
||
|
||
if cloud_ok and endpoint_ok:
|
||
print(f"\n🎉 生产环境准备就绪！")
|
||
print(f"\n📋 验证步骤:")
|
||
print("1. 确认ESP32已重启并显示新的服务器地址")
|
||
print("2. 手动触发报警（接触热成像传感器）")
|
||
print("3. 观察ESP32串口日志中的云端上报信息")
|
||
print("4. 检查云端数据库是否有新的alarm_records记录")
|
||
|
||
print(f"\n🔍 预期ESP32串口日志:")
|
||
print("I CloudAlarmReporter: 🔗 目标服务器: http://115.190.167.176:36988/device-alarms")
|
||
print("I AlarmManager: 🌐 准备上报报警到云端: [燃烧] 检测到燃烧...")
|
||
print("I AlarmManager: ☁️ 报警已成功加入云端上报队列")
|
||
print("I CloudAlarmReporter: 📨 开始上报报警: [燃烧报警] 206EF1B3AD74:xxx")
|
||
print("I CloudAlarmReporter: 📡 HTTP响应状态码: 200")
|
||
print("I CloudAlarmReporter: ✅ 报警上报成功")
|
||
|
||
print(f"\n⏳ 等待真实报警触发...")
|
||
print("按 Ctrl+C 结束监控")
|
||
|
||
try:
|
||
while True:
|
||
time.sleep(5)
|
||
new_count = monitor_esp32_alarm_history()
|
||
if new_count > initial_count:
|
||
print(f"🚨 检测到新报警! 数量从 {initial_count} 增加到 {new_count}")
|
||
print("✅ 请检查云端数据库是否有对应的alarm_records记录")
|
||
initial_count = new_count
|
||
except KeyboardInterrupt:
|
||
print(f"\n\n🎯 测试结束")
|
||
else:
|
||
print(f"\n❌ 生产环境存在问题，请检查:")
|
||
if not cloud_ok:
|
||
print("- 云端服务器无法访问")
|
||
if not endpoint_ok:
|
||
print("- 报警接收端点异常")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|