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

Question

让我们假设以下问题：我们有一个 Ipywidget 按钮和一个进度条。单击该按钮时，将执行一个函数 work()，该函数仅填充进度条直到完成，然后反转该过程并将其清空。就目前而言，这样的功能会持续运行。以下代码sn -p提供了对应的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

这是我通过 threading 包并基于 https://ipywidgets.readthedocs.io/en/latest/examples/Widget%20Asynchronous.html 中给出的 background-working-widget 示例得出的答案。它当然没有优化，可能不符合良好实践。欢迎任何提出更好答案的人提供。

# 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]))