Widgets

Utilities for ipywidgets: cleanup, async interaction, and blocking input

cleanupwidgets

Helper to properly cleanup ipywidget instances by closing their comms.

When working with ipywidgets in notebooks, each widget creates a comm channel with the kernel. During heavy development, it’s better to close the widgets, to avoid memory leaks and kernel issues.


source

close_widget


def close_widget(
    w:W.Widget, all:bool=True
):

source

cleanupwidgets


def cleanupwidgets(
    ws:VAR_POSITIONAL, mod:str | None=None, clear:bool=True, all:bool=True
):
_b = W.Button()
test_eq(_b.comm is not None, True)
cleanupwidgets('_b')
test_is(_b.comm, None)
_b = W.Button()
test_eq(_b.comm is not None, True)
cleanupwidgets('_b')
test_is(_b.comm, None)
import ipywidgets.widgets.widget
import ipywidgets as W
from IPython.core.getipython import get_ipython
ipywidgets.widgets.widget._instances
{}
def get_active_widget_comms():
    """Get "official" list of widget comms"""
    ip = get_ipython(); kernel = ip.kernel  # type: ignore
    if kernel:
        ks = W.Widget.get_manager_state()['state'].keys()
        for k,c in kernel.comm_manager.comms.items():
            if c.comm_id in ks:
                yield c
[*get_active_widget_comms()]
[]
W.Widget.close_all()

Clickable

Button subclass with a value trait.


source

Clickable


def Clickable(
    description:str='', kwargs:VAR_KEYWORD
):

Button with value.

cleanupwidgets('btn')
display(btn := Clickable())
cleanupwidgets('box')
box = W.HBox([sld := W.IntSlider(), btn := Clickable('>', layout={'width':'30px'})])
btn.on_click(lambda _: sld.set_trait('value', sld.value + 10))
display(box)

Asynchronous Widgets

Two common patterns for non-blocking widget interactions:

  1. Wait for user interaction: Pause code execution without blocking the kernel
  2. Update widgets in background: Progress bars, live updates while kernel remains responsive

These patterns leverage Python’s async/await or generator-based approaches.

Note: Examples below adapted from ipywidgets documentation

Two scenarios where we’d like widget-related code to run without blocking the kernel from acting on other execution requests.

  1. Pausing code to wait for user interaction with a widget in the frontend
  2. Updating a widget in the background
cleanupwidgets('slider')

display(slider := W.IntSlider(max=10))

def work(slider):
    start = time.time()
    print(f"waiting for slider to reach {slider.max}...", end='')
    while True:
        print('.', end='')
        time.sleep(0.5)
        if (time.time() - start) > 5: print('timeout'); break

work(slider)
waiting for slider to reach 10.............timeout

Try to change the slider. You can, the front-end is responsive, but the kernel is blocked from running other code, including handling messages from the front-end.

cleanupwidgets('progress')

display(progress := W.FloatProgress(value=0.0, min=0.0, max=1.0))

async def work(progress):
    total = 100
    for i in range(total):
        time.sleep(0.05)
        progress.value = float(i+1)/total

await work(progress)
print(progress.value)
1.0

Async doesn’t help, the kernel is still blocked.

Waiting for user interaction

Pausing code to wait for user interaction with a widget in the frontend

Event loop integration

If we take advantage of the event loop integration IPython offers, we can have a nice solution async/await syntax.

We define a new function that returns a future for when a widget attribute changes.


source

wait_for_change


def wait_for_change(
    widget:W.Widget, value:str
):

And we finally get to our function where we will wait for widget changes. We’ll do 10 units of work, and pause after each one until we observe a change in the widget. Notice that the widget’s value is available to us, since it is what the wait_for_change future has as a result.

Run this function, and change the slider 10 times.

cleanupwidgets('slider1', 'out1')

slider1 = W.IntSlider()
out1 = W.Output()

async def f():
    for i in range(10):
        out1.append_stdout('did work ' + str(i))
        x = await wait_for_change(slider1, 'value')
        out1.append_stdout(' - async function continued with value ' + str(x) + '\n')
asyncio.ensure_future(f())

display(slider1, out1)

Note that this is not blocking the kernel from running other code. We can run other cells, or even other widgets.

cleanupwidgets('slider2', 'out2')

slider2 = W.IntSlider()
out2 = W.Output()

def test():
    async def f():
        for i in range(10):
            out2.append_stdout('did work ' + str(i))
            x = await wait_for_change(slider2, 'value')
            out2.append_stdout(' - async function continued with value ' + str(x) + '\n')
    asyncio.ensure_future(f())

test()
display(slider2, out2)

Generator approach

Updating a widget in the background

If you can’t take advantage of the async/await syntax, or you don’t want to modify the event loop, you can also do this with generator functions.

First, we define a decorator which hooks a generator function up to widget change events.


source

yield_for_change


def yield_for_change(
    widget, attribute
):

Pause a generator to wait for a widget change event.

This is a decorator for a generator function which pauses the generator on yield until the given widget attribute changes. The new value of the attribute is sent to the generator and is the value of the yield.

Then we set up our generator.

cleanupwidgets('slider3')
display(slider3 := W.IntSlider())

@yield_for_change(slider3, 'value')
def f():
    for i in range(10):
        print('did work %s'%i, end=' ')
        x = yield
        print('- generator function continued with value %s'%x)
f();
did work 0 

Modifications

The above two approaches both waited on widget change events, but can be modified to wait for other things, such as button event messages (as in a “Continue” button), etc.

cleanupwidgets('btn')

# class Clickable(W.Button):
#     clicked = T.Int(0)
#     def __init__(self, *args, on_click=None, **kwargs):
#         super().__init__(*args, **kwargs)
#         if on_click is None:
#             self.on_click(lambda b: b.set_trait('clicked', b.clicked + 1))
#             T.dlink((self, 'clicked'), (self, 'description'), lambda x: f'{x}')
#         else:
#             self.on_click(on_click)

display(btn := Clickable())

@yield_for_change(btn, 'value')
def f():
    for i in range(10):
        print('did work %s'%i, end=' '  )
        x = yield
        print('- generator function continued with value %s'%x)
f();
did work 0 
cleanupwidgets('btn2', 'out3')
out3 = W.Output()

display(btn2 := Clickable(), out3)

def f():
    async def f():
        for i in range(10):
            out3.append_stdout('did work ' + str(i))
            x = await wait_for_change(btn2, 'value')
            out3.append_stdout(' - async function continued with value ' + str(x) + '\n')
    asyncio.ensure_future(f())

f()
cleanupwidgets('txt', 'out4')
out4 = W.Output()

display(txt := W.Text(continuous_update=False), out4)

def f():
    async def f():
        while True:
            x = await wait_for_change(txt, 'value')
            out4.append_stdout(' - async function continued with value ' + str(x) + '\n')
            if x == 'exit': break
            txt.value = ''
    asyncio.ensure_future(f())

f()

Updating a widget in the background

Sometimes you’d like to update a widget in the background, allowing the kernel to also process other execute requests. We can do this with threads. In the example below, the progress bar will update in the background and will allow the main kernel to do other computations.

cleanupwidgets('progress')

progress = W.FloatProgress(value=0.0, min=0.0, max=1.0)

def work(progress):
    total = 100
    for i in range(total):
        time.sleep(0.01)
        progress.value = float(i+1)/total

thread = threading.Thread(target=work, args=(progress,))
display(progress)
thread.start()
print(progress.value)
1.0

Blocking widgets

Sometimes you need the opposite: block the kernel until the user provides input. This is useful for:

  • Sequential workflows where each cell depends on previous user input
  • “Run All” notebooks that need user decisions
  • Modal-like dialogs that pause execution

The challenge: ipywidgets is designed to be non-blocking. Simple solutions don’t work because the kernel needs to process frontend messages while waiting.

Now we want to handle the inverse problem, where we want to block the kernel from running other code while waiting for a widget to change.

Consider Python input function, which blocks until the user enters some input.

name = input("Enter your name:")
print(f"It's very important to know your name, {name}!")
It's very important to know your name, qwerty!

If you’re running this on VSCode or forks, you’ll notice that input works as expected, but its UX is unwieldy with a nearly invisible popout up at the top of the window.

No problem, ipywidgets has a Text widget that can be used to get user input.

cleanupwidgets('text')

print('(Stop cell execution when you tire of waiting)')

text = W.Text(placeholder='Input your name; enter to submit', continuous_update=False)
display(text)

while True:
    if text.value: break
    time.sleep(0.5)
print(f"It's very important to know your name, {name}!")
(Stop cell execution when you tire of waiting)
---------------------------------------------------------------------------
KeyboardInterrupt                         Traceback (most recent call last)
Cell In[37], line 12
     10 while True:
     11     if text.value: break
---> 12     time.sleep(0.5)
     13 print(f"It's very important to know your name, {name}!")

KeyboardInterrupt: 

The widget is responsive, but as the kernel is blocked, the messages coming from the front-end are not processed.

No problem, Asyncio to the rescue.

cleanupwidgets('text')

async def wait_for_text(text):
    cnt = 0
    while cnt < 10:
        if text.value: return text.value
        await asyncio.sleep(0.5)
        cnt += 1
    return 'unknown'

text = W.Text(placeholder='Enter your name; enter to submit', continuous_update=False)
display(text)

await wait_for_text(text)
print(f"It's very important to know your name, {name}!")
It's very important to know your name, qwerty!

You can try lots of convoluted solutions, but by design is very difficult to block the kernel using widgets. Crafting modal UIs with ipywidgets is not simple.

TL;DR: Use jupyter-ui-poll
> Block Jupyter cell execution while interacting with widgets.

We want to solve the following problem:

  1. Display User Interface in Jupyter using ipywidgets or similar
  2. Wait for data to be entered (this step is surprisingly non-trivial to implement)
  3. Use entered data in cells below

You want to implement a notebook like the one below


   # cell 1
   ui = make_ui()
   display(ui)
   data = ui.wait_for_data()

   # cell 2
   do_things_with(data)

   # cell 3.
   do_more_tings()

And you want to be able to execute Cells -> Run All menu option and still get correct output.

Jupyter assists in implementing your custom ui.wait_for_data() poll loop. If you have tried implementing such workflow in the past you’ll know that it is not that simple. If you haven’t, see Technical Details for an explanation on why it’s hard and how jupyter-ui-poll solves it.

Quick, self contained example:

import time
from ipywidgets import Button
# 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 (up to 10 at a time)
        print('.', end='')
        time.sleep(0.1)
print('done')
.......................done

input with widgets

Now we can develop our pretty input function:


source

get_user_input


def get_user_input(
    prompt:str='', placeholder:str='Write something. Enter to submit', timeout:float=10.0, widget:NoneType=None,
    value:NoneType=None
):
name = get_user_input('You', placeholder='Input your name; enter to submit', timeout=5.)
Markdown(f"Your intervention has saved the Universe and beyond, **{name}**!")

You

Your intervention has saved the Universe and beyond, ****!