如何让Python目标检测脚本持续运行,无需重复加载libraries与checkpoints?
要解决每次调用都重新加载库和checkpoint的问题,核心思路是让脚本启动时一次性完成资源加载,之后常驻运行,通过接口、队列或长连接等方式接收Android传来的新数据,复用已加载的模型处理任务。以下是几种可行的实现方式:
方案一:搭建轻量Web服务(推荐用FastAPI)
这是适配移动端调用最便捷的方案,脚本启动后常驻内存,通过HTTP接口接收图片请求、执行检测并返回结果。
- 安装依赖(按需补充你的检测库,比如torch、yolov5等):
pip install fastapi uvicorn pillow opencv-python
- 编写服务代码:
from fastapi import FastAPI, UploadFile, File import cv2 import numpy as np # 替换成你的检测库导入语句 from ultralytics import YOLO # 启动阶段一次性加载模型和依赖 app = FastAPI() model = YOLO("yolov8n.pt") # 替换为你的checkpoint路径 @app.post("/detect") async def detect_object(file: UploadFile = File(...)): # 解析上传的图片 contents = await file.read() nparr = np.frombuffer(contents, np.uint8) img = cv2.imdecode(nparr, cv2.IMREAD_COLOR) # 复用已加载的模型执行检测 results = model(img) # 格式化检测结果(按需调整输出字段) result_list = [] for result in results: for box in result.boxes: x1, y1, x2, y2 = box.xyxy[0].tolist() class_name = model.names[int(box.cls[0].item())] result_list.append({ "bbox": [round(x1,2), round(y1,2), round(x2,2), round(y2,2)], "class": class_name, "confidence": round(box.conf[0].item(), 2) }) return {"detections": result_list} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)
- 启动服务:
python your_script.py,之后Android端通过POST请求上传图片到http://服务器IP:8000/detect,即可直接获取检测结果,无需重复加载资源。
方案二:消息队列实现异步处理
如果Android端不需要实时获取结果,可通过消息队列解耦任务提交与结果获取,脚本启动后持续监听队列任务。
- 安装依赖:
pip install redis rq
- 编写任务处理脚本(worker.py):
import redis from rq import Worker, Queue, Connection import cv2 import numpy as np import base64 from ultralytics import YOLO # 启动时加载模型 model = YOLO("yolov8n.pt") # 配置Redis连接 redis_conn = redis.Redis(host="localhost", port=6379, db=0) def process_detection(image_base64): # 解码Base64格式的图片 img_data = base64.b64decode(image_base64) nparr = np.frombuffer(img_data, np.uint8) img = cv2.imdecode(nparr, cv2.IMREAD_COLOR) # 执行检测并格式化结果 results = model(img) result_list = [] for result in results: for box in result.boxes: x1, y1, x2, y2 = box.xyxy[0].tolist() class_name = model.names[int(box.cls[0].item())] result_list.append({ "bbox": [round(x1,2), round(y1,2), round(x2,2), round(y2,2)], "class": class_name, "confidence": round(box.conf[0].item(), 2) }) return result_list if __name__ == "__main__": with Connection(redis_conn): task_queue = Queue("detection_tasks") worker = Worker([task_queue]) worker.work()
- 启动worker:
python worker.py,Android端将图片转为Base64后,通过Redis客户端把任务推入队列,worker会自动取出处理,结果可存入Redis或数据库,Android端再查询获取。
方案三:长连接Socket通信
如果需要极低延迟的交互,可采用TCP长连接,脚本作为服务端,Android客户端连接后保持会话,反复发送数据。
示例服务端代码:
import socket import json import base64 import cv2 import numpy as np from ultralytics import YOLO # 启动时加载模型 model = YOLO("yolov8n.pt") # 配置Socket服务 HOST = '0.0.0.0' PORT = 65432 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind((HOST, PORT)) s.listen() print("等待Android客户端连接...") conn, addr = s.accept() with conn: print(f"已建立连接:{addr}") while True: # 接收客户端数据(调整缓冲区大小适配图片尺寸) data = conn.recv(1024*1024) if not data: break # 解析JSON格式的请求 request = json.loads(data.decode('utf-8')) img_base64 = request.get('image') # 处理图片并执行检测 img_data = base64.b64decode(img_base64) nparr = np.frombuffer(img_data, np.uint8) img = cv2.imdecode(nparr, cv2.IMREAD_COLOR) results = model(img) # 格式化结果并返回 result_list = [] for result in results: for box in result.boxes: x1, y1, x2, y2 = box.xyxy[0].tolist() class_name = model.names[int(box.cls[0].item())] result_list.append({ "bbox": [round(x1,2), round(y1,2), round(x2,2), round(y2,2)], "class": class_name, "confidence": round(box.conf[0].item(), 2) }) response = json.dumps({"detections": result_list}).encode('utf-8') conn.sendall(response)
Android端只需建立一次Socket连接,之后每次发送图片数据即可,无需重新触发模型加载。
内容的提问来源于stack exchange,提问作者Leo
相关产品推荐
相关产品推荐

