16337
766ee6a69a
- notify_dispatch: _mark_false_alarm 使用 beijing_now() + 计算 duration_ms - alarm_event_service: handle_alarm 处理时自动计算 duration_ms 和 last_frame_time - notification_service: datetime.utcnow() 替换为 beijing_now() - device_service: datetime.now(timezone.utc) 替换为 beijing_now() Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
134 lines
3.8 KiB
Python
134 lines
3.8 KiB
Python
"""
|
|
设备服务
|
|
管理边缘设备状态
|
|
"""
|
|
from datetime import datetime, timezone, timedelta
|
|
from typing import Dict, Any, List, Optional
|
|
|
|
from app.models import EdgeDevice, DeviceStatus, get_session
|
|
from app.utils.logger import logger
|
|
from app.utils.timezone import beijing_now
|
|
|
|
|
|
class DeviceService:
|
|
"""设备服务"""
|
|
|
|
# 设备离线超时时间（秒）
|
|
OFFLINE_TIMEOUT = 90
|
|
|
|
def __init__(self):
|
|
self._devices: Dict[str, Dict[str, Any]] = {} # 内存缓存
|
|
|
|
def handle_heartbeat(self, heartbeat_data: Dict[str, Any]) -> Optional[EdgeDevice]:
|
|
"""
|
|
处理心跳消息（已废弃 - 边缘端未实现心跳机制）
|
|
|
|
保留此方法以兼容旧代码，实际上不会被调用。
|
|
边缘节点信息应通过手动创建或配置管理。
|
|
"""
|
|
logger.warning("handle_heartbeat 被调用，但心跳机制已废弃，忽略")
|
|
return None
|
|
|
|
def check_offline_devices(self) -> List[EdgeDevice]:
|
|
"""检查离线设备"""
|
|
db = get_session()
|
|
try:
|
|
threshold = beijing_now() - timedelta(seconds=self.OFFLINE_TIMEOUT)
|
|
|
|
# 查找需要标记为离线的设备
|
|
devices = db.query(EdgeDevice).filter(
|
|
EdgeDevice.status == DeviceStatus.ONLINE,
|
|
EdgeDevice.last_heartbeat < threshold
|
|
).all()
|
|
|
|
offline_devices = []
|
|
for device in devices:
|
|
device.status = DeviceStatus.OFFLINE
|
|
device.updated_at = beijing_now()
|
|
offline_devices.append(device)
|
|
logger.info(f"设备离线: {device.device_id}")
|
|
|
|
if offline_devices:
|
|
db.commit()
|
|
|
|
return offline_devices
|
|
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.error(f"检查离线设备失败: {e}")
|
|
return []
|
|
finally:
|
|
db.close()
|
|
|
|
def get_device(self, device_id: str) -> Optional[EdgeDevice]:
|
|
"""获取设备信息"""
|
|
db = get_session()
|
|
try:
|
|
return db.query(EdgeDevice).filter(
|
|
EdgeDevice.device_id == device_id
|
|
).first()
|
|
finally:
|
|
db.close()
|
|
|
|
def get_devices(
|
|
self,
|
|
status: Optional[str] = None,
|
|
page: int = 1,
|
|
page_size: int = 20
|
|
) -> tuple[List[EdgeDevice], int]:
|
|
"""获取设备列表"""
|
|
db = get_session()
|
|
try:
|
|
query = db.query(EdgeDevice)
|
|
|
|
if status:
|
|
query = query.filter(EdgeDevice.status == status)
|
|
|
|
total = query.count()
|
|
devices = (
|
|
query.order_by(EdgeDevice.last_heartbeat.desc())
|
|
.offset((page - 1) * page_size)
|
|
.limit(page_size)
|
|
.all()
|
|
)
|
|
|
|
return devices, total
|
|
finally:
|
|
db.close()
|
|
|
|
def get_statistics(self) -> Dict[str, Any]:
|
|
"""获取设备统计"""
|
|
db = get_session()
|
|
try:
|
|
total = db.query(EdgeDevice).count()
|
|
online = db.query(EdgeDevice).filter(
|
|
EdgeDevice.status == DeviceStatus.ONLINE
|
|
).count()
|
|
offline = db.query(EdgeDevice).filter(
|
|
EdgeDevice.status == DeviceStatus.OFFLINE
|
|
).count()
|
|
error = db.query(EdgeDevice).filter(
|
|
EdgeDevice.status == DeviceStatus.ERROR
|
|
).count()
|
|
|
|
return {
|
|
"total": total,
|
|
"online": online,
|
|
"offline": offline,
|
|
"error": error,
|
|
}
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
# 全局单例
|
|
_device_service = None
|
|
|
|
|
|
def get_device_service() -> DeviceService:
|
|
"""获取设备服务单例"""
|
|
global _device_service
|
|
if _device_service is None:
|
|
_device_service = DeviceService()
|
|
return _device_service
|