更优雅的定期检查事件/触发器的方法

Question

我有一个被动红外传感器，我想根据运动关闭和打开我的显示器。例如，如果5分钟没有动静，那么显示器应该关闭以节省电力。然而，如果有运动，不要关闭显示器，或将其重新打开。(不要问为什么屏保不能做到这一点。我要做的设备没有键盘和鼠标。它将只是一个独立的显示。)

我的想法是创建两个线程，一个生产者，一个消费者。生产者线程(PIR传感器)将消息放入队列，由消费者(控制显示)读取。这样我就可以把信号从一个发送到另一个。

下面有一个功能完整的代码(有一些解释)，它完成了前面描述的内容。我的问题是有没有一种更优雅的方式来达到这个目的?你觉得我的方法怎么样，还行吗，还粗俗吗?

#!/usr/bin/env python
import Queue
from threading import Thread
import RPi.GPIO as gpio
import time
import os
import sys
class PIRSensor:
    # PIR sensor's states
    current_state = 0
    previous_state = 0
    def __init__(self, pir_pin, timeout):
        # PIR GPIO pin
        self.pir_pin = pir_pin
        # Timeout between motion detections
        self.timeout = timeout
    def setup(self):
        gpio.setmode(gpio.BCM)
        gpio.setup(self.pir_pin, gpio.IN)
        # Wait for the PIR sensor to settle
        # (loop until PIR output is 0)
        while gpio.input(self.pir_pin) == 1:
            self.current_state = 0
    def report_motion(self, queue):
        try:
            self.setup()
            while True:
                self.current_state = gpio.input(self.pir_pin)
                if self.current_state == 1 and self.previous_state == 0:
                    # PIR sensor is triggered
                    queue.put(True)
                    # Record previous state
                    self.previous_state = 1
                elif self.current_state == 1 and self.previous_state == 1:
                    # Feed the queue since there is still motion
                    queue.put(True)
                elif self.current_state == 0 and self.previous_state == 1:
                    # PIR sensor has returned to ready state
                    self.previous_state = 0
                time.sleep(self.timeout)
        except KeyboardInterrupt:
            raise
class DisplayControl:
    # Display's status
    display_on = True
    def __init__(self, timeout):
        self.timeout = timeout
    def turn_off(self):
        # Turn off the display
        if self.display_on:
            os.system("/opt/vc/bin/tvservice -o > /dev/null 2>&1")
            self.display_on = False
    def turn_on(self):
        # Turn on the display
        if not self.display_on:
            os.system("{ /opt/vc/bin/tvservice -p && chvt 9 && chvt 7 ; } > /dev/null 2>&1")
            self.display_on = True
    def check_motion(self, queue):
        try:
            while True:
                try:
                    motion = queue.get(True, self.timeout)
                    if motion:
                        self.turn_on()
                except Queue.Empty:
                        self.turn_off()
        except KeyboardInterrupt:
            raise
if __name__ == "__main__":
    try:
        pir_sensor = PIRSensor(7, 0.25)
        display_control = DisplayControl(300)
        queue = Queue.Queue()
        producer = Thread(target=pir_sensor.report_motion, args=(queue,))
        consumer = Thread(target=display_control.check_motion, args=(queue,))
        producer.daemon = True
        consumer.daemon = True
        producer.start()
        consumer.start()
        while True:
            time.sleep(0.1)
    except KeyboardInterrupt:
        display_control.turn_on()
        # Reset GPIO settings
        gpio.cleanup()
        sys.exit(0)

生产者线程运行PIRSensor类实例的函数(report_motion)。PIRSensor类每秒读取被动红外传感器的状态四次，每当它感知到运动时，就将消息放入队列。

消费者线程运行DisplayControl类实例的(check_motion)函数。它在给定超时的情况下以阻塞模式读取前面提到的队列。可能发生以下情况:

• 如果显示打开并且队列中没有给定的消息时间，即超时超时，消费者线程将断电显示。
• 如果显示是关闭的，一个消息来了，线程将打开显示电源

Answer 1

我觉得这个主意不错。我对你的实现唯一的问题是为什么有消费者和生产者在子线程?你可以把消费者保持在主线程中，这样主线程中就不需要这个无意义的循环了。

while True:
    time.sleep(0.1)

只会浪费CPU周期。相反，您可以直接调用display_motion.check_motion(queue)。

Answer 2

我认为这是一个很好的解决方案。原因是您已经分离了不同类的关注点。一个类处理PIR传感器。一个人负责显示器。你今天把他们排成一列，这是一种方法。

这样做可以很容易地测试不同的类。

要扩展这一点(即使其可扩展)，您可能需要引入一个控制器。控制器获取事件(例如，从队列)并对事件进行操作(例如，告诉显示控制器关闭显示)。控制器知道传感器，也知道显示。但是传感器不应该知道显示，反之亦然。(这与MVC非常相似，在这种情况下，数据是模型(传感器)，显示是视图，控制器位于两者之间。

这种方法使设计可测试、可扩展、可维护。这样你就不是一个黑客了，你是在写真正的代码。