feat: 重构数据库schema分离空间与业务配置 - 新增algorithm_registry和roi_algo_bind表 - roi_configs简化为纯空间配置 - 新增AlgorithmInfo/ROIAlgoBind数据模型

config/config_models.py

@@ -215,7 +215,8 @@ class AlertInfo:
     alert_id: str
     camera_id: str
     roi_id: str
-    alert_type: str
+    bind_id: Optional[str] = None
+    alert_type: str = "detection"
     target_class: Optional[str] = None
     confidence: Optional[float] = None
     bbox: Optional[List[float]] = None
@@ -224,6 +225,7 @@ class AlertInfo:
     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]:
         """转换为字典"""
@@ -231,6 +233,7 @@ class AlertInfo:
             "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,
@@ -240,6 +243,7 @@ class AlertInfo:
             "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:
@@ -291,3 +295,184 @@ class ConfigVersion:
             "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