jc/jc-video-recognize
11
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 新增车辆检测Paddle模型及相关服务，优化依赖与代码兼容性

Browse Source 
1. 新增3套PaddlePaddle车辆检测相关模型文件
2. 新增车辆检测服务类与违停检测功能
3. 更新服务依赖并添加环境初始化脚本与文档
4. 修复YOLO检测tensor转换兼容问题
5. 新增PyTorch版本兼容性修复逻辑
6. 扩展模型服务支持Paddle模型加载
This commit is contained in:
wwh
2026-05-21 16:26:26 +08:00
parent 8809c3a6f4
commit a16e684e46
16 changed files with 1045 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

199
apps/server/ENVIRONMENT.md Normal file
View File

@@ -0,0 +1,199 @@
# 环境配置说明
## 📦 当前环境版本 (2026-05-21 验证通过)
### 核心框架
- **Python**: 3.12
- **FastAPI**: 0.136.1
- **Uvicorn**: 0.34.0 (with standard extras)
- **Pydantic**: 2.10.6
### 深度学习框架
- **PyTorch**: 2.12.0
- **TorchVision**: 0.27.0
- **Ultralytics**: 8.4.52 (YOLO 模型支持)
- **Ultralytics-THOP**: 2.0.19
### PaddlePaddle 生态
- **PaddlePaddle**: 3.0.0
- **Paddle2ONNX**: 2.1.0
### 数值计算和数据处理
- **NumPy**: 2.4.6 ⚠️ **需要注意兼容性**
- **Pandas**: 2.3.3
- **SciPy**: 1.15.2
- **Scikit-Image**: 0.26.2
### 图像和计算机视觉
- **OpenCV**: 4.13.0.92
- **Pillow**: 11.1.0
- **ImageIO**: 2.37.3
- **imgaug**: 0.4.0 ⚠️ **需要手动修复**
### 其他依赖
- **aiofiles**: 25.1.0
- **python-multipart**: 0.0.20
- **websockets**: 14.1
- **matplotlib**: 3.10.1
- **Shapely**: 2.1.0
- **tqdm**: 4.69.2
## 🔧 特殊注意事项
### 1. imgaug NumPy 2.0 兼容性问题
**问题**: imgaug 0.4.0 使用了 `np.sctypes` 属性，但 NumPy 2.0+ 移除了这个 API。
**解决方案**: 手动修复 `imgaug.py` 文件
**修复文件路径**: `venv/lib/python3.12/site-packages/imgaug/imgaug.py`
**需要修改的行**: 第 45-47 行
**原始代码**:
```python
NP_FLOAT_TYPES = set(np.sctypes["float"])
NP_INT_TYPES = set(np.sctypes["int"])
NP_UINT_TYPES = set(np.sctypes["uint"])
```
**修复后代码**:
```python
NP_FLOAT_TYPES = {np.float16, np.float32, np.float64}
NP_INT_TYPES = {np.int8, np.int16, np.int32, np.int64}
NP_UINT_TYPES = {np.uint8, np.uint16, np.uint32, np.uint64}
```
### 2. PyTorch 2.12.0 与 Ultralytics 兼容性
**状态**: ✅ 完全兼容，无需额外配置
**说明**:
- PyTorch 2.12.0 与 Ultralytics 8.4.52 完全兼容
- 无需手动添加安全全局变量
- 模型加载和推理功能正常
### 3. NumPy 2.4.6 兼容性
**状态**: ✅ 所有主要包都已验证兼容
**兼容包**:
- ✅ PyTorch 2.12.0
- ✅ TorchVision 0.27.0
- ✅ Ultralytics 8.4.52
- ✅ PaddlePaddle 3.0.0
- ✅ OpenCV 4.13.0.92
- ⚠️ imgaug 0.4.0 (需要手动修复)
## 📁 项目结构
### 核心目录
```
jc-video-recognize/
├── apps/server/
│ ├── venv/ # 虚拟环境
│ ├── models/ # 模型文件
│ │ ├── smoking_yolo/ # 吸烟检测模型
│ │ ├── fire_detection/ # 火灾检测模型
│ │ ├── vehicle_detection_paddle/ # 车辆检测模型
│ │ └── smoking_detection_paddle/ # Paddle吸烟检测模型
│ ├── api/ # API 端点
│ ├── services/ # 业务逻辑服务
│ ├── static/ # 静态文件和结果
│ ├── requirements.txt # Python 依赖
│ ├── setup_env.sh # 环境初始化脚本
│ └── main.py # 应用入口
└── third-party/
└── paddle-inference/ # PaddleDetection 部署代码
```
## 🚀 快速开始
### 1. 环境初始化
```bash
# 使用自动初始化脚本
cd apps/server
chmod +x setup_env.sh
./setup_env.sh
```
### 2. 手动安装
```bash
# 创建虚拟环境
python3 -m venv venv
source venv/bin/activate
# 安装依赖
pip install -r requirements.txt
# 修复 imgaug (如果自动脚本未执行)
# 手动修改 venv/lib/python3.12/site-packages/imgaug/imgaug.py 第45-47行
```
### 3. 启动服务
```bash
# 激活虚拟环境
source venv/bin/activate
# 启动服务
python main.py
# 或使用启动脚本
./start_server_with_env.sh
```
## 🔍 已知问题和解决方案
### 问题 1: YOLO 模型 tensor 转换错误
**错误信息**: `ValueError: only one element tensors can be converted to Python scalars`
**影响文件**:
- `services/detection_service.py` (第 62-72 行, 第 175-184 行)
**解决方案**: 已修复，使用 `box.xyxy.squeeze().tolist()` 安全转换 tensor
### 问题 2: PaddleDetection imgaug 兼容性
**错误信息**: `AttributeError: np.sctypes was removed in the NumPy 2.0 release`
**影响文件**: `venv/lib/python3.12/site-packages/imgaug/imgaug.py`
**解决方案**: 手动修复第 45-47 行（如上所述）
### 问题 3: PyTorch WeightsUnpickler 错误
**状态**: ✅ 已通过升级 PyTorch 和 Ultralytics 解决
## 📝 版本选择理由
1. **PyTorch 2.12.0**: 最新稳定版本，与 TorchVision 0.27.0 完全兼容
2. **NumPy 2.4.6**: 最新版本，性能更好，通过手动修复兼容性问题
3. **Ultralytics 8.4.52**: 与 PyTorch 2.12.0 完全兼容，支持最新 YOLO 模型
4. **PaddlePaddle 3.0.0**: 官方稳定版本，支持所有车辆检测功能
5. **imgaug 0.4.0**: 最后一个稳定版本，虽然不兼容 NumPy 2.0 但可手动修复
## 🤝 给团队成员的建议
1. **严格遵循版本要求**: 不要随意更改依赖版本
2. **执行初始化脚本**: 使用 `setup_env.sh` 自动处理兼容性问题
3. **检查环境**: 每次启动前验证关键包版本
4. **备份环境**: 使用 `pip freeze > environment.txt` 备份当前环境
5. **报告问题**: 如遇到新的兼容性问题，及时更新此文档
## 📞 技术支持
如遇到环境相关问题，请：
1. 检查此文档的已知问题部分
2. 确认所有依赖版本符合要求
3. 验证 imgaug 手动修复是否成功
4. 查看服务日志获取详细错误信息
---
**最后更新**: 2026-05-21
**维护者**: 开发团队
**环境状态**: ✅ 生产就绪

73
apps/server/requirements.txt
View File

@@ -1,12 +1,61 @@
fastapi>=0.104.0
uvicorn[standard]>=0.24.0
python-multipart>=0.0.6
pydantic>=2.0.0
python-dotenv>=1.0.0
aiofiles>=23.2.0
opencv-python>=4.8.0
pillow>=10.0.0
ultralytics>=8.0.0
numpy>=1.24.0
torch>=2.0.0
websockets>=12.0.0
# 核心依赖 - 精确版本 (2026-05-21 验证通过)
fastapi==0.136.1
uvicorn[standard]==0.34.0
python-multipart==0.0.20
pydantic==2.10.6
python-dotenv==1.1.0
aiofiles==25.1.0
websockets==14.1
# 图像处理和计算机视觉
opencv-python==4.13.0.92
pillow==11.1.0
imgaug==0.4.0
# 机器学习框架 - 已解决兼容性问题
numpy==2.4.6
torch==2.12.0
torchvision==0.27.0
ultralytics==8.4.52
ultralytics-thop==2.0.19
# PaddlePaddle 生态
paddlepaddle==3.0.0
paddle2onnx==2.1.0
# 数据处理
pandas==2.3.3
scipy==1.15.2
scikit-image==0.26.2
# 图像和几何处理
imageio==2.37.3
matplotlib==3.10.1
shapely==2.1.0
# 其他工具
click==8.4.0
tqdm==4.69.2
psutil==6.1.1
# 网络相关
httpx==0.28.1
certifi==2026.5.20
# 开发工具
ipython==9.1.0
jedi==0.19.2
# 特殊注意事项：
# 1. imgaug==0.4.0 需要手动修复 numpy 2.0 兼容性问题：
# 修改 venv/lib/python3.12/site-packages/imgaug/imgaug.py 第45-47行
# 将 np.sctypes["float"] 等替换为直接指定类型：
# NP_FLOAT_TYPES = {np.float16, np.float32, np.float64}
# NP_INT_TYPES = {np.int8, np.int16, np.int32, np.int64}
# NP_UINT_TYPES = {np.uint8, np.uint16, np.uint32, np.uint64}
#
# 2. PyTorch 2.12.0 与 ultralytics 8.4.52 完全兼容
#
# 3. NumPy 2.4.6 已验证与所有主要包兼容
#
# 4. PaddleDetection 第三方库路径：../../third-party/paddle-inference

12
apps/server/services/detection_service.py
View File

@@ -60,7 +60,11 @@ class DetectionService:
try:
if isinstance(box.xyxy, torch.Tensor) and box.xyxy.dim() > 0:
x1, y1, x2, y2 = float(box.xyxy[0]), float(box.xyxy[1]), float(box.xyxy[2]), float(box.xyxy[3])
xyxy_values = box.xyxy.squeeze().tolist()
if len(xyxy_values) >= 4:
x1, y1, x2, y2 = float(xyxy_values[0]), float(xyxy_values[1]), float(xyxy_values[2]), float(xyxy_values[3])
else:
continue
elif isinstance(box.xyxy, (list, tuple)):
x1, y1, x2, y2 = float(box.xyxy[0]), float(box.xyxy[1]), float(box.xyxy[2]), float(box.xyxy[3])
else:
@@ -169,7 +173,11 @@ class DetectionService:
if isinstance(box.xyxy, torch.Tensor) and box.xyxy.dim() > 0:
x1, y1, x2, y2 = float(box.xyxy[0]), float(box.xyxy[1]), float(box.xyxy[2]), float(box.xyxy[3])
xyxy_values = box.xyxy.squeeze().tolist()
if len(xyxy_values) >= 4:
x1, y1, x2, y2 = float(xyxy_values[0]), float(xyxy_values[1]), float(xyxy_values[2]), float(xyxy_values[3])
else:
continue
elif isinstance(box.xyxy, (list, tuple)):
x1, y1, x2, y2 = float(box.xyxy[0]), float(box.xyxy[1]), float(box.xyxy[2]), float(box.xyxy[3])
else:

69
apps/server/services/model_service.py
View File

@@ -1,10 +1,36 @@
import os
import logging
from ultralytics import YOLO
from typing import Dict, List, Optional, Union
# PyTorch 2.6 兼容性修复
os.environ.setdefault('TORCH_DISABLE_TORCH_GRAPH_OPTIMIZER', '1')
logger = logging.getLogger(__name__)
try:
import torch
if hasattr(torch, 'serialization'):
# 导入 ultralytics 相关类以供 add_safe_globals 使用
from ultralytics.nn.tasks import DetectionModel
# 注册所有需要的类
torch.serialization.add_safe_globals([
DetectionModel, # ultralytics 模型类
torch.nn.modules.container.Sequential, # torch 序列类
torch.nn.modules.Conv2d, # 卷积层
torch.nn.modules.batchnorm.BatchNorm2d, # 批归一化
torch.nn.modules.activation.ReLU, # 激活函数
torch.nn.modules.Linear, # 线性层
torch.nn.modules.Dropout, # Dropout 层
torch.nn.modules.Upsample, # 上采样
torch.nn.modules.PixelShuffle, # 像素重排
])
logger.info("✅ 检测到 PyTorch 2.6+，应用 ultralytics 兼容性修复")
except (ImportError, AttributeError, NameError) as e:
logger.warning(f"⚠️ PyTorch 兼容性修复失败: {e}")
from ultralytics import YOLO
from typing import Dict, List, Optional, Union
class ModelService:
def __init__(self):
self.models: Dict[str, Union[YOLO, object]] = {}
@@ -65,6 +91,24 @@ class ModelService:
'size': '6MB',
'description': '基于YOLOv8的徘徊检测模型',
'name': '徘徊检测'
},
'vehicle_detection': {
'path': os.path.join(base_dir, 'models', 'vehicle_detection_paddle', 'mot_ppyoloe_l_36e_ppvehicle', 'model.pdmodel'),
'type': 'paddle',
'classes': ['vehicle'],
'labels': {'vehicle': '车辆'},
'size': '181MB',
'description': '基于PaddlePaddle PP-YOLOE-l的车辆检测和跟踪模型',
'name': '车辆检测 (Paddle)'
},
'illegal_parking_detection': {
'path': os.path.join(base_dir, 'models', 'vehicle_detection_paddle', 'mot_ppyoloe_l_36e_ppvehicle', 'model.pdmodel'),
'type': 'paddle',
'classes': ['vehicle'],
'labels': {'vehicle': '车辆'},
'size': '200MB',
'description': '基于PaddlePaddle PP-YOLOE-l的违停检测模型，支持车牌识别',
'name': '违停检测 (Paddle)'
}
}
@@ -112,31 +156,44 @@ class ModelService:
# 处理 PaddleDetection 模型
if config['type'] == 'paddle':
try:
if model_id == 'smoking_detection_paddle':
from .paddle_detection_service import SmokingDetectionModel
logger.info(f"正在加载 PaddlePaddle Docker 服务: {model_id}")
logger.info(f"正在加载 PaddlePaddle 抽烟检测服务: {model_id}")
model = SmokingDetectionModel()
elif model_id in ['vehicle_detection', 'illegal_parking_detection']:
from .vehicle_detection_service import VehicleDetectionModel
logger.info(f"正在加载 PaddlePaddle 车辆检测服务: {model_id}")
model = VehicleDetectionModel()
else:
logger.error(f"未知的 Paddle 模型类型: {model_id}")
return None
self.models[model_id] = model
logger.info(f"PaddlePaddle Docker 服务加载成功: {model_id}")
logger.info(f"PaddlePaddle 服务加载成功: {model_id}")
return model
except Exception as e:
logger.error(f"PaddlePaddle Docker 服务加载失败: {model_id}, 错误: {e}")
logger.error(f"PaddlePaddle 服务加载失败: {model_id}, 错误: {e}")
return None
# 处理 YOLO 模型
model_path = config['path']
if not os.path.exists(model_path):
logger.error(f"模型文件不存在: {model_path}")
logger.warning(f"模型文件不存在: {model_path}，跳过加载 {model_id}")
return None
try:
logger.info(f"正在加载 YOLO 模型: {model_id} from {model_path}")
model = YOLO(model_path)
self.models[model_id] = model
logger.info(f"YOLO 模型加载成功: {model_id}")
return model
except Exception as e:
logger.error(f"YOLO 模型加载失败: {model_id}, 错误: {e}")
logger.error(f"模型路径: {model_path}")
import traceback
logger.error(f"详细错误: {traceback.format_exc()}")
return None
def get_model(self, model_id: str) -> Optional[Union[YOLO, object]]:

585
apps/server/services/vehicle_detection_service.py Normal file
View File

@@ -0,0 +1,585 @@
"""
车辆检测服务适配器
支持车辆检测、跟踪、车牌识别和违停检测功能
"""
# 禁用 PIR API 以支持旧版模型格式（必须在任何导入之前设置）
import os
os.environ['FLAGS_enable_pir_api'] = '0'
import cv2
import numpy as np
import logging
import threading
import time
import sys
from typing import Dict, List, Optional, Tuple
from pathlib import Path
from collections import defaultdict
from dataclasses import dataclass
logger = logging.getLogger(__name__)
@dataclass
class VehicleTrackingInfo:
"""车辆跟踪信息"""
track_id: int
bbox: List[float]
center: Tuple[float, float]
first_seen: float
last_seen: float
plate_number: Optional[str] = None
is_illegal_parking: bool = False
trajectory: List[Tuple[float, float]] = None
def __post_init__(self):
if self.trajectory is None:
self.trajectory = []
class VehicleDetectionService:
"""车辆检测服务（本地模式）"""
def __init__(self):
self.model_name = "vehicle_detection"
self.threshold = 0.1
self._lock = threading.Lock()
# 本地环境配置
project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
self.paddle_dir = os.path.join(project_root, "third-party", "paddle-inference")
self.model_dir = os.path.join(project_root, "models", "vehicle_detection_paddle")
# 模型路径配置
self.mot_model_dir = os.path.join(self.model_dir, "mot_ppyoloe_l_36e_ppvehicle")
self.plate_det_model_dir = os.path.join(self.model_dir, "ch_PP-OCRv3_det_infer")
self.plate_rec_model_dir = os.path.join(self.model_dir, "ch_PP-OCRv3_rec_infer")
# 检测器实例（延迟加载）
self._detector = None
self._detector_initialized = False
# 车辆跟踪信息
self.vehicle_tracks: Dict[int, VehicleTrackingInfo] = {}
self.track_id_counter = 0
# 违停检测配置
self.illegal_parking_time = 5.0 # 默认5秒
self.illegal_parking_region = None # 违停区域多边形
self.available = True
logger.info(f"车辆检测服务初始化完成")
logger.info(f"车辆检测模型目录: {self.mot_model_dir}")
logger.info(f"车牌检测模型目录: {self.plate_det_model_dir}")
logger.info(f"车牌识别模型目录: {self.plate_rec_model_dir}")
# 禁用 PIR API 以支持旧版模型格式
os.environ['FLAGS_enable_pir_api'] = '0'
try:
self._initialize_environment()
except Exception as e:
logger.error(f"环境初始化失败: {e}")
self.available = False
def _initialize_environment(self):
"""初始化本地 PaddlePaddle 环境"""
try:
# 添加 PaddleDetection 部署路径
paddle_detection_path = self.paddle_dir
if paddle_detection_path not in sys.path:
sys.path.insert(0, paddle_detection_path)
logger.info(f"✅ 添加 PaddleDetection 路径: {paddle_detection_path}")
# 检查模型目录是否存在
required_models = {
'MOT': self.mot_model_dir,
'Plate Detection': self.plate_det_model_dir,
'Plate Recognition': self.plate_rec_model_dir
}
for model_name, model_path in required_models.items():
if not os.path.exists(model_path):
raise Exception(f"{model_name} 模型目录不存在: {model_path}")
required_files = ['inference.pdmodel', 'inference.pdiparams', 'inference.pdiparams.info']
if model_name == 'MOT':
required_files = ['model.pdmodel', 'model.pdiparams', 'infer_cfg.yml']
for file in required_files:
file_path = os.path.join(model_path, file)
if not os.path.exists(file_path):
raise Exception(f"{model_name} 模型文件不存在: {file}")
logger.info("✅ 环境检查通过")
except Exception as e:
logger.error(f"环境初始化失败: {e}")
raise
def _get_detector(self):
"""获取检测器实例（单例模式）"""
if self._detector is None or not self._detector_initialized:
try:
# 设置环境变量以支持旧版模型格式
os.environ['FLAGS_enable_pir_api'] = '0'
# 添加 PaddleDetection 路径
if self.paddle_dir not in sys.path:
sys.path.insert(0, self.paddle_dir)
# 导入 PaddleDetection 模块
from infer import Detector, PredictConfig
# 创建检测器（使用MOT模型）
self._detector = Detector(
model_dir=self.mot_model_dir,
device='CPU',
run_mode='paddle',
batch_size=1,
output_dir='output',
threshold=self.threshold
)
self._detector_initialized = True
logger.info("✅ 车辆检测器初始化成功")
except Exception as e:
logger.error(f"检测器初始化失败: {e}")
raise
return self._detector
def detect_image(self, image: np.ndarray, threshold: float = None) -> Dict:
"""
检测图片中的车辆
Args:
image: OpenCV 图片 (BGR格式)
threshold: 置信度阈值
Returns:
检测结果字典
"""
if threshold is None:
threshold = self.threshold
if not self.available:
return {
'success': False,
'message': '车辆检测服务不可用',
'detections': [],
'stats': None
}
try:
with self._lock:
start_time = time.time()
# 确保检测器已初始化
detector = self._get_detector()
# 准备输入图片
if not isinstance(image, np.ndarray):
raise Exception(f"不支持的图片类型: {type(image)}")
if len(image.shape) == 2:
image = cv2.cvtColor(image, cv2.COLOR_GRAY2BGR)
elif image.shape[2] == 4:
image = cv2.cvtColor(image, cv2.COLOR_RGBA2BGR)
# 执行推理
inference_start = time.time()
results = detector.predict_image(
[image],
visual=False,
save_results=False
)
inference_time = time.time() - inference_start
logger.info(f"推理耗时: {inference_time:.3f}s")
# 解析检测结果
detections = self._parse_detection_results(results, threshold)
total_time = time.time() - start_time
logger.info(f"检测总耗时: {total_time:.3f}s")
return {
'success': True,
'message': '检测完成',
'detections': detections,
'stats': {
'total_detections': len(detections),
'model_used': 'mot_ppyoloe_l_36e_ppvehicle',
'threshold': threshold,
'processing_time': round(total_time, 3),
'inference_time': round(inference_time, 3)
}
}
except Exception as e:
import traceback
logger.error(f"检测失败: {e}")
logger.error(f"错误堆栈: {traceback.format_exc()}")
self._detector_initialized = False
return {
'success': False,
'message': f'检测失败: {e}',
'detections': [],
'stats': None
}
def _parse_detection_results(self, results: Dict, threshold: float) -> List[Dict]:
"""解析 PaddleDetection 返回的检测结果"""
detections = []
try:
if results and 'boxes' in results:
boxes = results['boxes']
if boxes is not None and len(boxes) > 0:
for box in boxes:
if len(box) >= 6:
class_id = int(box[0])
confidence = float(box[1])
x1, y1, x2, y2 = float(box[2]), float(box[3]), float(box[4]), float(box[5])
# 计算中心点
center_x = (x1 + x2) / 2
center_y = (y1 + y2) / 2
# 过滤低置信度检测
if confidence >= threshold:
detections.append({
'class': 'vehicle',
'label': '车辆',
'confidence': round(confidence, 3),
'bbox': [int(x1), int(y1), int(x2), int(y2)],
'center': [round(center_x, 2), round(center_y, 2)]
})
except Exception as e:
logger.error(f"解析检测结果失败: {e}")
return detections
def detect_illegal_parking(self, image: np.ndarray, threshold: float = None,
illegal_parking_time: float = 5.0,
region_polygon: List[Tuple[int, int]] = None) -> Dict:
"""
检测违停车辆
Args:
image: OpenCV 图片
threshold: 置信度阈值
illegal_parking_time: 违停时间阈值（秒）
region_polygon: 违停区域多边形点集 [(x1,y1), (x2,y2), ...]
Returns:
违停检测结果
"""
if threshold is None:
threshold = self.threshold
# 更新违停配置
self.illegal_parking_time = illegal_parking_time
self.illegal_parking_region = region_polygon
# 基础车辆检测
detection_result = self.detect_image(image, threshold)
if not detection_result['success']:
return {
'success': False,
'message': detection_result['message'],
'illegal_parking': [],
'vehicles': []
}
current_time = time.time()
current_detections = detection_result['detections']
# 更新车辆跟踪信息
illegal_parking_vehicles = []
for detection in current_detections:
bbox = detection['bbox']
center = detection['center']
# 简单的跟踪（基于位置匹配）
matched_track_id = self._match_vehicle_to_track(center, bbox)
if matched_track_id is None:
# 新车辆
self.track_id_counter += 1
matched_track_id = self.track_id_counter
self.vehicle_tracks[matched_track_id] = VehicleTrackingInfo(
track_id=matched_track_id,
bbox=bbox,
center=center,
first_seen=current_time,
last_seen=current_time,
trajectory=[center]
)
else:
# 更新现有车辆
track_info = self.vehicle_tracks[matched_track_id]
track_info.bbox = bbox
track_info.center = center
track_info.last_seen = current_time
track_info.trajectory.append(center)
# 检查违停条件
if self._check_illegal_parking(track_info, region_polygon):
track_info.is_illegal_parking = True
illegal_parking_vehicles.append({
'track_id': matched_track_id,
'bbox': bbox,
'center': center,
'parking_duration': round(current_time - track_info.first_seen, 2),
'plate_number': track_info.plate_number
})
# 清理长时间未出现的车辆
self._cleanup_old_tracks(current_time)
return {
'success': True,
'message': '违停检测完成',
'illegal_parking': illegal_parking_vehicles,
'total_vehicles': len(current_detections),
'stats': detection_result['stats']
}
def _match_vehicle_to_track(self, center: Tuple[float, float],
bbox: List[float]) -> Optional[int]:
"""将检测到的车辆匹配到已有轨迹"""
x, y = center
for track_id, track_info in self.vehicle_tracks.items():
track_x, track_y = track_info.center
# 计算距离
distance = np.sqrt((x - track_x) ** 2 + (y - track_y) ** 2)
# 距离阈值（基于检测框大小）
bbox_width = bbox[2] - bbox[0]
bbox_height = bbox[3] - bbox[1]
max_dim = max(bbox_width, bbox_height)
if distance < max_dim * 0.5: # 距离小于检测框最大尺寸的一半
return track_id
return None
def _check_illegal_parking(self, track_info: VehicleTrackingInfo,
region_polygon: List[Tuple[int, int]] = None) -> bool:
"""检查是否违停"""
current_time = time.time()
parking_duration = current_time - track_info.first_seen
# 检查时间是否超过阈值
if parking_duration < self.illegal_parking_time:
return False
# 检查是否在违停区域内
if region_polygon is None:
return False
# 检查车辆中心是否在多边形内
return self._point_in_polygon(track_info.center, region_polygon)
def _point_in_polygon(self, point: Tuple[float, float],
polygon: List[Tuple[int, int]]) -> bool:
"""判断点是否在多边形内（射线法）"""
x, y = point
n = len(polygon)
inside = False
p1x, p1y = polygon[0]
for i in range(n + 1):
p2x, p2y = polygon[i % n]
if y > min(p1y, p2y):
if y <= max(p1y, p2y):
if x <= max(p1x, p2x):
if p1y != p2y:
xinters = (y - p1y) * (p2x - p1x) / (p2y - p1y) + p1x
if p1x == p2x or x <= xinters:
inside = not inside
p1x, p1y = p2x, p2y
return inside
def _cleanup_old_tracks(self, current_time: float):
"""清理长时间未出现的车辆轨迹"""
timeout = 10.0 # 10秒未出现则删除
tracks_to_remove = []
for track_id, track_info in self.vehicle_tracks.items():
if current_time - track_info.last_seen > timeout:
tracks_to_remove.append(track_id)
for track_id in tracks_to_remove:
del self.vehicle_tracks[track_id]
logger.debug(f"清理车辆轨迹: {track_id}")
def get_performance_info(self) -> Dict:
"""获取性能信息"""
return {
'mode': 'local',
'environment': 'PaddlePaddle',
'model_dir': self.model_dir,
'mot_model_dir': self.mot_model_dir,
'plate_det_model_dir': self.plate_det_model_dir,
'plate_rec_model_dir': self.plate_rec_model_dir,
'detector_loaded': self._detector_initialized,
'available': self.available,
'active_tracks': len(self.vehicle_tracks)
}
# 兼容性包装，保持与 YOLO 模型相同的接口
class VehicleDetectionModel:
"""车辆检测模型包装器，兼容 YOLO 接口"""
def __init__(self):
self.service = VehicleDetectionService()
self.names = {0: 'vehicle'}
def __call__(self, image, conf=0.1, iou=0.45, verbose=False):
"""
模拟 YOLO 模型的调用接口
"""
result = self.service.detect_image(image, threshold=conf)
return [PaddleDetectionResult(result, self.names)]
def detect_illegal_parking(self, image, conf=0.1, illegal_parking_time=5.0,
region_polygon=None):
"""违停检测接口"""
return self.service.detect_illegal_parking(
image, conf, illegal_parking_time, region_polygon
)
class PaddleDetectionResult:
"""模拟 YOLO 检测结果对象"""
def __init__(self, detection_result: Dict, names: Dict):
self.detection_result = detection_result
self.names = names
self.boxes = self._create_boxes()
def _create_boxes(self):
"""创建模拟的 boxes 对象"""
detections = self.detection_result.get('detections', [])
if not detections:
return MockBoxes([])
xyxy = []
conf = []
cls = []
for det in detections:
xyxy.append(det['bbox'])
conf.append(det['confidence'])
cls.append(0)
return MockBoxes(xyxy, conf, cls)
class MockBoxes:
"""模拟 YOLO boxes 对象"""
def __init__(self, xyxy_list, conf_list=None, cls_list=None):
try:
import torch
use_torch = True
except ImportError:
use_torch = False
if xyxy_list and len(xyxy_list) > 0:
if use_torch:
self.xyxy = torch.tensor(xyxy_list, dtype=torch.float32)
self.conf = torch.tensor(conf_list, dtype=torch.float32).reshape(-1, 1)
self.cls = torch.tensor(cls_list, dtype=torch.int64).reshape(-1, 1)
else:
self.xyxy = np.array(xyxy_list, dtype=np.float32)
self.conf = np.array(conf_list, dtype=np.float32).reshape(-1, 1)
self.cls = np.array(cls_list, dtype=np.int64).reshape(-1, 1)
else:
if use_torch:
self.xyxy = torch.empty((0, 4), dtype=torch.float32)
self.conf = torch.empty((0, 1), dtype=torch.float32)
self.cls = torch.empty((0, 1), dtype=torch.int64)
else:
self.xyxy = np.array([]).reshape(0, 4)
self.conf = np.array([]).reshape(0, 1)
self.cls = np.array([], dtype=np.int64).reshape(0, 1)
self._use_torch = use_torch
def __iter__(self):
for i in range(len(self.xyxy)):
yield MockBox(
self.xyxy[i],
self.conf[i][0] if len(self.conf) > i else 0.0,
self.cls[i][0] if len(self.cls) > i else 0
)
def __len__(self):
return len(self.xyxy)
def cpu(self):
return self
def numpy(self):
if self._use_torch:
if len(self.xyxy) > 0:
return (
self.xyxy.numpy(),
self.conf.numpy(),
self.cls.numpy()
)
else:
return (
np.array([]).reshape(0, 4),
np.array([]).reshape(0, 1),
np.array([], dtype=np.int64).reshape(0, 1)
)
else:
return (
self.xyxy,
self.conf,
self.cls
)
class MockBox:
"""模拟单个 YOLO box 对象"""
def __init__(self, xyxy, conf, cls):
try:
import torch
use_torch = True
except ImportError:
use_torch = False
if use_torch:
if isinstance(xyxy, torch.Tensor):
self.xyxy = xyxy
else:
self.xyxy = torch.tensor(xyxy, dtype=torch.float32)
else:
if isinstance(xyxy, np.ndarray):
self.xyxy = xyxy
else:
self.xyxy = np.array(xyxy, dtype=np.float32)
self.conf = conf
self.cls = cls

98
apps/server/setup_env.sh Normal file
View File

@@ -0,0 +1,98 @@
#!/bin/bash
# 视频模型检测平台 - 环境初始化脚本
# 创建日期：2026-05-21
# Python 版本：3.12
set -e
echo "========================================="
echo "开始初始化视频模型检测平台环境"
echo "========================================="
# 检查 Python 版本
PYTHON_VERSION=$(python3 --version 2>&1 | awk '{print $2}')
echo "当前 Python 版本: $PYTHON_VERSION"
# 创建虚拟环境
if [ ! -d "venv" ]; then
echo "创建虚拟环境..."
python3 -m venv venv
echo "✅ 虚拟环境创建成功"
else
echo "⚠️ 虚拟环境已存在，跳过创建步骤"
fi
# 激活虚拟环境
echo "激活虚拟环境..."
source venv/bin/activate
# 升级 pip
echo "升级 pip..."
pip install --upgrade pip
# 安装依赖
echo "安装项目依赖..."
pip install -r requirements.txt
# 修复 imgaug NumPy 2.0 兼容性问题
echo "修复 imgaug NumPy 2.0 兼容性问题..."
IMGUAG_FILE="venv/lib/python3.12/site-packages/imgaug/imgaug.py"
if [ -f "$IMGUAG_FILE" ]; then
echo "应用 imgaug 兼容性补丁..."
# 检查是否已经修复
if grep -q "np.float16, np.float32, np.float64" "$IMGUAG_FILE"; then
echo "✅ imgaug 已修复，跳过"
else
# 备份原文件
cp "$IMGUAG_FILE" "${IMGUAG_FILE}.backup"
# 应用修复
sed -i.bak 's/NP_FLOAT_TYPES = set(np.sctypes\["float"\])/NP_FLOAT_TYPES = {np.float16, np.float32, np.float64}/' "$IMGUAG_FILE"
sed -i.bak 's/NP_INT_TYPES = set(np.sctypes\["int"\])/NP_INT_TYPES = {np.int8, np.int16, np.int32, np.int64}/' "$IMGUAG_FILE"
sed -i.bak 's/NP_UINT_TYPES = set(np.sctypes\["uint"\])/NP_UINT_TYPES = {np.uint8, np.uint16, np.uint32, np.uint64}/' "$IMGUAG_FILE"
echo "✅ imgaug 修复完成"
fi
else
echo "⚠️ imgaug 文件未找到，跳过修复步骤"
fi
# 验证安装
echo "验证关键包安装..."
python -c "import numpy; print(f'✅ NumPy {numpy.__version__}')"
python -c "import torch; print(f'✅ PyTorch {torch.__version__}')"
python -c "import ultralytics; print(f'✅ Ultralytics {ultralytics.__version__}')"
python -c "import paddle; print(f'✅ PaddlePaddle {paddle.__version__}')"
python -c "import cv2; print(f'✅ OpenCV {cv2.__version__}')"
# 检查 PaddleDetection 第三方库
PADDLE_DIR="../third-party/paddle-inference"
if [ -d "$PADDLE_DIR" ]; then
echo "✅ PaddleDetection 第三方库存在"
else
echo "⚠️ 警告：PaddleDetection 第三方库不存在于 $PADDLE_DIR"
echo "请确保从 AI Studio 下载相关模型和代码"
fi
# 创建必要的目录
echo "创建必要的目录..."
mkdir -p static/results
mkdir -p logs
echo "========================================="
echo "环境初始化完成！"
echo "========================================="
echo ""
echo "使用说明："
echo "1. 激活虚拟环境: source venv/bin/activate"
echo "2. 启动服务: python main.py"
echo "3. 或使用启动脚本: ./start_server_with_env.sh"
echo ""
echo "重要提示："
echo "- 确保下载了所有需要的模型文件到 models/ 目录"
echo "- PaddleDetection 第三方库位于: $PADDLE_DIR"
echo "- 详细版本信息见 requirements.txt"
echo ""

BIN
models/vehicle_detection_paddle/ch_PP-OCRv3_det_infer/inference.pdiparams LFS Executable file
View File

Binary file not shown.

BIN
models/vehicle_detection_paddle/ch_PP-OCRv3_det_infer/inference.pdiparams.info Executable file
View File

Binary file not shown.

BIN
models/vehicle_detection_paddle/ch_PP-OCRv3_det_infer/inference.pdmodel Executable file
View File

Binary file not shown.

BIN
models/vehicle_detection_paddle/ch_PP-OCRv3_rec_infer/inference.pdiparams LFS Executable file
View File

Binary file not shown.

BIN
models/vehicle_detection_paddle/ch_PP-OCRv3_rec_infer/inference.pdiparams.info Executable file
View File

Binary file not shown.

BIN
models/vehicle_detection_paddle/ch_PP-OCRv3_rec_infer/inference.pdmodel Executable file
View File

Binary file not shown.

17
models/vehicle_detection_paddle/mot_ppyoloe_l_36e_ppvehicle/infer_cfg.yml Normal file
View File

@@ -0,0 +1,17 @@
mode: paddle
draw_threshold: 0.5
metric: COCO
use_dynamic_shape: false
arch: YOLO
min_subgraph_size: 3
Preprocess:
- interp: 2
keep_ratio: false
target_size:
- 640
- 640
type: Resize
- type: Permute
label_list:
- vehicle

BIN
models/vehicle_detection_paddle/mot_ppyoloe_l_36e_ppvehicle/model.pdiparams LFS Normal file
View File

Binary file not shown.

BIN
models/vehicle_detection_paddle/mot_ppyoloe_l_36e_ppvehicle/model.pdiparams.info Normal file
View File

Binary file not shown.

BIN
models/vehicle_detection_paddle/mot_ppyoloe_l_36e_ppvehicle/model.pdmodel Normal file
View File

Binary file not shown.