h2o-py/h2o/demos.py
def _run_demo(body_fn, interactive, echo, testing):
    """
    Execute the demo, echoing commands and pausing for user input.

    :param body_fn: function that contains the sequence of demo's commands.
    :param interactive: If True, the user will be prompted to continue the demonstration after every segment.
    :param echo: If True, the python commands that are executed will be displayed.
    :param testing: Used for pyunit testing. h2o.init() will not be called if set to True.
    :type body_fn: function
    """
    import colorama
    from colorama import Style, Fore
    colorama.init()

    class StopExecution(Exception):
        """Helper class for cancelling the demo."""

    assert_is_type(body_fn, type(_run_demo))
    # Reformat description by removing extra spaces; then print it.
    if body_fn.__doc__:
        desc_lines = body_fn.__doc__.split("\n")
        while desc_lines[0].strip() == "":
            desc_lines = desc_lines[1:]
        while desc_lines[-1].strip() == "":
            desc_lines = desc_lines[:-1]
        strip_spaces = min(len(line) - len(line.lstrip(" ")) for line in desc_lines[1:] if line.strip() != "")
        maxlen = max(len(line) for line in desc_lines)
        print(Fore.CYAN)
        print("-" * maxlen)
        for line in desc_lines:
            print(line[strip_spaces:].rstrip())
        print("-" * maxlen)
        print(Style.RESET_ALL, end="")

    # Prepare the executor function
    def controller():
        """Print to console the next block of commands, and wait for keypress."""
        try:
            raise RuntimeError("Catch me!")
        except RuntimeError:
            print()
            # Extract and print lines that will be executed next
            if echo:
                tb = sys.exc_info()[2]
                fr = tb.tb_frame.f_back
                filename = fr.f_code.co_filename
                linecache.checkcache(filename)
                line = linecache.getline(filename, fr.f_lineno, fr.f_globals).rstrip()
                indent_len = len(line) - len(line.lstrip(" "))
                assert line[indent_len:] == "go()"
                i = fr.f_lineno
                output_lines = []
                n_blank_lines = 0
                while True:
                    i += 1
                    line = linecache.getline(filename, i, fr.f_globals).rstrip()
                    # Detect dedent
                    if line[:indent_len].strip() != "": break
                    line = line[indent_len:]
                    if line == "go()": break
                    style = Fore.LIGHTBLACK_EX if line.lstrip().startswith("#") else Style.BRIGHT
                    prompt = "... " if line.startswith(" ") else ">>> "
                    output_lines.append(Fore.CYAN + prompt + Fore.RESET + style + line + Style.RESET_ALL)
                    del style  # Otherwise exception print-outs may get messed-up...
                    if line.strip() == "":
                        n_blank_lines += 1
                        if n_blank_lines > 5: break  # Just in case we hit file end or something
                    else:
                        n_blank_lines = 0
                for line in output_lines[:-n_blank_lines]:
                    print(line)

            # Prompt for user input
            if interactive:
                print("\n" + Style.DIM + "(press any key)" + Style.RESET_ALL, end="")
                key = _wait_for_keypress()
                print("\r                     \r", end="")
                if key.lower() == "q":
                    raise StopExecution()

    # Replace h2o.init() with a stub when running in "test" mode
    _h2o_init = h2o.init
    if testing:
        h2o.init = lambda *args, **kwargs: None

    # Run the test
    try:
        body_fn(controller)
        print("\n" + Fore.CYAN + "---- End of Demo ----" + Style.RESET_ALL)
    except (StopExecution, KeyboardInterrupt):
        print("\n" + Fore.RED + "---- Demo aborted ----" + Style.RESET_ALL)

    # Clean-up
    if testing:
        h2o.init = _h2o_init
    print()
    colorama.deinit()
def _run_demo(body_fn, interactive, echo, testing):
    """
    Execute the demo, echoing commands and pausing for user input.

    :param body_fn: function that contains the sequence of demo's commands.
    :param interactive: If True, the user will be prompted to continue the demonstration after every segment.
    :param echo: If True, the python commands that are executed will be displayed.
    :param testing: Used for pyunit testing. h2o.init() will not be called if set to True.
    :type body_fn: function
    """
    import colorama
    from colorama import Style, Fore
    colorama.init()

    class StopExecution(Exception):
        """Helper class for cancelling the demo."""

    assert_is_type(body_fn, type(_run_demo))
    # Reformat description by removing extra spaces; then print it.
    if body_fn.__doc__:
        desc_lines = body_fn.__doc__.split("\n")
        while desc_lines[0].strip() == "":
            desc_lines = desc_lines[1:]
        while desc_lines[-1].strip() == "":
            desc_lines = desc_lines[:-1]
        strip_spaces = min(len(line) - len(line.lstrip(" ")) for line in desc_lines[1:] if line.strip() != "")
        maxlen = max(len(line) for line in desc_lines)
        print(Fore.CYAN)
        print("-" * maxlen)
        for line in desc_lines:
            print(line[strip_spaces:].rstrip())
        print("-" * maxlen)
        print(Style.RESET_ALL, end="")

    # Prepare the executor function
    def controller():
        """Print to console the next block of commands, and wait for keypress."""
        try:
            raise RuntimeError("Catch me!")
        except RuntimeError:
            print()
            # Extract and print lines that will be executed next
            if echo:
                tb = sys.exc_info()[2]
                fr = tb.tb_frame.f_back
                filename = fr.f_code.co_filename
                linecache.checkcache(filename)
                line = linecache.getline(filename, fr.f_lineno, fr.f_globals).rstrip()
                indent_len = len(line) - len(line.lstrip(" "))
                assert line[indent_len:] == "go()"
                i = fr.f_lineno
                output_lines = []
                n_blank_lines = 0
                while True:
                    i += 1
                    line = linecache.getline(filename, i, fr.f_globals).rstrip()
                    # Detect dedent
                    if line[:indent_len].strip() != "": break
                    line = line[indent_len:]
                    if line == "go()": break
                    style = Fore.LIGHTBLACK_EX if line.lstrip().startswith("#") else Style.BRIGHT
                    prompt = "... " if line.startswith(" ") else ">>> "
                    output_lines.append(Fore.CYAN + prompt + Fore.RESET + style + line + Style.RESET_ALL)
                    del style  # Otherwise exception print-outs may get messed-up...
                    if line.strip() == "":
                        n_blank_lines += 1
                        if n_blank_lines > 5: break  # Just in case we hit file end or something
                    else:
                        n_blank_lines = 0
                for line in output_lines[:-n_blank_lines]:
                    print(line)

            # Prompt for user input
            if interactive:
                print("\n" + Style.DIM + "(press any key)" + Style.RESET_ALL, end="")
                key = _wait_for_keypress()
                print("\r                     \r", end="")
                if key.lower() == "q":
                    raise StopExecution()

    # Replace h2o.init() with a stub when running in "test" mode
    _h2o_init = h2o.init
    if testing:
        h2o.init = lambda *args, **kwargs: None

    # Run the test
    try:
        body_fn(controller)
        print("\n" + Fore.CYAN + "---- End of Demo ----" + Style.RESET_ALL)
    except (StopExecution, KeyboardInterrupt):
        print("\n" + Fore.RED + "---- Demo aborted ----" + Style.RESET_ALL)

    # Clean-up
    if testing:
        h2o.init = _h2o_init
    print()
    colorama.deinit()
Execute the demo, echoing commands and pausing for user input.

    :param body_fn: function that contains the sequence of demo's commands.
    :param interactive: If True, the user will be prompted to continue the demonstration after every segment.
    :param echo: If True, the python commands that are executed will be displayed.
    :param testing: Used for pyunit testing. h2o.init() will not be called if set to True.
    :type body_fn: function