如何将给定树莓派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
相关产品推荐
相关产品推荐

