我认为问题就像所有具有长时间运行代码的Python脚本一样——它在一个线程中运行所有代码,并且当它运行时while True
循环(长时间运行的代码),那么它不能同时运行其他函数。
您可能必须在单独的线程中运行您的函数 - 然后主线程才能执行on_button_clicked
这个版本对我有用:
from IPython.display import display
import ipywidgets as widgets
import time
import threading
button = widgets.Button(description="STOP!")
output = widgets.Output()
display(button, output)
break_cicle = True
def on_button_clicked(event):
global break_cicle
break_cicle = False
print("Button pressed: break_cicle:", break_cicle)
button.on_click(on_button_clicked)
def function():
while break_cicle:
print("While loop: break_cicle:", break_cicle)
time.sleep(1)
print("Done")
threading.Thread(target=function).start()
也许 Jupyter 有其他方法可以解决这个问题 - 即。当你编写函数时async
那么你可以使用asyncio.sleep()
它允许Python在该函数休眠时运行其他函数。
EDIT:
在互联网上挖掘(使用谷歌)我在 Jupyter 论坛上找到了帖子
执行长时间运行的单元时的交互式小部件 - JupyterLab - Jupyter 社区论坛
并且有模块链接jupyter-ui-民意调查其中显示了类似的示例(while
-loop + Button
)并且它使用events
为了这。当函数pull()
执行(在每个循环中)然后 Jupyter 可以将事件发送到小部件并且它有时间执行on_click()
.
import time
from ipywidgets import Button
from jupyter_ui_poll import ui_events
# Set up simple GUI, button with on_click callback
# that sets ui_done=True and changes button text
ui_done = False
def on_click(btn):
global ui_done
ui_done = True
btn.description = '????'
btn = Button(description='Click Me')
btn.on_click(on_click)
display(btn)
# Wait for user to press the button
with ui_events() as poll:
while ui_done is False:
poll(10) # React to UI events (upto 10 at a time)
print('.', end='')
time.sleep(0.1)
print('done')
In 源代码我可以看到它使用asyncio
为了这。
EDIT:
版本有多重处理
进程不共享变量,因此需要Queue
将信息从一个进程发送到另一个进程。
示例发送消息来自button
to function
。如果您想从以下位置发送消息function
to button
那么最好使用第二个队列。
from IPython.display import display
import ipywidgets as widgets
import time
import multiprocessing
button = widgets.Button(description="STOP!")
output = widgets.Output()
display(button, output)
queue = multiprocessing.Queue()
def on_button_clicked(event):
queue.put('stop')
print("Button pressed")
button.on_click(on_button_clicked)
def function(queue):
while True:
print("While loop")
time.sleep(1)
if not queue.empty():
msg = queue.get()
if msg == 'stop':
break
#if msg == 'other text':
# ...other code...
print("Done")
multiprocessing.Process(target=function, args=(queue,)).start()
或更类似于之前的
def function(queue):
break_cicle = True
while break_cicle:
print("While loop: break_cicle:", break_cicle)
time.sleep(1)
if (not queue.empty()) and (queue.get() == 'stop'):
break_cicle = False
print("Done")
EDIT:
版本有asyncio
Jupyter 已经在运行asynio event loop
我添加async function
到这个循环。以及功能用途await
功能类似于asyncio.sleep
so asynio event loop
有时间运行其他函数 - 但如果函数只能运行标准(非异步)函数,那么它就无法工作。
from IPython.display import display
import ipywidgets as widgets
import asyncio
button = widgets.Button(description="STOP!")
output = widgets.Output()
display(button, output)
break_cicle = True
def on_button_clicked(event):
global break_cicle
break_cicle = False
print("Button pressed: break_cicle:", break_cicle)
button.on_click(on_button_clicked)
async def function(): # it has to be `async`
while break_cicle:
print("While loop: break_cicle:", break_cicle)
await asyncio.sleep(1) # it needs some `await` functions
print("Done")
loop = asyncio.get_event_loop()
t = loop.create_task(function()) # assign to variable if you don't want to see `<Task ...>` in output