notebook/examples/widgets/Nonblocking Console.ipynb

In [1]:

from subprocess import Popen, PIPE
import fcntl
import os

from IPython.html import  widgets
from IPython.display import display
from IPython.utils.py3compat import bytes_to_str, string_types

Create the output, input, and console toggle widgets.

In [2]:

console_container = widgets.ContainerWidget(visible=False)
console_container.set_css('padding', '10px')

console_style = {
    'font-family': 'monospace',
    'color': '#AAAAAA',
    'background': 'black',
    'width': '800px',
}

output_box = widgets.StringWidget(parent=console_container, default_view_name='TextAreaView')
output_box.set_css(console_style)
output_box.set_css('height', '400px')

input_box = widgets.StringWidget(parent=console_container)
input_box.set_css(console_style)

toggle_button = widgets.ButtonWidget(description="Start Console")
def toggle_console():
    console_container.visible = not console_container.visible
    if console_container.visible:
        toggle_button.description="Stop Console"
        input_box.disabled = False
    else:
        toggle_button.description="Start Console"
toggle_button.on_click(toggle_console)

Define function to run a process without blocking the input.

In [3]:

def read_process(process, append_output):
    """ Try to read the stdout and stderr of a process and render it using 
    the append_output method provided
    
    Parameters
    ----------
    process: Popen handle
    append_output: method handle
        Callback to render output.  Signature of
        append_output(output, [prefix=])"""
    
    try:
        stdout = process.stdout.read()
        if stdout is not None and len(stdout) > 0:
            append_output(stdout, prefix='    ')
    except:
        pass
        
    try:
        stderr = process.stderr.read()
        if stderr is not None and len(stderr) > 0:
            append_output(stderr, prefix='ERR ')
    except:
        pass


def set_pipe_nonblocking(pipe):
    """Set a pipe as non-blocking"""
    fl = fcntl.fcntl(pipe, fcntl.F_GETFL)
    fcntl.fcntl(pipe, fcntl.F_SETFL, fl | os.O_NONBLOCK)


kernel = get_ipython().kernel
def run_command(command, append_output, has_user_exited=None):
    """Run a command asyncronously
    
    Parameters
    ----------
    command: str
        Shell command to launch a process with.
    append_output: method handle
        Callback to render output.  Signature of
        append_output(output, [prefix=])
    has_user_exited: method handle
        Check to see if the user wants to stop the command.
        Must return a boolean."""
    
    # Echo input.
    append_output(command, prefix='>>> ')
    
    # Create the process.  Make sure the pipes are set as non-blocking.
    process = Popen(command, shell=True, stdout=PIPE, stderr=PIPE)
    set_pipe_nonblocking(process.stdout)
    set_pipe_nonblocking(process.stderr)
    
    # Only continue to read from the command 
    while (has_user_exited is None or not has_user_exited()) and process.poll() is None:
        read_process(process, append_output)
        kernel.do_one_iteration() # Run IPython iteration.  This is the code that
                                  # makes this operation non-blocking.  This will
                                  # allow widget messages and callbacks to be 
                                  # processed.
        
    # If the process is still running, the user must have exited.
    if process.poll() is None:
        process.kill()
    else:
        read_process(process, append_output) # Read remainer
        
    
        
        

Hook the process execution methods up to our console widgets.

In [4]:

def append_output(output, prefix):
    if isinstance(output, string_types):
        output_str = output
    else:
        output_str = bytes_to_str(output)
    output_lines = output_str.split('\n')
    formatted_output = '\n'.join([prefix + line for line in output_lines if len(line) > 0]) + '\n'
    output_box.value += formatted_output
    output_box.scroll_to_bottom()
    
def has_user_exited():
    return not console_container.visible

def handle_input(sender):
    sender.disabled = True
    try:
        command = sender.value
        sender.value = ''
        run_command(command, append_output=append_output, has_user_exited=has_user_exited)
    finally:
        sender.disabled = False
        
input_box.on_submit(handle_input)

Show the console

In [5]:

display(toggle_button)
display(console_container)