You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何将给定树莓派Python代码封装为可跨文件调用的函数

树莓派GPIO+MQTT功能封装改造方案

改造说明

原代码存在变量作用域错误、MQTT连接逻辑不合理、导入即自动运行无法复用的问题,改造后所有逻辑封装为独立可导入的控制器类,对外暴露启动、停止接口,保留全部原有功能,其他Python文件可直接导入调用。
核心改造点:

  • 修复原代码引脚变量定义在空函数内、全局无法访问的bug
  • 所有状态(引脚配置、MQTT客户端、运行标志)封装在类内部,不污染全局命名空间
  • 优化MQTT连接逻辑:初始化时一次性建立连接,启动后台网络线程,避免每次按钮触发都新建客户端
  • 增加可控的停止逻辑,支持外部触发时安全退出循环、清理GPIO资源
  • 保留if __name__ == '__main__'块,文件单独运行时和原逻辑行为一致

改造后完整代码

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

class PrivacyBtnController:
    def __init__(self, led_pin_1=7, led_pin_2=21, but_pin=18, mqtt_broker="mqtt.fluux.io", mqtt_topic="privacy", client_id="privacybtn"):
        # 支持自定义配置参数,默认值和原代码完全一致
        self.led_pin_1 = led_pin_1
        self.led_pin_2 = led_pin_2
        self.but_pin = but_pin
        self.mqtt_broker = mqtt_broker
        self.mqtt_topic = mqtt_topic
        self.client_id = client_id
        self.running = False
        self.mqtt_client = None

    def _blink_callback(self, channel):
        # 按钮按下内部回调,不对外暴露
        btn_state = GPIO.input(self.but_pin)
        print("blinked")
        # 保留原LED闪烁逻辑
        GPIO.output(self.led_pin_2, GPIO.HIGH)
        time.sleep(0.5)
        GPIO.output(self.led_pin_2, GPIO.LOW)
        time.sleep(0.5)
        # 发布MQTT消息
        if self.mqtt_client:
            self.mqtt_client.publish(self.mqtt_topic, btn_state)
            print(f"Just published {btn_state} to topic {self.mqtt_topic}")

    def start(self):
        # 对外暴露的启动接口
        if self.running:
            print("控制器已在运行中")
            return
        
        # GPIO初始化
        GPIO.setmode(GPIO.BOARD)
        GPIO.setup([self.led_pin_1, self.led_pin_2], GPIO.OUT)
        GPIO.setup(self.but_pin, GPIO.IN)
        GPIO.output(self.led_pin_1, GPIO.LOW)
        GPIO.output(self.led_pin_2, GPIO.LOW)

        # MQTT初始化
        self.mqtt_client = mqtt.Client(self.client_id)
        self.mqtt_client.connect(self.mqtt_broker, keepalive=60)
        self.mqtt_client.loop_start()

        # 绑定按钮中断事件
        GPIO.add_event_detect(self.but_pin, GPIO.FALLING, callback=self._blink_callback, bouncetime=10)
        
        self.running = True
        print("控制器启动,按CTRL+C可退出")
        try:
            while self.running:
                btn_state = GPIO.input(self.but_pin)
                print(btn_state)
                # 保留原LED1运行逻辑
                GPIO.output(self.led_pin_1, GPIO.HIGH)
                time.sleep(2)
        finally:
            self.stop()

    def stop(self):
        # 对外暴露的停止、资源清理接口
        self.running = False
        if self.mqtt_client:
            self.mqtt_client.loop_stop()
            self.mqtt_client.disconnect()
        GPIO.cleanup()
        print("控制器已停止,资源已清理")

# 单独运行本文件时执行原有逻辑
if __name__ == '__main__':
    controller = PrivacyBtnController()
    controller.start()

其他文件导入调用方法

把上述代码保存为privacy_btn.py,放在项目代码同级目录下,即可直接导入使用:

# 其他Python文件中导入模块
from privacy_btn import PrivacyBtnController
import time

# 初始化控制器,可按需修改引脚、MQTT配置参数
btn_controller = PrivacyBtnController()

# 如果需要后台运行不阻塞主线程,可以用线程启动
# run_thread = threading.Thread(target=btn_controller.start, daemon=True)
# run_thread.start()

# 阻塞式启动(和原代码运行效果一致)
btn_controller.start()

# 业务逻辑结束后可手动调用stop安全清理资源
# btn_controller.stop()

最简调用仅需3行代码:

from privacy_btn import PrivacyBtnController
ctl = PrivacyBtnController()
ctl.start()

内容的提问来源于stack exchange,提问作者Usitha Kodithuwakku Arachchi

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.09.01 22:54:38