停止按钮，用于在Ipywidgets生态系统中中断while循环

Question

让我们假设以下问题：我们有一个Ipywidget按钮和一个进度条。单击按钮后，将执行一个函数work((，它只填充进度条直到完成进度条，然后反转过程并清空进度条。就目前而言，这样一个函数是连续运行的。以下代码片段提供了相应的MWE:

# importing packages.
from IPython.display import display
import ipywidgets as widgets
import time
import functools
# setting 'progress', 'start_button' and 'Hbox' variables.
progress = widgets.FloatProgress(value=0.0, min=0.0, max=1.0)
start_button = widgets.Button(description="start fill")
Hbox = widgets.HBox(children=[start_button, progress])
# defining 'on_button_clicked_start()' function; executes 'work()' function.
def on_button_clicked_start(b, start_button, progress):
work(progress)
# call to 'on_button_clicked_start()' function when clicking the button.
start_button.on_click(functools.partial(on_button_clicked_start, start_button=start_button, progress=progress))
# defining 'work()' function.
def work(progress):
total = 100
i = 0
# while roop for continuous run.
while True:
# while loop for filling the progress bar.
while progress.value < 1.0:
time.sleep(0.01)
i += 1
progress.value = float(i)/total
# while loop for emptying the progress bar.
while progress.value > 0.0:
time.sleep(0.01)
i -= 1
progress.value = float(i)/total

# display statement.
display(Hbox)

其目的是包括"；停止"；以及"；简历"按钮，这样每当单击第一个时while循环就会中断，而当按下第二个时则会恢复执行。这可以在不使用线程、多处理或异步的情况下完成吗？

Answer 1

以下是我通过线程包得出的答案，基于中给出的后台工作小部件示例https://ipywidgets.readthedocs.io/en/latest/examples/Widget%20Asynchronous.html.它肯定没有经过优化，可能也不符合良好的实践。欢迎任何有更好答案的人提供。

# importing packages.
import threading
from IPython.display import display
import ipywidgets as widgets
import time
# defining progress bar 'progress', start, stop and resume buttons
# 'start_button', 'stop_button' and 'resume_button', and horizontal
# box 'Hbox'.
progress = widgets.FloatProgress(value=0.0, min=0.0, max=1.0)
start_button = widgets.Button(description="start fill")
stop_button = widgets.Button(description="stop fill/empty")
resume_button = widgets.Button(description="resume fill/empty")
Hbox = widgets.HBox(children=[start_button, stop_button, resume_button, progress])
# defining boolean flags 'pause' and 'resume'.
pause = False
restart = False
# defining 'on_button_clicked_start()' function.
def on_button_clicked_start(b):
# setting global variables.
global pause
global thread
global restart
# conditinoal for checking whether the thread is alive;
# if it isn't, then start it.
if not thread.is_alive():
thread.start()
# else, pause and set 'restart' to True for setting
# progress bar values to 0.
else:
pause = True
restart = True
time.sleep(0.1)
restart = False
# conditional for changing boolean flag 'pause'.
if pause:
pause = not pause

# defining 'on_button_clicked_stop()' function.    
def on_button_clicked_stop(b):
# defining global variables.
global pause
# conditional for changing boolean flag 'pause'.
if not pause:
pause = not pause

# defining 'on_button_clicked_resume()' function.
def on_button_clicked_resume(b):
# defining global variables.
global pause
global restart
# conditional for changing boolean flags 'pause' and 'restart'
# if necessary.
if pause:
if restart:
restart = False
pause = not pause
# call to 'on_button_clicked_start()' function when clicking the button.
start_button.on_click(on_button_clicked_start)
# call to 'on_button_clicked_stop()' function when clicking the button.
stop_button.on_click(on_button_clicked_stop)
# call to 'on_button_clicked_resume()' function when clicking the button.
resume_button.on_click(on_button_clicked_resume)
# defining the 'work()' function.
def work(progress):
# setting global variables.
global pause
i = 0
i_m1 = 0
# setting 'total' variable.
total = 100
# infinite loop.
while True:
# stop/resume conditional.
if not pause:
# filling the progress bar.
if (i == 0) or i > i_m1 and not pause:
time.sleep(0.1)
if i == i_m1:
pass
else:
i_m1 = i
i += 1
progress.value = float(i)/total
# emptying the progress bar.
if (i == 101) or i < i_m1 and not pause:
time.sleep(0.1)
if i == i_m1:
pass
else:
i_m1 = i
i -= 1
progress.value = float(i)/total
else:
if restart:
i = 0
i_m1 = 0
# setting the thread.
thread = threading.Thread(target=work, args=(progress,))
# displaying statement.
display(Hbox)

Answer 2

在尝试使用异步之后，我遇到了太多问题。更好的方法是使用生成器方法，文档中也提到了这一点https://ipywidgets.readthedocs.io/en/latest/examples/Widget%20Asynchronous.html

这是我想出的代码

from functools import wraps
from IPython.core.display import display
import ipywidgets as widgets
def yield_for_change(widget, attribute):
def f(iterator):
@wraps(iterator)
def inner():
i = iterator()
def next_i(change):
try:
print([w.description for w  in widget])
i.send(change)
except StopIteration as e:
for w in widget:
w.unobserve(next_i, attribute)
for w in widget:
w.on_click(next_i)
w.observe(next_i, attribute)
# start the generator
next(i)
return inner
return f
btn_true = widgets.Button(description="True",style={'button_color':'green'} )
btn_false = widgets.Button(description="False",style={'button_color':'red'})
btn_list = [btn_true, btn_false]
buttons = widgets.HBox(btn_list)
value_loop = widgets.Label()
out = widgets.Output()
@yield_for_change(btn_list, 'description')
def f():
for i in range(10):
print('did work %s'%i)
x = yield
print('generator function continued with value %s'%x)
f()
display(widgets.VBox([buttons, out]))