Python多线程传值：如何在RFID、温感、PIR传感器所属线程间传递数据

你可以通过线程安全的状态事件、共享变量加锁的方式实现需求，不需要复杂的线程通信机制，修改步骤如下：

1. 初始化线程安全的控制变量 #

在类中新增线程安全的控制对象与共享变量，避免多线程读写冲突，同时把数据库配置移到类属性，替换冗余的全局变量：

import threading
# 保留原有其他导入项
class Security:
    # 数据库配置移到类属性
    dbHost = 'localhost'
    dbUser = 'root'
    dbPass = 'iTdev#2020'
    dbNameEmp = 'empinfo'
    dbNameSecure = 'secureinfo'
    # PIR传感器默认禁用时长，单位秒，可根据需求调整
    PIR_DISABLE_SECONDS = 10

    def __init__(self):
        self.msg = ""
        # PIR禁用控制事件：事件处于set状态时PIR停止工作
        self.pir_disable_event = threading.Event()
        # 最新温度共享变量 + 读写锁，保证多线程读写的原子性
        self.latest_temperature = None
        self.temp_lock = threading.Lock()
        # PIR禁用倒计时计时器
        self.pir_disable_timer = None

2. 封装PIR禁用通用逻辑 #

把PIR临时禁用、到期自动恢复的逻辑封装为公用方法，避免RFID和温度模块重复编写代码：

def disable_pir_temp(self):
    """临时禁用PIR传感器，到达指定时长后自动恢复"""
    # 取消之前未完成的倒计时，避免多次触发冲突
    if self.pir_disable_timer and self.pir_disable_timer.is_alive():
        self.pir_disable_timer.cancel()
    # 激活PIR禁用状态
    self.pir_disable_event.set()
    # 启动倒计时，到期后自动解除禁用
    self.pir_disable_timer = threading.Timer(self.PIR_DISABLE_SECONDS, self.pir_disable_event.clear)
    self.pir_disable_timer.start()

3. 修改PirSensor方法响应禁用状态 #

调整PIR传感器的循环逻辑，检测到禁用事件激活时跳过运动检测：

def PirSensor(self):
    while True:
        # PIR处于禁用状态时休眠跳过检测
        if self.pir_disable_event.is_set():
            time.sleep(0.2)
            continue
        # 原有检测、报警逻辑完全保留
        led_blue.on()
        time.sleep(0.1)
        current_state = GPIO.input(pir_sensor)
        if current_state == 1:
            print("GPIO pin %s is %s" % (pir_sensor, current_state))
            GPIO.output(relay_alarm,True)
            led_blue.off()
            led_red.blink(0.1, 0.2)
            time.sleep(5)
            GPIO.output(relay_alarm,False)
            led_red.off()
            led_yellow.blink(0.5, 0.5)
            time.sleep(5)
            led_yellow.off()

4. 修改Temperature方法同步温度+触发PIR禁用 #

测温成功后更新共享温度变量，同时触发PIR禁用：

def Temperature(self):
    while True:
        # 原有测距逻辑完全保留
        GPIO.output(TRIG, False)
        time.sleep(0.5)
        GPIO.output(TRIG, True)
        time.sleep(0.01)
        GPIO.output(TRIG, False)
        while GPIO.input(ECHO) == 0 :
            pulse_start = time.time()
        while GPIO.input(ECHO) == 1 :
            pulse_end = time.time()
        pulse_duration = pulse_end - pulse_start
        distance = pulse_duration * 11150
        distance = round(distance,2)

        if distance > 0 and distance < 3 :
            # 原有测温、屏幕显示逻辑完全保留
            buzzer.on()
            time.sleep(0.1)
            buzzer.off()
            print ("Distance: %scm" % (distance))
            temp_val = sensor.get_object_1()
            print ("Temperature: %.1fc" % temp_val)
            msgTemp = ("%.1f" % temp_val)
            w, h = textsize(msgTemp, font=proportional(CP437_FONT))
            if w <= device.width:
                x = round((device.width - w) / 2)
                with canvas(device) as draw:
                    text(draw, (x, 0), msgTemp, fill="white", font=proportional(CP437_FONT))
            else:
                show_message(device, msgTemp, fill="white", font=proportional(CP437_FONT),scroll_delay=0.04)
            time.sleep(1)
            device.clear()

            # 新增逻辑：加锁更新共享温度+禁用PIR
            with self.temp_lock:
                self.latest_temperature = temp_val
            self.disable_pir_temp()

5. 修改rfid_callback方法读取温度+触发PIR禁用 #

RFID记录保存成功后，读取最新温度写入数据库，同时触发PIR禁用：

def rfid_callback(self, state, dev):
    rfid_presented = ""
    keys = "X^1234567890XXXXqwertzuiopXXXXasdfghjklXXXXXyxcvbnmXXXXXXXXXXXXXXXXXXXXXXX"
    while True:
        r,w,x = select([dev], [], [])
        for event in dev.read():
            if event.type==1 and event.value==1:
                if event.code==28:
                    rfid_presented = rfid_presented.replace("X", "")
                    travel = state.replace("Thread-", "")
                    # 原有拼接SQL的写法存在注入风险，改为参数化查询
                    dbConnEmp = mysql.connect(host=self.dbHost, user=self.dbUser, passwd=self.dbPass, db=self.dbNameEmp)
                    curEmp = dbConnEmp.cursor()
                    curEmp.execute("Select RFidEmp FROM RFidMaster WHERE (RFidStatus = 1) AND RFidNumber = %s", (rfid_presented,))
                    resultRFid = curEmp.fetchone()
                    if curEmp.rowcount != 1:
                        with canvas(device) as draw:
                            text(draw, (0, 0), "None", fill="white", font=proportional(CP437_FONT))
                        time.sleep(0.5)
                        device.clear()
                    else:
                        dbConnSecure = mysql.connect(host=self.dbHost, user=self.dbUser, passwd=self.dbPass, db=self.dbNameSecure)
                        curSecure = dbConnSecure.cursor()
                        curSecure.execute("SELECT EntraId,DATE_FORMAT(CreateDate, '%Y%m%d') AS CreateDate FROM entranlog ORDER BY EntraId DESC, CreateDate DESC LIMIT 1")
                        resultKey = curSecure.fetchone()
                        KeyDate = time.strftime("%Y%m%d")
                        if curSecure.rowcount != 1:
                            KeyId = KeyDate+"0001"
                        else:
                            if resultKey[1] == KeyDate:
                                KeyId = str(int(resultKey[0]) + 1)
                            else:
                                KeyId = KeyDate+"0001"

                        create_date = time.strftime('%Y-%m-%d %H:%M:%S')
                        # 新增：加锁读取最新温度
                        with self.temp_lock:
                            record_temp = self.latest_temperature
                        # 修改插入语句，增加温度字段（需提前在数据库entranlog表新增Temperature浮点型字段）
                        insertSecure = "INSERT INTO entranlog (EntraId,EntraAction,EntraStatus,CreateBy,CreateDate,Temperature) VALUES (%s,%s,%s,%s,%s,%s)"
                        valSecure = (KeyId,travel,1,resultRFid[0],create_date,record_temp)
                        curSecure.execute(insertSecure, valSecure)
                        dbConnSecure.commit()
                        with canvas(device) as draw:
                            text(draw, (0, 0), "Hello", fill="white", font=proportional(CP437_FONT))
                        print(curSecure.rowcount, "record inserted.")
                        time.sleep(0.8)
                        device.clear()
                        # 新增：RFID数据保存成功后禁用PIR
                        self.disable_pir_temp()
                    rfid_presented = ""
                else:
                    rfid_presented += keys[ event.code ]

额外注意 #

• 提前在数据库entranlog表新增Temperature浮点类型字段，用于存储温度记录
• 所有数据库操作建议统一使用参数化查询，避免SQL注入风险
• 可根据实际使用场景调整PIR_DISABLE_SECONDS的禁用时长

内容的提问来源于stack exchange，提问作者Mr. Search