# Edge_Inference_Service 工业级边缘AI推理服务开发计划 ## 一、项目概述 开发工业级Python推理服务,支持多路RTSP视频流接入、TensorRT GPU推理、实时告警上报,满足7×24小时稳定运行要求。 ## 二、技术环境要求 - **操作系统**: Windows 10/11 - **GPU环境**: CUDA 12.1 + cuDNN 8.9.7 + TensorRT 8.6.1 - **Python环境**: Conda虚拟环境 `yolo` - **核心依赖**: opencv-python, ultralytics, pycuda, paho-mqtt, sqlalchemy, redis ## 三、项目目录结构 ``` edge_inference_service/ ├── config/ # 配置模块 │ ├── __init__.py │ ├── settings.py # 全局配置(数据库、MQTT、Redis等) │ ├── database.py # 数据库连接配置 │ └── config_models.py # 数据模型定义 ├── core/ # 核心处理模块 │ ├── __init__.py │ ├── config_sync.py # 配置同步模块(第1阶段) │ ├── video_stream.py # 视频流接入模块(第2阶段) │ ├── preprocessor.py # 图像预处理流水线(第3阶段) │ ├── tensorrt_engine.py # TensorRT推理引擎(第4阶段) │ ├── postprocessor.py # 后处理与业务逻辑(第5阶段) │ └── result_reporter.py # 结果上报模块(第6阶段) ├── utils/ # 工具类 │ ├── __init__.py │ ├── logger.py # 分级日志系统 │ ├── common.py # 公共工具函数 │ └── version_control.py # 版本控制机制 ├── tests/ # 测试目录 │ ├── __init__.py │ ├── test_config_sync.py # 配置模块单元测试 │ ├── test_video_stream.py # 视频流模块单元测试 │ ├── test_preprocessor.py # 预处理模块单元测试 │ ├── test_tensorrt.py # TensorRT模块单元测试 │ ├── test_postprocessor.py # 后处理模块单元测试 │ └── test_integration.py # 集成测试 ├── logs/ # 日志目录 ├── models/ # 模型文件目录 ├── data/ # 数据目录 ├── main.py # 主入口 ├── requirements.txt # 依赖清单 └── CHANGELOG.md # 版本更新记录 ``` ## 四、详细实施计划 ### 第1阶段:配置同步模块 **目标**: 实现MySQL数据库连接管理、Redis Pub/Sub订阅、配置缓存与动态刷新 **任务分解**: 1.1 创建`config/settings.py`,定义数据库连接参数、Redis配置、MQTT配置 1.2 开发`config/database.py`,实现MySQL连接池管理类 1.3 创建`config/config_models.py`,定义CameraInfo和ROIConfig数据模型 1.4 开发`core/config_sync.py`: - 实现MySQL连接管理,读取camera_info和roi_config表 - 开发Redis Pub/Sub客户端,监听"config_update"主题 - 实现配置动态刷新机制,记录版本号 - 设计LRU缓存策略,减少数据库访问 1.5 编写配置模块单元测试,覆盖率≥80% ### 第2阶段:视频流接入模块 **目标**: 实现多路RTSP流拉取、动态抽帧、断线重连 **任务分解**: 2.1 开发`core/video_stream.py`: - 实现RTSP流拉取类,支持多线程并发 - 实现动态抽帧机制(可配置fps,如5fps) - 设计断线重连机制(指数退避策略) - 实现视频帧时间戳同步与缓存管理 2.2 编写视频流模块单元测试 ### 第3阶段:图像预处理流水线 **目标**: 实现ROI裁剪、Letterbox预处理、Batch打包 **任务分解**: 3.1 开发`core/preprocessor.py`: - 实现ROI裁剪模块(支持多边形和矩形) - 实现Letterbox预处理: - 等比例缩放至480×480 - 灰色(114,114,114)填充 - 保持原始比例 - 设计Batch打包机制: - 动态Batch大小(1-8) - NCHW格式转换 - FP16精度转换 - 连续内存块分配 3.2 编写预处理模块单元测试 ### 第4阶段:TensorRT推理引擎 **目标**: 实现引擎加载、显存优化、异步推理、性能监控 **任务分解**: 4.1 开发`core/tensorrt_engine.py`: - 实现TensorRT引擎管理类 - 实现engine文件加载与卸载 - 实现上下文管理,支持动态设置输入维度 - 优化显存管理: - 输入/输出显存绑定 - 异步推理模式execute_v2 - 显存池机制 - 实现推理性能监控(耗时、吞吐量) 4.2 编写TensorRT模块单元测试 ### 第5阶段:后处理与业务逻辑 **目标**: 实现NMS算法、坐标映射、ROI区域判定、告警状态机 **任务分解**: 5.1 开发`core/postprocessor.py`: - 实现高效NMS算法 - 开发坐标映射系统: - 480×480 → ROI空间 - ROI空间 → 原始帧坐标 - 实现ROI区域判定算法: - 多边形区域内点检测 - 目标中心点/Bottom-Center判定 - 设计告警状态机: - 连续N帧检测触发 - 可配置告警阈值 - 防止瞬时误报 5.2 编写后处理模块单元测试 ### 第6阶段:结果上报模块 **目标**: 实现MQTT客户端、告警存储、重试机制 **任务分解**: 6.1 开发`core/result_reporter.py`: - 实现MQTT客户端(云端业务平台连接) - 设计告警信息格式(JSON) - 实现MySQL告警记录存储 - 开发上报重试机制(指数退避) 6.2 编写结果上报模块单元测试 ### 第7阶段:质量保障与集成测试 **目标**: 完善异常处理、性能监控、代码质量 **任务分解**: 7.1 开发`utils/logger.py`: - 实现分级日志系统(DEBUG/INFO/WARNING/ERROR) - 记录关键性能指标 7.2 开发`utils/version_control.py`: - 实现代码更新记录机制 - 记录更新时间、内容、修改人、影响范围 7.3 完善各模块异常处理: - RTSP流异常断开检测与重连 - 显存溢出捕获与恢复 - 数据库连接异常处理 - MQTT连接状态监控与重连 7.4 编写集成测试: - 模块间集成测试 - 端到端测试 - 性能压力测试 7.5 创建`CHANGELOG.md`,记录版本更新 ## 五、开发规范 1. **编码规范**: 严格遵循PEP 8 2. **注释要求**: 关键步骤添加详细中文注释 3. **代码质量**: 单元测试覆盖率≥80% 4. **模块化**: 高内聚低耦合,支持独立测试 5. **异常处理**: 全面的try-except捕获与日志记录 6. **性能优化**: 显存管理、异步处理、批处理优化 ## 六、测试策略 1. **单元测试**: 每个模块独立测试,确保功能正确 2. **集成测试**: 模块间接口测试,确保数据流正确 3. **性能测试**: 端到端延迟、吞吐量、资源占用测试 4. **稳定性测试**: 长时间运行测试,验证7×24小时稳定性 ## 七、交付物清单 1. 完整的项目源代码 2. requirements.txt依赖清单 3. CHANGELOG.md版本更新记录 4. 单元测试代码(覆盖率≥80%) 5. 集成测试代码 6. 配置文件模板 --- **计划总周期**: 7个阶段,预计2-3周完成 **优先级**: 第1→7阶段顺序开发,可并行测试