Android应用与树莓派Python脚本交互及GPIO控制技术咨询

1. 本地网络低延迟控制：基于Flask的REST API实现 #

用Flask搭建轻量HTTP服务，Android通过本地IP访问API接口控制GPIO，延迟低且实现简单。

步骤1：安装Flask #

在树莓派上执行：

pip install flask

步骤2：修改现有Python代码，集成Flask与线程处理 #

Tkinter的mainloop为阻塞逻辑，需将Flask服务放在单独线程运行，避免UI卡死：

import RPi.GPIO as GPIO
import tkinter as tk
from flask import Flask, jsonify, request
import threading
from threading import Lock

# 全局状态变量
l1 = 1
l2 = 1
l3 = 1
lm = 1
# GPIO操作锁，避免多线程冲突
gpio_lock = Lock()

# 初始化GPIO
GPIO.setmode(GPIO.BCM)
GPIO.setup(20, GPIO.OUT)
GPIO.setup(26, GPIO.OUT)
GPIO.setup(21, GPIO.OUT)

# Flask应用初始化
app = Flask(__name__)

# 灯光控制函数（保持原有逻辑，加锁）
def light_1():
    global l1
    with gpio_lock:
        if l1 % 2 == 1:
            GPIO.output(26, True)
            l1 = 0
        else:
            GPIO.output(26, False)
            l1 = 1
    # 同步UI状态
    top.after(0, lambda: but1.config(text=f"Light 1: {'ON' if l1==0 else 'OFF'}"))

def light_2():
    global l2
    with gpio_lock:
        if l2 % 2 == 1:
            GPIO.output(21, True)
            l2 = 0
        else:
            GPIO.output(21, False)
            l2 = 1
    top.after(0, lambda: but3.config(text=f"Light 3: {'ON' if l2==0 else 'OFF'}"))

def light_3():
    global l3
    with gpio_lock:
        if l3 % 2 == 1:
            GPIO.output(20, True)
            l3 = 0
        else:
            GPIO.output(20, False)
            l3 = 1
    top.after(0, lambda: but2.config(text=f"Light 2: {'ON' if l3==0 else 'OFF'}"))

def master_switch():
    global lm
    with gpio_lock:
        if lm % 2 == 1:
            GPIO.output(20, True)
            GPIO.output(26, True)
            GPIO.output(21, True)
            lm = 0
        else:
            GPIO.output(20, False)
            GPIO.output(21, False)
            GPIO.output(26, False)
            lm = 1
    # 同步所有按钮状态
    top.after(0, lambda: but1.config(text=f"Light 1: {'ON' if l1==0 else 'OFF'}"))
    top.after(0, lambda: but2.config(text=f"Light 2: {'ON' if l3==0 else 'OFF'}"))
    top.after(0, lambda: but3.config(text=f"Light 3: {'ON' if l2==0 else 'OFF'}"))
    top.after(0, lambda: but4.config(text=f"Master: {'ON' if lm==0 else 'OFF'}"))

# REST API接口
@app.route('/control/<device>', methods=['POST'])
def control_device(device):
    if device == 'light1':
        light_1()
    elif device == 'light2':
        light_2()
    elif device == 'light3':
        light_3()
    elif device == 'master':
        master_switch()
    else:
        return jsonify({"status": "error", "message": "Invalid device"}), 400
    # 返回当前状态
    return jsonify({
        "status": "success",
        "l1": l1,
        "l2": l2,
        "l3": l3,
        "lm": lm
    })

@app.route('/status', methods=['GET'])
def get_status():
    return jsonify({
        "l1": l1,
        "l2": l2,
        "l3": l3,
        "lm": lm
    })

# 启动Flask线程
def run_flask():
    app.run(host='0.0.0.0', port=5000, debug=False)

# Tkinter UI初始化
top = tk.Tk()
top.title("IoT Control")
but1 = tk.Button(top, text="Light 1: OFF", command=light_1)
but2 = tk.Button(top, text="Light 2: OFF", command=light_3)
but3 = tk.Button(top, text="Light 3: OFF", command=light_2)
but4 = tk.Button(top, text="Master: OFF", command=master_switch)

but1.pack(pady=5)
but2.pack(pady=5)
but3.pack(pady=5)
but4.pack(pady=5)

# 启动Flask线程
flask_thread = threading.Thread(target=run_flask, daemon=True)
flask_thread.start()

try:
    top.mainloop()
finally:
    GPIO.cleanup()

步骤3：Android端实现 #

Android通过HttpURLConnection或OkHttp发送POST请求到树莓派本地IP（例如http://192.168.x.x:5000/control/light1）触发灯光开关，发送GET请求到/status接口获取当前状态。

2. 互联网远程控制：两种可行方案 #

方案A：内网穿透（适合个人小场景） #

1. 在内网穿透工具客户端配置文件中，将树莓派的5000端口映射到公网服务器的指定端口（例如remote_port = 6000）。
2. 树莓派启动内网穿透客户端，连接到公网服务器。
3. Android通过公网服务器IP+映射端口访问API（例如http://xxx.xxx.xxx.xxx:6000/control/light1）。

方案B：MQTT云中转（适合IoT规模化场景） #

用MQTT协议实现设备与云服务器的双向通信，延迟低且节省带宽。

步骤1：安装依赖

树莓派安装paho-mqtt：

pip install paho-mqtt

步骤2：修改Python代码集成MQTT

在原有代码基础上添加MQTT订阅逻辑：

# 新增MQTT相关代码
import paho.mqtt.client as mqtt
import json

# MQTT配置
MQTT_BROKER = "你的云服务器IP或公共MQTT broker地址"
MQTT_PORT = 1883
MQTT_CONTROL_TOPIC = "iot/control"
MQTT_STATUS_TOPIC = "iot/status"
MQTT_CLIENT_ID = "raspberrypi-iot"

def on_connect(client, userdata, flags, rc):
    print(f"Connected with result code {rc}")
    client.subscribe(MQTT_CONTROL_TOPIC)

def on_message(client, userdata, msg):
    payload = msg.payload.decode('utf-8')
    # 根据指令执行控制
    if payload == "light1":
        light_1()
    elif payload == "light2":
        light_2()
    elif payload == "light3":
        light_3()
    elif payload == "master":
        master_switch()
    # 发布状态更新到主题
    client.publish(MQTT_STATUS_TOPIC, json.dumps({
        "l1": l1,
        "l2": l2,
        "l3": l3,
        "lm": lm
    }))

mqtt_client = mqtt.Client(MQTT_CLIENT_ID)
mqtt_client.on_connect = on_connect
mqtt_client.on_message = on_message
mqtt_client.connect(MQTT_BROKER, MQTT_PORT, 60)

# 启动MQTT线程
def run_mqtt():
    mqtt_client.loop_forever()

mqtt_thread = threading.Thread(target=run_mqtt, daemon=True)
mqtt_thread.start()

步骤3：Android端实现

Android集成Eclipse Paho MQTT客户端库，订阅iot/status主题获取设备状态，发布指令到iot/control主题实现远程控制。

3. 状态同步实现 #

核心逻辑如下：

• 所有控制操作（本地UI、Android）统一调用同一组控制函数（如light_1()），函数内部自动更新全局状态变量l1/l2/l3/lm。
• Tkinter非线程安全，在Flask、MQTT等非UI线程中更新UI时，必须使用top.after(0, 回调函数)将UI操作放入Tkinter事件循环执行（如代码中更新按钮文本的逻辑）。
• Android端通过/status接口（REST）或订阅iot/status主题（MQTT）实时获取最新状态，确保两端状态一致。

内容的提问来源于stack exchange，提问作者Bhuvan R J