pylutron_caseta无法监听Lutron设备按钮按压事件问题求助

问题描述 #

使用pylutron_caseta Python包对接Lutron设备时，已完成桥接器配对、可正常获取全量按钮信息，但无法接收到按钮按压事件回调，按下Pico遥控器按钮时控制台无任何输出。
原实现代码如下：

import asyncio
from pylutron_caseta.smartbridge import Smartbridge
from pylutron_caseta.pairing import async_pair
from os.path import exists


bridgeIp = "192.168.1.40"

async def pair(host: str):
    def _ready():
        print("Press the small black button on the back of the bridge.")

    data = await async_pair(host, _ready)
    with open("caseta-bridge.crt", "w") as cacert:
        cacert.write(data["ca"])
    with open("caseta.crt", "w") as cert:
        cert.write(data["cert"])
    with open("caseta.key", "w") as key:
        key.write(data["key"])
    print(f"Successfully paired with {data['version']}")

async def registerButton(bridge,button_id):
    print("Press the small button on the button device.")

def printThis(x):
    print(x)


async def connect():
    bridge = Smartbridge.create_tls(bridgeIp, "caseta.key", "caseta.crt", "caseta-bridge.crt")
    await bridge.connect()
    buttons = bridge.get_buttons()
    print (buttons)
    for b in buttons:
        print (b)
        loopListen = asyncio.get_event_loop()
        asyncio.ensure_future(listen(bridge,int(b)))
    loopListen.run_forever()

async def listen(bridge,_buttonID):
    while True:
        bridge.add_button_subscriber(str(_buttonID), printThis)

#Program
if exists("caseta-bridge.crt"):
    print("found pair files")
    loop = asyncio.get_event_loop()
    loop.run_until_complete(connect())
else:
    loop = asyncio.get_event_loop()
    loop.run_until_complete(pair(bridgeIp))
    loop = asyncio.get_event_loop()
    loop.run_until_complete(connect())

问题根因 #

• 订阅注册逻辑错误：add_button_subscriber是一次性注册方法，原代码把它放在while True无限循环中重复执行，既无意义还会阻塞异步事件循环
• 事件循环调用违规：原代码在connect协程内部调用loop.run_forever()，协程本身运行在已启动的事件循环中，嵌套调用会直接导致事件循环卡死，注册的回调永远没有执行机会
• 冗余异步逻辑：专门实现的listen协程完全多余，按钮订阅注册是同步操作，不需要包装成异步任务、套无限循环执行

修正方案 #

连接桥接器成功后，一次性为所有按钮注册事件回调，通过异步等待事件的方式保持程序运行即可，不要嵌套调用事件循环、不要重复注册订阅。
修正后的可运行代码：

import asyncio
from pylutron_caseta.smartbridge import Smartbridge
from pylutron_caseta.pairing import async_pair
from os.path import exists

bridgeIp = "192.168.1.40"

async def pair(host: str):
    def _ready():
        print("Press the small black button on the back of the bridge.")
    data = await async_pair(host, _ready)
    with open("caseta-bridge.crt", "w") as cacert:
        cacert.write(data["ca"])
    with open("caseta.crt", "w") as cert:
        cert.write(data["cert"])
    with open("caseta.key", "w") as key:
        key.write(data["key"])
    print(f"Successfully paired with {data['version']}")

def button_event_callback(event_data):
    """按钮事件回调，按压、释放动作都会触发该方法"""
    print(f"Received button event: {event_data}")

async def connect():
    bridge = Smartbridge.create_tls(bridgeIp, "caseta.key", "caseta.crt", "caseta-bridge.crt")
    await bridge.connect()
    buttons = bridge.get_buttons()
    print(f"Discovered buttons: {list(buttons.keys())}")
    
    # 一次性为所有按钮注册订阅回调
    for button_id in buttons:
        bridge.add_button_subscriber(str(button_id), button_event_callback)
        print(f"Listener registered for button ID: {button_id}")
    
    # 永久挂起等待事件触发，不退出程序
    await asyncio.Event().wait()

if __name__ == "__main__":
    # 未配对时先执行配对流程
    if not exists("caseta-bridge.crt"):
        asyncio.run(pair(bridgeIp))
    # 启动连接并监听事件
    asyncio.run(connect())

回调返回的事件数据中包含动作类型编码，通常4代表按钮按压事件、3代表按钮释放事件，可根据业务需求过滤对应动作做后续处理。

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