python - 如果 GPIO 输入改变，倒计时线程 RPi 停止

Question

我不确定执行此操作的最佳方法，并且不适合我，因为它会在程序休眠时锁定程序，但不确定如何使其正常工作...

我试图用树莓派监控门何时打开和关闭，如果门打开超过 x 时间，发送某种警报(比如和电子邮件)，我遇到的问题是当门在倒计时结束后，它不会停止倒计时，这会导致线程停止，也没有实现警报方面，但如果代码是目前的代码，即使门在倒计时之前关闭，它也会触发警报。

目前我正在使用按钮而不是门传感器进行测试，最终我也会记录门的打开和关闭，但现在我想知道是否有更好的方法这个，我从这个 post 得到了我正在使用的代码

我的代码如下

#!/usr/bin/python

import threading, subprocess, sys, time, syslog
import RPi.GPIO as GPIO

lim = 2 # seconds until warning

# thread for countdown (should be interruptable)
class CountdownTask:
    global dooropen
    global countdone

    def __init__(self):
        #print("thread in")
        self._running = True

    def start(self):
        print("thread in")
        self._running = True

    def terminate(self):
        print("thread killed")
        self._running = False

    def run(self, n):
        while True:
            global countdone

            while self._running and dooropen == False and countdone:
                pass

            while self._running and dooropen == False and countdone == False:
                pass

            while self._running and dooropen and countdone:
                pass

            while self._running and dooropen and countdone == False:
                print("start timer")
                time.sleep(5)
                if dooropen:
                    ## action when timer isup
                    print("timer ended, send notify")
                    countdone = True


c = CountdownTask()
t = threading.Thread(target=c.run, args=(lim,))
t.daemon = True

REED = 23 # data pin of reed sensor (in)

# GPIO setup
GPIO.setwarnings(False)
GPIO.setmode(GPIO.BCM)
GPIO.setup(REED, GPIO.IN, pull_up_down=GPIO.PUD_DOWN)

dooropen = False # assuming door's closed when starting
countdone = True


def edge(channel):
    global dooropen
    global countdone

    if GPIO.input(REED): # * no longer reached
        if dooropen == False: # catch fridge compressor spike
            print("Detect open")
            countdone = False
            dooropen = True
        else:
            print("Door closed")
            dooropen = False

def main():
    GPIO.add_event_detect(REED, GPIO.RISING, callback=edge, bouncetime=300)
    t.start()
    while True:
        pass

#------------------------------------------------------------

if __name__ == "__main__": main()

更新:

看来我需要使用 threading.Event 并等待有人可以建议如何在我的代码上实现它吗？

Answer 1

我想我得到了一个有效的脚本

#!/usr/bin/python

import threading
import time
import logging
import RPi.GPIO as GPIO
import smtplib
from email.MIMEMultipart import MIMEMultipart
from email.MIMEText import MIMEText

logging.basicConfig(level=logging.DEBUG,format='(%(threadName)-9s) %(message)s',)

# GPIO setup
Input = 23 # data pin of Input sensor (in)

GPIO.setwarnings(False)
GPIO.setmode(GPIO.BCM)
GPIO.setup(Input, GPIO.IN, pull_up_down=GPIO.PUD_DOWN)

global button
button = False
global count
count = 1
#global t
t = 1
countdown = 5

def sendmail():
    fromaddr = "email"
    toaddr = "tomail"
    msg = MIMEMultipart()
    msg['From'] = fromaddr
    msg['To'] = toaddr
    msg['Subject'] = "DoorAlarm"

    body = "This is a test mail generated with python on a RPi"
    msg.attach(MIMEText(body, 'plain'))

    server = smtplib.SMTP('smtp.gmail.com', 587)
    server.starttls()
    server.login("username", "password")
    text = msg.as_string()
    server.sendmail(fromaddr, toaddr, text)
    server.quit()


def timeout(e, t):
    global count
    global button

    while True:
        while button:
            while not e.isSet() and count <= countdown:
                if count == 1: logging.debug('starting counter')
                event_is_set = e.wait(t)
                if event_is_set:
                    count = 1
                    logging.debug('Door closed before countdown')
                else:
                    count += 1
                if count == countdown:
                    logging.debug('countdown completed - notify')
                    #sendmail()

def edge(channel):
    global button
    global count

    if button == False: # catch fridge compressor spike
        button = True
        e.clear()
        print("log door open")
        if count != 1:
            count = 1

    else:
        button = False
        e.set()
        print("log door closed")

if __name__ == '__main__':
    e = threading.Event()

    t = threading.Thread(name='non-blocking',target=timeout,args=(e, t))
    t.start()

    logging.debug('Waiting before calling Event.set()')
    GPIO.add_event_detect(Input, GPIO.RISING, callback=edge, bouncetime=300)