使用 Button Jupyter Notebook 杀死循环?

Kill a loop with Button Jupyter Notebook?

提问人:Leo 提问时间:2/21/2022 最后编辑:WayneLeo 更新时间:3/1/2022 访问量:1991

问:

我想:

  • 从串口读取(无限循环)
  • 当按下“STOP”按钮时 --> 停止读取并绘制数据

如何用击键杀死 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 image description here

按钮 事件

评论

0赞 furas 2/21/2022
在内部,您必须使用 to 通知函数您要为全局/外部变量赋值。此时创建局部变量,并且它不会更改全局变量中的值on_button_clicked()global break_ciclebreak_cicleon_button_clicked()break_ciclebreak_cicle

答:

8赞 furas 2/22/2022 #1

我认为问题就像所有具有长时间运行代码的 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 有其他一些方法来解决这个问题 - 即。当你用 then 编写函数时,你可以使用它让 Python 在这个函数处于睡眠状态时运行其他函数。asyncasyncio.sleep()


编辑:

在互联网上挖掘(使用谷歌)我在 Jyputer 论坛上找到了帖子

执行长时间运行的单元格时的交互式小部件 - JupyterLab - Jupyter 社区论坛

并且有指向模块 jupyter-ui-poll 的链接,它显示了类似的示例 (-loop + ),它用于此。当函数被执行时(在每个循环中),Jupyter 可以将事件发送到小部件,并且它有时间执行。whileButtoneventspull()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


编辑:

具有多处理功能的版本

进程不共享变量,因此它需要将信息从一个进程发送到另一个进程。Queue

示例将消息从 发送到 。如果您想发送消息,则最好使用第二个队列。buttonfunctionfunctionbutton

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 已经在运行,我添加到这个循环中。函数使用函数,所以有时间运行其他函数 - 但如果函数只能运行标准(不是异步)函数,那么它就不会起作用。asynio event loopasync functionawaitasyncio.sleepasynio 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