XW-AIOT/iot-device-management-service
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(service): 删除MQTT旧代码 + 修复边缘节点重复显示问题

Browse Source 
**删除MQTT旧代码：**
- 删除 mqtt_service.py（已废弃的空壳）
- 从 config.py 删除 MQTTConfig 类和相关配置
- 从 schemas.py 删除 mqtt 字段
- 从 alert_service.py 删除 create_alert_from_mqtt 方法
- 告警上报已改为 HTTP + COS 方案，MQTT机制完全废弃

**修复边缘节点重复显示（方案A）：**
- 清理 edge_devices 表历史数据（删除 edge_device_001、edge_inference_device）
- 禁用 DeviceService 的 handle_heartbeat 自动创建设备功能
- 边缘端未实现心跳机制，告警数从 alarm_event 表统计
- 运行时长、处理帧数字段设为 null（无心跳机制，不可用）
- 添加 count_alarms_by_edge_node 方法统计边缘节点告警数

**影响范围：**
- /admin-api/aiot/edge/device/page 接口返回数据调整
- /admin-api/aiot/edge/device/get 接口返回数据调整
- 确保不破坏现有功能（告警上报已改为HTTP）
This commit is contained in:
16337
2026-02-25 10:30:01 +08:00
parent 3bac8be70e
commit f81cc81ce6
7 changed files with 48 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
app/config.py
View File

@@ -43,20 +43,6 @@ class AIModelConfig:
api_key: str = ""
@dataclass
class MQTTConfig:
"""MQTT 配置 (已废弃 - 告警上报已改为 HTTP + COS)"""
broker_host: str = "localhost"
broker_port: int = 1883
client_id: str = "alert_platform"
username: str = ""
password: str = ""
alert_topic: str = "edge/alert/#"
heartbeat_topic: str = "edge/alert/heartbeat/#"
qos: int = 1
enabled: bool = False # 默认禁用
@dataclass
class RedisConfig:
"""Redis 配置"""
@@ -103,7 +89,6 @@ class Settings(BaseModel):
cos: COSConfig = COSConfig()
app: AppConfig = AppConfig()
ai_model: AIModelConfig = AIModelConfig()
mqtt: MQTTConfig = MQTTConfig()
redis: RedisConfig = RedisConfig()
camera_name: CameraNameConfig = CameraNameConfig()
@@ -137,17 +122,6 @@ def load_settings() -> Settings:
endpoint=os.getenv("AI_MODEL_ENDPOINT", ""),
api_key=os.getenv("AI_MODEL_API_KEY", ""),
),
mqtt=MQTTConfig(
broker_host=os.getenv("MQTT_BROKER_HOST", "localhost"),
broker_port=int(os.getenv("MQTT_BROKER_PORT", "1883")),
client_id=os.getenv("MQTT_CLIENT_ID", "alert_platform"),
username=os.getenv("MQTT_USERNAME", ""),
password=os.getenv("MQTT_PASSWORD", ""),
alert_topic=os.getenv("MQTT_ALERT_TOPIC", "edge/alert/#"),
heartbeat_topic=os.getenv("MQTT_HEARTBEAT_TOPIC", "edge/alert/heartbeat/#"),
qos=int(os.getenv("MQTT_QOS", "1")),
enabled=os.getenv("MQTT_ENABLED", "false").lower() == "true",
),
redis=RedisConfig(
host=os.getenv("REDIS_HOST", "localhost"),
port=int(os.getenv("REDIS_PORT", "6379")),

36
app/routers/yudao_aiot_edge.py
View File

@@ -7,6 +7,9 @@ API 路径规范：
- /admin-api/aiot/edge/device/page - 分页查询设备
- /admin-api/aiot/edge/device/get - 获取设备详情
- /admin-api/aiot/edge/device/statistics - 设备统计
注意：边缘端未实现心跳机制，因此运行时长/处理帧数等实时指标不可用。
告警数统计从 alarm_event 表中提取 edge_node_id 字段。
"""
from fastapi import APIRouter, Query, Depends, HTTPException
@@ -14,6 +17,7 @@ from typing import Optional
from app.yudao_compat import YudaoResponse, get_current_user
from app.services.device_service import get_device_service, DeviceService
from app.services.alarm_event_service import get_alarm_event_service, AlarmEventService
router = APIRouter(prefix="/admin-api/aiot/edge", tags=["AIoT-边缘设备"])
@@ -24,9 +28,10 @@ async def get_device_page(
pageSize: int = Query(20, ge=1, le=100, description="每页大小"),
status: Optional[str] = Query(None, description="设备状态: online/offline/error"),
service: DeviceService = Depends(get_device_service),
alarm_service: AlarmEventService = Depends(get_alarm_event_service),
current_user: dict = Depends(get_current_user)
):
"""分页查询边缘设备列表"""
"""分页查询边缘设备列表（告警数从 alarm_event 表统计）"""
devices, total = service.get_devices(
status=status,
page=pageNo,
@@ -36,16 +41,21 @@ async def get_device_page(
device_list = []
for device in devices:
device_dict = device.to_dict()
device_id = device_dict.get("device_id")
# 从 alarm_event 表统计告警数
alerts_count = alarm_service.count_alarms_by_edge_node(device_id)
device_list.append({
"id": device_dict.get("id"),
"deviceId": device_dict.get("device_id"),
"deviceId": device_id,
"deviceName": device_dict.get("device_name"),
"status": device_dict.get("status"),
"statusName": _get_status_name(device_dict.get("status")),
"lastHeartbeat": device_dict.get("last_heartbeat"),
"uptimeSeconds": device_dict.get("uptime_seconds"),
"framesProcessed": device_dict.get("frames_processed"),
"alertsGenerated": device_dict.get("alerts_generated"),
"uptimeSeconds": None, # 无心跳机制，不可用
"framesProcessed": None, # 无心跳机制，不可用
"alertsGenerated": alerts_count, # 从 alarm_event 表统计
"ipAddress": device_dict.get("ip_address"),
"streamCount": device_dict.get("stream_count"),
"configVersion": device_dict.get("config_version"),
@@ -65,24 +75,30 @@ async def get_device_page(
async def get_device(
id: str = Query(..., description="设备ID"),
service: DeviceService = Depends(get_device_service),
alarm_service: AlarmEventService = Depends(get_alarm_event_service),
current_user: dict = Depends(get_current_user)
):
"""获取设备详情"""
"""获取设备详情（告警数从 alarm_event 表统计）"""
device = service.get_device(id)
if not device:
raise HTTPException(status_code=404, detail="设备不存在")
device_dict = device.to_dict()
device_id = device_dict.get("device_id")
# 从 alarm_event 表统计告警数
alerts_count = alarm_service.count_alarms_by_edge_node(device_id)
return YudaoResponse.success({
"id": device_dict.get("id"),
"deviceId": device_dict.get("device_id"),
"deviceId": device_id,
"deviceName": device_dict.get("device_name"),
"status": device_dict.get("status"),
"statusName": _get_status_name(device_dict.get("status")),
"lastHeartbeat": device_dict.get("last_heartbeat"),
"uptimeSeconds": device_dict.get("uptime_seconds"),
"framesProcessed": device_dict.get("frames_processed"),
"alertsGenerated": device_dict.get("alerts_generated"),
"uptimeSeconds": None, # 无心跳机制，不可用
"framesProcessed": None, # 无心跳机制，不可用
"alertsGenerated": alerts_count, # 从 alarm_event 表统计
"ipAddress": device_dict.get("ip_address"),
"streamCount": device_dict.get("stream_count"),
"configVersion": device_dict.get("config_version"),

1
app/schemas.py
View File

@@ -108,7 +108,6 @@ class DeviceStatisticsResponse(BaseModel):
class HealthResponse(BaseModel):
status: str
database: str
mqtt: Optional[str] = None
websocket_connections: Optional[int] = None

14
app/services/alarm_event_service.py
View File

@@ -588,6 +588,20 @@ class AlarmEventService:
finally:
db.close()
def count_alarms_by_edge_node(self, edge_node_id: str) -> int:
"""统计指定边缘节点的告警数量"""
db = get_session()
try:
count = db.query(AlarmEvent).filter(
AlarmEvent.edge_node_id == edge_node_id
).count()
return count
except Exception as e:
logger.error(f"统计边缘节点告警数失败: {e}")
return 0
finally:
db.close()
def save_llm_analysis(
self,
alarm_id: str,

60
app/services/alert_service.py
View File

@@ -1,6 +1,6 @@
"""
告警服务
处理告警 CRUD 和 MQTT 消息
处理告警 CRUD（旧版告警系统，已迁移至 alarm_event_service）
"""
import uuid
from datetime import datetime, timezone
@@ -63,64 +63,6 @@ class AlertService:
finally:
db.close()
def create_alert_from_mqtt(self, mqtt_data: Dict[str, Any]) -> Optional[Alert]:
"""从 MQTT 消息创建告警"""
db = get_session()
try:
# 解析时间
trigger_time_str = mqtt_data.get("timestamp")
if trigger_time_str:
try:
trigger_time = datetime.fromisoformat(trigger_time_str.replace("Z", "+00:00"))
except ValueError:
trigger_time = datetime.now(timezone.utc)
else:
trigger_time = datetime.now(timezone.utc)
# 解析置信度（MQTT 可能是 0-1，需要转为 0-100）
confidence = mqtt_data.get("confidence")
if confidence is not None:
if isinstance(confidence, float) and confidence <= 1:
confidence = int(confidence * 100)
else:
confidence = int(confidence)
# 解析持续时长
duration_minutes = mqtt_data.get("duration_minutes")
if duration_minutes is not None:
duration_minutes = int(float(duration_minutes))
alert = Alert(
alert_no=self.generate_alert_no(),
camera_id=mqtt_data.get("camera_id", "unknown"),
roi_id=mqtt_data.get("roi_id"),
bind_id=mqtt_data.get("bind_id"),
device_id=mqtt_data.get("device_id"),
alert_type=mqtt_data.get("alert_type", "unknown"),
algorithm=mqtt_data.get("algorithm", "YOLO"),
confidence=confidence,
duration_minutes=duration_minutes,
trigger_time=trigger_time,
message=mqtt_data.get("message"),
bbox=mqtt_data.get("bbox"),
status=AlertStatus.PENDING,
level=self._determine_level(mqtt_data),
)
db.add(alert)
db.commit()
db.refresh(alert)
logger.info(f"MQTT告警创建成功: {alert.alert_no}, type={alert.alert_type}")
return alert
except Exception as e:
db.rollback()
logger.error(f"创建MQTT告警失败: {e}")
return None
finally:
db.close()
def _determine_level(self, data: Dict[str, Any]) -> AlertLevel:
"""根据告警数据确定告警级别"""
alert_type = data.get("alert_type", "")

65
app/services/device_service.py
View File

@@ -19,66 +19,15 @@ class DeviceService:
self._devices: Dict[str, Dict[str, Any]] = {} # 内存缓存
def handle_heartbeat(self, heartbeat_data: Dict[str, Any]) -> Optional[EdgeDevice]:
"""处理心跳消息"""
device_id = heartbeat_data.get("device_id")
if not device_id:
logger.warning("心跳消息缺少 device_id")
"""
处理心跳消息（已废弃 - 边缘端未实现心跳机制）
保留此方法以兼容旧代码，实际上不会被调用。
边缘节点信息应通过手动创建或配置管理。
"""
logger.warning("handle_heartbeat 被调用，但心跳机制已废弃，忽略")
return None
status_info = heartbeat_data.get("status", {})
db = get_session()
try:
# 查找或创建设备
device = db.query(EdgeDevice).filter(
EdgeDevice.device_id == device_id
).first()
now = datetime.now(timezone.utc)
if device is None:
# 新设备
device = EdgeDevice(
device_id=device_id,
device_name=f"边缘设备_{device_id[:8]}",
status=DeviceStatus.ONLINE,
last_heartbeat=now,
)
db.add(device)
logger.info(f"发现新设备: {device_id}")
else:
# 更新状态
old_status = device.status
device.status = DeviceStatus.ONLINE
device.last_heartbeat = now
if old_status != DeviceStatus.ONLINE:
logger.info(f"设备上线: {device_id}")
# 更新运行信息
if status_info:
device.uptime_seconds = status_info.get("uptime_seconds")
device.frames_processed = status_info.get("frames_processed")
device.alerts_generated = status_info.get("alerts_generated")
device.extra_info = status_info.get("stream_stats")
device.updated_at = now
db.commit()
db.refresh(device)
# 更新内存缓存
self._devices[device_id] = device.to_dict()
return device
except Exception as e:
db.rollback()
logger.error(f"处理心跳失败: {e}")
return None
finally:
db.close()
def check_offline_devices(self) -> List[EdgeDevice]:
"""检查离线设备"""
db = get_session()

54
app/services/mqtt_service.py
View File

@@ -1,54 +0,0 @@
"""
MQTT 服务 - 已废弃
告警上报已改为 HTTP + COS 方案（边缘端直传）。
此文件保留为空壳，避免其他模块 import 报错。
后续版本将彻底删除此文件。
"""
from typing import Dict, Any, Optional
class MQTTService:
"""MQTT 服务 (已废弃，保留空壳兼容旧代码)"""
def __init__(self):
pass
@property
def is_connected(self) -> bool:
return False
def register_alert_handler(self, handler):
pass
def register_heartbeat_handler(self, handler):
pass
def start(self):
pass
def stop(self):
pass
def get_statistics(self) -> Dict[str, Any]:
return {
"messages_received": 0,
"alerts_received": 0,
"heartbeats_received": 0,
"errors": 0,
"connected": False,
"running": False,
"deprecated": True,
}
_mqtt_service: Optional[MQTTService] = None
def get_mqtt_service() -> MQTTService:
"""获取 MQTT 服务单例 (已废弃)"""
global _mqtt_service
if _mqtt_service is None:
_mqtt_service = MQTTService()
return _mqtt_service