AutoDL实例上的YOLOv8 Flask服务器实现

存档预览：# AutoDL实例上的YOLOv8 Flask服务器实现 下面是一个完整的Flask服务器实现，可以在AutoDL实例上运行，接收视频帧并使用YOLOv8进行检测后返回结果。 #…（全文以下方原始文本完整保留）

# AutoDL实例上的YOLOv8 Flask服务器实现

下面是一个完整的Flask服务器实现，可以在AutoDL实例上运行，接收视频帧并使用YOLOv8进行检测后返回结果。

## 完整服务器代码

```python
# yolo_server.py
from flask import Flask, request, Response, jsonify
import cv2
import numpy as np
from ultralytics import YOLO
import time
import threading

app = Flask(__name__)

# 加载YOLOv8模型（可以根据需要更换模型大小）
model = YOLO('yolov8n.pt')  # 会自动下载模型如果不存在

# 用于统计性能
request_count = 0
start_time = time.time()
lock = threading.Lock()

@app.route('/detect', methods=['POST'])
def detect():
    global request_count
    
    # 接收图像数据
    if 'image' not in request.files:
        return jsonify({'error': 'No image provided'}), 400
    
    try:
        # 读取并解码图像
        img_bytes = request.files['image'].read()
        img = cv2.imdecode(np.frombuffer(img_bytes, np.uint8), cv2.IMREAD_COLOR)
        
        if img is None:
            return jsonify({'error': 'Failed to decode image'}), 400
        
        # 运行YOLOv8检测
        results = model(img)
        
        # 获取检测结果
        boxes = results[0].boxes.xyxy.tolist()  # 边界框坐标
        classes = results[0].boxes.cls.tolist()  # 类别ID
        confidences = results[0].boxes.conf.tolist()  # 置信度
        
        # 获取类别名称
        class_names = results[0].names
        detected_objects = []
        for box, cls, conf in zip(boxes, classes, confidences):
            detected_objects.append({
                'class': class_names[int(cls)],
                'confidence': float(conf),
                'bbox': [float(x) for x in box]
            })
        
        # 更新请求计数
        with lock:
            request_count += 1
        
        # 返回JSON结果
        return jsonify({
            'objects': detected_objects,
            'image_size': {'width': img.shape[1], 'height': img.shape[0]}
        })
        
    except Exception as e:
        return jsonify({'error': str(e)}), 500

@app.route('/detect_image', methods=['POST'])
def detect_image():
    """返回带标注的图像而不是JSON"""
    if 'image' not in request.files:
        return jsonify({'error': 'No image provided'}), 400
    
    try:
        img_bytes = request.files['image'].read()
        img = cv2.imdecode(np.frombuffer(img_bytes, np.uint8), cv2.IMREAD_COLOR)
        
        if img is None:
            return jsonify({'error': 'Failed to decode image'}), 400
        
        # 运行检测并获取带标注的图像
        results = model(img)
        annotated_img = results[0].plot()  # 自动绘制检测结果
        
        # 编码图像为JPEG
        _, img_encoded = cv2.imencode('.jpg', annotated_img)
        
        return Response(img_encoded.tobytes(), mimetype='image/jpeg')
    
    except Exception as e:
        return jsonify({'error': str(e)}), 500

@app.route('/stats')
def stats():
    """获取服务器统计信息"""
    with lock:
        elapsed = time.time() - start_time
        return jsonify({
            'request_count': request_count,
            'uptime': elapsed,
            'requests_per_second': request_count / elapsed if elapsed > 0 else 0
        })

if __name__ == '__main__':
    # 在AutoDL上运行时，使用0.0.0.0监听所有网络接口
    # 端口号可以根据需要修改，但要确保AutoDL实例的安全组允许该端口
    app.run(host='0.0.0.0', port=5000, threaded=True)
```

## 服务器部署步骤

1. **在AutoDL实例上准备环境**：
   ```bash
   # 创建conda环境（可选）
   conda create -n yolo_flask python=3.8
   conda activate yolo_flask
   
   # 安装依赖
   pip install flask ultralytics opencv-python numpy
   ```

2. **上传代码**：
   - 将上面的代码保存为`yolo_server.py`
   - 上传到AutoDL实例

3. **运行服务器**：
   ```bash
   python yolo_server.py
   ```

4. **设置AutoDL端口映射**：
   - 在AutoDL控制台找到你的实例
   - 点击"自定义服务"，添加映射（例如将5000端口映射到外部）

## 客户端调用示例

### Python客户端调用示例

```python
# client.py
import cv2
import requests
import time

SERVER_URL = "http://你的AutoDL实例地址:映射端口/detect_image"  # 或/detect获取JSON结果

cap = cv2.VideoCapture(0)  # 本地摄像头

while True:
    ret, frame = cap.read()
    if not ret:
        break
    
    # 编码图像为JPEG
    _, img_encoded = cv2.imencode('.jpg', frame)
    
    # 发送到服务器
    start_time = time.time()
    response = requests.post(SERVER_URL, 
                           files={'image': img_encoded.tobytes()},
                           timeout=5)
    
    # 处理响应
    if response.status_code == 200:
        if SERVER_URL.endswith('/detect_image'):
            # 显示带标注的图像
            img = cv2.imdecode(np.frombuffer(response.content, np.uint8), cv2.IMREAD_COLOR)
            cv2.imshow('YOLOv8 Detection', img)
        else:
            # 处理JSON结果
            result = response.json()
            print("检测到对象:", result['objects'])
    else:
        print("请求失败:", response.text)
    
    # 显示FPS
    fps = 1 / (time.time() - start_time)
    print(f"FPS: {fps:.2f}")
    
    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

cap.release()
cv2.destroyAllWindows()
```

## 性能优化建议

1. **模型选择**：
   - 使用较小的模型(yolov8n.pt)获得更快速度
   - 如需更高精度，可使用yolov8s.pt或yolov8m.pt

2. **图像尺寸**：
   - 客户端可以调整图像尺寸后再发送，减小传输量
   ```python
   frame = cv2.resize(frame, (640, 480))  # 调整到较小尺寸
   ```

3. **多线程处理**：
   - Flask已启用多线程(threaded=True)
   - 对于高负载，可以考虑使用Gunicorn部署

4. **视频流优化**：
   - 考虑使用WebSocket代替HTTP请求减少延迟
   - 或者使用RTMP等视频流协议

这个实现提供了两种返回格式：JSON格式的检测结果和带标注的图像，你可以根据需求选择使用哪种方式。