Qwen3-4B-Instruct算力优化:CPU内存占用<6GB的4B模型轻量部署方案
批量检测虽未上线,但可脚本化!DAMO-YOLO图片循环检测自动化方案
1. 引言:从手动上传到自动化的跨越
你是不是也遇到过这样的场景?手头有一大堆图片需要检测里面有没有手机,但系统只支持一张一张上传,点得手都酸了。或者,你需要定时监控某个文件夹,看看有没有新的图片需要处理,但每次都得手动操作,费时费力。
我最近在用的这个DAMO-YOLO手机检测系统确实好用,准确率88.8%,速度也快,3.83毫秒就能处理一张图。但美中不足的是,它目前只支持单张图片上传检测,批量处理功能还没上线。
不过,作为技术人,我们总有办法。既然系统提供了Web界面和API,那我们能不能自己写个脚本,让它自动循环检测文件夹里的所有图片呢?答案是肯定的。
今天我就来分享一个实用的自动化方案,用Python脚本实现图片的批量循环检测。这个方案特别适合那些需要处理大量图片的场景,比如监控录像截图分析、考场照片批量检查、或者日常的图片素材管理。
2. 理解系统架构:WebUI背后的技术栈
在开始写脚本之前,我们先简单了解一下这个手机检测系统的技术架构,这样写脚本的时候心里更有底。
2.1 核心组件分析
这个系统基于几个关键技术构建:
-
DAMO-YOLO模型:这是阿里巴巴达摩院开发的轻量级目标检测模型,特点是"小、快、省"。模型文件大约125MB,在T4 GPU上每张图片只需要3.83毫秒就能完成检测,非常适合手机端或边缘设备这种算力有限的场景。
-
Gradio Web框架:系统用Gradio 6.5搭建了Web界面。Gradio有个很好的特性,它不仅能提供友好的用户界面,还会自动创建API接口。我们写脚本的时候,就是通过调用这些API来实现自动化。
-
Supervisor服务管理:系统用Supervisor来管理服务进程,确保服务稳定运行。我们的脚本需要先确认服务是否正常启动,才能开始批量处理。
2.2 系统工作流程
当你通过Web界面上传一张图片时,系统内部是这样工作的:
- 图片通过HTTP请求发送到服务器
- Gradio接收图片并调用DAMO-YOLO模型
- 模型进行推理,识别图片中的手机
- 结果以JSON格式返回,包含检测框位置和置信度
- Web界面显示带标注框的结果图片
我们的自动化脚本就是要模拟这个流程,但是批量进行。
3. 自动化脚本设计与实现
现在进入正题,我来详细讲解如何编写这个自动化检测脚本。我会从最简单的版本开始,逐步增加功能,让你能根据自己的需求进行调整。
3.1 基础版本:单次批量检测
我们先写一个最基础的脚本,它能一次性检测指定文件夹里的所有图片。
import os
import requests
import json
from PIL import Image
import time
class PhoneDetector:
def __init__(self, server_url="http://localhost:7860"):
"""
初始化检测器
:param server_url: 手机检测服务的地址
"""
self.server_url = server_url
self.api_url = f"{server_url}/api/predict"
def check_service(self):
"""检查服务是否正常运行"""
try:
response = requests.get(self.server_url, timeout=5)
return response.status_code == 200
except:
return False
def detect_single_image(self, image_path):
"""
检测单张图片
:param image_path: 图片文件路径
:return: 检测结果
"""
if not os.path.exists(image_path):
return {"error": f"文件不存在: {image_path}"}
try:
# 打开并验证图片
with Image.open(image_path) as img:
img.verify()
# 准备请求数据
files = {'image': open(image_path, 'rb')}
data = {'fn_index': 0} # Gradio API参数
# 发送请求
start_time = time.time()
response = requests.post(self.api_url, files=files, data=data)
processing_time = time.time() - start_time
if response.status_code == 200:
result = response.json()
return {
"success": True,
"image_path": image_path,
"detection_count": len(result.get('data', [])),
"confidence_scores": [item.get('confidence', 0) for item in result.get('data', [])],
"processing_time": processing_time,
"raw_result": result
}
else:
return {
"success": False,
"error": f"API请求失败: {response.status_code}",
"image_path": image_path
}
except Exception as e:
return {
"success": False,
"error": str(e),
"image_path": image_path
}
def batch_detect(self, image_folder, extensions=['.jpg', '.jpeg', '.png', '.bmp']):
"""
批量检测文件夹中的所有图片
:param image_folder: 图片文件夹路径
:param extensions: 支持的图片格式
:return: 所有图片的检测结果
"""
# 检查服务状态
if not self.check_service():
print("错误:检测服务未启动或无法访问")
print(f"请确保服务正在运行,并可以通过 {self.server_url} 访问")
return []
# 获取所有图片文件
image_files = []
for ext in extensions:
image_files.extend([f for f in os.listdir(image_folder)
if f.lower().endswith(ext)])
if not image_files:
print(f"在文件夹 {image_folder} 中未找到支持的图片文件")
return []
print(f"找到 {len(image_files)} 张图片,开始批量检测...")
results = []
for i, filename in enumerate(image_files, 1):
image_path = os.path.join(image_folder, filename)
print(f"正在处理第 {i}/{len(image_files)} 张: {filename}")
result = self.detect_single_image(image_path)
results.append(result)
if result.get('success'):
count = result['detection_count']
if count > 0:
avg_conf = sum(result['confidence_scores']) / count
print(f" 检测到 {count} 个手机,平均置信度: {avg_conf:.1%}")
else:
print(" 未检测到手机")
else:
print(f" 处理失败: {result.get('error')}")
# 添加短暂延迟,避免请求过快
time.sleep(0.1)
return results
# 使用示例
if __name__ == "__main__":
# 创建检测器实例
detector = PhoneDetector("http://localhost:7860")
# 指定要检测的图片文件夹
image_folder = "./test_images"
# 执行批量检测
results = detector.batch_detect(image_folder)
# 打印统计信息
total_images = len(results)
successful = sum(1 for r in results if r.get('success'))
total_phones = sum(r.get('detection_count', 0) for r in results if r.get('success'))
print(f"\n批量检测完成!")
print(f"总共处理: {total_images} 张图片")
print(f"成功处理: {successful} 张")
print(f"检测到手机总数: {total_phones} 个")
这个基础版本已经能完成基本的批量检测功能。它主要做了几件事:
- 检查服务状态:先确认检测服务是否正常运行
- 扫描图片文件夹:找出所有支持的图片文件
- 逐张发送检测请求:模拟Web界面的上传操作
- 收集和展示结果:统计检测到的手机数量和置信度
3.2 增强版本:添加更多实用功能
基础版本能用,但还不够好用。我们来给它增加一些实用功能,比如结果保存、进度显示、错误重试等。
import os
import requests
import json
import time
import csv
from datetime import datetime
from pathlib import Path
from PIL import Image
import threading
from queue import Queue
class EnhancedPhoneDetector(PhoneDetector):
def __init__(self, server_url="http://localhost:7860", max_retries=3):
super().__init__(server_url)
self.max_retries = max_retries
self.results_dir = Path("./detection_results")
self.results_dir.mkdir(exist_ok=True)
def detect_with_retry(self, image_path, retry_count=0):
"""带重试机制的检测函数"""
try:
return self.detect_single_image(image_path)
except Exception as e:
if retry_count < self.max_retries:
print(f" 检测失败,第 {retry_count + 1} 次重试...")
time.sleep(1) # 等待1秒后重试
return self.detect_with_retry(image_path, retry_count + 1)
else:
return {
"success": False,
"error": f"重试{self.max_retries}次后失败: {str(e)}",
"image_path": image_path
}
def save_results(self, results, output_format='csv'):
"""保存检测结果到文件"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
if output_format == 'csv':
csv_path = self.results_dir / f"detection_results_{timestamp}.csv"
with open(csv_path, 'w', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
# 写入表头
writer.writerow([
'图片文件', '检测状态', '手机数量',
'平均置信度', '处理时间(秒)', '错误信息'
])
# 写入数据
for result in results:
if result.get('success'):
count = result.get('detection_count', 0)
if count > 0:
avg_conf = sum(result.get('confidence_scores', [])) / count
else:
avg_conf = 0
writer.writerow([
result.get('image_path', ''),
'成功',
count,
f"{avg_conf:.2%}",
f"{result.get('processing_time', 0):.3f}",
''
])
else:
writer.writerow([
result.get('image_path', ''),
'失败',
0,
'0%',
0,
result.get('error', '未知错误')
])
print(f"结果已保存到: {csv_path}")
return csv_path
elif output_format == 'json':
json_path = self.results_dir / f"detection_results_{timestamp}.json"
with open(json_path, 'w', encoding='utf-8') as f:
json.dump(results, f, ensure_ascii=False, indent=2)
print(f"结果已保存到: {json_path}")
return json_path
def batch_detect_enhanced(self, image_folder, save_results=True,
output_format='csv', show_progress=True):
"""
增强版批量检测
:param save_results: 是否保存结果
:param output_format: 结果保存格式 (csv/json)
:param show_progress: 是否显示进度条
:return: 检测结果
"""
# 检查服务
if not self.check_service():
print("服务检查失败,请确认:")
print(f"1. 服务是否启动: supervisorctl status phone-detection")
print(f"2. 端口是否可访问: curl {self.server_url}")
return []
# 获取图片文件
image_files = []
extensions = ['.jpg', '.jpeg', '.png', '.bmp', '.webp']
for ext in extensions:
pattern = f"*{ext}"
image_files.extend(list(Path(image_folder).glob(pattern)))
image_files.extend(list(Path(image_folder).glob(pattern.upper())))
if not image_files:
print(f"在 {image_folder} 中未找到图片文件")
return []
print(f"开始批量检测,共 {len(image_files)} 张图片")
print("=" * 50)
results = []
start_time = time.time()
for i, image_path in enumerate(image_files, 1):
# 显示进度
if show_progress:
progress = i / len(image_files) * 100
print(f"[{i:3d}/{len(image_files)}] {progress:5.1f}% - {image_path.name}")
# 检测图片
result = self.detect_with_retry(image_path)
results.append(result)
# 显示简要结果
if result.get('success'):
count = result.get('detection_count', 0)
if count > 0:
avg_conf = sum(result.get('confidence_scores', [])) / count
print(f" 检测到 {count} 个手机,置信度: {avg_conf:.1%}")
else:
print(f" ✗ 失败: {result.get('error', '未知错误')}")
# 计算统计信息
total_time = time.time() - start_time
successful = sum(1 for r in results if r.get('success'))
total_phones = sum(r.get('detection_count', 0) for r in results if r.get('success'))
print("=" * 50)
print("批量检测完成!")
print(f"总耗时: {total_time:.1f}秒")
print(f"平均每张: {total_time/len(image_files):.2f}秒")
print(f"成功处理: {successful}/{len(image_files)} 张图片")
print(f"检测到手机总数: {total_phones} 个")
# 保存结果
if save_results and results:
self.save_results(results, output_format)
return results
# 使用示例
if __name__ == "__main__":
# 创建增强版检测器
detector = EnhancedPhoneDetector("http://localhost:7860")
# 执行批量检测并保存结果
results = detector.batch_detect_enhanced(
image_folder="./监控图片",
save_results=True,
output_format='csv',
show_progress=True
)
增强版本增加了几个实用功能:
- 重试机制:网络不稳定或服务短暂异常时自动重试
- 结果保存:支持CSV和JSON格式,方便后续分析
- 进度显示:实时显示处理进度和预估时间
- 详细统计:处理完成后显示详细的统计信息
3.3 高级版本:监控文件夹自动检测
如果你需要实时监控某个文件夹,有新图片就自动检测,可以试试这个版本。
import os
import time
import json
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class FolderMonitor:
"""监控文件夹变化,自动触发检测"""
def __init__(self, detector, watch_folder, processed_folder="./processed"):
self.detector = detector
self.watch_folder = Path(watch_folder)
self.processed_folder = Path(processed_folder)
self.processed_folder.mkdir(exist_ok=True)
# 记录已处理的文件
self.processed_files = set()
# 支持的图片格式
self.image_extensions = {'.jpg', '.jpeg', '.png', '.bmp', '.webp'}
def is_image_file(self, filepath):
"""检查是否为图片文件"""
return filepath.suffix.lower() in self.image_extensions
def process_image(self, image_path):
"""处理单张图片"""
print(f"\n检测到新图片: {image_path.name}")
# 执行检测
result = self.detector.detect_single_image(str(image_path))
if result.get('success'):
count = result.get('detection_count', 0)
if count > 0:
avg_conf = sum(result.get('confidence_scores', [])) / count
print(f" 检测到 {count} 个手机,平均置信度: {avg_conf:.1%}")
# 记录到日志文件
self.log_detection(image_path, count, avg_conf)
else:
print(" 未检测到手机")
else:
print(f" 检测失败: {result.get('error')}")
# 移动已处理的文件
self.move_to_processed(image_path)
def log_detection(self, image_path, count, avg_conf):
"""记录检测结果到日志"""
log_file = self.processed_folder / "detection_log.csv"
# 如果日志文件不存在,创建并写入表头
if not log_file.exists():
with open(log_file, 'w', encoding='utf-8') as f:
f.write("时间,图片文件,手机数量,平均置信度\n")
# 追加日志
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
with open(log_file, 'a', encoding='utf-8') as f:
f.write(f'{timestamp},{image_path.name},{count},{avg_conf:.2%}\n')
def move_to_processed(self, image_path):
"""移动已处理的文件到processed文件夹"""
try:
target_path = self.processed_folder / image_path.name
# 如果目标文件已存在,添加时间戳
if target_path.exists():
timestamp = datetime.now().strftime("%H%M%S")
new_name = f"{image_path.stem}_{timestamp}{image_path.suffix}"
target_path = self.processed_folder / new_name
image_path.rename(target_path)
print(f" 文件已移动到: {target_path}")
except Exception as e:
print(f" 移动文件失败: {e}")
def start_monitoring(self):
"""开始监控文件夹"""
print(f"开始监控文件夹: {self.watch_folder}")
print("支持的文件格式:", ', '.join(self.image_extensions))
print("按 Ctrl+C 停止监控")
print("-" * 50)
# 创建事件处理器
event_handler = ImageFileHandler(self)
# 创建观察者
observer = Observer()
observer.schedule(event_handler, str(self.watch_folder), recursive=False)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
print("\n监控已停止")
observer.join()
class ImageFileHandler(FileSystemEventHandler):
"""处理文件系统事件"""
def __init__(self, monitor):
self.monitor = monitor
def on_created(self, event):
"""当新文件创建时触发"""
if not event.is_directory:
filepath = Path(event.src_path)
if self.monitor.is_image_file(filepath):
# 等待文件完全写入
time.sleep(0.5)
self.monitor.process_image(filepath)
# 使用示例
if __name__ == "__main__":
# 创建检测器
detector = EnhancedPhoneDetector("http://localhost:7860")
# 创建文件夹监控器
monitor = FolderMonitor(
detector=detector,
watch_folder="./watch_folder", # 监控的文件夹
processed_folder="./processed_images" # 处理后的文件存放位置
)
# 开始监控
monitor.start_monitoring()
这个监控版本特别适合这些场景:
- 监控摄像头截图:摄像头定期截图保存到指定文件夹,脚本自动检测
- 批量导入处理:把需要检测的图片拖到监控文件夹,自动开始处理
- 实时监控系统:配合其他系统实现自动化流水线
4. 实际应用场景与优化建议
有了这些脚本,我们来看看在实际工作中怎么用,以及如何根据具体需求进行优化。
4.1 常见应用场景
场景一:考场监控图片批量分析
# 考场监控专用脚本
def exam_monitoring_analysis(image_folder, output_report="exam_report.csv"):
"""
专门用于考场监控图片分析的函数
特点:严格的时间记录、详细的统计报告
"""
detector = EnhancedPhoneDetector()
# 执行批量检测
results = detector.batch_detect_enhanced(image_folder)
# 生成考场专用报告
exam_stats = {
'total_images': len(results),
'images_with_phones': 0,
'total_phones_detected': 0,
'high_confidence_detections': 0, # 置信度>90%
'suspicious_images': [] # 检测到手机的图片列表
}
for result in results:
if result.get('success') and result.get('detection_count', 0) > 0:
exam_stats['images_with_phones'] += 1
exam_stats['total_phones_detected'] += result['detection_count']
# 检查是否有高置信度检测
if any(conf > 0.9 for conf in result.get('confidence_scores', [])):
exam_stats['high_confidence_detections'] += 1
exam_stats['suspicious_images'].append({
'image': result['image_path'],
'phone_count': result['detection_count'],
'max_confidence': max(result['confidence_scores'])
})
# 保存详细报告
save_exam_report(exam_stats, output_report)
return exam_stats
场景二:社交媒体内容审核
def social_media_content_check(image_urls, batch_size=10):
"""
社交媒体图片内容审核
特点:支持URL直接检测、批量处理、快速响应
"""
detector = EnhancedPhoneDetector()
all_results = []
# 分批处理,避免一次性请求太多
for i in range(0, len(image_urls), batch_size):
batch_urls = image_urls[i:i+batch_size]
print(f"处理批次 {i//batch_size + 1}/{(len(image_urls)-1)//batch_size + 1}")
batch_results = []
for url in batch_urls:
# 下载图片并检测
result = download_and_detect(url, detector)
batch_results.append(result)
all_results.extend(batch_results)
# 如果发现违规内容,立即处理
for result in batch_results:
if result.get('detection_count', 0) > 0:
handle_violation_content(result)
# 批次间延迟
time.sleep(1)
return all_results
4.2 性能优化建议
如果你的图片数量特别大,可以考虑这些优化:
1. 多线程处理
from concurrent.futures import ThreadPoolExecutor, as_completed
def batch_detect_parallel(image_folder, max_workers=4):
"""使用多线程加速批量检测"""
detector = EnhancedPhoneDetector()
image_files = get_image_files(image_folder)
results = []
with ThreadPoolExecutor(max_workers=max_workers) as executor:
# 提交所有任务
future_to_image = {
executor.submit(detector.detect_single_image, str(img_path)): img_path
for img_path in image_files
}
# 收集结果
for future in as_completed(future_to_image):
image_path = future_to_image[future]
try:
result = future.result()
results.append(result)
print(f"完成: {image_path.name}")
except Exception as e:
print(f"处理失败 {image_path.name}: {e}")
return results
2. 请求批量化
def batch_detect_api(images):
"""
如果服务端支持批量API,可以一次性发送多张图片
注意:当前版本可能不支持,但可以作为一个优化方向
"""
# 将多张图片打包成一个请求
files = []
for i, image_data in enumerate(images):
files.append(('images', (f'image_{i}.jpg', image_data, 'image/jpeg')))
response = requests.post(f"{self.server_url}/api/batch_predict", files=files)
return response.json()
3. 结果缓存
import hashlib
from functools import lru_cache
class CachedPhoneDetector(EnhancedPhoneDetector):
"""带缓存功能的检测器,避免重复检测相同图片"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.cache = {}
def get_image_hash(self, image_path):
"""计算图片的哈希值,用于缓存键"""
with open(image_path, 'rb') as f:
return hashlib.md5(f.read()).hexdigest()
@lru_cache(maxsize=100)
def detect_cached(self, image_path):
"""带缓存的检测函数"""
image_hash = self.get_image_hash(image_path)
if image_hash in self.cache:
print(f"使用缓存结果: {image_path}")
return self.cache[image_hash]
# 执行检测并缓存结果
result = super().detect_single_image(image_path)
self.cache[image_hash] = result
return result
4.3 错误处理与日志记录
在实际使用中,完善的错误处理和日志记录很重要:
import logging
from logging.handlers import RotatingFileHandler
def setup_logging():
"""配置日志系统"""
logger = logging.getLogger('PhoneDetector')
logger.setLevel(logging.INFO)
# 文件处理器(自动轮转)
file_handler = RotatingFileHandler(
'phone_detection.log',
maxBytes=10*1024*1024, # 10MB
backupCount=5
)
file_handler.setLevel(logging.INFO)
# 控制台处理器
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
# 格式
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)
logger.addHandler(file_handler)
logger.addHandler(console_handler)
return logger
# 在检测函数中使用日志
def detect_with_logging(image_path):
logger = logging.getLogger('PhoneDetector')
try:
logger.info(f"开始检测: {image_path}")
result = detector.detect_single_image(image_path)
if result['success']:
logger.info(f"检测成功: {image_path}, 找到{result['detection_count']}个手机")
else:
logger.error(f"检测失败: {image_path}, 错误: {result.get('error')}")
return result
except Exception as e:
logger.exception(f"检测异常: {image_path}, 异常信息: {str(e)}")
raise
5. 总结与下一步建议
5.1 方案总结
通过今天的分享,我们实现了一个完整的DAMO-YOLO图片循环检测自动化方案。这个方案的核心价值在于:
- 解决了批量处理的需求:虽然Web界面不支持批量上传,但通过脚本我们可以轻松处理成百上千张图片
- 提高了工作效率:自动化检测比手动操作快得多,特别适合大量图片的处理场景
- 灵活可扩展:脚本可以根据具体需求进行定制,比如添加监控功能、结果分析、自动报告等
- 易于集成:可以轻松集成到现有的工作流程或系统中
5.2 实际使用建议
根据我的使用经验,给你几个实用建议:
对于初学者:
- 先从基础版本开始,理解基本的API调用流程
- 用小批量图片(10-20张)测试,确保脚本正常工作
- 仔细查看日志,了解每张图片的处理结果
对于常规使用:
- 使用增强版本,它有更好的错误处理和结果保存功能
- 定期清理结果文件夹,避免占用太多磁盘空间
- 考虑使用监控版本,实现真正的自动化流水线
对于大规模使用:
- 考虑性能优化,比如多线程处理
- 实现结果缓存,避免重复检测相同图片
- 建立完善的日志和监控系统
5.3 可能的改进方向
如果你对这个方案还有更多需求,可以考虑这些改进:
- Web界面集成:把批量检测功能做成一个简单的Web界面,让非技术人员也能用
- 分布式处理:如果需要处理海量图片,可以考虑分布式部署
- 实时报警:检测到手机时,自动发送邮件或消息通知
- 结果可视化:生成检测结果的可视化报告,用图表展示统计信息
- 模型优化:如果对准确率有更高要求,可以考虑对DAMO-YOLO模型进行微调
5.4 开始你的自动化之旅
现在你已经掌握了DAMO-YOLO图片循环检测的自动化方案。无论你是要处理考场监控图片、分析社交媒体内容,还是管理自己的图片库,这个方案都能帮你节省大量时间。
记住,技术是为了解决问题而存在的。当现有工具不能满足需求时,不要等待功能更新,而是主动寻找解决方案。写脚本实现自动化,就是技术人员最直接的解决问题方式。
从今天开始,告别手动上传,拥抱自动化检测吧!
获取更多AI镜像
想探索更多AI镜像和应用场景?访问 CSDN星图镜像广场,提供丰富的预置镜像,覆盖大模型推理、图像生成、视频生成、模型微调等多个领域,支持一键部署。
更多推荐

所有评论(0)