1. YOLO11输出解析错误: 移除不存在的objectness行,正确使用class_scores.max() 2. CPU NMS逻辑错误: keep_mask同时标记保留和抑制框导致NMS失效,改用独立suppressed集合 3. 坐标映射缺失: _build_tracks中scale_info未使用,添加revert_boxes还原到ROI裁剪空间 4. batch=1限制: 恢复真正的动态batch推理(1~8),BatchPreprocessor支持多图stack 5. 帧率控制缺失: _read_frame添加time.monotonic()间隔控制,按target_fps跳帧 6. 拉流推理耦合: 新增独立推理线程(InferenceWorker),生产者-消费者模式解耦 7. 攒批形同虚设: 添加50ms攒批窗口+max_batch阈值,替代>=1立即处理 8. LeavePost双重等待: LEAVING确认后直接触发告警,不再进入OFF_DUTY二次等待 9. register_algorithm每帧调用: 添加_registered_keys缓存,O(1)快速路径跳过 10. GPU context线程安全: TensorRT infer()内部加锁,防止多线程CUDA context竞争 附带修复: - reset_algorithm中未定义algorithm_type变量(NameError) - update_roi_params中循环变量key覆盖外层key - AlertInfo缺少bind_id字段(TypeError) - _logger.log_alert在标准logger上不存在(AttributeError) - AlarmStateMachine死锁(Lock改为RLock) - ROICropper.create_mask坐标解析错误 - 更新测试用例适配新API Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
208 lines
6.1 KiB
Python
208 lines
6.1 KiB
Python
"""
|
|
预处理模块单元测试
|
|
"""
|
|
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
import sys
|
|
import os
|
|
import numpy as np
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
|
|
class TestROICropper(unittest.TestCase):
|
|
"""测试ROI裁剪器"""
|
|
|
|
def setUp(self):
|
|
"""设置测试环境"""
|
|
from core.preprocessor import ROICropper
|
|
self.cropper = ROICropper()
|
|
|
|
self.test_image = np.zeros((480, 640, 3), dtype=np.uint8)
|
|
self.test_image[100:200, 200:400] = 255
|
|
|
|
def test_crop_rectangle(self):
|
|
"""测试矩形裁剪"""
|
|
from config.config_models import ROIInfo, ROIType, AlgorithmType
|
|
from core.preprocessor import ROICropper
|
|
|
|
roi = ROIInfo(
|
|
roi_id="roi001",
|
|
camera_id="cam001",
|
|
roi_type=ROIType.RECTANGLE,
|
|
coordinates=[[200, 100], [400, 200]],
|
|
algorithm_type=AlgorithmType.LEAVE_POST,
|
|
)
|
|
|
|
cropper = ROICropper()
|
|
result = cropper.crop(self.test_image, roi)
|
|
|
|
self.assertIsNotNone(result)
|
|
self.assertEqual(result.shape[1], 200)
|
|
self.assertEqual(result.shape[0], 100)
|
|
|
|
def test_crop_polygon(self):
|
|
"""测试多边形裁剪"""
|
|
from config.config_models import ROIInfo, ROIType, AlgorithmType
|
|
from core.preprocessor import ROICropper
|
|
|
|
roi = ROIInfo(
|
|
roi_id="roi001",
|
|
camera_id="cam001",
|
|
roi_type=ROIType.POLYGON,
|
|
coordinates=[[200, 100], [400, 100], [400, 200], [200, 200]],
|
|
algorithm_type=AlgorithmType.LEAVE_POST,
|
|
)
|
|
|
|
cropper = ROICropper()
|
|
result = cropper.crop(self.test_image, roi)
|
|
|
|
self.assertIsNotNone(result)
|
|
|
|
def test_create_mask(self):
|
|
"""测试创建掩码"""
|
|
from config.config_models import ROIInfo, ROIType, AlgorithmType
|
|
from core.preprocessor import ROICropper
|
|
|
|
roi = ROIInfo(
|
|
roi_id="roi001",
|
|
camera_id="cam001",
|
|
roi_type=ROIType.RECTANGLE,
|
|
coordinates=[[100, 100], [200, 200]],
|
|
algorithm_type=AlgorithmType.LEAVE_POST,
|
|
)
|
|
|
|
cropper = ROICropper()
|
|
mask = cropper.create_mask((480, 640), roi)
|
|
|
|
self.assertEqual(mask.shape, (480, 640))
|
|
self.assertTrue(mask[150, 150] > 0)
|
|
|
|
|
|
class TestLetterboxPreprocessor(unittest.TestCase):
|
|
"""测试Letterbox预处理器"""
|
|
|
|
def test_preprocess_16_9(self):
|
|
"""测试16:9图像预处理"""
|
|
from core.preprocessor import LetterboxPreprocessor
|
|
|
|
preprocessor = LetterboxPreprocessor(target_size=(480, 480))
|
|
|
|
image = np.zeros((720, 1280, 3), dtype=np.uint8)
|
|
image[:, :] = [100, 120, 140]
|
|
|
|
result, scale_info = preprocessor.preprocess(image)
|
|
|
|
self.assertEqual(result.shape, (480, 480, 3))
|
|
self.assertEqual(len(scale_info), 4)
|
|
|
|
def test_preprocess_square(self):
|
|
"""测试正方形图像预处理"""
|
|
from core.preprocessor import LetterboxPreprocessor
|
|
|
|
preprocessor = LetterboxPreprocessor(target_size=(480, 480))
|
|
|
|
image = np.zeros((640, 640, 3), dtype=np.uint8)
|
|
|
|
result, scale_info = preprocessor.preprocess(image)
|
|
|
|
self.assertEqual(result.shape, (480, 480, 3))
|
|
|
|
def test_revert_coordinates(self):
|
|
"""测试坐标还原"""
|
|
from core.preprocessor import LetterboxPreprocessor
|
|
|
|
preprocessor = LetterboxPreprocessor(target_size=(480, 480))
|
|
|
|
scale = 0.5
|
|
pad_x = 60
|
|
pad_y = 60
|
|
|
|
scale_info = (scale, pad_x, pad_y, scale)
|
|
|
|
box = [100, 100, 200, 200]
|
|
reverted = preprocessor.revert_coordinates(box, scale_info)
|
|
|
|
self.assertEqual(len(reverted), 4)
|
|
self.assertGreater(reverted[0], 0)
|
|
|
|
|
|
class TestBatchPreprocessor(unittest.TestCase):
|
|
"""测试Batch预处理器"""
|
|
|
|
def test_preprocess_batch(self):
|
|
"""测试批次预处理"""
|
|
from core.preprocessor import BatchPreprocessor
|
|
|
|
preprocessor = BatchPreprocessor(
|
|
target_size=(480, 480),
|
|
fp16_mode=True
|
|
)
|
|
|
|
images = [
|
|
np.zeros((640, 640, 3), dtype=np.uint8)
|
|
for _ in range(2)
|
|
]
|
|
|
|
result, scale_info_list = preprocessor.preprocess_batch(images)
|
|
|
|
self.assertEqual(result.shape[0], 2)
|
|
self.assertEqual(len(scale_info_list), 2)
|
|
|
|
def test_max_batch_size(self):
|
|
"""测试最大batch大小"""
|
|
from core.preprocessor import BatchPreprocessor
|
|
|
|
preprocessor = BatchPreprocessor(
|
|
target_size=(480, 480),
|
|
fp16_mode=True
|
|
)
|
|
|
|
self.assertEqual(preprocessor.max_batch_size, 8)
|
|
|
|
|
|
class TestImagePreprocessor(unittest.TestCase):
|
|
"""测试图像预处理主类"""
|
|
|
|
def test_preprocess_single(self):
|
|
"""测试单张图像预处理"""
|
|
from core.preprocessor import ImagePreprocessor
|
|
|
|
preprocessor = ImagePreprocessor()
|
|
|
|
image = np.zeros((720, 1280, 3), dtype=np.uint8)
|
|
|
|
result, scale_info = preprocessor.preprocess_single(image)
|
|
|
|
self.assertEqual(result.shape, (480, 480, 3))
|
|
|
|
def test_preprocess_batch(self):
|
|
"""测试批次预处理"""
|
|
from core.preprocessor import ImagePreprocessor
|
|
|
|
preprocessor = ImagePreprocessor()
|
|
|
|
images = [
|
|
np.zeros((720, 1280, 3), dtype=np.uint8)
|
|
for _ in range(4)
|
|
]
|
|
|
|
result, scale_info_list = preprocessor.preprocess_batch(images)
|
|
|
|
self.assertEqual(result.shape[0], 4)
|
|
|
|
def test_get_statistics(self):
|
|
"""测试获取统计"""
|
|
from core.preprocessor import ImagePreprocessor
|
|
|
|
preprocessor = ImagePreprocessor()
|
|
stats = preprocessor.get_statistics()
|
|
|
|
self.assertIn("config", stats)
|
|
self.assertIn("batch_size", stats["config"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|