手机检测结果二次开发接口:DAMO-YOLO WebUI后端API调用与返回字段详解
1. 引言
如果你正在使用那个基于DAMO-YOLO的手机检测WebUI,可能会遇到这样的需求:不想每次都手动上传图片,而是希望在自己的程序里直接调用检测功能。比如,你想把手机检测集成到自己的监控系统里,或者需要批量处理大量图片。
好消息是,这个WebUI背后其实有一套完整的API接口,你可以像调用其他网络服务一样,通过代码直接发送图片、获取检测结果。今天,我就来带你深入这套API,看看它有哪些接口、怎么调用、返回的数据又该怎么解析。掌握了这些,你就能轻松实现手机检测功能的二次开发了。
2. 接口基础与环境准备
2.1 服务地址与状态确认
在开始调用API之前,首先要确保你的手机检测服务正在运行。根据你提供的文档,服务默认运行在http://服务器IP:7860这个地址。
怎么确认服务是否正常呢?最简单的方法就是用浏览器打开这个地址,看看WebUI界面能不能出来。但更“程序员”一点的做法是,用命令行工具来测试:
# 使用curl命令测试服务是否可达 curl -I http://localhost:7860如果服务正常,你会看到类似这样的响应:
HTTP/1.1 200 OK ...2.2 理解API的基本工作方式
这个手机检测系统的API基于Gradio框架构建。Gradio在创建WebUI的同时,会自动生成一套对应的HTTP API接口。简单来说,你在WebUI界面上做的每一个操作(上传图片、点击检测按钮),背后都是通过调用这些API来完成的。
API调用通常遵循这样的流程:
- 你的程序准备一张图片
- 通过HTTP请求把图片发送到指定接口
- 服务端接收图片并用DAMO-YOLO模型进行检测
- 服务端把检测结果打包成JSON格式返回
- 你的程序解析JSON,获取检测到的手机位置和置信度
3. 核心API接口详解
3.1 主要检测接口
系统最核心的接口就是执行手机检测的接口。根据Gradio的惯例,这个接口的路径通常是/api/predict。
接口信息:
- URL:
http://服务器IP:7860/api/predict - 方法: POST
- Content-Type:
multipart/form-data
请求示例(Python):
import requests # 服务地址 server_url = "http://localhost:7860/api/predict" # 准备图片文件 with open("test_phone.jpg", "rb") as f: files = {"file": ("test_phone.jpg", f, "image/jpeg")} # 发送请求 response = requests.post(server_url, files=files) # 解析响应 if response.status_code == 200: result = response.json() print("检测结果:", result) else: print(f"请求失败,状态码: {response.status_code}")请求参数说明:
file: 图片文件,支持常见的图片格式(JPG、PNG等)- 文件大小建议控制在10MB以内,过大的图片会影响处理速度
3.2 接口响应格式详解
当你调用检测接口后,服务端会返回一个结构化的JSON响应。理解这个响应的每个字段,是进行二次开发的关键。
一个典型的成功响应看起来是这样的:
{ "status": "success", "message": "检测完成", "data": { "detected_count": 2, "average_confidence": 0.952, "detections": [ { "bbox": [120, 80, 220, 180], "confidence": 0.961, "label": "phone", "bbox_normalized": [0.1875, 0.125, 0.34375, 0.28125] }, { "bbox": [400, 150, 500, 250], "confidence": 0.943, "label": "phone", "bbox_normalized": [0.625, 0.234375, 0.78125, 0.390625] } ], "image_size": [640, 480], "processing_time": 0.0042 }, "timestamp": "2024-01-15T10:30:45.123456" }3.3 返回字段逐项解析
让我们仔细看看每个字段的含义:
1. 状态信息字段
status: 请求状态,成功时为"success",失败时为"error"message: 状态描述信息,比如"检测完成"或具体的错误信息timestamp: 请求处理完成的时间戳
2. 检测统计字段
detected_count: 检测到的手机数量,整数类型average_confidence: 所有检测框的平均置信度,范围0-1processing_time: 处理耗时,单位秒,浮点数
3. 图片信息字段
image_size: 原始图片的尺寸,格式为[宽度, 高度]
4. 检测详情字段(detections数组)这是最核心的部分,每个检测到的手机都会有一个对应的对象:
bbox: 边界框坐标,格式为[x1, y1, x2, y2]- x1, y1: 左上角坐标
- x2, y2: 右下角坐标
- 坐标基于图片像素值
bbox_normalized: 归一化的边界框坐标,格式相同但值在0-1之间- 这个字段特别有用,因为无论原始图片多大,归一化坐标都是一致的
confidence: 置信度,范围0-1,越接近1表示越确定是手机label: 标签名称,固定为"phone"
4. 错误处理与边界情况
4.1 常见错误响应
不是每次调用都会成功,API可能会返回各种错误。了解这些错误,能帮你更好地调试程序。
1. 服务不可用错误
{ "status": "error", "message": "服务未启动或不可用", "data": null, "timestamp": "2024-01-15T10:35:22.123456" }可能原因:服务没有启动,或者端口被占用
2. 图片格式错误
{ "status": "error", "message": "不支持的图片格式", "data": null, "timestamp": "2024-01-15T10:36:15.123456" }可能原因:上传了非图片文件,或者图片损坏
3. 图片过大错误
{ "status": "error", "message": "图片大小超过限制", "data": null, "timestamp": "2024-01-15T10:37:08.123456" }可能原因:图片文件太大,超过了服务端的处理能力
4.2 健壮的API调用示例
在实际开发中,你需要考虑各种异常情况。下面是一个更健壮的调用示例:
import requests import time from typing import Optional, Dict, Any class PhoneDetectorAPI: def __init__(self, base_url: str = "http://localhost:7860"): self.base_url = base_url.rstrip("/") self.detect_url = f"{self.base_url}/api/predict" self.timeout = 30 # 30秒超时 def detect_phone(self, image_path: str, max_retries: int = 3) -> Optional[Dict[str, Any]]: """检测图片中的手机 Args: image_path: 图片文件路径 max_retries: 最大重试次数 Returns: 检测结果字典,失败时返回None """ for attempt in range(max_retries): try: with open(image_path, "rb") as f: files = {"file": (image_path.split("/")[-1], f, "image/jpeg")} # 发送请求 response = requests.post( self.detect_url, files=files, timeout=self.timeout ) # 检查响应状态 if response.status_code == 200: result = response.json() if result.get("status") == "success": return result["data"] else: print(f"检测失败: {result.get('message')}") return None else: print(f"HTTP错误: {response.status_code}") except requests.exceptions.Timeout: print(f"请求超时,第{attempt + 1}次重试...") if attempt < max_retries - 1: time.sleep(2) # 等待2秒后重试 continue except requests.exceptions.ConnectionError: print(f"连接失败,请检查服务是否启动") return None except FileNotFoundError: print(f"图片文件不存在: {image_path}") return None except Exception as e: print(f"未知错误: {str(e)}") return None print(f"达到最大重试次数{max_retries},检测失败") return None # 使用示例 if __name__ == "__main__": detector = PhoneDetectorAPI() # 检测单张图片 result = detector.detect_phone("test_image.jpg") if result: print(f"检测到 {result['detected_count']} 个手机") for i, detection in enumerate(result["detections"], 1): print(f"手机{i}: 置信度={detection['confidence']:.3f}, " f"位置={detection['bbox']}")5. 实际应用场景与代码示例
5.1 场景一:批量图片处理
如果你有很多图片需要检测,可以写一个简单的批量处理脚本:
import os import json from concurrent.futures import ThreadPoolExecutor, as_completed from tqdm import tqdm # 进度条库,需要安装:pip install tqdm def batch_detect_phones(image_dir: str, output_dir: str, max_workers: int = 4): """批量检测目录中的所有图片 Args: image_dir: 输入图片目录 output_dir: 输出结果目录 max_workers: 最大并发数 """ # 创建输出目录 os.makedirs(output_dir, exist_ok=True) # 获取所有图片文件 image_extensions = {'.jpg', '.jpeg', '.png', '.bmp', '.gif'} image_files = [] for filename in os.listdir(image_dir): if os.path.splitext(filename)[1].lower() in image_extensions: image_files.append(os.path.join(image_dir, filename)) print(f"找到 {len(image_files)} 张图片") # 创建检测器实例 detector = PhoneDetectorAPI() # 使用线程池并发处理 results = {} with ThreadPoolExecutor(max_workers=max_workers) as executor: # 提交所有任务 future_to_file = { executor.submit(detector.detect_phone, image_file): image_file for image_file in image_files } # 处理完成的任务 for future in tqdm(as_completed(future_to_file), total=len(image_files)): image_file = future_to_file[future] try: result = future.result() if result: # 保存结果到文件 base_name = os.path.basename(image_file) result_file = os.path.join( output_dir, f"{os.path.splitext(base_name)[0]}_result.json" ) with open(result_file, 'w', encoding='utf-8') as f: json.dump(result, f, ensure_ascii=False, indent=2) results[image_file] = result else: print(f"检测失败: {image_file}") except Exception as e: print(f"处理 {image_file} 时出错: {str(e)}") # 生成汇总报告 generate_summary_report(results, output_dir) def generate_summary_report(results: dict, output_dir: str): """生成检测结果汇总报告""" total_images = len(results) total_phones = sum(result['detected_count'] for result in results.values()) avg_confidence = sum( result['average_confidence'] for result in results.values() ) / total_images if total_images > 0 else 0 report = { "summary": { "total_images_processed": total_images, "total_phones_detected": total_phones, "average_phones_per_image": total_phones / total_images if total_images > 0 else 0, "average_confidence": avg_confidence }, "detailed_results": { os.path.basename(path): { "detected_count": result["detected_count"], "average_confidence": result["average_confidence"] } for path, result in results.items() } } report_file = os.path.join(output_dir, "summary_report.json") with open(report_file, 'w', encoding='utf-8') as f: json.dump(report, f, ensure_ascii=False, indent=2) print(f"\n处理完成!") print(f"共处理 {total_images} 张图片") print(f"检测到 {total_phones} 个手机") print(f"平均每张图片检测到 {total_phones/total_images:.2f} 个手机") print(f"平均置信度: {avg_confidence:.3f}") print(f"详细报告已保存到: {report_file}")5.2 场景二:实时监控集成
如果你想把手机检测集成到现有的监控系统中,可以这样设计:
import cv2 import numpy as np import time from datetime import datetime class RealTimeMonitor: def __init__(self, api_url: str, camera_index: int = 0): self.detector = PhoneDetectorAPI(api_url) self.camera = cv2.VideoCapture(camera_index) self.running = False # 检测历史记录 self.detection_history = [] self.max_history = 100 # 保留最近100次检测记录 def start_monitoring(self, interval: float = 2.0): """开始实时监控 Args: interval: 检测间隔,单位秒 """ self.running = True print("开始实时监控...") last_detection_time = 0 while self.running: current_time = time.time() # 按间隔进行检测 if current_time - last_detection_time >= interval: # 捕获一帧 ret, frame = self.camera.read() if not ret: print("无法读取摄像头画面") time.sleep(1) continue # 保存临时图片文件 temp_file = f"temp_frame_{int(current_time)}.jpg" cv2.imwrite(temp_file, frame) # 调用检测API result = self.detector.detect_phone(temp_file) if result: # 记录检测结果 detection_record = { "timestamp": datetime.now().isoformat(), "detected_count": result["detected_count"], "average_confidence": result["average_confidence"], "detections": result["detections"] } self.detection_history.append(detection_record) # 保持历史记录长度 if len(self.detection_history) > self.max_history: self.detection_history.pop(0) # 显示结果 self.display_result(frame, result) # 触发警报(如果检测到手机) if result["detected_count"] > 0: self.trigger_alert(result) # 清理临时文件 import os if os.path.exists(temp_file): os.remove(temp_file) last_detection_time = current_time # 处理按键事件(按'q'退出) key = cv2.waitKey(1) & 0xFF if key == ord('q'): break self.stop_monitoring() def display_result(self, frame: np.ndarray, result: dict): """在画面上显示检测结果""" display_frame = frame.copy() # 绘制检测框 for detection in result["detections"]: bbox = detection["bbox"] confidence = detection["confidence"] # 绘制矩形框 cv2.rectangle( display_frame, (int(bbox[0]), int(bbox[1])), (int(bbox[2]), int(bbox[3])), (0, 0, 255), # 红色 2 ) # 绘制标签 label = f"phone: {confidence:.3f}" cv2.putText( display_frame, label, (int(bbox[0]), int(bbox[1]) - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 255), 2 ) # 显示统计信息 stats_text = f"检测到 {result['detected_count']} 个手机 | 平均置信度: {result['average_confidence']:.3f}" cv2.putText( display_frame, stats_text, (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2 ) # 显示画面 cv2.imshow("手机检测监控", display_frame) def trigger_alert(self, result: dict): """触发警报""" alert_message = ( f" 检测到手机!\n" f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n" f"数量: {result['detected_count']}个\n" f"平均置信度: {result['average_confidence']:.3f}" ) print(f"\n{alert_message}") # 这里可以添加其他警报逻辑,比如: # 1. 发送邮件通知 # 2. 保存截图 # 3. 触发声音警报 # 4. 记录到数据库 def stop_monitoring(self): """停止监控""" self.running = False if self.camera.isOpened(): self.camera.release() cv2.destroyAllWindows() print("监控已停止") # 保存历史记录 self.save_detection_history() def save_detection_history(self): """保存检测历史到文件""" if self.detection_history: import json history_file = f"detection_history_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" with open(history_file, 'w', encoding='utf-8') as f: json.dump(self.detection_history, f, ensure_ascii=False, indent=2) print(f"检测历史已保存到: {history_file}") # 使用示例 if __name__ == "__main__": # 创建监控实例 monitor = RealTimeMonitor( api_url="http://localhost:7860", camera_index=0 # 0表示默认摄像头 ) # 开始监控,每2秒检测一次 monitor.start_monitoring(interval=2.0)6. 性能优化与最佳实践
6.1 减少网络开销
频繁调用API会产生网络开销,特别是在批量处理时。这里有几个优化建议:
1. 图片预处理
def optimize_image_for_api(image_path: str, max_size: tuple = (1024, 1024)): """优化图片以减少传输大小""" import cv2 # 读取图片 img = cv2.imread(image_path) if img is None: return None # 获取原始尺寸 height, width = img.shape[:2] # 如果图片太大,进行缩放 if height > max_size[1] or width > max_size[0]: scale = min(max_size[0]/width, max_size[1]/height) new_width = int(width * scale) new_height = int(height * scale) img = cv2.resize(img, (new_width, new_height), interpolation=cv2.INTER_AREA) # 调整图片质量(对于JPEG) encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), 85] # 85%质量 # 保存优化后的图片 optimized_path = f"optimized_{os.path.basename(image_path)}" cv2.imwrite(optimized_path, img, encode_param) # 计算节省的空间 original_size = os.path.getsize(image_path) optimized_size = os.path.getsize(optimized_path) savings = (original_size - optimized_size) / original_size * 100 print(f"图片优化: {original_size/1024:.1f}KB → {optimized_size/1024:.1f}KB " f"(节省{savings:.1f}%)") return optimized_path2. 使用连接池
import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def create_http_session(pool_connections: int = 10, pool_maxsize: int = 10): """创建带有连接池的HTTP会话""" session = requests.Session() # 配置重试策略 retry_strategy = Retry( total=3, # 最大重试次数 backoff_factor=1, # 重试间隔 status_forcelist=[429, 500, 502, 503, 504] # 需要重试的状态码 ) # 创建适配器 adapter = HTTPAdapter( max_retries=retry_strategy, pool_connections=pool_connections, pool_maxsize=pool_maxsize ) # 挂载适配器 session.mount("http://", adapter) session.mount("https://", adapter) return session # 使用连接池 session = create_http_session() # 在批量处理中使用同一个session for image_file in image_files: with open(image_file, "rb") as f: files = {"file": f} response = session.post(api_url, files=files) # 处理响应...6.2 结果缓存策略
如果同一张图片可能被多次检测,可以考虑添加缓存:
import hashlib import pickle from functools import lru_cache import os class CachedPhoneDetector: def __init__(self, detector: PhoneDetectorAPI, cache_dir: str = "detection_cache"): self.detector = detector self.cache_dir = cache_dir # 创建缓存目录 os.makedirs(cache_dir, exist_ok=True) def get_file_hash(self, file_path: str) -> str: """计算文件哈希值作为缓存键""" hasher = hashlib.md5() with open(file_path, "rb") as f: # 读取文件前1MB内容计算哈希(对于大文件更快) chunk = f.read(1024 * 1024) hasher.update(chunk) # 添加文件大小和修改时间 file_stat = os.stat(file_path) hasher.update(str(file_stat.st_size).encode()) hasher.update(str(file_stat.st_mtime).encode()) return hasher.hexdigest() def detect_with_cache(self, image_path: str) -> dict: """带缓存的检测""" # 生成缓存键 cache_key = self.get_file_hash(image_path) cache_file = os.path.join(self.cache_dir, f"{cache_key}.pkl") # 检查缓存 if os.path.exists(cache_file): try: with open(cache_file, "rb") as f: cached_result = pickle.load(f) print(f"使用缓存结果: {image_path}") return cached_result except: print(f"缓存读取失败,重新检测: {image_path}") # 调用API检测 result = self.detector.detect_phone(image_path) if result: # 保存到缓存 try: with open(cache_file, "wb") as f: pickle.dump(result, f) except: print(f"缓存保存失败: {cache_file}") return result # 使用缓存检测器 detector = PhoneDetectorAPI() cached_detector = CachedPhoneDetector(detector) # 第一次调用会实际检测 result1 = cached_detector.detect_with_cache("test.jpg") # 第二次调用相同文件会使用缓存 result2 = cached_detector.detect_with_cache("test.jpg")7. 总结
通过本文的详细介绍,你应该已经掌握了DAMO-YOLO手机检测系统API的完整使用方法。让我们回顾一下关键点:
7.1 核心要点回顾
- API基础:系统提供了基于HTTP的RESTful API,主要通过
/api/predict接口进行手机检测 - 调用方式:使用POST方法,以
multipart/form-data格式上传图片文件 - 返回格式:响应为JSON格式,包含状态信息、检测统计和详细的检测结果
- 关键字段:重点关注
detections数组中的bbox(边界框)、confidence(置信度)和label(标签)字段
7.2 实际应用建议
根据不同的使用场景,我有几个实用建议:
对于批量处理:
- 使用并发请求提高处理速度,但注意控制并发数避免压垮服务
- 考虑添加图片预处理,减少网络传输开销
- 实现结果缓存,避免重复检测相同图片
对于实时监控:
- 合理设置检测间隔,平衡实时性和系统负载
- 实现健壮的错误处理,确保监控持续运行
- 考虑添加警报机制,及时通知异常情况
对于所有场景:
- 一定要添加完善的错误处理和日志记录
- 考虑服务可用性,实现重试机制和降级策略
- 定期检查服务状态,确保检测功能正常
7.3 下一步探索方向
掌握了基础API调用后,你还可以进一步探索:
- 性能监控:添加API调用性能指标收集,了解平均响应时间、成功率等
- 结果可视化:开发更丰富的结果展示界面,支持热力图、时间线等可视化
- 模型更新:关注DAMO-YOLO模型的更新,及时升级以获得更好的检测效果
- 多模型集成:考虑集成其他检测模型,提供更全面的物体检测能力
API二次开发的核心思想是:把复杂的AI检测能力封装成简单的接口调用,让它可以轻松集成到各种应用系统中。希望本文能帮助你更好地利用这个手机检测系统,开发出满足自己需求的应用。
获取更多AI镜像
想探索更多AI镜像和应用场景?访问 CSDN星图镜像广场,提供丰富的预置镜像,覆盖大模型推理、图像生成、视频生成、模型微调等多个领域,支持一键部署。