如何在 Jupyter Notebook 中使用按钮停止循环?

12 浏览
0 Comments

如何在 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中按下按钮\"。

\"enter

admin 更改状态以发布 2023年5月24日
0
0 Comments

我认为问题就像所有长时间运行代码的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

0