从Demo到生产YOLOv5检测结果JSON化与业务系统集成实战在计算机视觉领域YOLOv5因其出色的速度和精度平衡成为工业界宠儿。但许多开发者止步于跑通demo当需要将检测结果接入业务系统时却无从下手——检测框坐标如何结构化存储置信度怎样传递给下游服务本文将手把手带你突破最后一公里实现从检测结果到业务价值的转化。1. 为什么需要定制化输出JSON默认的YOLOv5检测脚本detect.py提供txt格式输出但这种一行一个目标的简单结构难以满足业务系统需求。JSON作为现代Web服务的通用数据格式具有三大优势结构化嵌套支持多层数据嵌套如图像信息→多个目标→每个目标的属性和坐标类型丰富可保留浮点数精度如0.876而非文本化的0.87跨平台兼容几乎所有编程语言都内置JSON解析库典型业务场景需求示例{ image_id: factory_cam_20230501_084500.jpg, detections: [ { class: defective_product, confidence: 0.92, bbox: [325, 178, 412, 255], timestamp: 1682913900.45 } ] }2. 修改detect.py实现JSON输出2.1 核心修改点定位在detect.py中结果保存逻辑集中在run函数的以下位置# 原始代码片段约第240行 for *xyxy, conf, cls in reversed(det): if save_txt: # Write to file xywh (xyxy2xywh(torch.tensor(xyxy).view(1, 4)) / gn).view(-1).tolist() line (cls, *xywh, conf) if save_conf else (cls, *xywh) with open(txt_path .txt, a) as f: f.write((%g * len(line)).rstrip() % line \n)我们需要在此处插入JSON格式的生成逻辑。2.2 完整修改方案在脚本顶部参数解析部分添加新选项# 在parse_opt()函数中添加 parser.add_argument(--save-json, actionstore_true, helpsave results to *.json)然后在结果处理循环中插入以下代码# 在for *xyxy, conf, cls循环内部添加 if save_json: bbox_json { class_id: int(cls), class_name: names[int(cls)], confidence: float(conf), bbox: { xmin: int(xyxy[0]), ymin: int(xyxy[1]), xmax: int(xyxy[2]), ymax: int(xyxy[3]) }, normalized_bbox: [ # 标准化坐标可选 float((xyxy[0] xyxy[2]) / 2 / im0.shape[1]), # x_center float((xyxy[1] xyxy[3]) / 2 / im0.shape[0]), # y_center float((xyxy[2] - xyxy[0]) / im0.shape[1]), # width float((xyxy[3] - xyxy[1]) / im0.shape[0]) # height ] } if not hasattr(opt, json_output): opt.json_output [] opt.json_output.append(bbox_json)最后在循环结束后添加JSON文件保存逻辑# 在检测循环结束后添加 if save_json: import json from pathlib import Path json_path str(Path(save_dir) / f{Path(p).stem}.json) with open(json_path, w) as f: json.dump({ image_path: str(p), image_size: {width: im0.shape[1], height: im0.shape[0]}, detections: opt.json_output, timestamp: time.time() }, f, indent2)3. 业务系统对接方案3.1 轻量级API服务示例Flask创建接收JSON检测结果的基础服务from flask import Flask, request, jsonify import logging from datetime import datetime app Flask(__name__) app.route(/api/detections, methods[POST]) def handle_detection(): try: data request.json logging.info(fReceived detection at {datetime.now()}) # 业务逻辑示例缺陷产品告警 for det in data[detections]: if det[class_name] defective and det[confidence] 0.9: send_alert( image_iddata[image_path], defect_typedet[class_name], locationf{det[bbox][xmin]},{det[bbox][ymin]} ) return jsonify({status: success}), 200 except Exception as e: logging.error(fError: {str(e)}) return jsonify({error: str(e)}), 500 def send_alert(image_id, defect_type, location): # 实现告警逻辑邮件/短信/企业微信等 print(fALERT: {defect_type} detected in {image_id} at {location}) if __name__ __main__: app.run(host0.0.0.0, port5000)3.2 生产环境优化建议性能优化使用gunicorn部署替代开发服务器添加Redis缓存高频检测结果实现异步处理Celery RabbitMQ数据验证from pydantic import BaseModel, confloat, conint class BBox(BaseModel): xmin: conint(ge0) ymin: conint(ge0) xmax: conint(ge0) ymax: conint(ge0) class Detection(BaseModel): class_name: str confidence: confloat(ge0, le1) bbox: BBox安全措施添加JWT认证限制请求频率实现HTTPS加密传输4. 进阶动态配置与扩展4.1 可配置化输出模板通过YAML文件定义输出结构# config/output_template.yaml output_schema: required_fields: - image_info - detections detection_fields: - name: class_name type: string - name: confidence type: float precision: 4 - name: bbox type: object fields: - xmin - ymin - xmax - ymax metadata: include_timestamp: true custom_fields: - camera_id - production_line对应的Python加载代码import yaml def load_template(template_path): with open(template_path) as f: return yaml.safe_load(f) def format_output(detections, template): output {detections: []} if template[metadata][include_timestamp]: output[timestamp] time.time() for det in detections: formatted {} for field in template[detection_fields]: if field[name] in det: value det[field[name]] if field[type] float and precision in field: value round(value, field[precision]) formatted[field[name]] value output[detections].append(formatted) return output4.2 结果后处理管道构建可扩展的处理流程class DetectionPipeline: def __init__(self): self.processors [] def add_processor(self, processor): self.processors.append(processor) def run(self, detection_data): for processor in self.processors: detection_data processor.process(detection_data) return detection_data class ConfidenceFilter: def __init__(self, threshold0.5): self.threshold threshold def process(self, data): data[detections] [ d for d in data[detections] if d[confidence] self.threshold ] return data class BBoxNormalizer: def process(self, data): width data[image_size][width] height data[image_size][height] for det in data[detections]: bbox det[bbox] det[normalized_bbox] [ (bbox[xmin] bbox[xmax]) / 2 / width, (bbox[ymin] bbox[ymax]) / 2 / height, (bbox[xmax] - bbox[xmin]) / width, (bbox[ymax] - bbox[ymin]) / height ] return data5. 实战生产线缺陷检测案例5.1 场景描述某电子产品生产线需要实现实时检测产品外观缺陷划痕、污渍、装配异常将缺陷位置和类型记录到质量管理系统触发声光报警并暂停传送带5.2 系统架构YOLOv5检测节点 → JSON API → 业务处理中心 → [数据库存储 | 报警系统 | MES集成]关键实现代码# 检测节点定制化输出 def generate_production_output(detections, camera_id): return { production_batch: os.getenv(CURRENT_BATCH), camera_id: camera_id, timestamp: datetime.now().isoformat(), defects: [ { type: d[class_name], severity: critical if d[confidence] 0.9 else minor, location: { x: d[bbox][xmin], y: d[bbox][ymin], width: d[bbox][xmax] - d[bbox][xmin], height: d[bbox][ymax] - d[bbox][ymin] } } for d in detections ] } # 业务系统处理逻辑 def handle_production_defect(data): db_record { batch: data[production_batch], detection_time: data[timestamp], defects: [] } has_critical False for defect in data[defects]: db_record[defects].append({ type: defect[type], location: defect[location] }) if defect[severity] critical: has_critical True mongo_db.defects.insert_one(db_record) if has_critical: conveyor.stop() alert_system.trigger( locationdata[camera_id], defect_types[d[type] for d in data[defects]] )