Security_AI_integrated/api/roi.py

import json
from typing import List, Optional

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session

from db.crud import (
    create_roi,
    delete_roi,
    get_all_rois,
    get_roi_by_id,
    get_roi_points,
    update_roi,
)
from db.models import get_db
from inference.pipeline import get_pipeline
from inference.roi.cache_manager import get_roi_cache
from inference.roi.roi_filter import ROIFilter

router = APIRouter(prefix="/api/camera", tags=["ROI管理"])


def _invalidate_roi_cache(camera_id: int):
    pipeline = get_pipeline()
    pipeline.roi_filter.clear_cache(camera_id)
    get_roi_cache().invalidate(camera_id)


@router.get("/{camera_id}/rois", response_model=List[dict])
def list_rois(camera_id: int, db: Session = Depends(get_db)):
    roi_configs = get_all_rois(db, camera_id)
    return [
        {
            "id": roi.id,
            "roi_id": roi.roi_id,
            "name": roi.name,
            "type": roi.roi_type,
            "points": json.loads(roi.points),
            "rule": roi.rule_type,
            "direction": roi.direction,
            "stay_time": roi.stay_time,
            "enabled": roi.enabled,
            "threshold_sec": roi.threshold_sec,
            "confirm_sec": roi.confirm_sec,
            "return_sec": roi.return_sec,
        }
        for roi in roi_configs
    ]


@router.get("/{camera_id}/roi/{roi_id}", response_model=dict)
def get_roi(camera_id: int, roi_id: int, db: Session = Depends(get_db)):
    roi = get_roi_by_id(db, roi_id)
    if not roi:
        raise HTTPException(status_code=404, detail="ROI不存在")
    return {
        "id": roi.id,
        "roi_id": roi.roi_id,
        "name": roi.name,
        "type": roi.roi_type,
        "points": json.loads(roi.points),
        "rule": roi.rule_type,
        "direction": roi.direction,
        "stay_time": roi.stay_time,
        "enabled": roi.enabled,
        "threshold_sec": roi.threshold_sec,
        "confirm_sec": roi.confirm_sec,
        "return_sec": roi.return_sec,
    }


@router.post("/{camera_id}/roi", response_model=dict)
def add_roi(
    camera_id: int,
    roi_id: str,
    name: str,
    roi_type: str,
    points: List[List[float]],
    rule_type: str,
    direction: Optional[str] = None,
    stay_time: Optional[int] = None,
    threshold_sec: int = 360,
    confirm_sec: int = 30,
    return_sec: int = 5,
    db: Session = Depends(get_db),
):
    roi = create_roi(
        db,
        camera_id=camera_id,
        roi_id=roi_id,
        name=name,
        roi_type=roi_type,
        points=points,
        rule_type=rule_type,
        direction=direction,
        stay_time=stay_time,
        threshold_sec=threshold_sec,
        confirm_sec=confirm_sec,
        return_sec=return_sec,
    )

    _invalidate_roi_cache(camera_id)

    return {
        "id": roi.id,
        "roi_id": roi.roi_id,
        "name": roi.name,
        "type": roi.roi_type,
        "points": points,
        "rule": roi.rule_type,
        "enabled": roi.enabled,
    }


@router.put("/{camera_id}/roi/{roi_id}", response_model=dict)
def modify_roi(
    camera_id: int,
    roi_id: int,
    name: Optional[str] = None,
    points: Optional[List[List[float]]] = None,
    rule_type: Optional[str] = None,
    direction: Optional[str] = None,
    stay_time: Optional[int] = None,
    enabled: Optional[bool] = None,
    threshold_sec: Optional[int] = None,
    confirm_sec: Optional[int] = None,
    return_sec: Optional[int] = None,
    db: Session = Depends(get_db),
):
    roi = update_roi(
        db,
        roi_id=roi_id,
        name=name,
        points=points,
        rule_type=rule_type,
        direction=direction,
        stay_time=stay_time,
        enabled=enabled,
        threshold_sec=threshold_sec,
        confirm_sec=confirm_sec,
        return_sec=return_sec,
    )
    if not roi:
        raise HTTPException(status_code=404, detail="ROI不存在")

    _invalidate_roi_cache(camera_id)

    return {
        "id": roi.id,
        "roi_id": roi.roi_id,
        "name": roi.name,
        "type": roi.roi_type,
        "points": json.loads(roi.points),
        "rule": roi.rule_type,
        "enabled": roi.enabled,
    }


@router.delete("/{camera_id}/roi/{roi_id}")
def remove_roi(camera_id: int, roi_id: int, db: Session = Depends(get_db)):
    if not delete_roi(db, roi_id):
        raise HTTPException(status_code=404, detail="ROI不存在")

    _invalidate_roi_cache(camera_id)

    return {"message": "删除成功"}


@router.get("/{camera_id}/roi/validate")
def validate_roi(
    camera_id: int,
    roi_type: str,
    points: List[List[float]],
    db: Session = Depends(get_db),
):
    try:
        if roi_type == "polygon":
            from shapely.geometry import Polygon
            poly = Polygon(points)
            is_valid = poly.is_valid
            area = poly.area
        elif roi_type == "line":
            from shapely.geometry import LineString
            line = LineString(points)
            is_valid = True
            area = 0.0
        else:
            raise HTTPException(status_code=400, detail="不支持的ROI类型")

        return {
            "valid": is_valid,
            "area": area,
            "message": "有效" if is_valid else "无效：自交或自重叠",
        }
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))