3

我有这个代码 ipython 代码:

import ipywidgets as widgets
from IPython.display import display
import time

w = widgets.Dropdown(
    options=['Addition', 'Multiplication', 'Subtraction'],
    value='Addition',
    description='Task:',
)

def on_change(change):
    print("changed to %s" % change['new'])

w.observe(on_change)

display(w)

它按预期工作。当小部件的值发生变化时,该on_change函数被触发。但是,我想运行长时间的计算并定期检查小部件的更新。例如:

for i in range(100):
    time.sleep(1)
    # pull for changes to w here.
    # if w.has_changed:
    #     print(w.value)

我怎样才能做到这一点?

4

4 回答 4

2

作为参考,我似乎能够进行所需的轮询

import IPython
ipython = IPython.get_ipython()
ipython.kernel.do_one_iteration()

(我仍然希望得到一些反馈,说明这是偶然还是设计。)

于 2017-04-15T22:57:04.373 回答
0

我认为您需要使用线程并挂钩到 ZMQ 事件循环。这个要点说明了一个例子:

https://gist.github.com/maartenbreddels/3378e8257bf0ee18cfcbdacce6e6a77e

另请参阅https://github.com/jupyter-widgets/ipywidgets/issues/642

于 2017-04-14T23:45:20.050 回答
0

为了详细说明OP的自我回答,这确实有效。它强制小部件在循环中的任意点与内核同步。这可以在访问widget.value.

所以完整的解决方案是:

import IPython
ipython = IPython.get_ipython()

last_val = 0
for i in range(100):
    time.sleep(1)
    ipython.kernel.do_one_iteration()
    new_val = w.value
    if new_val != old_val:
        print(new_val)
        old_val = new_val
于 2019-07-24T01:14:24.467 回答
0

ipython.kernel.do_one_iteration对使用的调用略有改进

# Max iteration limit, in case I don't know what I'm doing here...
for _ in range(100):
  ipython.kernel.do_one_iteration()
  if ipython.kernel.msg_queue.empty():
    break

在我的例子中,我有许多 UI 元素,可以在调用之间多次单击do_one_iteration,这将一次处理一个,并且有 1 秒的时间延迟,这可能会很烦人。这将一次最多处理 100 个。通过多次捣碎一个按钮来测试它,现在它们都在sleep(1)结束时立即获得进程。

于 2019-12-25T19:40:15.497 回答