如何在 Jupyter Notebook 中使用按钮停止循环?
如何在 Jupyter Notebook 中使用按钮停止循环?
我想要:
- 从串口读取(无限循环)
- 当\"停止\"按钮按下时 - 停止读取并绘制数据
从如何使用按键中断while循环?中,我采用了使用键盘中断的示例,它有效,但我想使用一个按钮。
使用键盘中断的示例
weights = [] times = [] #open port ser = serial.Serial('COM3', 9600) try: while True: # read infinite loop #DO STUFF line = ser.readline() # read a byte string if line: weight_ = float(line.decode()) # convert the byte string to a unicode string time_ = time.time() weights.append(weight_) times.append(time_) print (weight_) #STOP it by keyboard interup and continue with program except KeyboardInterrupt: pass #Continue with plotting
然而,我希望使用一个显示的按钮(更容易让人们使用)。
我尝试制作一个按钮(在Jupiter Notebook 中),当按下时break_cicle = False,但当按下按钮时循环不会中断:
#make a button for stopping the while loop button = widgets.Button(description="STOP!") #STOP WHEN THIS BUTTON IS PRESSED output = widgets.Output() display(button, output) break_cicle=True def on_button_clicked(b): with output: break_cicle = False # Change break_cicle to False print(break_cicle) ser.close() button.on_click(on_button_clicked) ser = serial.Serial('COM3', 9600) try: while break_cicle: print (break_cicle) line = ser.readline() # read a byte string if line: weight_ = float(line.decode()) # convert the byte string to a unicode string time_ = time.time() weights.append(weight_) times.append(time_) print (weight_) except : pass ser.close()
使用全局变量的示例无法正常工作
from IPython.display import display import ipywidgets as widgets button = widgets.Button(description="STOP!") #STOP WHEN THIS BUTTON IS PRESSED output = widgets.Output() display(button, output) break_cicle=True def on_button_clicked(): global break_cicle #added global with output: break_cicle = False # Change break_cicle to False print ("Button pressed inside break_cicle", break_cicle) button.on_click(on_button_clicked) try: while break_cicle: print ("While loop break_cicle:", break_cicle) time.sleep(1) except : pass print ("done")
尽管我按了几次按钮,但从以下图像中可以看出,它从未打印出\"在break_cicle中按下按钮\"。
我认为问题就像所有长时间运行代码的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在该函数休眠时可以运行其他函数。
编辑:
在互联网上深入研究(使用Google)后,我在Jyputer论坛上找到了一篇文章
Interactive widgets while executing long-running cell - JupyterLab - Jupyter Community Forum
并且有一个指向模块jupyter-ui-poll的链接,它显示了类似的示例(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')
在源代码中,我可以看到它使用asyncio
实现这一点。
编辑:
使用multiprocessing实现的版本。
进程不共享变量,因此需要Queue
从一个进程发送信息到另一个进程。
示例将消息从button
发送到function
。如果你想从function
发送消息到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")
编辑:
使用asyncio实现的版本。
Jupyter已经在运行asyncio事件循环
,我向这个循环中添加了async函数
。函数使用await
函数,例如asyncio.sleep
,因此asyncio事件循环
有时间运行其他函数 - 但是如果函数仅能运行标准(非异步)函数,则将无法工作。
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 `` in output