树莓派RGB LED远程实时参数调整方案咨询

方案1：异步HTTP服务器 + 全局状态（推荐快速实现） #

之前用HTTP服务器卡死是因为同步框架（如Flask）的请求处理会阻塞LED的主循环。改用异步框架（比如FastAPI），让HTTP请求处理和LED闪烁循环在不同协程中运行，互不干扰。

步骤：

1. 安装依赖：pip install fastapi uvicorn
2. 定义全局变量存储当前的颜色和频率参数
3. 启动异步HTTP服务器处理参数修改请求
4. 单独的协程负责LED闪烁逻辑，实时读取全局参数

代码示例：

import asyncio
from fastapi import FastAPI
import RPi.GPIO as GPIO

# RGB LED引脚配置（根据实际接线修改）
RED_PIN = 17
GREEN_PIN = 22
BLUE_PIN = 24

# 全局状态参数
current_color = (255, 0, 0)  # 默认红色
current_freq = 1.0  # 默认1秒闪烁一次

app = FastAPI()

# 初始化GPIO
def setup_gpio():
    GPIO.setmode(GPIO.BCM)
    GPIO.setup(RED_PIN, GPIO.OUT)
    GPIO.setup(GREEN_PIN, GPIO.OUT)
    GPIO.setup(BLUE_PIN, GPIO.OUT)
    # 设置PWM控制亮度（RGB颜色通过PWM实现）
    global red_pwm, green_pwm, blue_pwm
    red_pwm = GPIO.PWM(RED_PIN, 1000)
    green_pwm = GPIO.PWM(GREEN_PIN, 1000)
    blue_pwm = GPIO.PWM(BLUE_PIN, 1000)
    red_pwm.start(0)
    green_pwm.start(0)
    blue_pwm.start(0)

def set_color(r, g, b):
    # 将0-255转换为0-100的占空比
    red_pwm.ChangeDutyCycle(r / 2.55)
    green_pwm.ChangeDutyCycle(g / 2.55)
    blue_pwm.ChangeDutyCycle(b / 2.55)

async def led_blink_loop():
    while True:
        # 亮灯
        set_color(*current_color)
        await asyncio.sleep(current_freq / 2)
        # 熄灯
        set_color(0, 0, 0)
        await asyncio.sleep(current_freq / 2)

@app.post("/update_params")
async def update_params(r: int, g: int, b: int, freq: float):
    global current_color, current_freq
    # 参数校验
    if 0 <= r <=255 and 0 <= g <=255 and 0 <= b <=255 and freq > 0:
        current_color = (r, g, b)
        current_freq = freq
        return {"status": "success", "current_params": {"color": current_color, "freq": current_freq}}
    else:
        return {"status": "error", "message": "参数无效：颜色值需在0-255之间，频率需大于0"}

if __name__ == "__main__":
    setup_gpio()
    try:
        # 启动LED循环协程
        asyncio.create_task(led_blink_loop())
        # 启动FastAPI服务器
        import uvicorn
        uvicorn.run(app, host="0.0.0.0", port=8000)
    except KeyboardInterrupt:
        # 清理GPIO资源
        red_pwm.stop()
        green_pwm.stop()
        blue_pwm.stop()
        GPIO.cleanup()

使用方式：移动端通过POST请求http://<你的RPI4 IP>:8000/update_params，携带参数r、g、b、freq，比如http://192.168.1.100:8000/update_params?r=0&g=255&b=0&freq=0.5即可切换为绿色、0.5秒闪烁一次。

方案2：WebSocket长连接（适合实时连续控制） #

如果需要移动端频繁发送指令（比如滑动调色），WebSocket长连接比多次HTTP请求更高效，避免重复建立连接的开销。

步骤：

1. 安装依赖：pip install fastapi uvicorn websockets
2. 建立WebSocket连接，实时接收移动端的参数指令
3. LED循环逻辑实时读取最新参数

代码示例：

import asyncio
from fastapi import FastAPI, WebSocket
import RPi.GPIO as GPIO

RED_PIN = 17
GREEN_PIN = 22
BLUE_PIN = 24

current_color = (255, 0, 0)
current_freq = 1.0

app = FastAPI()

def setup_gpio():
    GPIO.setmode(GPIO.BCM)
    GPIO.setup(RED_PIN, GPIO.OUT)
    GPIO.setup(GREEN_PIN, GPIO.OUT)
    GPIO.setup(BLUE_PIN, GPIO.OUT)
    global red_pwm, green_pwm, blue_pwm
    red_pwm = GPIO.PWM(RED_PIN, 1000)
    green_pwm = GPIO.PWM(GREEN_PIN, 1000)
    blue_pwm = GPIO.PWM(BLUE_PIN, 1000)
    red_pwm.start(0)
    green_pwm.start(0)
    blue_pwm.start(0)

def set_color(r, g, b):
    red_pwm.ChangeDutyCycle(r / 2.55)
    green_pwm.ChangeDutyCycle(g / 2.55)
    blue_pwm.ChangeDutyCycle(b / 2.55)

async def led_blink_loop():
    while True:
        set_color(*current_color)
        await asyncio.sleep(current_freq / 2)
        set_color(0, 0, 0)
        await asyncio.sleep(current_freq / 2)

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    global current_color, current_freq
    try:
        while True:
            # 接收移动端发送的JSON数据，格式示例：{"r":0,"g":255,"b":0,"freq":0.5}
            data = await websocket.receive_json()
            r = data.get("r", current_color[0])
            g = data.get("g", current_color[1])
            b = data.get("b", current_color[2])
            freq = data.get("freq", current_freq)
            if 0 <= r <=255 and 0 <= g <=255 and 0 <= b <=255 and freq > 0:
                current_color = (r, g, b)
                current_freq = freq
                await websocket.send_json({"status": "已更新", "params": {"color": current_color, "freq": current_freq}})
            else:
                await websocket.send_json({"status": "错误", "message": "参数无效"})
    except Exception as e:
        print(f"WebSocket连接断开: {e}")

if __name__ == "__main__":
    setup_gpio()
    try:
        asyncio.create_task(led_blink_loop())
        import uvicorn
        uvicorn.run(app, host="0.0.0.0", port=8000)
    except KeyboardInterrupt:
        red_pwm.stop()
        green_pwm.stop()
        blue_pwm.stop()
        GPIO.cleanup()

移动端用WebSocket客户端连接ws://<RPI4 IP>:8000/ws，随时发送参数指令即可实时生效。

方案3：MQTT客户端（物联网场景推荐） #

如果移动端已接入物联网生态（如搭配HomeAssistant），RPI作为MQTT客户端订阅特定主题，移动端发布消息到该主题修改参数。

步骤：

1. 安装依赖：pip install paho-mqtt
2. 配置MQTT服务器地址（本地搭建或公共服务器均可）
3. RPI订阅主题，接收参数消息
4. LED循环读取最新参数

代码示例：

import time
import json
import paho.mqtt.client as mqtt
import RPi.GPIO as GPIO

RED_PIN = 17
GREEN_PIN = 22
BLUE_PIN = 24

current_color = (255, 0, 0)
current_freq = 1.0

# MQTT配置（根据实际服务器修改）
MQTT_BROKER = "192.168.1.101"  # MQTT服务器IP
MQTT_PORT = 1883
MQTT_TOPIC = "rpi/led/params"

def setup_gpio():
    GPIO.setmode(GPIO.BCM)
    GPIO.setup(RED_PIN, GPIO.OUT)
    GPIO.setup(GREEN_PIN, GPIO.OUT)
    GPIO.setup(BLUE_PIN, GPIO.OUT)
    global red_pwm, green_pwm, blue_pwm
    red_pwm = GPIO.PWM(RED_PIN, 1000)
    green_pwm = GPIO.PWM(GREEN_PIN, 1000)
    blue_pwm = GPIO.PWM(BLUE_PIN, 1000)
    red_pwm.start(0)
    green_pwm.start(0)
    blue_pwm.start(0)

def set_color(r, g, b):
    red_pwm.ChangeDutyCycle(r / 2.55)
    green_pwm.ChangeDutyCycle(g / 2.55)
    blue_pwm.ChangeDutyCycle(b / 2.55)

# MQTT消息回调函数
def on_message(client, userdata, msg):
    global current_color, current_freq
    try:
        data = json.loads(msg.payload.decode())
        r = data.get("r", current_color[0])
        g = data.get("g", current_color[1])
        b = data.get("b", current_color[2])
        freq = data.get("freq", current_freq)
        if 0 <= r <=255 and 0 <= g <=255 and 0 <= b <=255 and freq > 0:
            current_color = (r, g, b)
            current_freq = freq
            print(f"已更新参数: 颜色={current_color}, 频率={current_freq}")
    except Exception as e:
        print(f"无效消息: {e}")

def main():
    setup_gpio()
    # 初始化MQTT客户端
    client = mqtt.Client()
    client.on_message = on_message
    client.connect(MQTT_BROKER, MQTT_PORT, 60)
    client.subscribe(MQTT_TOPIC)
    # 启动MQTT非阻塞网络循环
    client.loop_start()
    
    try:
        while True:
            set_color(*current_color)
            time.sleep(current_freq / 2)
            set_color(0, 0, 0)
            time.sleep(current_freq / 2)
    except KeyboardInterrupt:
        client.loop_stop()
        client.disconnect()
        red_pwm.stop()
        green_pwm.stop()
        blue_pwm.stop()
        GPIO.cleanup()

if __name__ == "__main__":
    main()

移动端用MQTT客户端发布JSON格式消息到rpi/led/params主题，即可实时修改LED参数。

方案选择建议 #

• 快速实现选异步HTTP服务器，依赖简单，移动端只需普通HTTP请求；
• 频繁实时控制选WebSocket，低延迟、连接开销小；
• 物联网生态下选MQTT，适配各类智能家居平台。

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