Stop button for breaking while loop within Ipywidgets ecosystem

Question

Let's assume the following problem: we have an Ipywidget button and a progress bar. On clicking the button, a function work() is executed, which merely fills the progress bar until completing it, then reverses the process and empties it out. As it stands, such a function runs continuously. The following code snippet provides the corresponding 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)

The aim is to include "Stop" and "Resume" buttons, so that the while loops are broken whenever the first is clicked, and the execution is resumed when pressing the second one. Can this be done without employing threading, multiprocessing or asynchronicity?

Answer 1

Here's an answer I derived by means of the threading package and based on the background-working-widget example given in https://ipywidgets.readthedocs.io/en/latest/examples/Widget%20Asynchronous.html. It's certainly not optimized, and probably not good-practice-compliant. Anybody coming up with a better answer is welcomed to provide it.

# 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

After trying using asynchronous, I run into too many problems. Much better approach is to use generator approach, also mentioned in documentation https://ipywidgets.readthedocs.io/en/latest/examples/Widget%20Asynchronous.html

Here is the code I came up with

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