如何在使用GPIO.wait_for_edge时实现脚本的优雅终止?
如何在使用GPIO.wait_for_edge时实现脚本的优雅终止?
兄弟,我懂你这痛点!GPIO.wait_for_edge的阻塞特性确实会让Ctrl+C或者系统终止信号没法及时响应,就像程序被“焊死”在那一步了。别担心,有几种简单的方法能解决这个问题,而且完全符合你不想占用CPU的需求~
方案一:信号捕获+带超时的wait_for_edge
这种方法不用大改现有代码,核心思路是给wait_for_edge加个超时时间,让它每隔一段时间“喘口气”,这样程序就能有机会检查是否收到退出信号。
首先我们要捕获Ctrl+C对应的SIGINT信号,还有系统发送的SIGTERM终止信号,设置一个退出标志。然后把原来的阻塞等待改成带超时的循环,每次超时后检查退出标志,确认要退出就优雅收尾。
示例代码:
import RPi.GPIO as GPIO import time import signal # 定义退出标志,初始为False exit_flag = False def handle_exit(signum, frame): """信号处理函数,收到终止信号时设置退出标志""" global exit_flag print("收到终止信号,准备优雅退出...") exit_flag = True # 注册信号处理:捕获Ctrl+C(SIGINT)和系统终止(SIGTERM) signal.signal(signal.SIGINT, handle_exit) signal.signal(signal.SIGTERM, handle_exit) WPIN = 你的引脚号 # 替换成你的实际引脚 WARNST = 你的等待时间 # 替换成你的实际等待时长 try: GPIO.setmode(GPIO.BCM) # 根据你的引脚编号方式调整 GPIO.setup(WPIN, GPIO.IN) while not exit_flag: try: Stat1 = GPIO.input(WPIN) time.sleep(WARNST) Stat2 = GPIO.input(WPIN) if Stat1 == Stat2: SendWarnStat(Stat2) # 你的状态上报函数 # 带超时的等待边缘变化,超时时间设为1秒(可根据需求调整) # 超时后返回None,此时可以检查exit_flag Event = GPIO.wait_for_edge(WPIN, GPIO.BOTH, timeout=1000) if Event is None: # 超时后直接进入下一轮循环,检查是否需要退出 continue except Exception as e: print(f"运行出错: {e}") break finally: # 收尾操作:清理GPIO资源,避免引脚状态异常 GPIO.cleanup() print("脚本已优雅退出")
关键说明:
handle_exit函数会在收到终止信号时触发,把exit_flag设为True,告诉主循环该退出了。GPIO.wait_for_edge的timeout=1000表示最多等待1秒,超时后程序会回到循环开头检查退出标志,不会一直卡死。finally块确保不管程序是正常退出还是出错退出,都会清理GPIO资源,这是个好习惯。
方案二:用线程分离GPIO等待逻辑
如果你想用线程的方式(就是你提到的把等待逻辑放到子线程,主线程处理退出),也很容易实现。我们用threading.Event对象来做主线程和子线程的同步,主线程负责监听终止信号,子线程专注于GPIO状态检测。
示例代码:
import RPi.GPIO as GPIO import time import signal import threading # 创建Event对象,用来通知子线程退出 exit_event = threading.Event() def handle_exit(signum, frame): print("收到终止信号,准备优雅退出...") exit_event.set() # 注册信号处理 signal.signal(signal.SIGINT, handle_exit) signal.signal(signal.SIGTERM, handle_exit) WPIN = 你的引脚号 WARNST = 你的等待时间 def gpio_monitor(): """子线程:负责GPIO状态检测和边缘等待""" while not exit_event.is_set(): try: Stat1 = GPIO.input(WPIN) time.sleep(WARNST) Stat2 = GPIO.input(WPIN) if Stat1 == Stat2: SendWarnStat(Stat2) # 带超时的边缘等待,超时后检查退出信号 Event = GPIO.wait_for_edge(WPIN, GPIO.BOTH, timeout=1000) except Exception as e: print(f"监控线程出错: {e}") break try: GPIO.setmode(GPIO.BCM) GPIO.setup(WPIN, GPIO.IN) # 启动GPIO监控子线程 monitor_thread = threading.Thread(target=gpio_monitor) monitor_thread.start() # 主线程等待退出信号被触发 exit_event.wait() # 等待子线程完成收尾工作 monitor_thread.join() finally: GPIO.cleanup() print("脚本已优雅退出")
关键说明:
- 子线程专注于GPIO的检测逻辑,主线程只负责监听终止信号并通知子线程退出,职责分离更清晰。
exit_event.set()会触发子线程的退出检查,exit_event.wait()让主线程不占用CPU,一直等到退出信号到来。
为什么原来的代码会卡死?
你原来的代码里,GPIO.wait_for_edge是无限阻塞的,只有引脚状态变化时才会返回。这时候如果收到Ctrl+C,Python的信号处理函数要等到wait_for_edge返回后才能执行,所以看起来程序像卡住了没法终止。加超时或者用线程,都是为了给程序留出响应信号的机会。
备注:内容来源于stack exchange,提问作者Marus Gradinaru
相关产品推荐
相关产品推荐

