security-ai-edge/config/config_models.py

"""
数据模型定义模块
定义配置同步相关的核心数据模型
"""

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
    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,
        }