如何在非响应式模块的回调中设置Shiny for Python响应式值？

我正在用Shiny for Python结合Tinkerforge硬件模块开发应用，已经把所有硬件相关代码封装成类，启动应用时创建实例。Tinkerforge支持给按钮状态变化事件注册回调，按钮状态改变时可以执行对应函数。

我想在按钮按下时触发应用的响应链，同时获取按压时长，但发现回调里能用全局变量，却没法设置响应式值，会提示“无当前响应上下文（No current reactive context）”。怎么通过这个回调触发Shiny应用里的事件？

回调注册正常，按钮按下/释放的打印逻辑能执行，但设置响应式值时应用崩溃，调试时会出现RuntimeError: No current reactive context错误。最初用reactive.invalidate_later每50ms轮询按钮状态，但回调方式明显更简洁。

更新：用asyncio做了无Tinkerforge的模拟按钮示例，也出现了相同错误。有意思的是，“无当前响应上下文”错误只在读取响应式值时触发，设置时不会。不在hw_button_callback()末尾调用print(triggerMe.get())不会崩溃，但界面也不更新，一直加载。

原始代码示例 #

TinkerforgeModule.py #

from tinkerforge.ip_connection import IPConnection
from tinkerforge.bricklet_rgb_led_button import BrickletRGBLEDButton

class TinkerforgeModule (object):
   def __init__(self):
      ipcon = IPConnection() # 创建和Tinkerforge的IP连接
      self.rgb_button = BrickletRGBLEDButton('24fu', ipcon)
      ipcon.connect('127.0.0.1', 4223)

   def register_hw_button_cb(self, func):
      self.rgb_button.register_callback(
         BrickletRGBLEDButton.CALLBACK_BUTTON_STATE_CHANGED, 
         func
      )

app.py #

from shiny import App, ui, render, reactive
from datetime import datetime
import TinkerforgeModule

app_ui = ui.page_auto(
    ui.output_text("result_output")
)

def server(input, output, session):

    hw_button_last_press = None
    triggerMe = reactive.value(None)

    def hw_button_callback(state):
        global hw_button_last_press
        if state == 1:      # BrickletRGBLEDButton.BUTTON_STATE_RELEASED = 1
            press_time = datetime.now() - hw_button_last_press
            print(f"Button released after {press_time} seconds.")
            
            # 下面这行无法工作
            # triggerMe.set(f"Button released after {press_time} seconds.") 

        else:               # BrickletRGBLEDButton.BUTTON_STATE_PRESSED = 0
            hw_button_last_press = datetime.now()
            print("Button pressed.")

            # 下面这行无法工作
            # triggerMe.set("Button pressed.") 

    tfm = TinkerforgeModule.TinkerforgeModule()
    tfm.register_hw_button_cb(hw_button_callback)

    @output
    @render.text
    def result_output():
        return(f"Message: {triggerMe()}")

app = App(app_ui, server)

模拟测试代码（更新） #

FakeTinkerforgeModule.py #

import asyncio
from datetime import datetime

class FakeTinkerforgeModule (object):
   def __init__(self):
      asyncio.create_task(self.fake_button_trigger())

   def register_hw_button_cb(self, func):
      self.cb_func = func

   async def fake_button_trigger(self):
      await asyncio.sleep(2)
      self.cb_func(0)
      await asyncio.sleep(1)
      self.cb_func(1)

app.py（模拟版） #

from shiny import App, ui, render, reactive
from datetime import datetime
import FakeTinkerforgeModule

app_ui = ui.page_auto(
    ui.output_text("result_output")
)

def server(input, output, session):

    hw_button_last_press = None
    triggerMe = reactive.value(None)

    def hw_button_callback(state):
        global hw_button_last_press
        print(f"callback called, state: {state}")
        if state == 1:      # BrickletRGBLEDButton.BUTTON_STATE_RELEASED = 1
            press_time = datetime.now() - hw_button_last_press
            print(f"Button released after {press_time} seconds.")
            triggerMe.set(f"Button released after {press_time} seconds.") # 这部分看似没效果，但也不报错

        else:               # BrickletRGBLEDButton.BUTTON_STATE_PRESSED = 0
            hw_button_last_press = datetime.now()
            print("Button pressed.")
            triggerMe.set("Button pressed.") # 这部分看似没效果，但也不报错
        print(triggerMe.get()) # 这行触发RuntimeError("No current reactive context")

    @output
    @render.text
    def result_output():
        return(f"Message: {triggerMe()}")
    
    tfm = FakeTinkerforgeModule.FakeTinkerforgeModule()
    tfm.register_hw_button_cb(hw_button_callback)

app = App(app_ui, server)

解决方法 #

问题根源：Tinkerforge（或模拟代码）的回调是在Shiny的响应上下文之外执行的，直接操作响应式值会因为未绑定到Shiny的反应系统而失效，读取响应式值还会触发上下文错误。

最可靠的解决方式是使用session.queue()，把更新响应式值的操作放到Shiny的内部事件队列中，确保操作在正确的响应上下文里执行。

修改后的核心代码（app.py） #

from shiny import App, ui, render, reactive
from datetime import datetime
import TinkerforgeModule

app_ui = ui.page_auto(
    ui.output_text("result_output")
)

def server(input, output, session):

    hw_button_last_press = None
    triggerMe = reactive.value("Waiting for button action...")

    def hw_button_callback(state):
        nonlocal hw_button_last_press
        if state == 1:      # 按钮释放
            press_time = datetime.now() - hw_button_last_press
            msg = f"Button released after {press_time.total_seconds():.2f} seconds."
            print(msg)
            # 将响应式值更新操作放入Shiny事件队列
            session.queue(lambda: triggerMe.set(msg))

        else:               # 按钮按下
            hw_button_last_press = datetime.now()
            msg = "Button pressed."
            print(msg)
            session.queue(lambda: triggerMe.set(msg))

    tfm = TinkerforgeModule.TinkerforgeModule()
    tfm.register_hw_button_cb(hw_button_callback)

    @output
    @render.text
    def result_output():
        return f"Message: {triggerMe()}"

app = App(app_ui, server)

关键说明 #

1. session.queue()：该方法会把传入的lambda函数放到Shiny的事件循环中执行，确保操作处于Shiny的响应上下文内，既不会触发上下文错误，也能正确触发响应链更新UI。
2. nonlocal替代global：hw_button_last_press是server函数内的局部变量，用nonlocal更贴合作用域规则，避免全局变量污染。
3. 避免在回调中读取响应式值：如果必须读取响应式值，也要把读取操作放到session.queue()的lambda函数里执行。

模拟代码适配 #

给模拟版的app.py做相同修改，将triggerMe.set()放入session.queue()，并去掉回调内的print(triggerMe.get())，即可正常更新UI。

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