树莓派4遥控小车Flask视频流服务器崩溃问题求助

用RPI4搭建了一台遥控小车，通过Flask服务器将摄像头视频流传输至PC，同时PC可对小车进行控制。但当小车快速移动时，视频流偶尔会冻结，重新连接视频流时会出现“Internal server error. this might be caused because of an internal error or the server is overloaded”错误。已尝试降低小车移动速度和视频流质量，但问题仍未解决。

树莓派端服务端脚本 #

from gpiozero import Motor, Servo, Device
from gpiozero.pins.pigpio import PiGPIOFactory
import socket
from flask import Flask, Response
import cv2
from imutils.video import VideoStream
import threading
from threading import Lock

# Flask app for streaming video
app = Flask(__name__)

# Setup for GPIO
Device.pin_factory = PiGPIOFactory()

# Initialize the motors and servo
motor_left = Motor(forward=6, backward=13)
motor_right = Motor(forward=19, backward=26)
servo = Servo(18)

# 线程锁，保护GPIO操作
gpio_lock = Lock()

# Video stream initialization
vs = VideoStream(src=0).start()

def generate():
    """Video streaming generator function with optimizations."""
    while True:
        try:
            frame = vs.read()
            if frame is None:
                # 空帧时尝试重启视频流
                vs.stop()
                vs = VideoStream(src=0).start()
                continue
            # Resize the frame to reduce size
            frame = cv2.resize(frame, (540, 380))  # Example resize, adjust as needed
            
            # Convert the frame to grayscale to reduce size further (optional)
            # frame = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
            
            # Adjust JPEG quality to reduce size
            ret, jpeg = cv2.imencode('.jpg', frame, [int(cv2.IMWRITE_JPEG_QUALITY), 50])
            
            if not ret:
                continue
            
            yield (b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' + jpeg.tobytes() + b'\r\n')
        except Exception as e:
            print(f"Video stream error: {e}")
            # 异常时短暂休眠，避免占用过多CPU
            cv2.waitKey(100)
            continue

@app.route('/video_feed')
def video_feed():
    """Video streaming route."""
    return Response(generate(), mimetype='multipart/x-mixed-replace; boundary=frame')

# Thread to run the Flask app
def run_flask_app():
    app.run(host='0.0.0.0', port=5000, debug=False, threaded=True, use_reloader=False)

flask_thread = threading.Thread(target=run_flask_app)
flask_thread.start()

# Motor control functions
def forward():
    with gpio_lock:
        motor_left.forward()
        motor_right.forward()

def backward():
    with gpio_lock:
        motor_left.backward()
        motor_right.backward()

def right():
    with gpio_lock:
        motor_left.backward()
        motor_right.forward()

def left():
    with gpio_lock:
        motor_left.forward()
        motor_right.backward()

def stop():
    with gpio_lock:
        motor_left.stop()
        motor_right.stop()

# Updated servo control functions
def servo_move_left():
    with gpio_lock:
        servo.max()

def servo_move_right():
    with gpio_lock:
        servo.min()

def servo_stop():
    with gpio_lock:
        # Updated to explicitly set servo to neutral position
        servo.value = 0  # Neutral position

# Command mapping with added 'servo_neutral' command
commands = {
    "forward": forward,
    "backward": backward,
    "left": left,
    "right": right,
    "stop": stop,
    "servo_move_left": servo_move_left,
    "servo_move_right": servo_move_right,
    "servo_neutral": servo_stop,  # Use "servo_neutral" to center the servo
}

def execute_command(command):
    action = commands.get(command)
    if action:
        action()
    else:
        print(f"Unknown command: {command}")

def run_socket_server():
    host = '0.0.0.0'
    port = 12345

    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind((host, port))
        s.listen()
        print("Server listening on port", port)

        while True:
            conn, addr = s.accept()
            with conn:
                print(f"Connected by {addr}")
                while True:
                    try:
                        data = conn.recv(1024)
                        if not data:
                            break
                        command = data.decode('utf-8').strip().lower()
                        execute_command(command)
                    except Exception as e:
                        print(f"Socket command error: {e}")
                        break

# 启动Socket服务线程
socket_thread = threading.Thread(target=run_socket_server)
socket_thread.start()

if __name__ == "__main__":
    flask_thread.join()
    socket_thread.join()

PC端控制代码 #

import socket
import tkinter as tk
import webbrowser

# 保持长连接
socket_conn = None

def connect_to_pi():
    """建立与树莓派的长连接"""
    global socket_conn
    try:
        if socket_conn is None or socket_conn.fileno() == -1:
            socket_conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            socket_conn.connect(("192.168.1.44", 12345))
            print("Connected to Raspberry Pi")
    except Exception as e:
        print(f"Connection error: {e}")
        socket_conn = None

def send_command(command):
    """通过长连接向树莓派发送控制命令"""
    global socket_conn
    if socket_conn is None:
        connect_to_pi()
    if socket_conn:
        try:
            socket_conn.sendall(command.encode())
        except Exception as e:
            print(f"Error sending command: {e}")
            socket_conn = None

def open_video_stream():
    """在默认浏览器打开视频流"""
    stream_url = "http://192.168.1.44:5000/video_feed"  # 替换为你的树莓派IP
    webbrowser.open(stream_url)

def on_key_press(event):
    """按键映射为控制命令"""
    commands = {
        'w': 'forward',
        's': 'backward',
        'a': 'left',
        'd': 'right',
        'Left': 'servo_move_left',  # 方向键控制舵机
        'Right': 'servo_move_right',
        'Up': 'servo_neutral',  # 上方向键归位舵机
    }
    command = commands.get(event.keysym) or commands.get(event.char)  # 兼容特殊按键与普通字符
    if command:
        send_command(command)

def on_key_release(event):
    """放开按键时停止小车动作"""
    control_keys = ['w', 's', 'a', 'd', 'Left', 'Right']
    if event.keysym in control_keys:
        send_command('stop')

root = tk.Tk()
root.title("树莓派小车与舵机控制")

# 程序启动时自动连接树莓派
root.after(100, connect_to_pi)

# 打开视频流按钮
stream_button = tk.Button(root, text="打开视频流", command=open_video_stream)
stream_button.pack(pady=20)

# 绑定按键事件
root.bind("<KeyPress>", on_key_press)
root.bind("<KeyRelease>", on_key_release)

# 使用说明
instructions = tk.Label(root, text="WASD控制小车移动，方向键控制舵机，上方向键归位舵机\n放开控制键自动停止小车动作\n点击'打开视频流'查看摄像头画面")
instructions.pack(pady=20)

root.mainloop()

# 关闭连接
if socket_conn:
    socket_conn.close()

关键优化点说明 #

• 线程分离：将Flask视频流服务、Socket控制服务分别放到独立线程，避免主线程阻塞导致资源抢占。
• GPIO线程安全：添加线程锁保护GPIO操作，防止多线程同时控制硬件引发冲突。
• 视频流异常处理：在生成器中捕获空帧和异常，自动重启视频流，避免生成器中断引发500错误。
• 长连接优化：PC端改用长连接发送控制命令，减少频繁创建Socket连接的开销，降低树莓派负载。

内容的提问来源于stack exchange，提问作者Adidev Narayan