使用 Button Jupyter Notebook 终止循环?
Kill a loop with Button Jupyter Notebook?
我想:
- 从串口读取(无限循环)
- 按下“停止”按钮时 --> 停止读取并绘制数据
来自 How to kill a while loop with a keystroke? 我已经使用键盘中断作为中断示例,这有效,但我想使用按钮。
键盘中断示例
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
但是我想用一个显示的按钮来做(更容易让人们使用)。
我尝试制作一个按钮(in 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")
尽管我按了几次按钮,从下图中你可以看到它从不打印“Button pressed inside break_cicle”。
我认为问题就像在所有 Python 脚本中一样 long-running 代码 - 它 运行 是一个线程中的所有代码,当它 运行 是 while True
循环(long-running 代码)然后它不能同时 运行 其他函数。
您可能必须 运行 在单独的线程中执行您的函数 - 然后主线程可以执行 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 论坛上找到 post
Interactive widgets while executing long-running cell - JupyterLab - Jupyter Community Forum
模块 jupyter-ui-poll 有 link,它显示了类似的示例(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')
在 source code 中,我可以看到它为此使用了 asyncio
。
编辑:
进程不共享变量,因此需要 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 已经 运行ning asynio event loop
并且我将 async function
添加到此循环中。并且函数使用 await
函数,例如 asyncio.sleep
所以 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
我想:
- 从串口读取(无限循环)
- 按下“停止”按钮时 --> 停止读取并绘制数据
来自 How to kill a while loop with a keystroke? 我已经使用键盘中断作为中断示例,这有效,但我想使用按钮。
键盘中断示例
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
但是我想用一个显示的按钮来做(更容易让人们使用)。 我尝试制作一个按钮(in 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")
尽管我按了几次按钮,从下图中你可以看到它从不打印“Button pressed inside break_cicle”。
我认为问题就像在所有 Python 脚本中一样 long-running 代码 - 它 运行 是一个线程中的所有代码,当它 运行 是 while True
循环(long-running 代码)然后它不能同时 运行 其他函数。
您可能必须 运行 在单独的线程中执行您的函数 - 然后主线程可以执行 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 论坛上找到 post
Interactive widgets while executing long-running cell - JupyterLab - Jupyter Community Forum
模块 jupyter-ui-poll 有 link,它显示了类似的示例(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')
在 source code 中,我可以看到它为此使用了 asyncio
。
编辑:
进程不共享变量,因此需要 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 已经 运行ning asynio event loop
并且我将 async function
添加到此循环中。并且函数使用 await
函数,例如 asyncio.sleep
所以 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