Python Firebase+Pusher双变量跌倒检测通知触发失败排查

我来帮你搞定这个跌倒检测通知的问题，你的代码里有几个关键问题导致最后if语句完全没生效，咱们一步步拆解：

核心问题分析 #

1. 全局if语句仅执行一次
   你写的if value == 1 & value1 == 1:是在程序启动时就执行的，这时候value和value1都是初始的0，所以这个判断直接跳过了。之后即使Firebase的stream更新了这两个变量，这个if也不会再重新检查了——它只会运行一次。

2. 逻辑运算符用错了
   你用了&（按位与运算符），但这里需要的是逻辑与，应该用and（Python里推荐用这个）。虽然1&1结果也是1，但这是错误的用法，逻辑判断必须用逻辑运算符。

3. 用is判断数值相等不严谨
   在Python里is是判断两个对象是否是同一个，而不是值相等。虽然小整数（比如1）可能会被缓存复用，但判断值相等应该用==，避免出现意外的bug。

修复后的完整代码 #

import pyrebase
from pusher_push_notifications import PushNotifications

config = {
    'apiKey': "***********************************",
    'authDomain': "arfduinopushnotification.firebaseapp.com",
    'databaseURL': "https://arduinopushnotification.firebaseio.com",
    'projectId': "arduinopushnotification",
    'storageBucket': "arduinopushnotification.appspot.com",
    'messagingSenderId': "************"
}
firebase = pyrebase.initialize_app(config)
db = firebase.database()

pn_client = PushNotifications(
    instance_id='*****************************',
    secret_key='**************************',
)

value = 0
value1 = 0

def check_and_send_notification():
    # 用正确的逻辑与判断两个状态是否都为1
    if value == 1 and value1 == 1:
        response = pn_client.publish(
            interests=['hello'],
            publish_body={
                'apns': {
                    'aps': {
                        'alert': 'Hello!',
                    },
                },
                'fcm': {
                    'notification': {
                        'title': 'Notification',
                        'body': 'Fall Detected !!',
                    },
                },
            },
        )
        print(f"Notification sent! Publish ID: {response['publishId']}")

def stream_handler(message):
    global value
    print(message)
    # 用==判断值相等，同时处理状态变回0的情况
    if message['data'] == 1:
        value = message['data']
    else:
        value = 0
    # 每次更新变量后检查是否满足通知条件
    check_and_send_notification()

def stream_handler1(message):
    global value1
    print(message)
    if message['data'] == 1:
        value1 = message['data']
    else:
        value1 = 0
    # 每次更新变量后检查是否满足通知条件
    check_and_send_notification()

# 启动数据流监听
my_stream = db.child("Fall_Detection_Status").stream(stream_handler)
my_stream1 = db.child("Fall_Detection_Status1").stream(stream_handler1)

关键修改说明 #

• 新增了check_and_send_notification()函数，把通知发送逻辑封装进去，这样每次任意一个状态变量更新时，都能触发检查。
• 把原来的全局if语句移到了这个函数里，保证每次变量变化都会重新判断。
• 把&改成了and，用正确的逻辑与判断。
• 把is 1改成了== 1，同时添加了else分支，确保状态变回0时变量能正确更新（避免之前的状态一直保留为1）。
• 每个stream回调函数更新变量后，都会调用check_and_send_notification()，这样不管哪个状态变化，都能立即检查是否满足双1的条件。

这样修改后，只要Fall_Detection_Status和Fall_Detection_Status1同时为1，就会自动触发推送通知，而且print语句也会正常输出啦。

内容的提问来源于stack exchange，提问作者Moustapha Sabbagh