479 lines
16 KiB
Python
479 lines
16 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
|
||
working_hours: Optional[List[Dict]] = None # 工作时间段
|
||
confirm_on_duty_sec: int = 10 # 在岗确认时间
|
||
confirm_leave_sec: int = 10 # 离岗确认时间
|
||
cooldown_sec: int = 300 # 告警冷却时间
|
||
target_class: str = "person" # 目标类别
|
||
|
||
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,
|
||
"working_hours": self.working_hours,
|
||
"confirm_on_duty_sec": self.confirm_on_duty_sec,
|
||
"confirm_leave_sec": self.confirm_leave_sec,
|
||
"cooldown_sec": self.cooldown_sec,
|
||
"target_class": self.target_class,
|
||
}
|
||
|
||
@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
|
||
|
||
working_hours = data.get("working_hours")
|
||
if isinstance(working_hours, str):
|
||
import json
|
||
try:
|
||
working_hours = json.loads(working_hours)
|
||
except:
|
||
working_hours = None
|
||
|
||
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"),
|
||
working_hours=working_hours,
|
||
confirm_on_duty_sec=data.get("confirm_on_duty_sec", 10),
|
||
confirm_leave_sec=data.get("confirm_leave_sec", 10),
|
||
cooldown_sec=data.get("cooldown_sec", 300),
|
||
target_class=data.get("target_class", "person"),
|
||
)
|
||
|
||
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
|
||
bind_id: Optional[str] = None
|
||
alert_type: str = "detection"
|
||
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
|
||
detections: Optional[str] = None
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
"""转换为字典"""
|
||
return {
|
||
"alert_id": self.alert_id,
|
||
"camera_id": self.camera_id,
|
||
"roi_id": self.roi_id,
|
||
"bind_id": self.bind_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,
|
||
"detections": self.detections,
|
||
}
|
||
|
||
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,
|
||
}
|
||
|
||
|
||
@dataclass
|
||
class AlgorithmInfo:
|
||
"""算法配置信息数据模型"""
|
||
algo_code: str
|
||
algo_name: str
|
||
target_class: str = "person"
|
||
param_schema: Optional[Dict[str, Any]] = None
|
||
description: Optional[str] = None
|
||
is_active: bool = True
|
||
created_at: Optional[str] = None
|
||
updated_at: Optional[str] = None
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
"""转换为字典"""
|
||
return {
|
||
"algo_code": self.algo_code,
|
||
"algo_name": self.algo_name,
|
||
"target_class": self.target_class,
|
||
"param_schema": self.param_schema,
|
||
"description": self.description,
|
||
"is_active": self.is_active,
|
||
"created_at": self.created_at,
|
||
"updated_at": self.updated_at,
|
||
}
|
||
|
||
@classmethod
|
||
def from_dict(cls, data: Dict[str, Any]) -> 'AlgorithmInfo':
|
||
"""从字典创建实例"""
|
||
return cls(
|
||
algo_code=data.get("algo_code", ""),
|
||
algo_name=data.get("algo_name", ""),
|
||
target_class=data.get("target_class", "person"),
|
||
param_schema=data.get("param_schema"),
|
||
description=data.get("description"),
|
||
is_active=data.get("is_active", True),
|
||
created_at=data.get("created_at"),
|
||
updated_at=data.get("updated_at"),
|
||
)
|
||
|
||
|
||
@dataclass
|
||
class ROIAlgoBind:
|
||
"""ROI与算法绑定关系数据模型"""
|
||
bind_id: str
|
||
roi_id: str
|
||
algo_code: str
|
||
params: Dict[str, Any] = field(default_factory=dict)
|
||
priority: int = 0
|
||
enabled: bool = True
|
||
created_at: Optional[str] = None
|
||
updated_at: Optional[str] = None
|
||
algo_name: Optional[str] = None
|
||
target_class: Optional[str] = None
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
"""转换为字典"""
|
||
return {
|
||
"bind_id": self.bind_id,
|
||
"roi_id": self.roi_id,
|
||
"algo_code": self.algo_code,
|
||
"params": self.params,
|
||
"priority": self.priority,
|
||
"enabled": self.enabled,
|
||
"created_at": self.created_at,
|
||
"updated_at": self.updated_at,
|
||
"algo_name": self.algo_name,
|
||
"target_class": self.target_class,
|
||
}
|
||
|
||
@classmethod
|
||
def from_dict(cls, data: Dict[str, Any]) -> 'ROIAlgoBind':
|
||
"""从字典创建实例"""
|
||
return cls(
|
||
bind_id=data.get("bind_id", ""),
|
||
roi_id=data.get("roi_id", ""),
|
||
algo_code=data.get("algo_code", ""),
|
||
params=data.get("params", {}),
|
||
priority=data.get("priority", 0),
|
||
enabled=data.get("enabled", True),
|
||
created_at=data.get("created_at"),
|
||
updated_at=data.get("updated_at"),
|
||
algo_name=data.get("algo_name"),
|
||
target_class=data.get("target_class"),
|
||
)
|
||
|
||
|
||
@dataclass
|
||
class ROIInfoNew:
|
||
"""ROI区域信息数据模型（新版本，包含绑定信息）"""
|
||
roi_id: str
|
||
camera_id: str
|
||
roi_type: ROIType
|
||
coordinates: List[List[float]]
|
||
enabled: bool = True
|
||
priority: int = 0
|
||
extra_params: Optional[Dict[str, Any]] = None
|
||
bindings: List[ROIAlgoBind] = field(default_factory=list)
|
||
updated_at: Optional[str] = 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,
|
||
"enabled": self.enabled,
|
||
"priority": self.priority,
|
||
"extra_params": self.extra_params,
|
||
"bindings": [b.to_dict() for b in self.bindings],
|
||
"updated_at": self.updated_at,
|
||
}
|
||
|
||
@classmethod
|
||
def from_dict(cls, data: Dict[str, Any]) -> 'ROIInfoNew':
|
||
"""从字典创建实例"""
|
||
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
|
||
|
||
bindings = data.get("bindings", [])
|
||
if isinstance(bindings, list):
|
||
bindings = [ROIAlgoBind.from_dict(b) for b in bindings]
|
||
|
||
return cls(
|
||
roi_id=data.get("roi_id", ""),
|
||
camera_id=data.get("camera_id", ""),
|
||
roi_type=roi_type,
|
||
coordinates=data.get("coordinates", []),
|
||
enabled=data.get("enabled", True),
|
||
priority=data.get("priority", 0),
|
||
extra_params=data.get("extra_params"),
|
||
bindings=bindings,
|
||
updated_at=data.get("updated_at"),
|
||
)
|
||
|
||
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
|