271 lines
8.3 KiB
Python
271 lines
8.3 KiB
Python
|
"""
|
||
|
|
数据模型定义模块
|
||
|
|
定义配置同步相关的核心数据模型
|
||
|
|
"""
|
||
|
|
|
||
|
|
from dataclasses import dataclass, field
|
||
|
|
from typing import Any, Dict, List, Optional
|
||
|
|
from enum import Enum
|
||
|
|
import json
|
||
|
|
|
||
|
|
|
||
|
|
class ROIType(str, Enum):
|
||
|
|
"""ROI类型枚举"""
|
||
|
|
POLYGON = "polygon"
|
||
|
|
RECTANGLE = "rectangle"
|
||
|
|
|
||
|
|
|
||
|
|
class AlgorithmType(str, Enum):
|
||
|
|
"""算法类型枚举"""
|
||
|
|
LEAVE_POST = "leave_post"
|
||
|
|
INTRUSION = "intrusion"
|
||
|
|
CROWD_DETECTION = "crowd_detection"
|
||
|
|
FACE_RECOGNITION = "face_recognition"
|
||
|
|
|
||
|
|
|
||
|
|
class AlertLevel(str, Enum):
|
||
|
|
"""告警级别枚举"""
|
||
|
|
LOW = "low"
|
||
|
|
MEDIUM = "medium"
|
||
|
|
HIGH = "high"
|
||
|
|
CRITICAL = "critical"
|
||
|
|
|
||
|
|
|
||
|
|
class DeviceStatus(str, Enum):
|
||
|
|
"""设备状态枚举"""
|
||
|
|
ONLINE = "online"
|
||
|
|
OFFLINE = "offline"
|
||
|
|
MAINTAINING = "maintaining"
|
||
|
|
ERROR = "error"
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class CameraInfo:
|
||
|
|
"""摄像头信息数据模型"""
|
||
|
|
camera_id: str
|
||
|
|
rtsp_url: str
|
||
|
|
camera_name: Optional[str] = None
|
||
|
|
status: bool = True
|
||
|
|
enabled: bool = True
|
||
|
|
location: Optional[str] = None
|
||
|
|
extra_params: Optional[Dict[str, Any]] = None
|
||
|
|
|
||
|
|
def to_dict(self) -> Dict[str, Any]:
|
||
|
|
"""转换为字典"""
|
||
|
|
return {
|
||
|
|
"camera_id": self.camera_id,
|
||
|
|
"camera_name": self.camera_name,
|
||
|
|
"rtsp_url": self.rtsp_url,
|
||
|
|
"status": self.status,
|
||
|
|
"enabled": self.enabled,
|
||
|
|
"location": self.location,
|
||
|
|
"extra_params": self.extra_params,
|
||
|
|
}
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def from_dict(cls, data: Dict[str, Any]) -> 'CameraInfo':
|
||
|
|
"""从字典创建实例"""
|
||
|
|
return cls(
|
||
|
|
camera_id=data.get("camera_id", ""),
|
||
|
|
camera_name=data.get("camera_name"),
|
||
|
|
rtsp_url=data.get("rtsp_url", ""),
|
||
|
|
status=data.get("status", True),
|
||
|
|
enabled=data.get("enabled", True),
|
||
|
|
location=data.get("location"),
|
||
|
|
extra_params=data.get("extra_params"),
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class CoordinatePoint:
|
||
|
|
"""坐标点数据模型"""
|
||
|
|
x: float
|
||
|
|
y: float
|
||
|
|
|
||
|
|
def to_list(self) -> List[float]:
|
||
|
|
"""转换为列表"""
|
||
|
|
return [self.x, self.y]
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def from_list(cls, data: List[float]) -> 'CoordinatePoint':
|
||
|
|
"""从列表创建实例"""
|
||
|
|
return cls(x=data[0], y=data[1]) if len(data) >= 2 else cls(x=0, y=0)
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class ROIInfo:
|
||
|
|
"""ROI区域信息数据模型"""
|
||
|
|
roi_id: str
|
||
|
|
camera_id: str
|
||
|
|
roi_type: ROIType
|
||
|
|
coordinates: List[List[float]] # 多边形顶点或矩形坐标
|
||
|
|
algorithm_type: AlgorithmType
|
||
|
|
alert_threshold: int = 3
|
||
|
|
alert_cooldown: int = 300
|
||
|
|
enabled: bool = True
|
||
|
|
extra_params: Optional[Dict[str, Any]] = None
|
||
|
|
|
||
|
|
def to_dict(self) -> Dict[str, Any]:
|
||
|
|
"""转换为字典"""
|
||
|
|
return {
|
||
|
|
"roi_id": self.roi_id,
|
||
|
|
"camera_id": self.camera_id,
|
||
|
|
"roi_type": self.roi_type.value if isinstance(self.roi_type, ROIType) else self.roi_type,
|
||
|
|
"coordinates": self.coordinates,
|
||
|
|
"algorithm_type": self.algorithm_type.value if isinstance(self.algorithm_type, AlgorithmType) else self.algorithm_type,
|
||
|
|
"alert_threshold": self.alert_threshold,
|
||
|
|
"alert_cooldown": self.alert_cooldown,
|
||
|
|
"enabled": self.enabled,
|
||
|
|
"extra_params": self.extra_params,
|
||
|
|
}
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def from_dict(cls, data: Dict[str, Any]) -> 'ROIInfo':
|
||
|
|
"""从字典创建实例"""
|
||
|
|
roi_type_str = data.get("roi_type", "polygon")
|
||
|
|
roi_type = ROIType(roi_type_str) if roi_type_str in [e.value for e in ROIType] else ROIType.POLYGON
|
||
|
|
|
||
|
|
algo_type_str = data.get("algorithm_type", "leave_post")
|
||
|
|
algo_type = AlgorithmType(algo_type_str) if algo_type_str in [e.value for e in AlgorithmType] else AlgorithmType.LEAVE_POST
|
||
|
|
|
||
|
|
return cls(
|
||
|
|
roi_id=data.get("roi_id", ""),
|
||
|
|
camera_id=data.get("camera_id", ""),
|
||
|
|
roi_type=roi_type,
|
||
|
|
coordinates=data.get("coordinates", []),
|
||
|
|
algorithm_type=algo_type,
|
||
|
|
alert_threshold=data.get("alert_threshold", 3),
|
||
|
|
alert_cooldown=data.get("alert_cooldown", 300),
|
||
|
|
enabled=data.get("enabled", True),
|
||
|
|
extra_params=data.get("extra_params"),
|
||
|
|
)
|
||
|
|
|
||
|
|
def is_point_inside(self, point: List[float]) -> bool:
|
||
|
|
"""判断点是否在ROI区域内"""
|
||
|
|
if self.roi_type == ROIType.RECTANGLE:
|
||
|
|
return self._is_point_in_rectangle(point)
|
||
|
|
elif self.roi_type == ROIType.POLYGON:
|
||
|
|
return self._is_point_in_polygon(point)
|
||
|
|
return False
|
||
|
|
|
||
|
|
def _is_point_in_rectangle(self, point: List[float]) -> bool:
|
||
|
|
"""判断点是否在矩形区域内"""
|
||
|
|
if len(self.coordinates) < 2:
|
||
|
|
return False
|
||
|
|
|
||
|
|
x, y = point[0], point[1]
|
||
|
|
x1, y1 = self.coordinates[0]
|
||
|
|
x2, y2 = self.coordinates[1]
|
||
|
|
|
||
|
|
left = min(x1, x2)
|
||
|
|
right = max(x1, x2)
|
||
|
|
top = min(y1, y2)
|
||
|
|
bottom = max(y1, y2)
|
||
|
|
|
||
|
|
return left <= x <= right and top <= y <= bottom
|
||
|
|
|
||
|
|
def _is_point_in_polygon(self, point: List[float]) -> bool:
|
||
|
|
"""判断点是否在多边形区域内(射线法)"""
|
||
|
|
if len(self.coordinates) < 3:
|
||
|
|
return False
|
||
|
|
|
||
|
|
x, y = point[0], point[1]
|
||
|
|
n = len(self.coordinates)
|
||
|
|
inside = False
|
||
|
|
|
||
|
|
j = n - 1
|
||
|
|
for i in range(n):
|
||
|
|
xi, yi = self.coordinates[i]
|
||
|
|
xj, yj = self.coordinates[j]
|
||
|
|
|
||
|
|
if ((yi > y) != (yj > y)) and (x < (xj - xi) * (y - yi) / (yj - yi) + xi):
|
||
|
|
inside = not inside
|
||
|
|
|
||
|
|
j = i
|
||
|
|
|
||
|
|
return inside
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class AlertInfo:
|
||
|
|
"""告警信息数据模型"""
|
||
|
|
alert_id: str
|
||
|
|
camera_id: str
|
||
|
|
roi_id: str
|
||
|
|
alert_type: str
|
||
|
|
target_class: Optional[str] = None
|
||
|
|
confidence: Optional[float] = None
|
||
|
|
bbox: Optional[List[float]] = None
|
||
|
|
message: Optional[str] = None
|
||
|
|
screenshot: Optional[str] = None
|
||
|
|
level: AlertLevel = AlertLevel.MEDIUM
|
||
|
|
timestamp: Optional[str] = None
|
||
|
|
extra_data: Optional[Dict[str, Any]] = None
|
||
|
|
|
||
|
|
def to_dict(self) -> Dict[str, Any]:
|
||
|
|
"""转换为字典"""
|
||
|
|
return {
|
||
|
|
"alert_id": self.alert_id,
|
||
|
|
"camera_id": self.camera_id,
|
||
|
|
"roi_id": self.roi_id,
|
||
|
|
"alert_type": self.alert_type,
|
||
|
|
"target_class": self.target_class,
|
||
|
|
"confidence": self.confidence,
|
||
|
|
"bbox": self.bbox,
|
||
|
|
"message": self.message,
|
||
|
|
"screenshot": self.screenshot,
|
||
|
|
"level": self.level.value if isinstance(self.level, AlertLevel) else self.level,
|
||
|
|
"timestamp": self.timestamp,
|
||
|
|
"extra_data": self.extra_data,
|
||
|
|
}
|
||
|
|
|
||
|
|
def to_json(self) -> str:
|
||
|
|
"""转换为JSON字符串"""
|
||
|
|
return json.dumps(self.to_dict(), ensure_ascii=False)
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def from_dict(cls, data: Dict[str, Any]) -> 'AlertInfo':
|
||
|
|
"""从字典创建实例"""
|
||
|
|
level = data.get("level", "medium")
|
||
|
|
if isinstance(level, str) and level in [e.value for e in AlertLevel]:
|
||
|
|
level = AlertLevel(level)
|
||
|
|
else:
|
||
|
|
level = AlertLevel.MEDIUM
|
||
|
|
|
||
|
|
return cls(
|
||
|
|
alert_id=data.get("alert_id", ""),
|
||
|
|
camera_id=data.get("camera_id", ""),
|
||
|
|
roi_id=data.get("roi_id", ""),
|
||
|
|
alert_type=data.get("alert_type", ""),
|
||
|
|
target_class=data.get("target_class"),
|
||
|
|
confidence=data.get("confidence"),
|
||
|
|
bbox=data.get("bbox"),
|
||
|
|
message=data.get("message"),
|
||
|
|
screenshot=data.get("screenshot"),
|
||
|
|
level=level,
|
||
|
|
timestamp=data.get("timestamp"),
|
||
|
|
extra_data=data.get("extra_data"),
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class ConfigVersion:
|
||
|
|
"""配置版本信息模型"""
|
||
|
|
version: str
|
||
|
|
update_time: str
|
||
|
|
update_type: str # 'full', 'incremental'
|
||
|
|
updated_by: str
|
||
|
|
description: str
|
||
|
|
affected_items: List[str] = field(default_factory=list)
|
||
|
|
|
||
|
|
def to_dict(self) -> Dict[str, Any]:
|
||
|
|
"""转换为字典"""
|
||
|
|
return {
|
||
|
|
"version": self.version,
|
||
|
|
"update_time": self.update_time,
|
||
|
|
"update_type": self.update_type,
|
||
|
|
"updated_by": self.updated_by,
|
||
|
|
"description": self.description,
|
||
|
|
"affected_items": self.affected_items,
|
||
|
|
}
|